osint-investigation

1#!/usr/bin/env python32"""Cross-source entity resolution (stdlib-only).3 4Given two CSV files with name columns, find candidate matches using three5tiers of normalization:6 7  1. exact          — normalized strings equal8  2. fuzzy          — sorted-token (word-bag) match9  3. token_overlap  — >=60% Jaccard overlap on >=4-char tokens, >=2 shared10 11Adapted from ShinMegamiBoson/OpenPlanter (MIT) but generalized: no Boston-12specific record types, no contribution-code filters, no fixed schemas.13 14Output CSV columns:15    match_type, confidence, left_name, right_name,16    left_normalized, right_normalized, left_row, right_row,17    overlap_ratio, shared_tokens18"""19from __future__ import annotations20 21import argparse22import csv23import sys24from pathlib import Path25 26# Allow running directly or as a module.27sys.path.insert(0, str(Path(__file__).parent))28from _normalize import (  # noqa: E40229    normalize_name,30    normalize_aggressive,31    token_overlap_ratio,32)33 34CONFIDENCE = {35    "exact": "high",36    "fuzzy": "medium",37    "token_overlap": "low",38}39 40 41def _read_csv(path: str, name_col: str) -> list[dict[str, str]]:42    rows = []43    with open(path, newline="", encoding="utf-8") as fh:44        reader = csv.DictReader(fh)45        if name_col not in (reader.fieldnames or []):46            raise SystemExit(47                f"Column {name_col!r} not in {path}. "48                f"Available: {reader.fieldnames}"49            )50        for i, row in enumerate(reader):51            row["__row__"] = str(i)52            rows.append(row)53    return rows54 55 56def _build_index(rows: list[dict[str, str]], name_col: str):57    """Index by exact-normalized and aggressive (sorted-token) form."""58    exact: dict[str, list[dict[str, str]]] = {}59    aggressive: dict[str, list[dict[str, str]]] = {}60    for row in rows:61        raw = row.get(name_col, "")62        n = normalize_name(raw)63        if n:64            exact.setdefault(n, []).append(row)65        a = normalize_aggressive(raw)66        if a:67            aggressive.setdefault(a, []).append(row)68    return exact, aggressive69 70 71def _emit(72    out_rows: list[dict[str, str]],73    seen: set[tuple],74    match_type: str,75    left_row: dict[str, str],76    right_row: dict[str, str],77    left_col: str,78    right_col: str,79    ratio: float = 0.0,80    shared: int = 0,81):82    left_raw = left_row.get(left_col, "")83    right_raw = right_row.get(right_col, "")84    key = (85        left_row["__row__"],86        right_row["__row__"],87        match_type,88    )89    if key in seen:90        return91    seen.add(key)92    out_rows.append(93        {94            "match_type": match_type,95            "confidence": CONFIDENCE[match_type],96            "left_name": left_raw,97            "right_name": right_raw,98            "left_normalized": normalize_name(left_raw),99            "right_normalized": normalize_name(right_raw),100            "left_row": left_row["__row__"],101            "right_row": right_row["__row__"],102            "overlap_ratio": f"{ratio:.3f}" if ratio else "",103            "shared_tokens": str(shared) if shared else "",104        }105    )106 107 108def resolve(109    left_path: str,110    left_col: str,111    right_path: str,112    right_col: str,113    out_path: str,114    overlap_threshold: float = 0.60,115    min_shared: int = 2,116    skip_overlap: bool = False,117) -> int:118    left_rows = _read_csv(left_path, left_col)119    right_rows = _read_csv(right_path, right_col)120 121    right_exact, right_aggressive = _build_index(right_rows, right_col)122 123    out_rows: list[dict[str, str]] = []124    seen: set[tuple] = set()125 126    # Pass 1+2: exact / fuzzy via index lookup.127    for lrow in left_rows:128        raw = lrow.get(left_col, "")129        n = normalize_name(raw)130        if not n:131            continue132        for rrow in right_exact.get(n, []):133            _emit(out_rows, seen, "exact", lrow, rrow, left_col, right_col)134        a = normalize_aggressive(raw)135        if a:136            for rrow in right_aggressive.get(a, []):137                _emit(out_rows, seen, "fuzzy", lrow, rrow, left_col, right_col)138 139    if not skip_overlap:140        # Pass 3: token overlap (O(N*M) — expensive; allow opt-out).141        for lrow in left_rows:142            l_raw = lrow.get(left_col, "")143            if not normalize_name(l_raw):144                continue145            for rrow in right_rows:146                ratio, shared = token_overlap_ratio(147                    l_raw, rrow.get(right_col, "")148                )149                if ratio >= overlap_threshold and shared >= min_shared:150                    _emit(151                        out_rows,152                        seen,153                        "token_overlap",154                        lrow,155                        rrow,156                        left_col,157                        right_col,158                        ratio=ratio,159                        shared=shared,160                    )161 162    fieldnames = [163        "match_type",164        "confidence",165        "left_name",166        "right_name",167        "left_normalized",168        "right_normalized",169        "left_row",170        "right_row",171        "overlap_ratio",172        "shared_tokens",173    ]174    with open(out_path, "w", newline="", encoding="utf-8") as fh:175        writer = csv.DictWriter(fh, fieldnames=fieldnames)176        writer.writeheader()177        writer.writerows(out_rows)178    return len(out_rows)179 180 181def main() -> int:182    p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)183    p.add_argument("--left", required=True, help="Left CSV path")184    p.add_argument(185        "--left-name-col", required=True, help="Name column in left CSV"186    )187    p.add_argument("--right", required=True, help="Right CSV path")188    p.add_argument(189        "--right-name-col",190        required=True,191        help="Name column in right CSV",192    )193    p.add_argument("--out", required=True, help="Output CSV path")194    p.add_argument(195        "--overlap-threshold",196        type=float,197        default=0.60,198        help="Jaccard overlap threshold for token_overlap tier (default 0.60)",199    )200    p.add_argument(201        "--min-shared",202        type=int,203        default=2,204        help="Minimum shared tokens for token_overlap tier (default 2)",205    )206    p.add_argument(207        "--skip-overlap",208        action="store_true",209        help="Skip the O(N*M) token_overlap pass (much faster on large CSVs)",210    )211    args = p.parse_args()212 213    count = resolve(214        left_path=args.left,215        left_col=args.left_name_col,216        right_path=args.right,217        right_col=args.right_name_col,218        out_path=args.out,219        overlap_threshold=args.overlap_threshold,220        min_shared=args.min_shared,221        skip_overlap=args.skip_overlap,222    )223    print(f"Wrote {count} match rows to {args.out}")224    return 0225 226 227if __name__ == "__main__":228    raise SystemExit(main())229