"""Small CSV profiling helper for Project F experiment dry-run."""
|
|
from __future__ import annotations
|
|
import argparse
|
import csv
|
import json
|
from collections import Counter
|
from pathlib import Path
|
from typing import Any
|
|
|
def _read_rows(input_path: Path) -> tuple[list[str], list[dict[str, str]]]:
|
with input_path.open("r", encoding="utf-8-sig", newline="") as fh:
|
reader = csv.DictReader(fh)
|
if not reader.fieldnames:
|
raise ValueError("input CSV has no header")
|
rows = [dict(row) for row in reader]
|
return list(reader.fieldnames), rows
|
|
|
def build_profile(input_path: Path) -> dict[str, Any]:
|
columns, rows = _read_rows(input_path)
|
missing_by_column: dict[str, int] = {}
|
for column in columns:
|
missing_by_column[column] = sum(1 for row in rows if not (row.get(column) or "").strip())
|
|
id_values = [(row.get("id") or "").strip() for row in rows]
|
nonempty_ids = [value for value in id_values if value]
|
duplicate_id_count = len(nonempty_ids) - len(set(nonempty_ids))
|
|
group_counts: dict[str, int] = {}
|
if "group" in columns:
|
group_counts = dict(sorted(Counter((row.get("group") or "").strip() or "MISSING" for row in rows).items()))
|
|
quality_pass = bool(rows) and duplicate_id_count == 0 and all(
|
missing_by_column.get(column, 0) == 0 for column in ("id", "group", "value", "event_date") if column in columns
|
)
|
|
return {
|
"input_path": str(input_path),
|
"row_count": len(rows),
|
"column_count": len(columns),
|
"columns": columns,
|
"missing_by_column": missing_by_column,
|
"duplicate_id_count": duplicate_id_count,
|
"group_counts": group_counts,
|
"quality_pass": quality_pass,
|
}
|
|
|
def write_outputs(profile: dict[str, Any], output_dir: Path) -> None:
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
(output_dir / "summary.json").write_text(
|
json.dumps(profile, ensure_ascii=False, indent=2) + "\n",
|
encoding="utf-8",
|
)
|
|
with (output_dir / "column_quality.csv").open("w", encoding="utf-8", newline="") as fh:
|
writer = csv.DictWriter(fh, fieldnames=["column", "missing_count"])
|
writer.writeheader()
|
for column, missing_count in profile["missing_by_column"].items():
|
writer.writerow({"column": column, "missing_count": missing_count})
|
|
readout = [
|
"# CSV Profile Readout",
|
"",
|
"创建人员：csv_profile_experiment.py",
|
"文件职责：记录 RUN-F-DEV-001 的 CSV 质量统计摘要。",
|
"管理规范/模板：exp-doc/实验存储体系.md；exp-doc/实验设计.md。",
|
"引用文件：summary.json；column_quality.csv；exp-data/raw/EXP-F-DEV-001/input_sample.csv。",
|
"记录方式：实验结果包 readout；由 helper 运行时生成。",
|
"",
|
f"row_count: {profile['row_count']}",
|
f"column_count: {profile['column_count']}",
|
f"duplicate_id_count: {profile['duplicate_id_count']}",
|
f"quality_pass: {str(profile['quality_pass']).lower()}",
|
"",
|
"group_counts:",
|
]
|
for group, count in profile["group_counts"].items():
|
readout.append(f"- {group}: {count}")
|
(output_dir / "readout.md").write_text("\n".join(readout) + "\n", encoding="utf-8")
|
|
|
def main() -> int:
|
parser = argparse.ArgumentParser(description="Profile a CSV and write a small experiment result package.")
|
parser.add_argument("--input", required=True, type=Path)
|
parser.add_argument("--output-dir", required=True, type=Path)
|
args = parser.parse_args()
|
|
profile = build_profile(args.input)
|
write_outputs(profile, args.output_dir)
|
return 0
|
|
|
if __name__ == "__main__":
|
raise SystemExit(main())
|
