#!/usr/bin/env python3 """ MCRO Language Analytics Report ------------------------------ Scans JSON files and emits language-focused CSV reports. Usage: python mcro_language_report.py --input /path/to/json_dir --outdir /path/to/report_language [--recurse] [--pattern *.json] [--focus_case 27-CR-23-1886] Outputs (CSV): - terms_overall.csv - terms_by_case.csv - terms_by_cluster.csv - ubiquitous_terms_all_cases.csv - ubiquitous_terms_all_clusters.csv - focal_case_terms.csv - focal_case_instances.csv - focal_term_uniqueness.csv - focal_term_links_by_case.csv - overview_metrics.csv - README.txt Notes: - Uses language.terms.quantity as term occurrence count per doc. - Instances are provided for page-level context (language.instances). - Uniqueness score: 1 / (1 + others_case_count) and 1 / (1 + others_cluster_count). """ 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 safe_str(x): return "" if x is None else str(x) def main(): ap = argparse.ArgumentParser() ap.add_argument("--input", required=True, help="Directory with JSON files") ap.add_argument("--outdir", default="report_language", help="Output folder") ap.add_argument("--pattern", default="*.json", help="Glob for JSON files") ap.add_argument("--recurse", action="store_true", help="Recurse into subdirs") ap.add_argument("--focus_case", default="27-CR-23-1886", help="Case ID to center uniqueness analysis on") 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) files = sorted(in_dir.rglob(args.pattern) if args.recurse else in_dir.glob(args.pattern)) if not files: print(f"[warn] No JSON files matched {args.pattern} in {in_dir}") sys.exit(0) outdir = Path(args.outdir) outdir.mkdir(parents=True, exist_ok=True) # ---- Extract minimal doc, case, language tables ---- docs_rows = [] terms_rows = [] inst_rows = [] for fp in files: try: data = json.loads(fp.read_text(encoding="utf-8")) except Exception as e: continue 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", "") case = data.get("case") or {} case_id = case.get("case_id") cluster_id = case.get("cluster_id") cluster_name = case.get("cluster_name") docs_rows.append({ "doc_sha256": doc_sha256, "filename": filename, "filing_type": filing_type, "filing_date": filing_date, "case_id": case_id, "cluster_id": cluster_id, "cluster_name": cluster_name, }) lang = data.get("language") or {} for t in as_list(lang.get("terms")): terms_rows.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")): inst_rows.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"), }) docs = pd.DataFrame(docs_rows) terms = pd.DataFrame(terms_rows) instances = pd.DataFrame(inst_rows) # Guard for empty language if terms.empty: (outdir / "README.txt").write_text("No language terms found in the provided JSONs.\n", encoding="utf-8") print("[warn] No terms found. Exiting.") sys.exit(0) # Normalize types for col in ["search_group", "search_term"]: terms[col] = terms[col].astype("string") terms["quantity"] = pd.to_numeric(terms["quantity"], errors="coerce").fillna(0).astype(int) # Join term/doc mapping docs_light = docs[["doc_sha256","filename","filing_type","filing_date","case_id","cluster_id","cluster_name"]] t = terms.merge(docs_light, on="doc_sha256", how="left") # ---- Metrics ---- total_docs = t["doc_sha256"].nunique() total_cases = t["case_id"].dropna().nunique() total_clusters = t["cluster_id"].dropna().nunique() # terms_overall terms_overall = ( t.groupby(["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), ) .sort_values(["search_group","search_term"]) ) # terms_by_case def list_join(vals): vals = [v for v in vals if pd.notna(v)] return "; ".join(sorted(set(map(str, vals)))) terms_by_case = ( t.groupby(["case_id","cluster_id","cluster_name","search_group","search_term"], dropna=False, as_index=False) .agg( total_quantity=("quantity","sum"), doc_count=("doc_sha256","nunique"), filenames=("filename", list_join), ) .sort_values(["case_id","search_group","search_term"]) ) # terms_by_cluster terms_by_cluster = ( t.groupby(["cluster_id","cluster_name","search_group","search_term"], dropna=False, as_index=False) .agg( total_quantity=("quantity","sum"), doc_count=("doc_sha256","nunique"), case_count=("case_id","nunique"), filenames=("filename", list_join), ) .sort_values(["cluster_id","search_group","search_term"]) ) # ubiquitous terms ub_all_cases = terms_overall.loc[terms_overall["case_count"] == total_cases].copy() ub_all_clusters = terms_overall.loc[terms_overall["cluster_count"] == total_clusters].copy() # ---- Focal case analysis ---- focus_case = args.focus_case t_focus = t.loc[t["case_id"] == focus_case].copy() # Focal terms with in-case stats focal_terms = ( t_focus.groupby(["case_id","cluster_id","cluster_name","search_group","search_term"], as_index=False) .agg( in_case_quantity=("quantity","sum"), in_case_doc_count=("doc_sha256","nunique"), filenames=("filename", list_join), ) ) # For each focal term, compute "others" metrics (excluding focal case and focal cluster(s)) all_case_counts = ( t.groupby(["search_group","search_term"], as_index=False) .agg( total_case_count=("case_id","nunique"), total_cluster_count=("cluster_id","nunique"), total_doc_count=("doc_sha256","nunique") ) ) # Determine focal cluster set focal_clusters = set(t_focus["cluster_id"].dropna().unique().tolist()) # Case/cluster counts excluding focus others_case_counts = ( t.loc[t["case_id"] != focus_case] .groupby(["search_group","search_term"], as_index=False) .agg(others_case_count=("case_id","nunique")) ) t_excl_focal_clusters = t.loc[~t["cluster_id"].isin(focal_clusters)] if focal_clusters else t.copy() others_cluster_counts = ( t_excl_focal_clusters.groupby(["search_group","search_term"], as_index=False) .agg(others_cluster_count=("cluster_id","nunique")) ) others_doc_counts = ( t.loc[t["case_id"] != focus_case] .groupby(["search_group","search_term"], as_index=False) .agg(others_doc_count=("doc_sha256","nunique")) ) # Merge others into focal_terms focal_term_uniqueness = ( focal_terms.merge(all_case_counts, on=["search_group","search_term"], how="left") .merge(others_case_counts, on=["search_group","search_term"], how="left") .merge(others_cluster_counts, on=["search_group","search_term"], how="left") .merge(others_doc_counts, on=["search_group","search_term"], how="left") ) for col in ["others_case_count","others_cluster_count","others_doc_count"]: focal_term_uniqueness[col] = focal_term_uniqueness[col].fillna(0).astype(int) # Uniqueness score: higher is more unique to the focus case focal_term_uniqueness["uniqueness_score_cases"] = 1.0 / (1 + focal_term_uniqueness["others_case_count"]) focal_term_uniqueness["uniqueness_score_clusters"] = 1.0 / (1 + focal_term_uniqueness["others_cluster_count"]) # Rank for convenience focal_term_uniqueness = focal_term_uniqueness.sort_values( ["uniqueness_score_cases","uniqueness_score_clusters","in_case_quantity"], ascending=[False, False, False] ) # Focal instances (page-level) focal_instances = instances.merge(docs_light, on="doc_sha256", how="left") focal_instances = focal_instances.loc[focal_instances["case_id"] == focus_case].copy() focal_instances = focal_instances[ ["case_id","cluster_id","cluster_name","filename","search_group","search_term","page_num","num_pages","full_text_row","search_term_url"] ].sort_values(["filename","search_group","search_term","page_num"]) # Focal term links: enumerate other cases/clusters where each focal term appears t_focus_terms = focal_terms[["search_group","search_term"]].drop_duplicates() t_others = t.merge(t_focus_terms, on=["search_group","search_term"], how="inner") t_others = t_others.loc[t_others["case_id"] != focus_case].copy() def filenames_join(g): return "; ".join(sorted(set(map(str, g)))) focal_term_links_by_case = ( t_others.groupby(["search_group","search_term","case_id","cluster_id","cluster_name"], as_index=False) .agg( other_case_quantity=("quantity","sum"), other_case_doc_count=("doc_sha256","nunique"), other_filenames=("filename", filenames_join), ) .sort_values(["search_group","search_term","other_case_quantity"], ascending=[True, True, False]) ) # ---- Overview metrics ---- metrics = pd.DataFrame([{ "total_docs_with_terms": int(total_docs), "total_cases_with_terms": int(total_cases), "total_clusters_with_terms": int(total_clusters), "distinct_terms": int(terms_overall.shape[0]), "ubiquitous_terms_all_cases": int(ub_all_cases.shape[0]), "ubiquitous_terms_all_clusters": int(ub_all_clusters.shape[0]), "focus_case": args.focus_case, "focus_case_distinct_terms": int(focal_terms.shape[0]), }]) # ---- Write outputs ---- outdir.mkdir(parents=True, exist_ok=True) terms_overall.to_csv(outdir / "terms_overall.csv", index=False) terms_by_case.to_csv(outdir / "terms_by_case.csv", index=False) terms_by_cluster.to_csv(outdir / "terms_by_cluster.csv", index=False) ub_all_cases.to_csv(outdir / "ubiquitous_terms_all_cases.csv", index=False) ub_all_clusters.to_csv(outdir / "ubiquitous_terms_all_clusters.csv", index=False) focal_terms.to_csv(outdir / "focal_case_terms.csv", index=False) focal_instances.to_csv(outdir / "focal_case_instances.csv", index=False) focal_term_uniqueness.to_csv(outdir / "focal_term_uniqueness.csv", index=False) focal_term_links_by_case.to_csv(outdir / "focal_term_links_by_case.csv", index=False) metrics.to_csv(outdir / "overview_metrics.csv", index=False) readme = f""" MCRO Language Analytics Report ============================== Files: - terms_overall.csv Per-term totals across the corpus with doc/case/cluster coverage and first/last seen dates. - terms_by_case.csv Per case_id/cluster_id/cluster_name, per term: total quantity, doc count, filenames. - terms_by_cluster.csv Per cluster, per term: total quantity, doc count, case coverage, filenames. - ubiquitous_terms_all_cases.csv Terms that appear in ALL cases with language. - ubiquitous_terms_all_clusters.csv Terms that appear in ALL clusters with language. - focal_case_terms.csv All terms observed in the focus case ({args.focus_case}) with in-case totals. - focal_case_instances.csv Page-level instances (quotes/rows) for the focus case ({args.focus_case}). - focal_term_uniqueness.csv For each focus-case term, how widely it appears OUTSIDE the focus case: others_case_count, others_cluster_count, others_doc_count, plus uniqueness_score_* = 1/(1+others_*). Higher scores imply the term is more unique to the focus case. - focal_term_links_by_case.csv Expansion list: for each focus-case term, which other cases/clusters contain it, with per-case quantities and filenames. - overview_metrics.csv Single-row corpus metrics for quick reference. Notes: - Quantity is sourced from language.terms.quantity per doc. - If you prefer instance-based counts, substitute language_instances with per-doc counts. - Uniqueness score uses an inverse transform for interpretability: 1/(1+n). """.strip() + "\n" (outdir / "README.txt").write_text(readme, encoding="utf-8") print(f"[ok] Wrote language reports to: {outdir.resolve()}") if __name__ == "__main__": main()