#!/usr/bin/env python3 """Reproduce a fictional survey-cleaning handover with Python's standard library. All input records are synthetic. This is a teaching demonstration, not client research, and its rules are illustrative rather than a general survey protocol. """ import argparse import csv import json from collections import Counter from decimal import Decimal, InvalidOperation from pathlib import Path RAW_FIELDS = ( "record_id", "age_years", "delivery_mode", "study_hours_week", "attended_session", ) CLEAN_FIELDS = ( "source_row", "record_id_raw", "record_id", "age_years_raw", "age_years", "delivery_mode_raw", "delivery_mode", "study_hours_week_raw", "study_hours_week", "attended_session_raw", "attended_session", "review_flags", "passes_checks", ) LOG_FIELDS = ("source_row", "record_id", "event", "field", "before", "after", "note") MODE_MAP = { "online": "online", "on-line": "online", "remote": "online", "in person": "in_person", "in-person": "in_person", "in_person": "in_person", } ATTENDANCE_MAP = { "yes": "yes", "y": "yes", "1": "yes", "no": "no", "n": "no", "0": "no", } def write_csv(path, fields, rows): with path.open("w", encoding="utf-8", newline="") as output: writer = csv.DictWriter(output, fieldnames=fields, lineterminator="\n") writer.writeheader() writer.writerows(rows) def clean(input_path, output_dir): with input_path.open(encoding="utf-8", newline="") as source: reader = csv.DictReader(source) if reader.fieldnames != list(RAW_FIELDS): raise ValueError("The CSV header must match the documented five raw fields.") raw_rows = list(reader) if any(None in row or any(value is None for value in row.values()) for row in raw_rows): raise ValueError("Every input record must contain exactly five fields.") # Exact duplicates use every original field, before any recoding or trimming. # source_row is a 1-based CSV record index including the header as record 1. seen = {} retained = [] log = [] for source_row, raw in enumerate(raw_rows, start=2): key = tuple(raw[field] for field in RAW_FIELDS) if key in seen: log.append({ "source_row": source_row, "record_id": raw["record_id"].strip(), "event": "exact_duplicate_removed", "field": "all_fields", "before": "present", "after": "removed", "note": f"All five raw values match source_row {seen[key]}; keep the first record.", }) continue seen[key] = source_row retained.append((source_row, raw)) id_counts = Counter(raw["record_id"].strip() for _, raw in retained if raw["record_id"].strip()) cleaned = [] flag_counts = Counter() normalisations = Counter() for source_row, raw in retained: record = {"source_row": source_row} flags = [] record_id = raw["record_id"].strip() def event(kind, field, before, after, note): log.append({ "source_row": source_row, "record_id": record_id, "event": kind, "field": field, "before": before, "after": after, "note": note, }) def flag(code, field, before, note): flags.append(code) flag_counts[code] += 1 event("review_required", field, before, "", note) def number(field, minimum, maximum, integer=False): original = raw[field] value = original.strip() if not value: flag("missing_" + field, field, original, "Missing response; left blank without imputation.") return "" try: parsed = Decimal(value) except InvalidOperation: parsed = None if parsed is None or not parsed.is_finite() or (integer and parsed != parsed.to_integral_value()): flag("invalid_" + field, field, original, "Value does not match the documented numeric type; raw value retained.") return "" if not minimum <= parsed <= maximum: flag(field + "_out_of_range", field, original, f"Outside the demo range {minimum}–{maximum}; raw value retained.") return "" canonical = format(parsed, "f").rstrip("0").rstrip(".") if "." in format(parsed, "f") else format(parsed, "f") if canonical != original: event("numeric_format_normalised", field, original, canonical, "Numeric value unchanged.") return canonical def category(field, mapping): original = raw[field] value = original.strip().lower() if not value: flag("missing_" + field, field, original, "Missing response; left blank without imputation.") return "" if value not in mapping: flag("unrecognised_" + field, field, original, "Outside the demo's documented categories; raw value retained for review.") return "" canonical = mapping[value] if canonical != original: normalisations[field] += 1 event("category_normalised", field, original, canonical, "Apply the explicit codebook mapping.") return canonical for field in RAW_FIELDS: record[field + "_raw"] = raw[field] record["record_id"] = record_id if not record_id: flag("missing_record_id", "record_id", raw["record_id"], "Identifier absent; retain the record and request source verification.") elif id_counts[record_id] > 1: flag("repeated_record_id", "record_id", raw["record_id"], "Identifier occurs in different retained records; retain all for review.") if record_id != raw["record_id"]: event("identifier_trimmed", "record_id", raw["record_id"], record_id, "Remove surrounding whitespace only.") record["age_years"] = number("age_years", Decimal(18), Decimal(100), integer=True) record["delivery_mode"] = category("delivery_mode", MODE_MAP) record["study_hours_week"] = number("study_hours_week", Decimal(0), Decimal(20)) record["attended_session"] = category("attended_session", ATTENDANCE_MAP) record["review_flags"] = ";".join(flags) record["passes_checks"] = "yes" if not flags else "no" cleaned.append(record) passing = [record for record in cleaned if record["passes_checks"] == "yes"] mean_hours = ( sum(Decimal(record["study_hours_week"]) for record in passing) / len(passing) if passing else None ) summary = { "data_status": "entirely fictional demonstration; no real respondents or client outcomes", "raw_records": len(raw_rows), "exact_duplicate_records_removed": len(raw_rows) - len(cleaned), "retained_records": len(cleaned), "records_passing_all_demo_checks": len(passing), "records_requiring_review": len(cleaned) - len(passing), "review_flags_by_type": dict(sorted(flag_counts.items())), "category_values_normalised": dict(sorted(normalisations.items())), "passing_records_summary": { "denominator": len(passing), "delivery_mode_counts": dict(sorted(Counter(row["delivery_mode"] for row in passing).items())), "attended_session_counts": dict(sorted(Counter(row["attended_session"] for row in passing).items())), "mean_study_hours_week": float(mean_hours.quantize(Decimal("0.01"))) if mean_hours is not None else None, }, } output_dir.mkdir(parents=True, exist_ok=True) write_csv(output_dir / "cleaned-survey.csv", CLEAN_FIELDS, cleaned) write_csv(output_dir / "cleaning-log.csv", LOG_FIELDS, log) (output_dir / "summary.json").write_text( json.dumps(summary, indent=2, ensure_ascii=False) + "\n", encoding="utf-8" ) return summary def main(): base_dir = Path(__file__).resolve().parent parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--input", type=Path, default=base_dir / "raw-survey.csv") parser.add_argument("--output-dir", type=Path, default=base_dir / "reproduced") arguments = parser.parse_args() summary = clean(arguments.input.resolve(), arguments.output_dir.resolve()) print(json.dumps(summary, indent=2, ensure_ascii=False)) if __name__ == "__main__": main()