osint-investigation

1#!/usr/bin/env python32"""Search the GDELT 2.0 DOC API for news mentions.3 4GDELT monitors world news in 100+ languages and indexes the full text.5Free, anonymous, ~15-minute update frequency. Covers ~2015→present.6 7Useful for surfacing news mentions of a person, company, or topic across8international media — much wider net than Google News.9"""10from __future__ import annotations11 12import argparse13import csv14import sys15import time16import urllib.parse17from pathlib import Path18 19sys.path.insert(0, str(Path(__file__).parent))20from _http import get_json  # noqa: E40221 22BASE = "https://api.gdeltproject.org/api/v2/doc/doc"23 24COLUMNS = [25    "title",26    "url",27    "seen_date",28    "domain",29    "language",30    "source_country",31    "tone",32    "social_image",33]34 35 36def fetch(37    query: str,38    mode: str,39    timespan: str | None,40    start_datetime: str | None,41    end_datetime: str | None,42    source_country: str | None,43    source_lang: str | None,44    limit: int,45    out_path: str,46) -> int:47    params: dict[str, str] = {48        "query": query,49        "mode": mode,50        "format": "json",51        "maxrecords": str(min(limit, 250)),52        "sort": "datedesc",53    }54    if timespan:55        params["timespan"] = timespan56    if start_datetime:57        params["startdatetime"] = start_datetime.replace("-", "").replace(":", "").replace(" ", "")58    if end_datetime:59        params["enddatetime"] = end_datetime.replace("-", "").replace(":", "").replace(" ", "")60    if source_country:61        params["sourcecountry"] = source_country62    if source_lang:63        params["sourcelang"] = source_lang64 65    url = f"{BASE}?{urllib.parse.urlencode(params)}"66    payload: dict | list = {}67    for attempt in range(3):68        try:69            payload = get_json(url)70            break71        except RuntimeError as e:72            # GDELT requires 1 request per 5 seconds; back off and retry.73            if "429" in str(e) and attempt < 2:74                print(75                    f"GDELT throttle hit; sleeping 6s before retry "76                    f"(attempt {attempt + 1}/3)",77                    file=sys.stderr,78                )79                time.sleep(6)80                continue81            print(f"GDELT error: {e}", file=sys.stderr)82            payload = {}83            break84        except Exception as e:  # noqa: BLE00185            print(f"GDELT error: {e}", file=sys.stderr)86            payload = {}87            break88 89    rows: list[dict[str, str]] = []90    if isinstance(payload, dict):91        articles = payload.get("articles", []) or []92        for a in articles[:limit]:93            seen = (a.get("seendate") or "")94            # GDELT format: 20260319T083000Z → 2026-03-19 08:30:00Z95            if len(seen) == 16 and "T" in seen:96                seen = f"{seen[0:4]}-{seen[4:6]}-{seen[6:8]} {seen[9:11]}:{seen[11:13]}:{seen[13:15]}Z"97            rows.append(98                {99                    "title": (a.get("title") or "").replace("\n", " ").strip(),100                    "url": a.get("url") or "",101                    "seen_date": seen,102                    "domain": a.get("domain") or "",103                    "language": a.get("language") or "",104                    "source_country": a.get("sourcecountry") or "",105                    "tone": str(a.get("tone") or ""),106                    "social_image": a.get("socialimage") or "",107                }108            )109 110    Path(out_path).parent.mkdir(parents=True, exist_ok=True)111    with open(out_path, "w", newline="", encoding="utf-8") as fh:112        w = csv.DictWriter(fh, fieldnames=COLUMNS)113        w.writeheader()114        w.writerows(rows)115    if not rows:116        print(117            f"GDELT: 0 articles for query={query!r}. "118            "GDELT indexes ~2015→present. Try widening the timespan or "119            "checking the query syntax (https://blog.gdeltproject.org/gdelt-doc-2-0-api-debuts/).",120            file=sys.stderr,121        )122    return len(rows)123 124 125def main() -> int:126    p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)127    p.add_argument("--query", required=True, help='Search query (supports GDELT operators: quoted phrases, AND/OR/NOT, sourcecountry:, theme:)')128    p.add_argument(129        "--mode",130        default="ArtList",131        choices=["ArtList", "ImageCollage", "TimelineVol", "TimelineTone", "ToneChart"],132        help="GDELT mode (default ArtList for article list)",133    )134    p.add_argument(135        "--timespan",136        help="Relative window: e.g. '1d', '1w', '1m', '3m', '1y' (overrides start/end)",137    )138    p.add_argument("--start", help="Absolute start YYYY-MM-DD or YYYY-MM-DDTHH:MM:SS")139    p.add_argument("--end", help="Absolute end YYYY-MM-DD or YYYY-MM-DDTHH:MM:SS")140    p.add_argument("--source-country", help="2-letter source country (e.g. US, UK)")141    p.add_argument("--source-lang", help="Source language (e.g. English, Spanish)")142    p.add_argument("--limit", type=int, default=100)143    p.add_argument("--out", required=True)144    a = p.parse_args()145    n = fetch(146        query=a.query,147        mode=a.mode,148        timespan=a.timespan,149        start_datetime=a.start,150        end_datetime=a.end,151        source_country=a.source_country,152        source_lang=a.source_lang,153        limit=a.limit,154        out_path=a.out,155    )156    print(f"Wrote {n} GDELT article rows to {a.out}")157    return 0158 159 160if __name__ == "__main__":161    raise SystemExit(main())162