Build a CSV-to-JSON Classification Service

Scenario

In machine learning ingestion pipelines, unstructured text data often arrives via CSV files and must be cleaned, classified via a model API, and output as structured JSON Lines (.jsonl) for downstream annotation tasks.

Implement a backend processing function process_and_classify that reads a CSV file, classifies each text payload against a restricted list of allowed categories using a provided classification client, and writes valid JSON records to an output file while handling errors gracefully.

Requirements

1. Validation & Categorization: For each record, send the text to classifier.classify(text). The returned category must be validated against allowed_categories. If the category returned is not allowed or the API raises a ClassificationError, retry the call up to max_retries times.
2. Failure Handling: If a row exhausts all retries or has invalid/missing required columns, write a record to the output file with "status": "FAILED" and an "error" message, rather than aborting the entire process.
3. Output Order: The output records written to output_path must match the original row order of the CSV.
4. Summary: Return a summary dictionary containing processing counts.

Interface

from typing import Protocol, List, Dict, Any

class ClassificationError(Exception):
    pass

class ClassifierAPI(Protocol):
    def classify(self, text: str) -> str:
        """Calls an external LLM/model. May raise ClassificationError."""
        ...

def process_and_classify(
    input_csv_path: str,
    output_jsonl_path: str,
    classifier: ClassifierAPI,
    allowed_categories: List[str],
    max_retries: int = 2
) -> Dict[str, int]:
    """
    Returns a summary dict:
    {
        "total": int,
        "success": int,
        "failed": int
    }
    """
    pass

Format & Schema

Input CSV Schema:

• id: Unique string identifier for the record.
• text: The raw text content to be classified.

Output JSONL Schema (each line is a standalone JSON object):

• Successful row: {"id": "123", "text": "...", "category": "billing", "status": "SUCCESS"}
• Failed row: {"id": "124", "text": "...", "category": null, "status": "FAILED", "error": ""}

Example

Input CSV (tasks.csv):

id,text
rec_1,Where is my refund?
rec_2,System error code 500
rec_3,

Configuration:

• allowed_categories = ["billing", "technical_support", "general"]
• classifier.classify("Where is my refund?") -> "billing"
• classifier.classify("System error code 500") -> raises ClassificationError on all retries

Output JSONL (output.jsonl):

{"id": "rec_1", "text": "Where is my refund?", "category": "billing", "status": "SUCCESS"}
{"id": "rec_2", "text": "System error code 500", "category": null, "status": "FAILED", "error": "ClassificationError: rate limit reached"}
{"id": "rec_3", "text": "", "category": null, "status": "FAILED", "error": "Empty text"}

Return Value:

{"total": 3, "success": 1, "failed": 2}

Constraints

• allowed_categories contains 1 to 50 unique string values.
• Number of rows in CSV is between 0 and 100,000.
• Empty or missing id or text fields should immediately be marked as FAILED without calling the classifier.