#!/usr/bin/env python3 # (script body unchanged from previous cell; only environment imports fixed) 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 list_join(vals): vals = [v for v in vals if pd.notna(v)] return "; ".join(sorted(set(map(str, vals)))) def ensure_string_cols(df, cols): for c in cols: if c in df.columns: df[c] = df[c].astype("string") return df def load_terms_csv(path: Path) -> pd.DataFrame: df = pd.read_csv(path) # normalize column names colmap = {} for c in df.columns: lc = c.lower().strip() if lc == "search_term": colmap[c] = "search_term" elif lc == "search_group": colmap[c] = "search_group" if "search_term" not in colmap.values(): raise ValueError("terms_csv must contain a 'search_term' column") df = df.rename(columns=colmap) keep = [c for c in ["search_group","search_term"] if c in df.columns] df = df[keep].copy() for c in keep: df[c] = df[c].astype("string").str.strip() df = df.dropna(subset=["search_term"]).drop_duplicates().reset_index(drop=True) return df def main(): ap = argparse.ArgumentParser() ap.add_argument("--input", required=True, help="Directory with JSON files") ap.add_argument("--terms_csv", required=True, help="CSV file with search_term (and optional search_group)") ap.add_argument("--outdir", default="report_language_list", 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=None, help="Optional focal case: or ALL") args = ap.parse_args() in_dir = Path(args.input) 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) # Load term list term_list = load_terms_csv(Path(args.terms_csv)) has_group = "search_group" in term_list.columns key_cols = ["search_group","search_term"] if has_group else ["search_term"] # ---- Extract minimal doc, case, language ---- docs_rows, terms_rows, inst_rows = [], [], [] for fp in files: try: data = json.loads(fp.read_text(encoding="utf-8")) except Exception: 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) # Standardize types terms = ensure_string_cols(terms, ["search_group","search_term"]) instances = ensure_string_cols(instances, ["search_group","search_term"]) if "quantity" in terms.columns: terms["quantity"] = pd.to_numeric(terms["quantity"], errors="coerce").fillna(0).astype(int) # Join doc context t = terms.merge(docs, on="doc_sha256", how="left") t = ensure_string_cols(t, ["search_group","search_term"]) # Filter to provided terms term_list_key = ensure_string_cols(term_list.copy(), key_cols) t_filtered = t.merge(term_list_key.drop_duplicates(), on=key_cols, how="inner") # ---- GLOBAL reports (all cases; includes focal case automatically) ---- def list_join_unique(vals): vals = [v for v in vals if pd.notna(v)] return "; ".join(sorted(set(map(str, vals)))) terms_overall = ( t_filtered.groupby(key_cols, 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)) ) by_case = ( t_filtered.groupby(["case_id","cluster_id","cluster_name"] + key_cols, as_index=False, dropna=False) .agg(total_quantity=("quantity","sum"), doc_count=("doc_sha256","nunique"), filenames=("filename", list_join_unique)) ) by_cluster = ( t_filtered.groupby(["cluster_id","cluster_name"] + key_cols, as_index=False, dropna=False) .agg(total_quantity=("quantity","sum"), doc_count=("doc_sha256","nunique"), case_count=("case_id","nunique"), filenames=("filename", list_join_unique)) ) # Page-level instances for filtered terms if not instances.empty: inst_all = instances.merge(docs[["doc_sha256","filename","filing_type","filing_date","case_id","cluster_id","cluster_name"]], on="doc_sha256", how="left") inst_all = ensure_string_cols(inst_all, key_cols) instances_filtered = inst_all.merge(term_list_key.drop_duplicates(), on=key_cols, how="inner") instances_filtered = instances_filtered[[ "filename","filing_type","filing_date","case_id","cluster_id","cluster_name" ] + key_cols + ["page_num","num_pages","full_text_row","search_term_url"]].sort_values(["filename"] + key_cols + ["page_num"]) else: instances_filtered = pd.DataFrame(columns=["filename","filing_type","filing_date","case_id","cluster_id","cluster_name"] + key_cols + ["page_num","num_pages","full_text_row","search_term_url"]) # Matrices across ALL cases/clusters (filtered terms only) matrix_cases_qty = by_case.pivot_table( index=key_cols, columns="case_id", values="total_quantity", fill_value=0, aggfunc="sum" ).reset_index() matrix_cases_presence = matrix_cases_qty.copy() matrix_cases_presence.iloc[:, len(key_cols):] = (matrix_cases_presence.iloc[:, len(key_cols):] > 0).astype(int) matrix_clusters_qty = by_cluster.pivot_table( index=key_cols, columns="cluster_id", values="total_quantity", fill_value=0, aggfunc="sum" ).reset_index() matrix_clusters_presence = matrix_clusters_qty.copy() matrix_clusters_presence.iloc[:, len(key_cols):] = (matrix_clusters_presence.iloc[:, len(key_cols):] > 0).astype(int) # Global rarity weights per term term_case_counts = by_case.groupby(key_cols, as_index=False).agg(case_count=("case_id","nunique")) term_cluster_counts = by_cluster.groupby(key_cols, as_index=False).agg(cluster_count=("cluster_id","nunique")) case_weights = term_case_counts.copy() case_weights["weight_cases"] = case_weights["case_count"].apply(lambda k: (1.0/(k-1)) if pd.notna(k) and k>=2 else 0.0) cluster_weights = term_cluster_counts.copy() cluster_weights["weight_clusters"] = cluster_weights["cluster_count"].apply(lambda k: (1.0/(k-1)) if pd.notna(k) and k>=2 else 0.0) def apply_weight_matrix(pres_df, weight_df, weight_col): core = pres_df.set_index(key_cols) weights = weight_df.set_index(key_cols)[weight_col].reindex(core.index).fillna(0).astype(float) weighted = core.copy().astype(float) for col in core.columns: weighted[col] = core[col].astype(float).where(core[col]==0, weights) return weighted.reset_index() rarity_weight_matrix_cases = apply_weight_matrix(matrix_cases_presence, case_weights, "weight_cases") rarity_weight_matrix_clusters = apply_weight_matrix(matrix_clusters_presence, cluster_weights, "weight_clusters") # Overview metrics = pd.DataFrame([{ "term_list_rows": int(term_list_key.drop_duplicates().shape[0]), "terms_found_in_corpus": int(terms_overall.shape[0]), "docs_with_list_terms": int(t_filtered["doc_sha256"].nunique()), "cases_with_list_terms": int(t_filtered["case_id"].nunique()), "clusters_with_list_terms": int(t_filtered["cluster_id"].nunique()), }]) # Write GLOBAL outputs outdir.mkdir(parents=True, exist_ok=True) terms_overall.to_csv(outdir/"terms_overall_filtered.csv", index=False) by_case.to_csv(outdir/"terms_by_case_filtered.csv", index=False) by_cluster.to_csv(outdir/"terms_by_cluster_filtered.csv", index=False) instances_filtered.to_csv(outdir/"instances_filtered.csv", index=False) matrix_cases_qty.to_csv(outdir/"matrix_cases_quantity.csv", index=False) matrix_cases_presence.to_csv(outdir/"matrix_cases_presence.csv", index=False) matrix_clusters_qty.to_csv(outdir/"matrix_clusters_quantity.csv", index=False) matrix_clusters_presence.to_csv(outdir/"matrix_clusters_presence.csv", index=False) rarity_weight_matrix_cases.to_csv(outdir/"rarity_weight_matrix_cases.csv", index=False) rarity_weight_matrix_clusters.to_csv(outdir/"rarity_weight_matrix_clusters.csv", index=False) metrics.to_csv(outdir/"overview_metrics.csv", index=False) # ---- FOCAL MODES ---- focus_arg = (args.focus_case or "").strip() if focus_arg and focus_arg.upper() != "ALL": # single-case focus focus = focus_arg t_focus = t_filtered[t_filtered["case_id"] == focus].copy() if t_focus.empty: for name in [ "focal_compare_summary.csv", "focal_links_by_case.csv", "focal_links_by_case_docs.csv", "focal_links_by_case_instances.csv", "focal_by_case_matrix_quantity.csv", "focal_by_case_matrix_presence.csv", "focal_by_case_matrix_weight.csv", "focal_by_cluster_matrix_quantity.csv", "focal_by_cluster_matrix_presence.csv", "focal_by_cluster_matrix_weight.csv", ]: (outdir/name).write_text("", encoding="utf-8") else: focal_terms = ( t_focus.groupby(key_cols, as_index=False) .agg(in_case_quantity=("quantity","sum"), in_case_doc_count=("doc_sha256","nunique"), focal_filenames=("filename", list_join)) ) cinfo = t_focus[["cluster_id","cluster_name"]].dropna().head(1) if not cinfo.empty: focal_terms["cluster_id"] = cinfo.iloc[0]["cluster_id"] focal_terms["cluster_name"] = cinfo.iloc[0]["cluster_name"] focal_terms["case_id"] = focus t_others = t_filtered[t_filtered["case_id"] != focus].copy() others_case_counts = ( t_others.groupby(key_cols, as_index=False) .agg(others_case_count=("case_id","nunique")) ) others_cluster_counts = ( t_others.groupby(key_cols, as_index=False) .agg(others_cluster_count=("cluster_id","nunique")) ) focal_compare = focal_terms.merge(others_case_counts, on=key_cols, how="left") \ .merge(others_cluster_counts, on=key_cols, how="left") focal_compare["others_case_count"] = focal_compare["others_case_count"].fillna(0).astype(int) focal_compare["others_cluster_count"] = focal_compare["others_cluster_count"].fillna(0).astype(int) focal_compare = focal_compare[focal_compare["others_case_count"] >= 1].copy() focal_compare["weight_cases"] = focal_compare["others_case_count"].apply(lambda k: 1.0/k if k>=1 else 0.0) focal_compare["weight_clusters"] = focal_compare["others_cluster_count"].apply(lambda k: 1.0/k if k>=1 else 0.0) focus_terms_set = focal_compare[key_cols].drop_duplicates() links = t_others.merge(focus_terms_set, on=key_cols, how="inner") def agg_types(x): vals = [v for v in x if pd.notna(v)] return "; ".join(sorted(set(vals))) focal_links_by_case = ( links.groupby(key_cols + ["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", list_join), other_filing_types=("filing_type", agg_types), 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)) ) focal_links_by_case_docs = ( links.groupby(key_cols + ["doc_sha256","filename","filing_type","filing_date","case_id","cluster_id","cluster_name"], as_index=False) .agg(other_doc_quantity=("quantity","sum")) ) if not instances.empty: inst_all = instances.merge(docs[["doc_sha256","filename","filing_type","filing_date","case_id","cluster_id","cluster_name"]], on="doc_sha256", how="left") inst_all = inst_all.merge(focus_terms_set, on=key_cols, how="inner") focal_links_by_case_instances = inst_all[inst_all["case_id"] != focus].copy() focal_links_by_case_instances = focal_links_by_case_instances[[ "case_id","cluster_id","cluster_name","filename" ] + key_cols + ["page_num","num_pages","full_text_row","search_term_url"]].sort_values(["case_id","filename"] + key_cols + ["page_num"]) else: focal_links_by_case_instances = pd.DataFrame(columns=["case_id","cluster_id","cluster_name","filename"] + key_cols + ["page_num","num_pages","full_text_row","search_term_url"]) case_qty = focal_links_by_case.pivot_table( index=key_cols, columns="case_id", values="other_case_quantity", fill_value=0, aggfunc="sum" ).reset_index() case_presence = case_qty.copy() case_presence.iloc[:, len(key_cols):] = (case_presence.iloc[:, len(key_cols):] > 0).astype(int) term_weights = focal_compare.set_index(key_cols)["weight_cases"] core = case_presence.set_index(key_cols) weighted = core.copy().astype(float) for col in core.columns: weighted[col] = core[col].astype(float).where(core[col]==0, term_weights.reindex(core.index).fillna(0).astype(float)) case_weight = weighted.reset_index() cl_qty = focal_links_by_case.groupby(key_cols + ["cluster_id"], as_index=False)["other_case_quantity"].sum() cl_qty = cl_qty.pivot_table(index=key_cols, columns="cluster_id", values="other_case_quantity", fill_value=0, aggfunc="sum").reset_index() cl_presence = cl_qty.copy() cl_presence.iloc[:, len(key_cols):] = (cl_presence.iloc[:, len(key_cols):] > 0).astype(int) term_cluster_weights = focal_compare.set_index(key_cols)["weight_clusters"] corec = cl_presence.set_index(key_cols) weightedc = corec.copy().astype(float) for col in corec.columns: weightedc[col] = corec[col].astype(float).where(corec[col]==0, term_cluster_weights.reindex(corec.index).fillna(0).astype(float)) cl_weight = weightedc.reset_index() # write focal outputs focal_compare.to_csv(outdir/"focal_compare_summary.csv", index=False) focal_links_by_case.to_csv(outdir/"focal_links_by_case.csv", index=False) focal_links_by_case_docs.to_csv(outdir/"focal_links_by_case_docs.csv", index=False) focal_links_by_case_instances.to_csv(outdir/"focal_links_by_case_instances.csv", index=False) case_qty.to_csv(outdir/"focal_by_case_matrix_quantity.csv", index=False) case_presence.to_csv(outdir/"focal_by_case_matrix_presence.csv", index=False) case_weight.to_csv(outdir/"focal_by_case_matrix_weight.csv", index=False) cl_qty.to_csv(outdir/"focal_by_cluster_matrix_quantity.csv", index=False) cl_presence.to_csv(outdir/"focal_by_cluster_matrix_presence.csv", index=False) cl_weight.to_csv(outdir/"focal_by_cluster_matrix_weight.csv", index=False) elif focus_arg.upper() == "ALL": # Build vectors for all-case focal-style summary & matrices present_case = by_case[["case_id"] + key_cols].copy() term_case_counts = present_case.groupby(key_cols, as_index=False).agg(case_count=("case_id","nunique")) term_case_counts["others_case_count"] = term_case_counts["case_count"].apply(lambda k: max(int(k)-1, 0)) term_case_counts["weight_cases"] = term_case_counts["others_case_count"].apply(lambda k: (1.0/k) if k>=1 else 0.0) all_focals = present_case.merge(term_case_counts, on=key_cols, how="left") case_cluster = by_case.groupby(["case_id"], as_index=False).agg( focus_cluster_id=("cluster_id", lambda s: s.dropna().iloc[0] if len(s.dropna()) else pd.NA), focus_cluster_name=("cluster_name", lambda s: s.dropna().iloc[0] if len(s.dropna()) else pd.NA), ) all_focals = all_focals.merge(case_cluster, on="case_id", how="left") all_focals = all_focals.rename(columns={"case_id":"focus_case"}) all_focals = all_focals[[ "focus_case", "focus_cluster_id", "focus_cluster_name"] + key_cols + ["others_case_count","weight_cases"]] all_focals.to_csv(outdir/"all_focals_compare_summary.csv", index=False) present_cluster = by_cluster[["cluster_id"] + key_cols].copy() term_cluster_counts = present_cluster.groupby(key_cols, as_index=False).agg(cluster_count=("cluster_id","nunique")) term_cluster_counts["others_cluster_count"] = term_cluster_counts["cluster_count"].apply(lambda k: max(int(k)-1, 0)) term_cluster_counts["weight_clusters"] = term_cluster_counts["others_cluster_count"].apply(lambda k: (1.0/k) if k>=1 else 0.0) # Weight matrices (all cases / all clusters) case_presence = matrix_cases_presence.copy() core = case_presence.set_index(key_cols) weights_series = term_case_counts.set_index(key_cols)["weight_cases"].reindex(core.index).fillna(0).astype(float) weighted = core.copy().astype(float) for col in core.columns: weighted[col] = core[col].astype(float).where(core[col]==0, weights_series) weighted.reset_index().to_csv(outdir/"all_focals_weight_matrix_cases.csv", index=False) cl_presence = matrix_clusters_presence.copy() corec = cl_presence.set_index(key_cols) weightsc = term_cluster_counts.set_index(key_cols)["weight_clusters"].reindex(corec.index).fillna(0).astype(float) weightedc = corec.copy().astype(float) for col in corec.columns: weightedc[col] = corec[col].astype(float).where(corec[col]==0, weightsc) weightedc.reset_index().to_csv(outdir/"all_focals_weight_matrix_clusters.csv", index=False) # README readme = f""" MCRO Language Report from Term List (v2) ======================================= Filtered to terms from: {args.terms_csv} • Global outputs ALWAYS include ALL cases (so any focal is included). • --focus_case : adds single-case focal comparisons (rarity & links) • --focus_case ALL : computes focal-style comparisons for EVERY case. All outputs are restricted strictly to your term list. """.strip() + "\\n" (outdir/"README.txt").write_text(readme, encoding="utf-8") print(f"[ok] Wrote filtered language reports to: {outdir.resolve()}") if __name__ == "__main__": main()