#!/usr/bin/env python3 """ MCRO JSON Normalizer -------------------- Scans a directory for JSON files and emits normalized CSV tables. Usage: python mcro_json_normalizer.py --input /path/to/json_dir --outdir /path/to/report_json [--recurse] [--pattern *.json] Notes: - No external deps beyond the standard library + pandas. - Handles missing fields gracefully. - Joins list fields like observations/conflicts with "; " for compact CSVs. """ import argparse import json from pathlib import Path import sys import pandas as pd def as_list(x): if x is None: return [] if isinstance(x, list): return x return [x] def join_list(x): if not isinstance(x, list): return x return "; ".join([str(i) for i in x if pd.notna(i)]) def normalize_one(fp: Path, accum): try: data = json.loads(fp.read_text(encoding="utf-8")) except Exception as e: accum["errors"].append({"file": str(fp), "error": str(e)}) return # Safe getters doc_sha256 = data.get("doc_sha256") or data.get("sha256") or "" filename = data.get("filename", "") filing_type = data.get("filing_type", "") filing_date = data.get("filing_date", "") pdf_page_count = data.get("pdf_page_count", None) size_bytes = data.get("size_bytes", None) # documents accum["documents"].append({ "doc_sha256": doc_sha256, "filename": filename, "filing_type": filing_type, "filing_date": filing_date, "pdf_page_count": pdf_page_count, "size_bytes": size_bytes, }) # metadata md = data.get("metadata") or {} accum["metadata"].append({ "doc_sha256": doc_sha256, "xmp_author": md.get("xmp_author"), "xmp_title": md.get("xmp_title"), "xmp_company": md.get("xmp_company"), "xmp_creator": md.get("xmp_creator"), "xmp_creator_tool": md.get("xmp_creator_tool"), "xmp_producer": md.get("xmp_producer"), "xmp_toolkit": md.get("xmp_toolkit"), "xmp_create_date": md.get("xmp_create_date"), "xmp_metadata_date": md.get("xmp_metadata_date"), "efile_date": md.get("efile_date"), "document_id": md.get("document_id"), "instance_id": md.get("instance_id"), "filename_ts": md.get("filename_ts"), "downloaded_at_ts": md.get("downloaded_at_ts"), "derived_create_vs_metadata_days": md.get("derived_create_vs_metadata_days"), "derived_metadata_newer_by_days": md.get("derived_metadata_newer_by_days"), "derived_create_newer_by_days": md.get("derived_create_newer_by_days"), }) # case c = data.get("case") or {} accum["case"].append({ "doc_sha256": doc_sha256, "case_id": c.get("case_id"), "case_status": c.get("case_status"), "case_assigned_judge": c.get("case_assigned_judge"), "defendant_name": c.get("defendant_name"), "filing_date": c.get("filing_date"), "case_year": c.get("case_year"), "in_past_flag": c.get("in_past_flag"), "cluster_id": c.get("cluster_id"), "cluster_size": c.get("cluster_size"), "cluster_name": c.get("cluster_name"), }) # attorneys - defense for a in as_list(c.get("attorney_defense")): accum["case_attorneys_defense"].append({ "doc_sha256": doc_sha256, "case_id": c.get("case_id"), "attorney_name": a.get("attorney_name"), "attorney_status": a.get("attorney_status"), "attorney_is_lead_flag": a.get("attorney_is_lead_flag"), "guertin_flag": a.get("guertin_flag"), "observations": join_list(a.get("observations") or []), "conflicts": join_list(a.get("conflicts") or []), }) # attorneys - prosecution for a in as_list(c.get("attorney_prosecution")): accum["case_attorneys_prosecution"].append({ "doc_sha256": doc_sha256, "case_id": c.get("case_id"), "attorney_name": a.get("attorney_name"), "attorney_status": a.get("attorney_status"), "attorney_is_lead_flag": a.get("attorney_is_lead_flag"), "guertin_flag": a.get("guertin_flag"), "observations": join_list(a.get("observations") or []), "conflicts": join_list(a.get("conflicts") or []), }) # signatures sig = data.get("signatures") or {} for s in as_list(sig.get("full_report")): accum["signatures"].append({ "doc_sha256": doc_sha256, "chrono_rank": s.get("chrono_rank"), "field_name": s.get("field_name"), "signer_name": s.get("signer_name"), "sign_time": s.get("sign_time"), "sign_time_normalized": s.get("sign_time_normalized"), "signed_bytes": s.get("signed_bytes"), "total_sig_bytes": s.get("total_sig_bytes"), "coverage_pct": s.get("coverage_pct"), "sig_subfilter": s.get("sig_subfilter"), "sig_crypto_valid_flag": s.get("sig_crypto_valid_flag"), "edits_after_sig_flag": s.get("edits_after_sig_flag"), }) # signatures - mcro_authentication mcro = sig.get("mcro_authentication") or {} if mcro: accum["signatures_mcro"].append({ "doc_sha256": doc_sha256, "field_name": mcro.get("field_name"), "sign_time": mcro.get("sign_time"), "sign_time_normalized": mcro.get("sign_time_normalized"), "signed_bytes": mcro.get("signed_bytes"), "total_sig_bytes": mcro.get("total_sig_bytes"), "coverage_pct": mcro.get("coverage_pct"), "sig_subfilter": mcro.get("sig_subfilter"), "sig_crypto_valid_flag": mcro.get("sig_crypto_valid_flag"), "edits_after_sig_flag": mcro.get("edits_after_sig_flag"), }) # objects for o in as_list(data.get("objects")): accum["objects"].append({ "doc_sha256": doc_sha256, "object_sha256": o.get("object_sha256"), "object_type": o.get("object_type"), "object_family": o.get("object_family"), "object_size_bytes": o.get("object_size_bytes"), "object_font_name": o.get("object_font_name"), "object_font_code": o.get("object_font_code"), "object_person_name": o.get("object_person_name"), "object_person_role": o.get("object_person_role"), "object_image_w": o.get("object_image_w"), "object_image_h": o.get("object_image_h"), "object_date": o.get("object_date"), "object_time": o.get("object_time"), "object_url": o.get("object_url"), }) # language terms/instances lang = data.get("language") or {} for t in as_list(lang.get("terms")): accum["language_terms"].append({ "doc_sha256": doc_sha256, "search_group": t.get("search_group"), "search_term": t.get("search_term"), "quantity": t.get("quantity"), }) for inst in as_list(lang.get("instances")): accum["language_instances"].append({ "doc_sha256": doc_sha256, "search_group": inst.get("search_group"), "search_term": inst.get("search_term"), "page_num": inst.get("page_num"), "num_pages": inst.get("num_pages"), "full_text_row": inst.get("full_text_row"), "search_term_url": inst.get("search_term_url"), }) # flags for f in as_list(data.get("flags")): label = f.get("label") if isinstance(f, dict) else f accum["flags"].append({"doc_sha256": doc_sha256, "label": label}) # tracking tr = data.get("tracking") or {} if tr: accum["tracking"].append({ "doc_sha256": doc_sha256, "tracked_flag": tr.get("tracked_flag"), "pdf_group": tr.get("pdf_group"), "font_group": tr.get("font_group"), "font_hash_date": tr.get("font_hash_date"), "font1_name": tr.get("font1_name"), "font1_code": tr.get("font1_code"), "font1_sha256": tr.get("font1_sha256"), "font1_url": tr.get("font1_url"), "font2_name": tr.get("font2_name"), "font2_code": tr.get("font2_code"), "font2_sha256": tr.get("font2_sha256"), "font2_url": tr.get("font2_url"), }) # JavaScript has_js = data.get("has_js") if has_js is not None or data.get("javascript") or data.get("sources"): accum["javascript"].append({ "doc_sha256": doc_sha256, "has_js": has_js, }) for sn in as_list(data.get("javascript")): accum["javascript_snippets"].append({ "doc_sha256": doc_sha256, "file_name": sn.get("file_name"), "code": sn.get("code"), }) sources = data.get("sources") or {} tables = sources.get("tables") or {} if tables: accum["javascript_sources_tables"].append({ "doc_sha256": doc_sha256, "actions_count": tables.get("actions_count"), "js_entries_count": tables.get("js_entries_count"), "has_openaction_aa_flag": tables.get("has_openaction_aa_flag"), "has_page_open_js_flag": tables.get("has_page_open_js_flag"), "targets_page_open": join_list(tables.get("targets_page_open") or []), "targets_annot_aa": join_list(tables.get("targets_annot_aa") or []), "targets_field_js": join_list(tables.get("targets_field_js") or []), }) # sources URLs src = data.get("source_url") or {} if src: accum["sources_urls"].append({ "doc_sha256": doc_sha256, "file_url": src.get("file_url"), "file_decompressed_url": src.get("file_decompressed_url"), "exiftool_metadata_url": src.get("exiftool_metadata_url"), "pdfsig_data_url": src.get("pdfsig_data_url"), "exploded_objects_folder_url": src.get("exploded_objects_folder_url"), "exploded_objects_zip_url": src.get("exploded_objects_zip_url"), "ots_timestamp_url": src.get("ots_timestamp_url"), "docket_url": src.get("docket_url"), }) def main(): ap = argparse.ArgumentParser() ap.add_argument("--input", required=True, help="Directory containing JSON files") ap.add_argument("--outdir", default="report_json", help="Output directory for CSVs") ap.add_argument("--pattern", default="*.json", help="Glob pattern for JSON files (default: *.json)") ap.add_argument("--recurse", action="store_true", help="Recurse into subdirectories") args = ap.parse_args() in_dir = Path(args.input) if not in_dir.exists() or not in_dir.is_dir(): print(f"[error] Input directory not found: {in_dir}", file=sys.stderr) sys.exit(1) if args.recurse: files = sorted(in_dir.rglob(args.pattern)) else: files = sorted(in_dir.glob(args.pattern)) if not files: print(f"[warn] No files matched pattern {args.pattern} in {in_dir}") sys.exit(0) outdir = Path(args.outdir) outdir.mkdir(parents=True, exist_ok=True) # Accumulators (one list per table) accum = { "documents": [], "metadata": [], "case": [], "case_attorneys_defense": [], "case_attorneys_prosecution": [], "signatures": [], "signatures_mcro": [], "objects": [], "language_terms": [], "language_instances": [], "flags": [], "tracking": [], "javascript": [], "javascript_snippets": [], "javascript_sources_tables": [], "sources_urls": [], "errors": [], } # Process n = len(files) print(f"[info] Processing {n} JSON files from {in_dir} (recurse={args.recurse}, pattern={args.pattern})") for i, fp in enumerate(files, 1): normalize_one(fp, accum) if i % 200 == 0 or i == n: print(f" - {i}/{n} ...") # Emit CSVs def emit(name, rows): df = pd.DataFrame(rows) df.to_csv(outdir / f"{name}.csv", index=False) return df for key in [ "documents", "metadata", "case", "case_attorneys_defense", "case_attorneys_prosecution", "signatures", "signatures_mcro", "objects", "language_terms", "language_instances", "flags", "tracking", "javascript", "javascript_snippets", "javascript_sources_tables", "sources_urls", "errors" ]: emit(key, accum[key]) print(f"[ok] Wrote normalized tables to: {outdir.resolve()}") print("[done]") if __name__ == "__main__": main()