News-ML-Decoder/data_collect/crawl_google_news.py

#!/usr/bin/env python3
"""Collect historical headlines from Google News RSS into data/history.db.

One request per (query, date window), e.g.
``gold site:reuters.com after:2019-07-22 before:2019-07-29``. Google
returns at most 100 items per request. The crawl is
resumable: finished (query, window) pairs are recorded in ``crawl_log``
and skipped on the next run. Requests are sequential with a fixed
delay; on HTTP 429/503 the crawler backs off and stops after repeated
refusals rather than hammering the service.

Headlines are stored once per normalised text (``hash``); the queries
that found a headline are kept in ``queries``. Nothing here labels or
filters against the golden set; label_history.py and the trainer do.

Usage:
    python data_collect/crawl_google_news.py                      # precise set
    python data_collect/crawl_google_news.py --query-set broad    # off-topic side
    python data_collect/crawl_google_news.py --query-set growth --window-days 30   # growth data + equities
    python data_collect/crawl_google_news.py --limit 20           # smoke test
"""

from __future__ import annotations

import argparse
import email.utils
import hashlib
import html
import random
import re
import sqlite3
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
import xml.etree.ElementTree as ET
from datetime import date, datetime, timedelta, timezone
from pathlib import Path

ML_NEWS_DIR = Path(__file__).resolve().parent.parent
DEFAULT_DB = ML_NEWS_DIR / "data" / "history.db"
FEED_URL = "https://news.google.com/rss/search?q={q}&hl=en-US&gl=US&ceid=US:en"
USER_AGENT = "Mozilla/5.0 (X11; Linux x86_64) ML-NEWS research crawler"

#: Scoped to financial publishers: unscoped queries return mostly
#: general news ("gold price" -> phone prices, "safe haven" -> baby-box
#: laws). Reuters answers each weekly window with up to 100 market
#: headlines. Topics cover gold doctrine (Fed, dollar, yields,
#: inflation, jobs, central banks, geopolitics, risk appetite) plus
#: off-topic controls (oil, crypto, single stocks) so the relevance head
#: sees news that does not move gold.
BROAD_QUERIES = (
    "gold site:reuters.com",
    "Fed site:reuters.com",
    "dollar site:reuters.com",
    "yields site:reuters.com",
    "inflation site:reuters.com",
    "payrolls OR jobless site:reuters.com",
    "\"central bank\" site:reuters.com",
    "stocks site:reuters.com",
    "war OR sanctions OR tensions site:reuters.com",
    "euro OR yen OR sterling site:reuters.com",
    "oil site:reuters.com",
    "bitcoin OR crypto site:reuters.com",
    "gold site:cnbc.com",
    "markets site:cnbc.com",
    "site:marketwatch.com",
)

#: A keyword in the query matches anywhere in the article, so the broad
#: set is ~93% headlines that do not move gold (2026-09-23 pilot: 70 of
#: 1,000 DeepSeek-relevant). ``intitle:`` restricts the match to the
#: headline: 3-15 items per week per query, mostly gold doctrine. The
#: broad set supplies the not-relevant side, this one the relevant side.
PRECISE_QUERIES = tuple(
    f"intitle:{term} site:{site}"
    for site in ("reuters.com", "cnbc.com", "marketwatch.com")
    for term in ("gold", "bullion", "Fed", "Powell", "dollar", "yields",
                 "inflation", "payrolls", '"rate cut"', '"rate hike"', '"safe haven"')
    if site == "reuters.com" or term in ("gold", "Fed", "dollar", "yields", "inflation")
)

#: The two golden-set categories the 2026-09-27 model still fails are
#: strong-US-data (3/7) and risk-on (1/4): "GDP beats", "stocks at record
#: highs", "ceasefire lifts markets" should all be bearish for gold, but
#: no precise query covers growth data or equities. ``intitle:`` again,
#: so label_history.py puts these in the precise pool.
GROWTH_TERMS = ("GDP", '"retail sales"', "ISM", "PMI", '"consumer confidence"',
                '"jobless claims"', '"consumer spending"', "manufacturing")
