数据验证脚本编写指南

Below is a self-contained PySpark script that validates the described dataset for deduplication, partition completeness, time-window bounds, enum legality, JSON structure, and required non-null fields. It produces two outputs: a “clean” deduplicated dataset and an “invalid” dataset with violation reasons. It also optionally fails if expected partitions are missing.

Assumptions:

• Input is a partitioned Parquet dataset with column names: event_id (string), event_time (timestamp), user_id (string), event_type (string), properties (string containing JSON), dt (date partition).
• Enum values are case-insensitive; we normalize to lowercase before validation.
• Required non-null fields: event_id, event_time, dt.
• JSON field “properties” is nullable. If present, it must be a valid JSON object. Required JSON keys are configurable.
• Time window rule: to_date(event_time) must be within [dt - late_lag_days, dt + future_drift_days] (inclusive).

You can adjust configuration via command-line arguments.

Python script (PySpark):

#!/usr/bin/env python3
import argparse
import sys
from datetime import datetime, date, timedelta

from pyspark.sql import SparkSession, Window
from pyspark.sql.functions import (
    col, lower, trim, to_date, date_add, date_sub, expr, when, lit,
    array, array_union, array_except, size, from_json, map_keys, element_at,
    greatest, least, coalesce
)
from pyspark.sql.types import StringType, MapType

def parse_args():
    p = argparse.ArgumentParser(description="ETL integrity validation for event dataset")
    p.add_argument("--input-path", required=True, help="Input dataset path (Parquet), partitioned by dt")
    p.add_argument("--output-valid-path", required=True, help="Output path for valid, deduplicated dataset")
    p.add_argument("--output-invalid-path", required=True, help="Output path for invalid records with reasons")
    p.add_argument("--start-dt", required=True, help="Expected partition range start (YYYY-MM-DD)")
    p.add_argument("--end-dt", required=True, help="Expected partition range end (YYYY-MM-DD)")
    p.add_argument("--allowed-event-types", default="view,click,purchase", help="Comma-separated allowed event types")
    p.add_argument("--late-lag-days", type=int, default=1, help="Max allowed lateness (days) wrt dt")
    p.add_argument("--future-drift-days", type=int, default=0, help="Max allowed future drift (days) wrt dt")
    p.add_argument("--required-json-keys", default="", help="Comma-separated keys required within properties JSON (if properties is not null)")
    p.add_argument("--fail-on-partition-missing", action="store_true", help="Fail job if any expected dt partitions are missing")
    p.add_argument("--repartition", type=int, default=0, help="Optional number of partitions for output")
    return p.parse_args()

def daterange(start_dt: date, end_dt: date):
    d = start_dt
    out = []
    while d <= end_dt:
        out.append(d)
        d += timedelta(days=1)
    return out

