Split a full-name CSV into CRM columns with Python

import argparse
import csv
import io
import sys
from pathlib import Path

from parseapi import ParseAPI, ParseAPIError

DESCRIPTION = "Split a full-name CSV into suggested CRM fields."
REQUIRED = ["full_name"]
PARTS = ["name", "prefix", "first", "middle", "last", "suffix"]
COLUMNS = ["parse_" + name for name in PARTS] + ["parse_status", "parse_error", "parse_http_status"]
SAMPLE = """id,full_name,note
001,GRACE HOPPER,Original spelling retained
002,"Smith, John",Comma order
003,OSCAR DE LA HOYA,Family-name particle
004,GRACE HOPPER,Repeated name
005,,Missing name
006,test test,Review this row
"""

# example:read
def read_rows(stream):
    reader = csv.reader(stream, strict=True)
    fields = next(reader, None)
    if (not fields or not set(REQUIRED).issubset(fields)
            or any(not field.strip() for field in fields)
            or len(fields) != len(set(fields))):
        raise ValueError("Use distinct, nonempty headers including " + ", ".join(REQUIRED) + ".")
    if set(fields) & set(COLUMNS):
        raise ValueError("The input already contains output columns. Use the original CSV.")
    rows = []
    for cells in reader:
        if len(cells) != len(fields):
            raise ValueError("Every row must have the same number of cells as the header.")
        rows.append(dict(zip(fields, cells)))
        if len(rows) > 200:
            raise ValueError("Use 1 to 200 data rows for this example.")
    if not rows:
        raise ValueError("The CSV needs at least one data row.")
    return fields, rows
# /example:read


def outcome(status, error="", http_status="", **values):
    return {**dict.fromkeys(COLUMNS, ""), "parse_status": status,
            "parse_error": error, "parse_http_status": http_status, **values}


def write_rows(stream, fields, rows):
    writer = csv.DictWriter(stream, fieldnames=fields + COLUMNS)
    writer.writeheader()
    writer.writerows(rows)

def parsed(data):
    if not isinstance(data, dict) or type(data.get("valid")) is not bool:
        return outcome("error", "unexpected_response")
    if any(key not in data or (data[key] is not None and
           (not isinstance(data[key], str) or not data[key].strip())) for key in PARTS):
        return outcome("error", "unexpected_response")
    if not data["valid"]:
        return outcome("invalid_name", http_status=200)
    if not data["name"]:
        return outcome("error", "unexpected_response")
    # example:result
    status = "parsed" if data["first"] and data["last"] else "review_name_parts"
    return outcome(status, http_status=200,
                   **{"parse_" + key: data[key] or "" for key in PARTS})
    # /example:result


def enrich(rows, parse):
    results, cache, stopped = [], {}, False
    for row in rows:
        value = row["full_name"].strip()
        if not value:
            result = outcome("missing_name")
        elif len(value.encode("utf-16-le")) // 2 > 120:
            result = outcome("review_name_length")
        elif value in cache:
            result = cache[value]
        elif stopped:
            result = outcome("not_attempted")
        else:
            try:
                # example:request
                result = parsed(parse.name(value))
                # /example:request
            except ParseAPIError as error:
                result = outcome("error", error.code, error.status)
                stopped = error.status in (401, 403, 429)
            except Exception:
                result = outcome("error", "request_failed")
            cache[value] = result
        results.append({**row, **result})
    return results

def main():
    parser = argparse.ArgumentParser(description=DESCRIPTION)
    parser.add_argument("input", nargs="?", help="UTF-8 CSV filename, or omit to run the sample")
    parser.add_argument("--output", help="New CSV filename, or omit for standard output")

    args = parser.parse_args()
    try:

        if args.output and (Path(args.output).exists() or Path(args.output).is_symlink()):
            raise ValueError("The output path already exists. Choose a new filename.")
        if args.input:
            path = Path(args.input)
            if path.stat().st_size > 65536:
                raise ValueError("Use a CSV no larger than 64 KiB.")
            with path.open(encoding="utf-8-sig", newline="") as stream:
                fields, rows = read_rows(stream)
        else:
            fields, rows = read_rows(io.StringIO(SAMPLE))
        # Reads PARSEAPI_KEY. Reuse one client without automatic retries.
        with ParseAPI(timeout=10.0, retries=0) as parse:
            results = enrich(rows, parse)
        if args.output:
            # Exclusive creation protects against a file appearing during the lookups.
            with Path(args.output).open("x", encoding="utf-8", newline="") as stream:
                write_rows(stream, fields, results)
        else:
            write_rows(sys.stdout, fields, results)
    except (ValueError, OSError, csv.Error) as error:
        print(f"Could not complete: {error}", file=sys.stderr)
        return 1
    review = sum(row["parse_status"] != "parsed" for row in results)
    print(f"{len(results)} rows written, {review} need review.", file=sys.stderr)
    return 2 if review else 0


if __name__ == "__main__":
    sys.exit(main())