RISK_TERMS = ('"record high"', '"risk appetite"', '"Wall Street"', '"S&P 500"',
              "ceasefire", '"trade deal"', "equities")
GROWTH_QUERIES = tuple(
    f"intitle:{term} site:{site}"
    for site in ("reuters.com", "cnbc.com")
    for term in GROWTH_TERMS + RISK_TERMS
    if site == "reuters.com" or term in ("GDP", '"retail sales"', '"jobless claims"',
                                          '"record high"', '"S&P 500"')
)

QUERY_SETS = {"broad": BROAD_QUERIES, "precise": PRECISE_QUERIES, "growth": GROWTH_QUERIES}

SCHEMA = """
CREATE TABLE IF NOT EXISTS headlines (
    hash TEXT PRIMARY KEY,
    headline TEXT NOT NULL,
    raw_title TEXT NOT NULL,
    publisher TEXT,
    published_at TEXT,
    first_query TEXT NOT NULL,
    queries TEXT NOT NULL,
    link TEXT,
    crawled_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_headlines_published ON headlines (published_at);
CREATE TABLE IF NOT EXISTS crawl_log (
    query TEXT NOT NULL,
    window_start TEXT NOT NULL,
    items INTEGER NOT NULL,
    new_headlines INTEGER NOT NULL,
    crawled_at TEXT NOT NULL,
    PRIMARY KEY (query, window_start)
);
"""

_SPACE = re.compile(r"\s+")


def normalise(text: str) -> str:
    return _SPACE.sub(" ", html.unescape(text)).strip()


def headline_key(headline: str) -> str:
    """Hash of the case-folded, whitespace-collapsed headline."""
    return hashlib.sha256(normalise(headline).casefold().encode("utf-8")).hexdigest()


def strip_publisher(title: str, publisher: str | None) -> str:
    """Google appends ' - Publisher' to every title; drop it."""
    title = normalise(title)
    if publisher:
        suffix = f" - {normalise(publisher)}"
        if title.endswith(suffix):
            return title[: -len(suffix)].strip()
    return title


def windows_of(start: date, end: date, days: int):
    day = start
    while day < end:
        yield day, min(day + timedelta(days=days), end)
        day += timedelta(days=days)


def fetch(url: str, timeout: float) -> bytes:
    request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
    with urllib.request.urlopen(request, timeout=timeout) as response:
        return response.read()


def parse_items(payload: bytes):
    root = ET.fromstring(payload)
    for item in root.findall("./channel/item"):
        title = item.findtext("title") or ""
        publisher = item.findtext("source")
        published = item.findtext("pubDate")
        published_at = None
        if published:
            try:
                published_at = email.utils.parsedate_to_datetime(published).astimezone(
                    timezone.utc).isoformat()
            except (TypeError, ValueError):
                published_at = None
        headline = strip_publisher(title, publisher)
        if headline:
            yield {
                "headline": headline,
                "raw_title": normalise(title),
                "publisher": publisher,
                "published_at": published_at,
                "link": item.findtext("link"),
            }