def main():
    args = parse_args()

    start_dt = datetime.strptime(args.start_dt, "%Y-%m-%d").date()
    end_dt = datetime.strptime(args.end_dt, "%Y-%m-%d").date()
    if start_dt > end_dt:
        print("Error: start-dt must be <= end-dt", file=sys.stderr)
        sys.exit(2)

    allowed_event_types = [s.strip().lower() for s in args.allowed_event_types.split(",") if s.strip()]
    required_json_keys = [s.strip() for s in args.required_json_keys.split(",") if s.strip()]

    spark = (
        SparkSession.builder
        .appName("event-integrity-validation")
        .getOrCreate()
    )

    # Read input dataset
    df = spark.read.parquet(args.input_path)

    # Normalize types and values
    # - ensure event_type lowercase for enum validation
    # - ensure dt is date, event_time is timestamp
    df = df \
        .withColumn("event_type_norm", lower(trim(col("event_type")))) \
        .withColumn("dt", to_date(col("dt"))) \
        .withColumn("event_time", col("event_time").cast("timestamp")) \
        .withColumn("event_id", col("event_id").cast(StringType())) \
        .withColumn("user_id", col("user_id").cast(StringType())) \
        .withColumn("properties", col("properties").cast(StringType()))

    # Deduplicate by primary key event_id:
    # Keep the row with the latest event_time; in tie, keep the lexicographically max JSON to stabilize.
    # Mark dropped duplicates for invalid reporting.
    w = Window.partitionBy("event_id").orderBy(
        col("event_time").desc_nulls_last(),
        col("properties").desc_nulls_last()
    )
    df_with_rownum = df.withColumn("rn", expr("row_number() OVER (PARTITION BY event_id ORDER BY event_time DESC NULLS LAST, properties DESC NULLS LAST)"))
    canonical = df_with_rownum.filter(col("rn") == 1).drop("rn")
    duplicates_removed = df_with_rownum.filter(col("rn") > 1).drop("rn") \
        .withColumn("violations", array(lit("duplicate_removed")))

    # Core validations on canonical set

    # Required non-null fields
    required_viol = array()
    required_viol = when(col("event_id").isNull(), array_union(required_viol, array(lit("missing_required:event_id")))).otherwise(required_viol)
    required_viol = when(col("event_time").isNull(), array_union(required_viol, array(lit("missing_required:event_time")))).otherwise(required_viol)
    required_viol = when(col("dt").isNull(), array_union(required_viol, array(lit("missing_required:dt")))).otherwise(required_viol)

    # Enum validation (event_type cannot be null and must be in allowed set)
    enum_valid = col("event_type_norm").isin(allowed_event_types)
    enum_viol = when(col("event_type_norm").isNull(), array(lit("invalid_enum:null"))) \
        .otherwise(when(enum_valid, array()).otherwise(array(lit("invalid_enum:event_type"))))

    # Time window validation:
    # Allowed event_time date must be within [dt - late_lag_days, dt + future_drift_days]
    event_date = to_date(col("event_time"))
    min_allowed_date = date_sub(col("dt"), args.late_lag_days)
    max_allowed_date = date_add(col("dt"), args.future_drift_days)
    time_viol = when(
        col("event_time").isNull(), array(lit("time_window:event_time_null"))
    ).otherwise(
        when((event_date < min_allowed_date) | (event_date > max_allowed_date),
             array(lit("time_window:out_of_range"))).otherwise(array())
    )

    # JSON validation: properties is nullable; if present must be a valid JSON object.
    # We parse as Map<String,String> for structural validation. If parse fails => null.
    json_schema = MapType(StringType(), StringType(), containsNull=True)
    parsed_props = from_json(trim(col("properties")), json_schema)
    json_viol = when(
        col("properties").isNull(), array()
    ).otherwise(
        when(parsed_props.isNull(), array(lit("invalid_json:parse_error"))).otherwise(array())
    )

    # Required JSON keys (if provided): keys must exist and be non-empty when properties is not null
    if required_json_keys:
        # For each required key, check existence and non-empty value
        for k in required_json_keys:
            json_viol = when(
                col("properties").isNull(), json_viol  # no requirement if properties is null
            ).otherwise(
                when(
                    (parsed_props.isNull()) | (element_at(parsed_props, lit(k)).isNull()) | (trim(element_at(parsed_props, lit(k))) == ""),
                    array_union(json_viol, array(lit(f"missing_json_key:{k}")))
                ).otherwise(json_viol)
            )

    # Aggregate violations for canonical records
    violations = array()
    violations = array_union(violations, required_viol)
    violations = array_union(violations, enum_viol)
    violations = array_union(violations, time_viol)
    violations = array_union(violations, json_viol)

    canonical_with_viol = canonical.withColumn("violations", violations)

    # Split valid vs invalid
    invalid_canonical = canonical_with_viol.filter(size(col("violations")) > 0)
    valid_canonical = canonical_with_viol.filter(size(col("violations")) == 0).drop("violations")

    # Combine invalid sets (duplicates + invalid canonical)
    invalid_all = invalid_canonical.unionByName(duplicates_removed, allowMissingColumns=True)

    # Partition completeness check
    expected_dates = [d.strftime("%Y-%m-%d") for d in daterange(start_dt, end_dt)]
    actual_dates = [r["dt"].strftime("%Y-%m-%d") for r in df.select("dt").distinct().collect() if r["dt"] is not None]
    expected_set = set(expected_dates)
    actual_set = set(actual_dates)
    missing_partitions = sorted(expected_set - actual_set)

    # Write outputs
    if args.repartition > 0:
        valid_canonical = valid_canonical.repartition(args.repartition, col("dt"))
        invalid_all = invalid_all.repartition(args.repartition)

    # Write valid dataset partitioned by dt to preserve downstream partitioning
    valid_canonical.write.mode("overwrite").partitionBy("dt").parquet(args.output_valid_path)
    invalid_all.write.mode("overwrite").parquet(args.output_invalid_path)

    # Metrics and summary
    total_in = df.count()
    total_valid = valid_canonical.count()
    total_invalid = invalid_all.count()
    total_duplicates = duplicates_removed.count()

    print("Validation Summary")
    print(f"- Input rows: {total_in}")
    print(f"- Valid (deduplicated) rows: {total_valid}")
    print(f"- Invalid rows: {total_invalid}")
    print(f"- Duplicates removed: {total_duplicates}")
    print(f"- Expected dt partitions: {len(expected_set)} [{args.start_dt}..{args.end_dt}]")
    print(f"- Actual dt partitions: {len(actual_set)}")
    if missing_partitions:
        print(f"- Missing dt partitions ({len(missing_partitions)}): {', '.join(missing_partitions)}")
        if args.fail_on_partition_missing:
            print("Failing due to missing partitions.", file=sys.stderr)
            spark.stop()
            sys.exit(1)
    else:
        print("- Partition completeness: OK")

    spark.stop()

