osint-investigation

1#!/usr/bin/env python32"""Search NYC property records via ACRIS (Automated City Register Information System).3 4Uses the city's Socrata-backed open data API. No auth required for read access.5 6Datasets:7  bnx9-e6tj — Real Property Master (one row per recorded document)8  636b-3b5g — Real Property Parties (names — grantor, grantee, etc.)9  8h5j-fqxa — Real Property Legal (lot / property identifiers)10  uqqa-hym2 — Real Property References11 12The Parties dataset has the names. We search by name and optionally join to13Master to get the doc type and date.14"""15from __future__ import annotations16 17import argparse18import csv19import sys20import urllib.parse21from pathlib import Path22 23sys.path.insert(0, str(Path(__file__).parent))24from _http import get_json  # noqa: E40225 26PARTIES_URL = "https://data.cityofnewyork.us/resource/636b-3b5g.json"27MASTER_URL = "https://data.cityofnewyork.us/resource/bnx9-e6tj.json"28 29PARTY_TYPE = {30    "1": "grantor (seller / mortgagor / debtor)",31    "2": "grantee (buyer / mortgagee / creditor)",32    "3": "other party",33}34 35BOROUGH = {36    "1": "Manhattan",37    "2": "Bronx",38    "3": "Brooklyn",39    "4": "Queens",40    "5": "Staten Island",41}42 43COLUMNS = [44    "document_id",45    "name",46    "party_type",47    "party_role",48    "address_1",49    "address_2",50    "city",51    "state",52    "zip",53    "country",54    "doc_type",55    "doc_date",56    "recorded_date",57    "borough",58    "amount",59    "filing_url",60]61 62 63def _filing_url(document_id: str) -> str:64    if not document_id:65        return ""66    return (67        f"https://a836-acris.nyc.gov/DS/DocumentSearch/DocumentImageView?doc_id={document_id}"68    )69 70 71def fetch(72    name: str | None,73    address: str | None,74    party_type: str | None,75    limit: int,76    out_path: str,77    enrich: bool = True,78) -> int:79    if not (name or address):80        raise SystemExit("must supply --name or --address")81 82    where_clauses: list[str] = []83    if name:84        safe = name.upper().replace("'", "''")85        where_clauses.append(f"upper(name) like '%{safe}%'")86    if address:87        safe_addr = address.upper().replace("'", "''")88        where_clauses.append(f"upper(address_1) like '%{safe_addr}%'")89    if party_type and party_type in {"1", "2", "3"}:90        where_clauses.append(f"party_type='{party_type}'")91 92    params = {93        "$where": " AND ".join(where_clauses),94        "$limit": str(limit),95    }96    url = f"{PARTIES_URL}?{urllib.parse.urlencode(params)}"97    parties = get_json(url)98    if not isinstance(parties, list):99        raise SystemExit(f"Unexpected ACRIS response: {parties!r}")100 101    # Enrich with master record (doc_type, dates, borough, amount).102    doc_ids: list[str] = sorted({103        d for d in (p.get("document_id") for p in parties) if d104    })105    masters: dict[str, dict] = {}106    if enrich and doc_ids:107        # Batch up to 100 doc_ids per request (Socrata IN-list is fine for this).108        for i in range(0, len(doc_ids), 100):109            chunk = doc_ids[i : i + 100]110            id_list = ",".join(f"'{d}'" for d in chunk)111            master_params = {112                "$where": f"document_id in ({id_list})",113                "$limit": "100",114            }115            url = f"{MASTER_URL}?{urllib.parse.urlencode(master_params)}"116            try:117                rows = get_json(url)118            except Exception as e:  # noqa: BLE001119                print(f"ACRIS master lookup failed for chunk: {e}", file=sys.stderr)120                continue121            if isinstance(rows, list):122                for r in rows:123                    did = r.get("document_id", "")124                    if did:125                        masters[did] = r126 127    out_rows: list[dict[str, str]] = []128    for p in parties:129        did = p.get("document_id", "") or ""130        m = masters.get(did, {})131        out_rows.append(132            {133                "document_id": did,134                "name": p.get("name", "") or "",135                "party_type": p.get("party_type", "") or "",136                "party_role": PARTY_TYPE.get(p.get("party_type", ""), ""),137                "address_1": p.get("address_1", "") or "",138                "address_2": p.get("address_2", "") or "",139                "city": p.get("city", "") or "",140                "state": p.get("state", "") or "",141                "zip": p.get("zip", "") or "",142                "country": p.get("country", "") or "",143                "doc_type": m.get("doc_type", "") or "",144                "doc_date": (m.get("document_date", "") or "")[:10],145                "recorded_date": (m.get("recorded_datetime", "") or "")[:10],146                "borough": BOROUGH.get(m.get("recorded_borough", ""), m.get("recorded_borough", "")),147                "amount": m.get("document_amt", "") or "",148                "filing_url": _filing_url(did),149            }150        )151 152    Path(out_path).parent.mkdir(parents=True, exist_ok=True)153    with open(out_path, "w", newline="", encoding="utf-8") as fh:154        w = csv.DictWriter(fh, fieldnames=COLUMNS)155        w.writeheader()156        w.writerows(out_rows)157 158    if not out_rows:159        filters = []160        if name:161            filters.append(f"name={name!r}")162        if address:163            filters.append(f"address={address!r}")164        print(165            f"NYC ACRIS: 0 records for {', '.join(filters)}. "166            "ACRIS covers ONLY NYC (5 boroughs). For property records elsewhere, "167            "search the relevant county recorder directly.",168            file=sys.stderr,169        )170    return len(out_rows)171 172 173def main() -> int:174    p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)175    p.add_argument("--name", help="Party name substring (case-insensitive)")176    p.add_argument("--address", help="Address line 1 substring")177    p.add_argument(178        "--party-type",179        choices=["1", "2", "3"],180        help="Filter party type: 1=grantor (seller/mortgagor), 2=grantee (buyer/mortgagee), 3=other",181    )182    p.add_argument("--limit", type=int, default=200)183    p.add_argument(184        "--no-enrich",185        action="store_true",186        help="Skip the master-document lookup that adds doc_type/date/amount",187    )188    p.add_argument("--out", required=True)189    a = p.parse_args()190    n = fetch(191        name=a.name,192        address=a.address,193        party_type=a.party_type,194        limit=a.limit,195        out_path=a.out,196        enrich=not a.no_enrich,197    )198    print(f"Wrote {n} NYC ACRIS rows to {a.out}")199    return 0200 201 202if __name__ == "__main__":203    raise SystemExit(main())204