#!/usr/bin/env python3 """ MCRO Font Tracker Report ======================== Analyze font-like object types (".cff", ".cid", ".otf", ".pfa", ".ttf") in MCRO JSON files to uncover tracking patterns: - Within a cluster: font object_sha256 combos cloned across many docs/cases - Across clusters: whether those combos appear in multiple clusters or are unique to a single cluster (per-cluster tracker fonts). ASSUMED JSON STRUCTURE ---------------------- Top-level keys (all optional, script is robust to missing ones): doc_sha256 (or sha256) filename filing_type filing_date case.case_id case.cluster_id case.cluster_name objects -> list of dicts, each like: { "object_type": ".ttf", # we filter: .cff, .cid, .otf, .pfa, .ttf "object_sha256": "...", # or "sha256" "object_font_code": "...", # optional, e.g. "UNIZJF" "object_font_name": "...", # optional, e.g. "UNIZJF+ArialMT" ... } We ignore non-font object_types. OUTPUTS ------- All CSVs are written to --outdir: 1) font_cluster_hash_detail.csv One row per (cluster, font combo, document), but ONLY for combos that appear in >= 2 distinct docs within that cluster. Combo key (cluster-level): cluster_id, cluster_name, object_type, object_sha256, object_font_code, object_font_name Columns: cluster_id cluster_name object_type object_sha256 object_font_code object_font_name combo_n_docs # how many distinct filenames in this cluster combo_n_cases # how many cases in this cluster combo_n_instances # total objects of this combo in this cluster combo_first_filing_date combo_last_filing_date global_n_docs # docs across ALL clusters global_n_cases # cases across ALL clusters global_n_clusters # clusters across ALL clusters global_cluster_ids # list of clusters (ids) where combo appears global_cluster_names # list of cluster_names where combo appears is_cluster_unique # True if global_n_clusters == 1 case_id filename filing_type filing_date doc_n_instances # how many objects of this combo in this doc doc_sha256 => This is your "visual clones" view. 2) font_cluster_combo_summary.csv One row per (cluster + font combo), again only combos with combo_n_docs >= 2. Columns: cluster_id cluster_name object_type object_sha256 object_font_code object_font_name combo_n_docs combo_n_cases combo_n_instances combo_first_filing_date combo_last_filing_date global_n_docs global_n_cases global_n_clusters global_cluster_ids global_cluster_names is_cluster_unique USAGE ===== python3 mcro_font_tracker_report.py \ --input /path/to/json/documents \ --outdir /path/to/font_tracker_report \ [--pattern "*.json"] \ [--recurse] """ 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 join_unique(vals): """Join unique, non-null values with '; '.""" vals = [v for v in vals if pd.notna(v)] if not vals: return "" return "; ".join(sorted(set(map(str, vals)))) def main(): ap = argparse.ArgumentParser(description="MCRO font tracker report (font-like object types).") ap.add_argument("--input", required=True, help="Directory containing MCRO 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 under --input.") 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) # Font-like object types we care about (lowercase) tracked_types = {".cff", ".cid", ".otf", ".pfa", ".ttf"} obj_rows: List[Dict[str, Any]] = [] # ----------------------------------------------------- # Collect object-level rows for font-like object types # ----------------------------------------------------- for fp in files: try: text = fp.read_text(encoding="utf-8") data = json.loads(text) except Exception as e: print(f"[warn] Failed to read/parse JSON: {fp} ({e})", file=sys.stderr) continue if not isinstance(data, dict): print(f"[warn] Skipping {fp}: JSON root is not an object", file=sys.stderr) 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", "") objects = data.get("objects") or [] if not isinstance(objects, list): continue for obj in objects: if not isinstance(obj, dict): continue otype_raw = obj.get("object_type", "") otype = str(otype_raw).strip().lower() if otype not in tracked_types: continue osha = obj.get("object_sha256") or obj.get("sha256") or "" if not osha: # without a hash, it can't function as a tracker continue font_code = obj.get("object_font_code", "") font_name = obj.get("object_font_name", "") obj_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, "object_type": otype, "object_sha256": osha, "object_font_code": font_code, "object_font_name": font_name, } ) if not obj_rows: print("[info] No font-like objects found in any JSON files.") pd.DataFrame().to_csv(outdir / "font_cluster_hash_detail.csv", index=False) pd.DataFrame().to_csv(outdir / "font_cluster_combo_summary.csv", index=False) print(f"[ok] Wrote empty font tracker CSVs to {outdir.resolve()}") sys.exit(0) objs_df = pd.DataFrame(obj_rows) # Normalize fields for col in [ "cluster_id", "cluster_name", "case_id", "filing_date", "object_type", "object_sha256", "object_font_code", "object_font_name", "filename", ]: objs_df[col] = objs_df[col].fillna("") # ----------------------------------------------------- # Subset: only rows with non-empty cluster_id for cluster analysis # ----------------------------------------------------- objs_cluster_df = objs_df[objs_df["cluster_id"].astype(str).str.strip() != ""].copy() if objs_cluster_df.empty: print("[info] No rows with cluster_id found; cluster-level report will be empty.") pd.DataFrame().to_csv(outdir / "font_cluster_hash_detail.csv", index=False) pd.DataFrame().to_csv(outdir / "font_cluster_combo_summary.csv", index=False) sys.exit(0) # ----------------------------------------------------- # 1) Cluster-level combo stats # ----------------------------------------------------- combo_cols_cluster = [ "cluster_id", "cluster_name", "object_type", "object_sha256", "object_font_code", "object_font_name", ] cluster_combo_stats = ( objs_cluster_df.groupby(combo_cols_cluster, as_index=False) .agg( combo_n_instances=("object_sha256", "size"), combo_n_docs=("filename", lambda s: int(pd.Series(s).dropna().nunique())), combo_n_cases=("case_id", lambda s: int(pd.Series(s).dropna().nunique())), combo_first_filing_date=("filing_date", lambda s: s.dropna().min() if len(s.dropna()) else ""), combo_last_filing_date=("filing_date", lambda s: s.dropna().max() if len(s.dropna()) else ""), ) ) # ----------------------------------------------------- # 2) Global combo stats across ALL clusters # ----------------------------------------------------- combo_cols_global = [ "object_type", "object_sha256", "object_font_code", "object_font_name", ] global_combo_stats = ( objs_df.groupby(combo_cols_global, as_index=False) .agg( global_n_instances=("object_sha256", "size"), global_n_docs=("filename", lambda s: int(pd.Series(s).dropna().nunique())), global_n_cases=("case_id", lambda s: int(pd.Series(s).dropna().nunique())), global_n_clusters=("cluster_id", lambda s: int(pd.Series(s).dropna().nunique())), global_cluster_ids=("cluster_id", join_unique), global_cluster_names=("cluster_name", join_unique), ) ) # ----------------------------------------------------- # Join global stats onto cluster-level combo stats # ----------------------------------------------------- cluster_combo_stats = cluster_combo_stats.merge( global_combo_stats, on=combo_cols_global, how="left", ) # Determine if a combo is unique to a single cluster cluster_combo_stats["is_cluster_unique"] = ( cluster_combo_stats["global_n_clusters"].fillna(0).astype(int) == 1 ) # ----------------------------------------------------- # Filter to combos that appear in >= 2 docs in that cluster # ----------------------------------------------------- cluster_combo_dup = cluster_combo_stats[cluster_combo_stats["combo_n_docs"] >= 2].copy() # If nothing qualifies, write empty shells and bail if cluster_combo_dup.empty: print("[info] No font combos appear in >= 2 docs within any cluster.") pd.DataFrame().to_csv(outdir / "font_cluster_hash_detail.csv", index=False) cluster_combo_stats.to_csv(outdir / "font_cluster_combo_summary.csv", index=False) print(f"[ok] Wrote empty font_cluster_hash_detail.csv and full font_cluster_combo_summary.csv to {outdir.resolve()}") sys.exit(0) # ----------------------------------------------------- # 3) font_cluster_combo_summary.csv # ----------------------------------------------------- combo_summary_cols = [ "cluster_id", "cluster_name", "object_type", "object_sha256", "object_font_code", "object_font_name", "combo_n_docs", "combo_n_cases", "combo_n_instances", "combo_first_filing_date", "combo_last_filing_date", "global_n_docs", "global_n_cases", "global_n_clusters", "global_cluster_ids", "global_cluster_names", "is_cluster_unique", ] combo_summary = cluster_combo_dup[combo_summary_cols].sort_values( by=[ "cluster_id", "object_type", "object_sha256", "object_font_code", "object_font_name", ], kind="mergesort", ) combo_summary.to_csv(outdir / "font_cluster_combo_summary.csv", index=False) # ----------------------------------------------------- # 4) font_cluster_hash_detail.csv (per doc rows for these combos) # ----------------------------------------------------- objs_for_combos = objs_cluster_df.merge( cluster_combo_dup[combo_cols_cluster], on=combo_cols_cluster, how="inner", ) group_cols_doc = combo_cols_cluster + [ "case_id", "filename", "filing_type", "filing_date", "doc_sha256", ] doc_detail = ( objs_for_combos.groupby(group_cols_doc, as_index=False) .agg( doc_n_instances=("object_sha256", "size"), ) ) # Merge the cluster combo stats (which now include global stats + uniqueness) doc_detail = doc_detail.merge( cluster_combo_dup, on=combo_cols_cluster, how="left", ) # Ensure integer types where appropriate for col in [ "combo_n_instances", "combo_n_docs", "combo_n_cases", "global_n_instances", "global_n_docs", "global_n_cases", "global_n_clusters", "doc_n_instances", ]: if col in doc_detail.columns: doc_detail[col] = doc_detail[col].fillna(0).astype(int) # Final column order for detail detail_cols = [ "cluster_id", "cluster_name", "object_type", "object_sha256", "object_font_code", "object_font_name", "combo_n_docs", "combo_n_cases", "combo_n_instances", "combo_first_filing_date", "combo_last_filing_date", "global_n_docs", "global_n_cases", "global_n_clusters", "global_cluster_ids", "global_cluster_names", "is_cluster_unique", "case_id", "filename", "filing_type", "filing_date", "doc_n_instances", "doc_sha256", ] for c in detail_cols: if c not in doc_detail.columns: doc_detail[c] = pd.NA doc_detail = doc_detail[detail_cols].sort_values( by=[ "cluster_id", "object_type", "object_sha256", "object_font_code", "object_font_name", "case_id", "filename", ], kind="mergesort", ) doc_detail.to_csv(outdir / "font_cluster_hash_detail.csv", index=False) print(f"[ok] Font tracker report written to: {outdir.resolve()}") print(f" font_cluster_combo_summary.csv : {len(combo_summary)} combos (>=2 docs per cluster)") print(f" font_cluster_hash_detail.csv : {len(doc_detail)} rows (per-doc detail)") if __name__ == "__main__": main()