if __name__ == "__main__":
    main()

Usage example:

• Validate a dataset on s3 with 7-day completeness window, allowing 1 day late arrival and no future drift; require properties keys “item_id” and “price” when properties exists; fail if any partition missing.

python validate_events.py \
  --input-path s3://bucket/events_parquet/ \
  --output-valid-path s3://bucket/events_valid/ \
  --output-invalid-path s3://bucket/events_invalid/ \
  --start-dt 2025-09-20 \
  --end-dt 2025-09-26 \
  --allowed-event-types view,click,purchase \
  --late-lag-days 1 \
  --future-drift-days 0 \
  --required-json-keys item_id,price \
  --fail-on-partition-missing \
  --repartition 200

Notes:

• Deduplication retains a single canonical row per event_id (latest event_time). Removed duplicates are captured in the invalid output with reason “duplicate_removed”.
• The JSON structure check ensures the “properties” field, if present, parses into an object (map). If parsing fails, the record is flagged.
• Time-window validation operates on event_date relative to dt; adjust late_lag_days/future_drift_days to fit your pipeline’s allowed lateness.
• The script prints a concise summary and can hard-fail on missing partitions for strong SLA enforcement.

Below is a self-contained Python script that validates the integrity of a campaign reporting dataset. It checks schema conformance, funnel consistency (conversion <= click <= impression), date intervals, missing and anomalous values, channel enumeration, and currency consistency (if a currency column is present or specified).

You can save it as validate_campaign_report.py and run: python validate_campaign_report.py --input path/to/data.csv --output path/to/validation_report.csv --currency-col currency --expected-currency USD

If no currency column exists or is not specified, the currency check is skipped with a warning.

#!/usr/bin/env python3
"""
Validate campaign reporting dataset integrity.

Required columns:
- campaign_id: string (non-null)
- channel: enum in {"email", "ad", "app"}
- impression: int >= 0
- click: int >= 0
- conversion: int >= 0
- cost: decimal >= 0
- start_date: date
- end_date: date

Checks:
- Missing values in required fields
- Type and range: non-negative, integer for counts
- Funnel consistency: conversion <= click <= impression
- Date parsing and interval: start_date <= end_date
- Channel enumeration correctness
- Currency consistency (if currency column provided or detected):
    - Non-null currency values
    - Single unique currency or match expected currency

Outputs a CSV of row-level errors and prints a summary to stdout.
"""

import argparse
import sys
import pandas as pd
import numpy as np
from typing import Dict, List, Optional, Tuple

REQUIRED_COLUMNS = [
    "campaign_id", "channel", "impression", "click",
    "conversion", "cost", "start_date", "end_date"
]
ALLOWED_CHANNELS = {"email", "ad", "app"}

def coerce_numeric(df: pd.DataFrame, col: str) -> Tuple[pd.Series, pd.Series]:
    """Return (parsed_numeric, is_valid_numeric) with NaN for invalid entries."""
    s = pd.to_numeric(df[col], errors="coerce")
    valid = ~s.isna()
    return s, valid

def coerce_date(df: pd.DataFrame, col: str) -> Tuple[pd.Series, pd.Series]:
    """Return (parsed_datetime, is_valid_date) with NaT for invalid entries."""
    s = pd.to_datetime(df[col], errors="coerce", utc=False, infer_datetime_format=True)
    valid = ~s.isna()
    return s, valid

