A useful file-processing pipeline should be predictable before it is fast. Given a directory of input files, it should select the intended files, handle malformed inputs visibly, and produce outputs without silently changing the originals. Python's pathlib and context managers provide building blocks for that design. They do not automatically supply a retry policy, transaction boundary, or security sandbox.
This tutorial develops a local, read-only CSV inspection workflow. The example is deliberately modest: inspect exported files and write a separate summary. It can serve as a deterministic preparation stage before an AI assistant receives selected, sanitized text. No model call is needed for file discovery or row counting. The code is illustrative and should first be tried against disposable fixtures, rather than a production directory.
Define the input and output contract
Choose one dedicated input directory and a different output directory. For the exercise, accept only CSV files directly inside the input directory. Do not recurse into subdirectories. State that the expected format is UTF-8 text with a header row. If an exporter produces another encoding or delimiter, handle that as an explicit format variation instead of quietly guessing.
The output is a list of inspected filenames and data-row counts. An unreadable file should appear in an error list. A run with an error must not be described as complete. Keep the original exports unchanged so that a person can reproduce the inspection and investigate discrepancies later.
Write down what is excluded: changing transaction values, deleting processed inputs, following arbitrary links, and sending files to an external service. A clear contract is useful even for a small script because later improvements often expand its access unintentionally.
Discover files with pathlib
The official pathlib documentation describes Path objects and filesystem operations. In this example, Path.glob("*.csv") expresses the filename selection rule. Sort the results to make processing order easier to inspect. A deterministic order helps when comparing two reports, although it does not guarantee that directory contents remain unchanged while the script runs.
from pathlib import Path
input_dir = Path("sample-input")
if not input_dir.is_dir():
raise ValueError("Create the fixture directory first")
paths = sorted(input_dir.glob("*.csv"))
Treat an empty selection as a meaningful result. It might mean there were no exports this week, the directory was wrong, or the naming convention changed. Report the number of discovered candidates. Do not invent a successful processing count merely because the loop completed without raising an exception.
Read CSV records with a context manager
Use Python's CSV parser rather than splitting lines on commas. A quoted field may contain a comma, and a record may contain a quoted newline. The official csv documentation explains the reader interface and recommends opening file objects with newline="" for CSV processing.
import csv
def count_rows(path: Path) -> int:
with path.open("r", encoding="utf-8", newline="") as handle:
reader = csv.reader(handle, strict=True)
header = next(reader, None)
if not header:
raise ValueError("Missing header row")
return sum(1 for _ in reader)
The with statement gives the file object a defined cleanup boundary. Python's language reference documents that protocol. This does not make the rest of the workflow atomic: an output file written earlier will still exist if a later input fails. Resource cleanup and business-level rollback are separate design decisions.
Make errors part of the report
Keep a successful-result list and a separate error list. Catch only the failures you expect to classify. For this inspection example, file access, text decoding, CSV parsing, and an absent header are expected categories. Unexpected programming errors should remain visible during development rather than being converted into apparently ordinary skipped files.
results = []
errors = []
for path in paths:
if path.is_symlink() or not path.is_file():
errors.append((path.name, "Not an accepted regular file"))
continue
try:
results.append((path.name, count_rows(path)))
except (OSError, UnicodeError, csv.Error, ValueError) as error:
errors.append((path.name, type(error).__name__))
Checking a path and then opening it is not a security boundary against another process changing the filesystem. Keep this example limited to a directory you control. If untrusted people can modify input paths concurrently, design a stronger ingestion service rather than treating these checks as a complete defense.
Validate content before involving an AI model
Row counting establishes that a file can be parsed, but not that its contents are suitable for analysis. Add a separate validation stage for required column names, allowed date formats, missing identifiers, and the meaning of blank values. Report the rule version with the output so that changed rules can explain changed results.
For an AI-assisted workflow, create a narrow export containing only fields needed for the question. Remove account numbers, personal contact details, and unrelated free text. Keep source row identifiers so that a reviewer can trace an answer back to the local data. Let deterministic code calculate counts and totals; use the assistant for drafting an explanation or suggesting categories that a person can review.
The separation makes debugging easier. A wrong count belongs to ingestion or validation. An unsupported narrative belongs to generation or review. Combining both stages into one large prompt makes it harder to identify which part actually failed.
Test predictable failure cases
Build a disposable fixture directory with a valid CSV, an empty file, a missing header, and a file containing a quoted comma. Add a filename with spaces and a non-CSV file that should remain unselected. Write down the expected count or error for every fixture before running the example.
Then test reruns. Running a read-only inspection twice on unchanged fixtures should produce equivalent results. If a later version writes outputs into the input directory, ensure those outputs cannot become inputs to the next run. Also check that error reporting does not reveal sensitive cell contents in shared logs.
Measure larger files only after correctness is established. A streaming reader avoids loading the entire CSV into a list, but parsing still consumes time and memory. Establish file-size and runtime limits appropriate to your own environment, and report a limit failure explicitly.
Keep destructive operations separate
Do not add deletion, compression, or mass renaming to this inspection loop simply because the paths are already available. Those actions need their own collision checks, backup plan, and reviewable preview. A proposed rename list should show both the source and destination before any change occurs.
For recurring ingestion, decide how the application recognizes an already processed export. A filename alone may be insufficient if files can be replaced. Choose a record of accepted inputs and rule versions that suits your workflow. If the job crashes, that record should help an operator distinguish completed work from work that needs attention.
Completion checklist
A useful first version produces a reproducible candidate list, row counts for accepted files, explicit errors, and unchanged originals. Its report names the selection rule and validation assumptions. It does not claim that the data is correct merely because parsing succeeded.
For the next stage, use the AI ticket-classifier evaluation worksheet to plan test cases, or the workflow design guide to define where AI adds value. Keep ingestion deterministic, keep evidence traceable, and expand the workflow only after its current behavior is understandable.