#!/usr/bin/env python3 """ MCRO PDF Group Report ===================== Scans a directory of JSON document descriptors, filters to documents with tracking.tracked_flag == true, groups them by tracking.pdf_group, and emits a set of CSV "data dumps" plus per-group statistics. FOCUS: • Only JSON files where tracking.tracked_flag is true (bool True or "true"). • Grouping is by tracking.pdf_group. • Everything is driven by the JSON structure previously used in the MCRO pipeline (docs, objects, language.terms, signatures.pdfsig, metadata, tracking, case). EXTRACTED FIELDS ================ From the top-level document and nested structures, for each *tracked* document: filename filing_type filing_date pdf_page_count objects[*] -> flattened (one row per object) language.terms[*] -> search_group, search_term, quantity signatures.pdfsig -> all keys (one row per document) metadata.* -> all keys as columns tracking.* -> all keys as columns EXCEPT tracking.tracked_flag case.* -> all keys EXCEPT case.cluster_size, with attorney_defense / attorney_prosecution condensed to CSV-friendly columns. LANGUAGE: • language.instances is NOT included. FLAGS: • flags.* is NOT included (you said you'll handle with a CSV later). SIGNATURES: • signatures.pdfsig is included (all keys). • signatures.full_report and signatures.mcro_authentication are NOT included in the data-dump tables. OUTPUT FILES (all CSV, written into --outdir) ============================================= 1) pdf_group_documents.csv One row per tracked document. Columns include: - pdf_group, font_group, font_hash_date, font1_* etc (tracking.*) - doc_sha256, filename, filing_type, filing_date, pdf_page_count - metadata.* keys as columns - case.* keys as columns (excluding cluster_size) - aggregated attorney columns: * case_attorney_defense_names * case_attorney_defense_lead_names * case_attorney_prosecution_names * case_attorney_prosecution_lead_names * case_attorney_defense_raw (JSON string) * case_attorney_prosecution_raw (JSON string) 2) pdf_group_objects.csv One row per (pdf_group, doc_sha256, object_sha256). Includes all object-level keys plus doc-level context. 3) pdf_group_terms.csv One row per (pdf_group, doc_sha256, search_group, search_term). Includes quantity plus doc and case context. 4) pdf_group_signatures_pdfsig.csv One row per tracked document. All signatures.pdfsig keys flattened as columns, plus pdf_group/doc context. 5) pdf_group_overview.csv One row per pdf_group, with high-level statistics: - pdf_group - n_docs - n_cases - n_clusters - case_ids - cluster_ids - cluster_names - min_filing_date, max_filing_date - min_pdf_page_count, max_pdf_page_count, avg_pdf_page_count - n_unique_filing_types, filing_types - n_unique_terms, n_term_occurrences - n_unique_search_groups - n_objects, n_unique_object_sha256, n_unique_object_font_names - n_docs_with_pdfsig 6) pdf_group_term_stats.csv One row per (pdf_group, search_group, search_term): - total_quantity - doc_count - case_count - cluster_count - first_seen_date - last_seen_date 7) README.txt Brief description of what's in the folder. USAGE ===== python mcro_pdf_group_report.py \ --input /path/to/json_dir \ --outdir /path/to/pdf_group_report \ [--pattern "*.json"] \ [--recurse] The script is robust to missing keys and will simply leave those cells blank. """ import argparse import json from pathlib import Path from typing import Any, Dict, List import sys import pandas as pd def as_list(x): if x is None: return [] if isinstance(x, list): return x return [x] def ensure_string_cols(df: pd.DataFrame, cols: List[str]) -> pd.DataFrame: for c in cols: if c in df.columns: df[c] = df[c].astype("string") return df def join_unique(vals): vals = [v for v in vals if pd.notna(v)] return "; ".join(sorted(set(map(str, vals)))) def safe_get(d: Dict[str, Any], key: str, default=None): if isinstance(d, dict): return d.get(key, default) return default def flatten_attorneys(att_list: Any) -> Dict[str, str]: """ Given case.attorney_defense or case.attorney_prosecution (list of dicts), produce CSV-friendly aggregates: - names (all) - lead_names (attorney_is_lead_flag True in top-level or observations) - raw JSON string for full details """ if not isinstance(att_list, list): return { "names": "", "lead_names": "", "raw": json.dumps(att_list, ensure_ascii=False) if att_list is not None else "" } names = [] lead_names = [] for att in att_list: if not isinstance(att, dict): continue name = att.get("attorney_name") or att.get("name") or "" if name: names.append(name) # Determine lead flag: check top-level and any observations lead_flag = bool(att.get("attorney_is_lead_flag")) for obs in as_list(att.get("observations")): if isinstance(obs, dict) and obs.get("attorney_is_lead_flag"): lead_flag = True if lead_flag and name: lead_names.append(name) return { "names": "; ".join(sorted(set(names))) if names else "", "lead_names": "; ".join(sorted(set(lead_names))) if lead_names else "", "raw": json.dumps(att_list, ensure_ascii=False) } def main(): ap = argparse.ArgumentParser() ap.add_argument("--input", required=True, help="Directory containing JSON files") ap.add_argument("--outdir", required=True, 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 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} under {in_dir}", file=sys.stderr) sys.exit(0) outdir = Path(args.outdir) outdir.mkdir(parents=True, exist_ok=True) # Data collectors docs_rows = [] objects_rows = [] terms_rows = [] sig_rows = [] # For overview/term stats all_terms_for_stats = [] for fp in files: try: data = json.loads(fp.read_text(encoding="utf-8")) except Exception as e: print(f"[warn] Failed to parse JSON: {fp} ({e})", file=sys.stderr) continue tracking = data.get("tracking") or {} tracked_flag = tracking.get("tracked_flag") # Accept True, "true", "TRUE" if not (tracked_flag is True or (isinstance(tracked_flag, str) and tracked_flag.strip().lower() == "true")): continue # skip untracked docs pdf_group = tracking.get("pdf_group") if pdf_group is None: # If there's no pdf_group, we skip; they are out-of-scope for this report continue pdf_group = str(pdf_group) font_group = tracking.get("font_group") 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) metadata = data.get("metadata") or {} case = data.get("case") or {} signatures = data.get("signatures") or {} # Case flatten (excluding cluster_size) case_flat = {} case_keys = [k for k in case.keys() if k not in ("attorney_defense","attorney_prosecution","cluster_size")] for k in case_keys: case_flat[f"case_{k}"] = case.get(k) # Attorneys - defense att_def = flatten_attorneys(case.get("attorney_defense")) case_flat["case_attorney_defense_names"] = att_def["names"] case_flat["case_attorney_defense_lead_names"] = att_def["lead_names"] case_flat["case_attorney_defense_raw"] = att_def["raw"] # Attorneys - prosecution att_pros = flatten_attorneys(case.get("attorney_prosecution")) case_flat["case_attorney_prosecution_names"] = att_pros["names"] case_flat["case_attorney_prosecution_lead_names"] = att_pros["lead_names"] case_flat["case_attorney_prosecution_raw"] = att_pros["raw"] # Tracking flatten (all keys except tracked_flag) tracking_flat = {} for k, v in tracking.items(): if k == "tracked_flag": continue tracking_flat[f"tracking_{k}"] = v # Metadata flatten (all keys) metadata_flat = {} for k, v in metadata.items(): metadata_flat[f"metadata_{k}"] = v # Document-level row doc_row = { "pdf_group": pdf_group, "font_group": font_group, "doc_sha256": doc_sha256, "filename": filename, "filing_type": filing_type, "filing_date": filing_date, "pdf_page_count": pdf_page_count, "size_bytes": size_bytes, } doc_row.update(tracking_flat) doc_row.update(metadata_flat) doc_row.update(case_flat) docs_rows.append(doc_row) # Objects rows for obj in as_list(data.get("objects")): if not isinstance(obj, dict): continue o_row = { "pdf_group": pdf_group, "doc_sha256": doc_sha256, "filename": filename, "case_id": case.get("case_id"), } for k, v in obj.items(): o_row[k] = v objects_rows.append(o_row) # Terms rows (language.terms only) lang = data.get("language") or {} for term in as_list(lang.get("terms")): if not isinstance(term, dict): continue search_group = term.get("search_group") search_term = term.get("search_term") quantity = term.get("quantity") t_row = { "pdf_group": pdf_group, "doc_sha256": doc_sha256, "filename": filename, "case_id": case.get("case_id"), "cluster_id": case.get("cluster_id"), "cluster_name": case.get("cluster_name"), "filing_date": filing_date, "search_group": search_group, "search_term": search_term, "quantity": quantity, } terms_rows.append(t_row) all_terms_for_stats.append(t_row) # Signatures.pdfsig row pdfsig = safe_get(signatures, "pdfsig", {}) if isinstance(pdfsig, dict): s_row = { "pdf_group": pdf_group, "doc_sha256": doc_sha256, "filename": filename, "case_id": case.get("case_id"), } # flatten all pdfsig keys for k, v in pdfsig.items(): s_row[f"pdfsig_{k}"] = v sig_rows.append(s_row) # Convert to DataFrames docs_df = pd.DataFrame(docs_rows) objects_df = pd.DataFrame(objects_rows) terms_df = pd.DataFrame(terms_rows) sig_df = pd.DataFrame(sig_rows) stats_terms_df = pd.DataFrame(all_terms_for_stats) # Write main dumps if not docs_df.empty: docs_df.to_csv(outdir / "pdf_group_documents.csv", index=False) else: (outdir / "pdf_group_documents.csv").write_text("", encoding="utf-8") if not objects_df.empty: objects_df.to_csv(outdir / "pdf_group_objects.csv", index=False) else: (outdir / "pdf_group_objects.csv").write_text("", encoding="utf-8") if not terms_df.empty: terms_df.to_csv(outdir / "pdf_group_terms.csv", index=False) else: (outdir / "pdf_group_terms.csv").write_text("", encoding="utf-8") if not sig_df.empty: sig_df.to_csv(outdir / "pdf_group_signatures_pdfsig.csv", index=False) else: (outdir / "pdf_group_signatures_pdfsig.csv").write_text("", encoding="utf-8") # ---- Per-group overview ---- if not docs_df.empty: # Basic group stats group = docs_df.copy() # Merge term stats for counts if not stats_terms_df.empty: # total occurrences and unique terms per group term_group = stats_terms_df.groupby("pdf_group", as_index=False).agg( n_term_occurrences=("quantity", lambda s: int(pd.to_numeric(s, errors="coerce").fillna(0).sum())), n_unique_terms=("search_term", lambda s: int(s.dropna().nunique())), n_unique_search_groups=("search_group", lambda s: int(pd.Series(s).dropna().nunique())) ) else: term_group = pd.DataFrame(columns=["pdf_group","n_term_occurrences","n_unique_terms","n_unique_search_groups"]) # object stats if not objects_df.empty: obj_group = objects_df.groupby("pdf_group", as_index=False).agg( n_objects=("object_sha256", "count") if "object_sha256" in objects_df.columns else ("doc_sha256","count"), n_unique_object_sha256=("object_sha256", lambda s: int(pd.Series(s).dropna().nunique())) if "object_sha256" in objects_df.columns else ("doc_sha256", lambda s: int(pd.Series(s).dropna().nunique())), n_unique_object_font_names=("object_font_name", lambda s: int(pd.Series(s).dropna().nunique())) if "object_font_name" in objects_df.columns else ("doc_sha256", lambda s: 0) ) else: obj_group = pd.DataFrame(columns=["pdf_group","n_objects","n_unique_object_sha256","n_unique_object_font_names"]) # pdfsig stats if not sig_df.empty: sig_group = sig_df.groupby("pdf_group", as_index=False).agg( n_docs_with_pdfsig=("doc_sha256","nunique") ) else: sig_group = pd.DataFrame(columns=["pdf_group","n_docs_with_pdfsig"]) # assemble overview overview = ( docs_df.groupby("pdf_group", as_index=False) .agg( n_docs=("doc_sha256","nunique"), n_cases=("case_case_id","nunique") if "case_case_id" in docs_df.columns else ("doc_sha256","count"), n_clusters=("case_cluster_id","nunique") if "case_cluster_id" in docs_df.columns else ("doc_sha256","count"), case_ids=("case_case_id", join_unique) if "case_case_id" in docs_df.columns else ("doc_sha256", join_unique), cluster_ids=("case_cluster_id", join_unique) if "case_cluster_id" in docs_df.columns else ("doc_sha256", join_unique), cluster_names=("case_cluster_name", join_unique) if "case_cluster_name" in docs_df.columns else ("doc_sha256", join_unique), min_filing_date=("filing_date", lambda s: s.dropna().min() if len(s.dropna()) else pd.NA), max_filing_date=("filing_date", lambda s: s.dropna().max() if len(s.dropna()) else pd.NA), min_pdf_page_count=("pdf_page_count", lambda s: pd.to_numeric(s, errors="coerce").min()), max_pdf_page_count=("pdf_page_count", lambda s: pd.to_numeric(s, errors="coerce").max()), avg_pdf_page_count=("pdf_page_count", lambda s: float(pd.to_numeric(s, errors="coerce").mean() or 0.0)), n_unique_filing_types=("filing_type", lambda s: int(pd.Series(s).dropna().nunique())), filing_types=("filing_type", join_unique), ) ) # merge term / object / sig summary overview = overview.merge(term_group, on="pdf_group", how="left") overview = overview.merge(obj_group, on="pdf_group", how="left") overview = overview.merge(sig_group, on="pdf_group", how="left") # fill NaNs with sensible defaults for col in ["n_term_occurrences","n_unique_terms","n_unique_search_groups", "n_objects","n_unique_object_sha256","n_unique_object_font_names", "n_docs_with_pdfsig"]: if col in overview.columns: overview[col] = overview[col].fillna(0).astype(int) overview.to_csv(outdir / "pdf_group_overview.csv", index=False) else: (outdir / "pdf_group_overview.csv").write_text("", encoding="utf-8") # ---- Term stats per group ---- if not stats_terms_df.empty: stats_terms_df["quantity"] = pd.to_numeric(stats_terms_df["quantity"], errors="coerce").fillna(0).astype(int) term_stats = ( stats_terms_df.groupby(["pdf_group","search_group","search_term"], as_index=False) .agg( total_quantity=("quantity","sum"), doc_count=("doc_sha256","nunique"), case_count=("case_id","nunique"), cluster_count=("cluster_id","nunique"), first_seen_date=("filing_date", lambda s: s.dropna().min() if len(s.dropna()) else pd.NA), last_seen_date=("filing_date", lambda s: s.dropna().max() if len(s.dropna()) else pd.NA), ) ) term_stats.to_csv(outdir / "pdf_group_term_stats.csv", index=False) else: (outdir / "pdf_group_term_stats.csv").write_text("", encoding="utf-8") # ---- README ---- readme = """ MCRO PDF Group Report ===================== This folder was generated by mcro_pdf_group_report.py. Scope: • Only JSON files where tracking.tracked_flag == true. • Only documents that have tracking.pdf_group set. • Grouping and stats are all based on tracking.pdf_group. Main dumps: • pdf_group_documents.csv — 1 row per tracked document. • pdf_group_objects.csv — 1 row per object with pdf_group/doc context. • pdf_group_terms.csv — 1 row per doc/term with quantities. • pdf_group_signatures_pdfsig.csv — 1 row per doc with pdfsig fields flattened. Stats: • pdf_group_overview.csv — high-level metrics per pdf_group. • pdf_group_term_stats.csv — per pdf_group/search_term aggregates. You can now use these CSVs to drive deeper analyses: • Per-group flag overlays (using your flags CSV). • Font hash uniqueness/probabilities per group. • Cross-group language, signature, and metadata comparisons. """.strip() + "\n" (outdir / "README.txt").write_text(readme, encoding="utf-8") print(f"[ok] Wrote PDF group report to: {outdir.resolve()}") if __name__ == "__main__": main()