osint-investigation

1#!/usr/bin/env python32"""Fetch SEC EDGAR filings index for a given CIK or company name.3 4SEC requires a User-Agent header with contact info. Set SEC_USER_AGENT,5e.g. SEC_USER_AGENT="Research example@example.com".6 7Filings JSON is published at:8    https://data.sec.gov/submissions/CIK<10-digit-padded>.json9 10Company lookup uses:11    https://www.sec.gov/cgi-bin/browse-edgar?action=getcompany&company=<name>&output=atom12"""13from __future__ import annotations14 15import argparse16import csv17import os18import re19import sys20from pathlib import Path21 22sys.path.insert(0, str(Path(__file__).parent))23from _http import get, get_json  # noqa: E40224 25SUBMISSIONS_URL = "https://data.sec.gov/submissions/CIK{cik}.json"26COLUMNS = [27    "cik",28    "company_name",29    "form_type",30    "filing_date",31    "accession_number",32    "primary_document",33    "filing_url",34    "reporting_period",35]36 37 38def _ua() -> str:39    ua = os.environ.get("SEC_USER_AGENT", "").strip()40    if not ua:41        raise SystemExit(42            "SEC requires a User-Agent with contact info. "43            "Set SEC_USER_AGENT='Your Name your@email'."44        )45    return ua46 47 48def _resolve_cik(company: str) -> tuple[str, str]:49    """Resolve a company name to a CIK via EDGAR's atom feed.50 51    Returns (cik, resolved_company_name). The feed entries also reveal whether52    the match is an individual filer (Form 3/4/5 only) — surfaced in the53    return value so callers can warn.54    """55    url = "https://www.sec.gov/cgi-bin/browse-edgar"56    params = {"action": "getcompany", "company": company, "output": "atom", "owner": "include"}57    body = get(url, params=params, user_agent=_ua()).decode("utf-8", errors="replace")58    m = re.search(r"CIK=(\d{10})", body)59    if not m:60        raise SystemExit(f"Could not resolve CIK for company={company!r}")61    cik = m.group(1)62    name_m = re.search(r"<title>([^<]+)\s*\((\d{10})\)</title>", body)63    resolved = name_m.group(1).strip() if name_m else ""64    return cik, resolved65 66 67def fetch(68    cik: str | None,69    company: str | None,70    types: list[str],71    since: str | None,72    out_path: str,73) -> int:74    resolved_name = ""75    if not cik and company:76        try:77            cik, resolved_name = _resolve_cik(company)  # type: ignore[assignment]78        except SystemExit as e:79            # Write empty CSV with header so downstream tools still work,80            # and tell the user clearly.81            print(f"SEC EDGAR: {e}", file=sys.stderr)82            Path(out_path).parent.mkdir(parents=True, exist_ok=True)83            with open(out_path, "w", newline="", encoding="utf-8") as fh:84                csv.DictWriter(fh, fieldnames=COLUMNS).writeheader()85            return 086        if resolved_name:87            print(88                f"Resolved company={company!r} → CIK {cik} ({resolved_name})",89                file=sys.stderr,90            )91    if not cik:92        raise SystemExit("must supply --cik or --company")93    cik = cik.zfill(10)94    url = SUBMISSIONS_URL.format(cik=cik)95    payload = get_json(url, user_agent=_ua())96    if not isinstance(payload, dict):97        raise SystemExit(f"Unexpected EDGAR response shape for CIK {cik}")98    name = payload.get("name", "")99    recent = (payload.get("filings", {}) or {}).get("recent", {}) or {}100    form = recent.get("form", [])101    date = recent.get("filingDate", [])102    accession = recent.get("accessionNumber", [])103    primary_doc = recent.get("primaryDocument", [])104    period = recent.get("reportDate", [])105 106    # Histogram of available filing types — useful for surfacing why a filter107    # returned 0 (e.g. user asked for 10-K on an individual Form 4 filer).108    type_hist: dict[str, int] = {}109    for ftype in form:110        type_hist[ftype] = type_hist.get(ftype, 0) + 1111 112    type_set = {t.strip().upper() for t in types} if types else None113    rows: list[dict[str, str]] = []114    for i, ftype in enumerate(form):115        if type_set and ftype.upper() not in type_set:116            continue117        fdate = date[i] if i < len(date) else ""118        if since and fdate and fdate < since:119            continue120        acc = accession[i] if i < len(accession) else ""121        pdoc = primary_doc[i] if i < len(primary_doc) else ""122        acc_nodash = acc.replace("-", "")123        filing_url = (124            f"https://www.sec.gov/Archives/edgar/data/{int(cik)}/{acc_nodash}/{pdoc}"125            if acc and pdoc126            else ""127        )128        rows.append(129            {130                "cik": cik,131                "company_name": name,132                "form_type": ftype,133                "filing_date": fdate,134                "accession_number": acc,135                "primary_document": pdoc,136                "filing_url": filing_url,137                "reporting_period": period[i] if i < len(period) else "",138            }139        )140 141    Path(out_path).parent.mkdir(parents=True, exist_ok=True)142    with open(out_path, "w", newline="", encoding="utf-8") as fh:143        w = csv.DictWriter(fh, fieldnames=COLUMNS)144        w.writeheader()145        w.writerows(rows)146 147    if not rows and type_hist:148        top = sorted(type_hist.items(), key=lambda kv: -kv[1])[:8]149        hist_str = ", ".join(f"{t}={n}" for t, n in top)150        print(151            f"Warning: SEC EDGAR CIK {cik} ({name}) has {sum(type_hist.values())} "152            f"recent filings but NONE match types={types}. "153            f"Available form types: {hist_str}.",154            file=sys.stderr,155        )156        # Insider-filer heuristic: only Form 3/4/5 → individual person, not a company.157        company_types = {"10-K", "10-Q", "8-K", "20-F", "DEF 14A", "S-1"}158        if not (set(type_hist.keys()) & company_types):159            print(160                f"Note: CIK {cik} appears to be an INDIVIDUAL filer "161                f"(insider Form 3/4/5 only), not a corporate registrant. "162                f"The resolver may have matched an officer/director named "163                f"{company!r} rather than a company.",164                file=sys.stderr,165            )166    return len(rows)167 168 169def main() -> int:170    p = argparse.ArgumentParser(description=__doc__)171    p.add_argument("--cik", help="Central Index Key (will be 10-digit zero-padded)")172    p.add_argument("--company", help="Resolve to CIK by company name")173    p.add_argument("--types", default="", help="Comma-separated form types (e.g. 10-K,10-Q,8-K)")174    p.add_argument("--since", help="Skip filings before YYYY-MM-DD")175    p.add_argument("--out", required=True)176    a = p.parse_args()177    types = [t for t in (a.types or "").split(",") if t.strip()]178    n = fetch(cik=a.cik, company=a.company, types=types, since=a.since, out_path=a.out)179    print(f"Wrote {n} EDGAR filing rows to {a.out}")180    return 0181 182 183if __name__ == "__main__":184    raise SystemExit(main())185