osint-investigation

1#!/usr/bin/env python32"""Search OpenCorporates company registry data.3 4OpenCorporates aggregates ~200M companies from 130+ jurisdictions. The5public API requires an API token (free tier: 500 calls/month). Set6OPENCORPORATES_API_TOKEN in env or pass --token.7 8Without a token, this script falls back to scraping the public HTML9search page (limited fields, more brittle, no jurisdiction filter).10"""11from __future__ import annotations12 13import argparse14import csv15import os16import re17import sys18import urllib.parse19from pathlib import Path20 21sys.path.insert(0, str(Path(__file__).parent))22from _http import get, get_json  # noqa: E40223 24API_URL = "https://api.opencorporates.com/v0.4/companies/search"25HTML_URL = "https://opencorporates.com/companies"26 27COLUMNS = [28    "name",29    "company_number",30    "jurisdiction_code",31    "jurisdiction_name",32    "incorporation_date",33    "dissolution_date",34    "company_type",35    "status",36    "registered_address",37    "opencorporates_url",38    "officers_count",39    "source",40]41 42 43def _via_api(query: str, jurisdiction: str | None, token: str, limit: int) -> list[dict]:44    params = {45        "q": query,46        "api_token": token,47        "per_page": str(min(limit, 100)),48    }49    if jurisdiction:50        params["jurisdiction_code"] = jurisdiction51    url = f"{API_URL}?{urllib.parse.urlencode(params)}"52    payload = get_json(url)53    if not isinstance(payload, dict):54        return []55    results = payload.get("results", {}).get("companies", []) or []56    return [r.get("company", {}) for r in results if isinstance(r, dict)]57 58 59def _via_html(query: str, limit: int) -> list[dict]:60    """Best-effort HTML fallback when no API token is available."""61    params = {"q": query, "utf8": "✓"}62    url = f"{HTML_URL}?{urllib.parse.urlencode(params)}"63    body = get(url, user_agent="Mozilla/5.0 hermes-osint").decode("utf-8", errors="replace")64    # Each result is in <li class="company"> ... </li> with name, url, status65    pattern = re.compile(66        r'<li[^>]*class="[^"]*company[^"]*"[^>]*>.*?'67        r'<a[^>]+href="(?P<url>/companies/[^"]+)"[^>]*>(?P<name>[^<]+)</a>'68        r'(?:.*?<span[^>]*class="[^"]*jurisdiction[^"]*"[^>]*>(?P<jur>[^<]+)</span>)?'69        r"(?:.*?<dt[^>]*>(?:Company\s+Number|Number)</dt>\s*<dd[^>]*>(?P<num>[^<]+)</dd>)?",70        re.DOTALL | re.IGNORECASE,71    )72    out = []73    for m in pattern.finditer(body):74        if len(out) >= limit:75            break76        url_path = m.group("url").strip()77        out.append(78            {79                "name": (m.group("name") or "").strip(),80                "opencorporates_url": f"https://opencorporates.com{url_path}",81                "jurisdiction_code": (m.group("jur") or "").strip(),82                "company_number": (m.group("num") or "").strip(),83                "_via": "html",84            }85        )86    return out87 88 89def fetch(90    query: str,91    jurisdiction: str | None,92    token: str | None,93    limit: int,94    out_path: str,95) -> int:96    if token:97        try:98            companies = _via_api(query, jurisdiction, token, limit)99            source_tag = "api"100        except Exception as e:  # noqa: BLE001101            print(102                f"OpenCorporates API call failed ({e}); falling back to HTML.",103                file=sys.stderr,104            )105            companies = _via_html(query, limit)106            source_tag = "html-fallback"107    else:108        print(109            "OPENCORPORATES_API_TOKEN not set — using HTML fallback (limited fields). "110            "Get a free token at https://opencorporates.com/api_accounts/new",111            file=sys.stderr,112        )113        companies = _via_html(query, limit)114        source_tag = "html"115 116    rows: list[dict[str, str]] = []117    for c in companies[:limit]:118        if c.get("_via") == "html":119            rows.append(120                {121                    "name": c.get("name", ""),122                    "company_number": c.get("company_number", ""),123                    "jurisdiction_code": c.get("jurisdiction_code", ""),124                    "jurisdiction_name": "",125                    "incorporation_date": "",126                    "dissolution_date": "",127                    "company_type": "",128                    "status": "",129                    "registered_address": "",130                    "opencorporates_url": c.get("opencorporates_url", ""),131                    "officers_count": "",132                    "source": source_tag,133                }134            )135            continue136        addr = c.get("registered_address_in_full") or ""137        rows.append(138            {139                "name": c.get("name", "") or "",140                "company_number": c.get("company_number", "") or "",141                "jurisdiction_code": c.get("jurisdiction_code", "") or "",142                "jurisdiction_name": "",143                "incorporation_date": c.get("incorporation_date", "") or "",144                "dissolution_date": c.get("dissolution_date", "") or "",145                "company_type": c.get("company_type", "") or "",146                "status": c.get("current_status", "") or c.get("inactive", "") or "",147                "registered_address": addr,148                "opencorporates_url": c.get("opencorporates_url", "") or "",149                "officers_count": str(c.get("officers", {}).get("total_count", "") if c.get("officers") else ""),150                "source": source_tag,151            }152        )153 154    Path(out_path).parent.mkdir(parents=True, exist_ok=True)155    with open(out_path, "w", newline="", encoding="utf-8") as fh:156        w = csv.DictWriter(fh, fieldnames=COLUMNS)157        w.writeheader()158        w.writerows(rows)159    if not rows:160        print(161            f"OpenCorporates: 0 matches for query={query!r}"162            f"{f' jurisdiction={jurisdiction!r}' if jurisdiction else ''}.",163            file=sys.stderr,164        )165    return len(rows)166 167 168def main() -> int:169    p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)170    p.add_argument("--query", required=True, help="Company name search")171    p.add_argument(172        "--jurisdiction",173        help="Jurisdiction code, e.g. 'us_ny', 'us_de', 'gb', 'sg' (lowercased OpenCorporates style)",174    )175    p.add_argument("--limit", type=int, default=50)176    p.add_argument("--token", default=os.environ.get("OPENCORPORATES_API_TOKEN"))177    p.add_argument("--out", required=True)178    a = p.parse_args()179    n = fetch(180        query=a.query,181        jurisdiction=a.jurisdiction,182        token=a.token,183        limit=a.limit,184        out_path=a.out,185    )186    print(f"Wrote {n} OpenCorporates rows to {a.out}")187    return 0188 189 190if __name__ == "__main__":191    raise SystemExit(main())192