def validate_campaign_report(
    df: pd.DataFrame,
    currency_col: Optional[str] = None,
    expected_currency: Optional[str] = None
) -> Dict:
    errors: List[Dict] = []

    # Normalize column names by stripping whitespace
    df = df.copy()
    df.columns = [c.strip() for c in df.columns]

    # Check presence of required columns
    missing_cols = [c for c in REQUIRED_COLUMNS if c not in df.columns]
    if missing_cols:
        return {
            "fatal": True,
            "message": f"Missing required columns: {missing_cols}",
            "errors": pd.DataFrame(columns=["row_index", "campaign_id", "field", "code", "message"]),
            "summary": {}
        }

    n_rows = len(df)

    # Validate campaign_id non-null and non-empty
    cid_null = df["campaign_id"].isna() | (df["campaign_id"].astype(str).str.strip() == "")
    for idx in df.index[cid_null]:
        errors.append({
            "row_index": idx,
            "campaign_id": None if pd.isna(df.loc[idx, "campaign_id"]) else str(df.loc[idx, "campaign_id"]),
            "field": "campaign_id",
            "code": "missing",
            "message": "campaign_id is null or empty"
        })

    # Validate channel enumeration and missing
    ch_null = df["channel"].isna() | (df["channel"].astype(str).str.strip() == "")
    for idx in df.index[ch_null]:
        errors.append({
            "row_index": idx,
            "campaign_id": str(df.loc[idx, "campaign_id"]) if not pd.isna(df.loc[idx, "campaign_id"]) else None,
            "field": "channel",
            "code": "missing",
            "message": "channel is null or empty"
        })
    # allowed values
    ch_str = df["channel"].astype(str).str.strip().str.lower()
    ch_invalid = ~ch_str.isin(ALLOWED_CHANNELS)
    # Skip invalid flag for rows already missing
    ch_invalid = ch_invalid & ~ch_null
    for idx in df.index[ch_invalid]:
        errors.append({
            "row_index": idx,
            "campaign_id": str(df.loc[idx, "campaign_id"]) if not pd.isna(df.loc[idx, "campaign_id"]) else None,
            "field": "channel",
            "code": "enum_violation",
            "message": f"channel '{df.loc[idx, 'channel']}' not in {sorted(ALLOWED_CHANNELS)}"
        })

    # Numeric columns: impression, click, conversion, cost
    numeric_cols = ["impression", "click", "conversion", "cost"]
    parsed_numeric = {}
    valid_numeric = {}

    for col in numeric_cols:
        s_parsed, s_valid = coerce_numeric(df, col)
        parsed_numeric[col] = s_parsed
        valid_numeric[col] = s_valid

        # Missing or non-numeric
        invalid_mask = ~s_valid
        for idx in df.index[invalid_mask]:
            errors.append({
                "row_index": idx,
                "campaign_id": str(df.loc[idx, "campaign_id"]) if not pd.isna(df.loc[idx, "campaign_id"]) else None,
                "field": col,
                "code": "type_error",
                "message": f"{col} is not numeric"
            })

        # Non-negative constraint
        nonneg_mask = s_parsed < 0
        for idx in df.index[nonneg_mask & s_valid]:
            errors.append({
                "row_index": idx,
                "campaign_id": str(df.loc[idx, "campaign_id"]) if not pd.isna(df.loc[idx, "campaign_id"]) else None,
                "field": col,
                "code": "range_violation",
                "message": f"{col} must be >= 0"
            })

        # Integer constraint for count columns
        if col in {"impression", "click", "conversion"}:
            # Allow NaN failures already recorded; check integer-ness for valid values
            int_mask = s_parsed.dropna().apply(lambda x: float(x).is_integer())
            int_mask = int_mask.reindex(df.index, fill_value=False)
            non_integer = ~int_mask & s_valid
            for idx in df.index[non_integer]:
                errors.append({
                    "row_index": idx,
                    "campaign_id": str(df.loc[idx, "campaign_id"]) if not pd.isna(df.loc[idx, "campaign_id"]) else None,
                    "field": col,
                    "code": "type_error",
                    "message": f"{col} must be an integer count"
                })

    # Funnel consistency: conversion <= click <= impression
    conv = parsed_numeric["conversion"]
    clk = parsed_numeric["click"]
    imp = parsed_numeric["impression"]
    # Only evaluate on rows where all three are valid and non-negative integers
    funnel_valid = valid_numeric["conversion"] & valid_numeric["click"] & valid_numeric["impression"]
    # Apply comparison
    funnel_violation = funnel_valid & ~((conv <= clk) & (clk <= imp))
    for idx in df.index[funnel_violation]:
        errors.append({
            "row_index": idx,
            "campaign_id": str(df.loc[idx, "campaign_id"]) if not pd.isna(df.loc[idx, "campaign_id"]) else None,
            "field": "funnel",
            "code": "consistency_error",
            "message": f"conversion({conv.loc[idx]}) <= click({clk.loc[idx]}) <= impression({imp.loc[idx]}) violated"
        })

    # Date columns
    start_parsed, start_valid = coerce_date(df, "start_date")
    end_parsed, end_valid = coerce_date(df, "end_date")

    for idx in df.index[~start_valid]:
        errors.append({
            "row_index": idx,
            "campaign_id": str(df.loc[idx, "campaign_id"]) if not pd.isna(df.loc[idx, "campaign_id"]) else None,
            "field": "start_date",
            "code": "type_error",
            "message": "start_date is invalid or cannot be parsed"
        })
    for idx in df.index[~end_valid]:
        errors.append({
            "row_index": idx,
            "campaign_id": str(df.loc[idx, "campaign_id"]) if not pd.isna(df.loc[idx, "campaign_id"]) else None,
            "field": "end_date",
            "code": "type_error",
            "message": "end_date is invalid or cannot be parsed"
        })

    # Date interval: start_date <= end_date
    date_interval_valid = start_valid & end_valid
    interval_violation = date_interval_valid & (start_parsed > end_parsed)
    for idx in df.index[interval_violation]:
        errors.append({
            "row_index": idx,
            "campaign_id": str(df.loc[idx, "campaign_id"]) if not pd.isna(df.loc[idx, "campaign_id"]) else None,
            "field": "date_range",
            "code": "range_violation",
            "message": f"start_date({start_parsed.loc[idx].date()}) > end_date({end_parsed.loc[idx].date()})"
        })

    # Currency consistency (optional)
    currency_check_status = "skipped"
    currency_unique_values = None
    if currency_col is None:
        # Try to auto-detect common currency column names
        for candidate in ["currency", "cost_currency"]:
            if candidate in df.columns:
                currency_col = candidate
                break

    if currency_col is not None and currency_col in df.columns:
        currency_check_status = "performed"
        cur_series = df[currency_col].astype(str).str.strip()
        cur_missing = cur_series.isna() | (cur_series == "") | (cur_series.str.lower() == "nan")
        for idx in df.index[cur_missing]:
            errors.append({
                "row_index": idx,
                "campaign_id": str(df.loc[idx, "campaign_id"]) if not pd.isna(df.loc[idx, "campaign_id"]) else None,
                "field": currency_col,
                "code": "missing",
                "message": f"{currency_col} is null or empty"
            })

        currency_unique_values = sorted(cur_series[~cur_missing].unique().tolist())
        if expected_currency is not None:
            mism = cur_series[~cur_missing] != expected_currency
            for idx in df.index[mism]:
                errors.append({
                    "row_index": idx,
                    "campaign_id": str(df.loc[idx, "campaign_id"]) if not pd.isna(df.loc[idx, "campaign_id"]) else None,
                    "field": currency_col,
                    "code": "consistency_error",
                    "message": f"Currency '{cur_series.loc[idx]}' != expected '{expected_currency}'"
                })
        else:
            # Require single currency across dataset
            if len(set(cur_series[~cur_missing])) > 1:
                # Report a dataset-level inconsistency with per-row notes
                for idx in df.index[~cur_missing]:
                    # Only flag rows that contribute to inconsistency (i.e., not the modal value)
                    # For simplicity, flag all non-missing rows when multiple currencies exist
                    errors.append({
                        "row_index": idx,
                        "campaign_id": str(df.loc[idx, "campaign_id"]) if not pd.isna(df.loc[idx, "campaign_id"]) else None,
                        "field": currency_col,
                        "code": "consistency_error",
                        "message": f"Multiple currencies present in dataset: {currency_unique_values}"
                    })
    else:
        currency_check_status = "skipped_no_column"

    # Build errors DataFrame
    errors_df = pd.DataFrame(errors, columns=["row_index", "campaign_id", "field", "code", "message"])
    errors_df.sort_values(by=["row_index", "field"], inplace=True)

    # Summary metrics
    summary = {
        "row_count": int(n_rows),
        "error_count": int(len(errors_df)),
        "missing_campaign_id": int(((df["campaign_id"].isna()) | (df["campaign_id"].astype(str).str.strip() == "")).sum()),
        "invalid_channel": int(ch_invalid.sum()),
        "missing_channel": int(ch_null.sum()),
        "invalid_impression": int((~valid_numeric["impression"]).sum()),
        "invalid_click": int((~valid_numeric["click"]).sum()),
        "invalid_conversion": int((~valid_numeric["conversion"]).sum()),
        "invalid_cost": int((~valid_numeric["cost"]).sum()),
        "negative_impression": int((parsed_numeric["impression"] < 0).sum()),
        "negative_click": int((parsed_numeric["click"] < 0).sum()),
        "negative_conversion": int((parsed_numeric["conversion"] < 0).sum()),
        "negative_cost": int((parsed_numeric["cost"] < 0).sum()),
        "funnel_inconsistency": int(funnel_violation.sum()),
        "invalid_start_date": int((~start_valid).sum()),
        "invalid_end_date": int((~end_valid).sum()),
        "date_range_violation": int(interval_violation.sum()),
        "currency_check_status": currency_check_status,
        "currency_unique_values": currency_unique_values if currency_unique_values is not None else []
    }

    return {
        "fatal": False,
        "message": "Validation completed",
        "errors": errors_df,
        "summary": summary
    }