def store(conn: sqlite3.Connection, query: str, items) -> tuple[int, int]:
    now = datetime.now(timezone.utc).isoformat()
    seen = new = 0
    for it in items:
        seen += 1
        key = headline_key(it["headline"])
        row = conn.execute("SELECT queries FROM headlines WHERE hash = ?", (key,)).fetchone()
        if row is None:
            conn.execute(
                "INSERT INTO headlines (hash, headline, raw_title, publisher, published_at,"
                " first_query, queries, link, crawled_at) VALUES (?,?,?,?,?,?,?,?,?)",
                (key, it["headline"], it["raw_title"], it["publisher"], it["published_at"],
                 query, query, it["link"], now))
            new += 1
        elif query not in row[0].split("|"):
            conn.execute("UPDATE headlines SET queries = ? WHERE hash = ?",
                         (row[0] + "|" + query, key))
    return seen, new


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    ap.add_argument("--db", type=Path, default=DEFAULT_DB)
    ap.add_argument("--start", type=date.fromisoformat, default=date(2019, 1, 7))
    ap.add_argument("--end", type=date.fromisoformat, default=date(2026, 9, 1))
    ap.add_argument("--delay", type=float, default=2.0, help="seconds between requests")
    ap.add_argument("--timeout", type=float, default=20.0)
    ap.add_argument("--limit", type=int, default=0, help="stop after N requests (0 = all)")
    ap.add_argument("--query-set", choices=sorted(QUERY_SETS), default="precise")
    ap.add_argument("--window-days", type=int, default=0,
                    help="search window length (default: 7 broad, 14 precise; precise "
                         "queries return 3-15 items a week, far under the 100 cap)")
    ap.add_argument("--max-refusals", type=int, default=5,
                    help="stop after this many consecutive 429/503 responses")
    args = ap.parse_args()

    args.db.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(args.db)
    conn.executescript(SCHEMA)
    done = {(q, w) for q, w in conn.execute("SELECT query, window_start FROM crawl_log")}
    queries = QUERY_SETS[args.query_set]
    days = args.window_days or (14 if args.query_set == "precise" else 7)
    windows = list(windows_of(args.start, args.end, days))
    # Seeded shuffle, window-major: an interrupted crawl still covers
    # every year and every query evenly, so a pilot on a partial crawl
    # is not all 2019.
    random.Random(0).shuffle(windows)
    todo = [(q, a, b) for a, b in windows for q in queries if (q, a.isoformat()) not in done]
    print(f"{args.query_set}: {len(windows)} windows of {days} days x {len(queries)} queries; "
          f"{len(done)} done, {len(todo)} to go", flush=True)

    requests = refusals = 0
    started = time.monotonic()
    for query, a, b in todo:
        if args.limit and requests >= args.limit:
            break
        q = f"{query} after:{a.isoformat()} before:{b.isoformat()}"
        url = FEED_URL.format(q=urllib.parse.quote_plus(q))
        try:
            payload = fetch(url, args.timeout)
        except urllib.error.HTTPError as error:
            if error.code in (429, 503):
                refusals += 1
                wait = min(600, 30 * 2 ** (refusals - 1))
                print(f"HTTP {error.code} on {q!r}; refusal {refusals}, sleeping {wait}s",
                      flush=True)
                if refusals >= args.max_refusals:
                    print("too many refusals, stopping; rerun later to resume", flush=True)
                    break
                time.sleep(wait)
                continue
            print(f"HTTP {error.code} on {q!r}; skipped", flush=True)
            time.sleep(args.delay)
            continue
        except (urllib.error.URLError, TimeoutError, OSError) as error:
            print(f"network error on {q!r}: {error}; skipped", flush=True)
            time.sleep(args.delay * 5)
            continue
        refusals = 0
        requests += 1
        try:
            seen, new = store(conn, query, parse_items(payload))
        except ET.ParseError as error:
            print(f"bad XML for {q!r}: {error}; skipped", flush=True)
            time.sleep(args.delay)
            continue
        conn.execute(
            "INSERT OR REPLACE INTO crawl_log VALUES (?,?,?,?,?)",
            (query, a.isoformat(), seen, new, datetime.now(timezone.utc).isoformat()))
        conn.commit()
        if requests % 100 == 0:
            total = conn.execute("SELECT count(*) FROM headlines").fetchone()[0]
            rate = requests / (time.monotonic() - started)
            print(f"{requests} requests, window {a}, {total} unique headlines, "
                  f"{rate * 60:.0f} req/min", flush=True)
        time.sleep(args.delay)

    total = conn.execute("SELECT count(*) FROM headlines").fetchone()[0]
    print(f"done: {requests} requests this run, {total} unique headlines in {args.db}",
          flush=True)
    conn.close()
    return 0


if __name__ == "__main__":
    sys.exit(main())