def main():
    parser = argparse.ArgumentParser(description="Validate campaign report dataset integrity.")
    parser.add_argument("--input", required=True, help="Path to input CSV file")
    parser.add_argument("--output", required=False, help="Path to output CSV file for row-level errors")
    parser.add_argument("--currency-col", required=False, help="Currency column name (e.g., 'currency')")
    parser.add_argument("--expected-currency", required=False, help="Expected currency code (e.g., 'USD')")
    parser.add_argument("--delimiter", required=False, default=",", help="CSV delimiter (default ',')")
    parser.add_argument("--encoding", required=False, default="utf-8", help="File encoding (default 'utf-8')")
    args = parser.parse_args()

    try:
        df = pd.read_csv(args.input, delimiter=args.delimiter, encoding=args.encoding)
    except Exception as e:
        print(f"Failed to read input file: {e}", file=sys.stderr)
        sys.exit(2)

    result = validate_campaign_report(df, currency_col=args.currency_col, expected_currency=args.expected_currency)

    if result.get("fatal", False):
        print(f"Fatal error: {result.get('message')}", file=sys.stderr)
        sys.exit(1)

    # Print summary
    summary = result["summary"]
    print("Validation Summary")
    print(f"Rows: {summary['row_count']}")
    print(f"Total errors: {summary['error_count']}")
    print(f"Missing campaign_id: {summary['missing_campaign_id']}")
    print(f"Missing channel: {summary['missing_channel']}")
    print(f"Invalid channel enum: {summary['invalid_channel']}")
    print(f"Invalid impression: {summary['invalid_impression']} | Negative impression: {summary['negative_impression']}")
    print(f"Invalid click: {summary['invalid_click']} | Negative click: {summary['negative_click']}")
    print(f"Invalid conversion: {summary['invalid_conversion']} | Negative conversion: {summary['negative_conversion']}")
    print(f"Invalid cost: {summary['invalid_cost']} | Negative cost: {summary['negative_cost']}")
    print(f"Funnel inconsistencies: {summary['funnel_inconsistency']}")
    print(f"Invalid start_date: {summary['invalid_start_date']}")
    print(f"Invalid end_date: {summary['invalid_end_date']}")
    print(f"Date range violations: {summary['date_range_violation']}")
    print(f"Currency check: {summary['currency_check_status']}")
    if summary["currency_unique_values"]:
        print(f"Currencies detected: {summary['currency_unique_values']}")

    # Write per-row errors if requested
    if args.output:
        try:
            result["errors"].to_csv(args.output, index=False)
            print(f"Row-level errors written to {args.output}")
        except Exception as e:
            print(f"Failed to write output errors CSV: {e}", file=sys.stderr)
            sys.exit(3)

if __name__ == "__main__":
    main()