thermograph/indexnow.py

"""IndexNow — instantly notify Bing, DuckDuckGo, Yandex, Seznam and friends of new
or changed URLs so they (re)crawl within minutes instead of days.

Note: Google does NOT use IndexNow. For Google, submit the sitemap in Search
Console; this module covers everyone else (and DuckDuckGo, which is powered by
Bing's index).

A per-host key authorises submissions: it is served as a plain-text file at the
site root (``/{key}.txt``, wired up in content.register) and echoed in every
submission so the search engine can confirm we own the host. Key resolution order
mirrors push.py's VAPID handling: env var → gitignored file → generated once and
persisted.

Usage (run after a content rebuild, or once to seed a fresh site):

    cd backend && python indexnow.py https://thermograph.org
    # or:  make indexnow URL=https://thermograph.org
"""
import hashlib
import logging
import os
import secrets
import threading
from urllib.parse import urlparse

import httpx

log = logging.getLogger("thermograph.indexnow")

_DATA_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "data"))
_KEY_PATH = os.environ.get("THERMOGRAPH_INDEXNOW_KEY_FILE") or os.path.join(_DATA_DIR, "indexnow_key.txt")
_STATE_PATH = os.environ.get("THERMOGRAPH_INDEXNOW_STATE_FILE") or os.path.join(_DATA_DIR, "indexnow_state.txt")
_ENDPOINT = "https://api.indexnow.org/indexnow"   # a shared endpoint; it fans out to all participants
_BATCH = 10000                                    # IndexNow's max URLs per request

_lock = threading.Lock()
_key = None


def key() -> str:
    """The IndexNow key (env → file → generate-and-persist), cached for the process."""
    global _key
    if _key:
        return _key
    with _lock:
        if _key:
            return _key
        env = os.environ.get("THERMOGRAPH_INDEXNOW_KEY", "").strip()
        if env:
            _key = env
            return _key
        try:
            with open(_KEY_PATH, encoding="utf-8") as f:
                k = f.read().strip()
            if k:
                _key = k
                return _key
        except OSError:
            pass
        _key = secrets.token_hex(16)   # 32 hex chars — within IndexNow's 8–128 range
        try:
            os.makedirs(_DATA_DIR, exist_ok=True)
            with open(_KEY_PATH, "w", encoding="utf-8") as f:
                f.write(_key)
            os.chmod(_KEY_PATH, 0o600)
        except OSError:
            log.warning("could not persist IndexNow key to %s; using an in-memory key", _KEY_PATH)
        return _key


def submit(urls, host: str, key_location: str, scheme: str = "https", timeout: float = 30) -> dict:
    """POST ``urls`` (deduped, in batches of 10k) to IndexNow. Every URL must be on
    ``host``; ``key_location`` is the public URL of our key file. Returns a summary."""
    urls = list(dict.fromkeys(urls))
    if not urls:
        return {"submitted": 0, "total": 0, "batches": 0}
    k = key()
    submitted, statuses = 0, []
    for i in range(0, len(urls), _BATCH):
        batch = urls[i:i + _BATCH]
        payload = {"host": host, "key": k, "keyLocation": key_location, "urlList": batch}
        try:
            r = httpx.post(_ENDPOINT, json=payload, timeout=timeout,
                           headers={"Content-Type": "application/json; charset=utf-8"})
        except httpx.HTTPError as e:
            log.warning("IndexNow request failed: %s", e)
            statuses.append("error")
            continue
        statuses.append(r.status_code)
        if r.status_code in (200, 202):
            submitted += len(batch)
        else:
            log.warning("IndexNow batch rejected: %s %s", r.status_code, r.text[:200])
    return {"submitted": submitted, "total": len(urls),
            "batches": len(statuses), "statuses": statuses, "key_location": key_location}


def submit_all(site_base_url: str, **kw) -> dict:
    """Submit every indexable page. ``site_base_url`` is the public origin + base
    path, e.g. ``https://thermograph.org`` (root) or ``https://host/thermograph``."""
    import content   # lazy: avoids a content ↔ indexnow import cycle
    site = site_base_url.rstrip("/")
    parsed = urlparse(site)
    urls = [site + path for path in content.public_paths()]
    return submit(urls, host=parsed.hostname, scheme=parsed.scheme or "https",
                  key_location=f"{site}/{key()}.txt", **kw)


def url_signature() -> str:
    """A stable fingerprint of the indexable URL set — changes only when pages are
    added or removed (e.g. a new city), not on code-only deploys. Used to skip
    resubmitting an unchanged set (see the --if-changed CLI flag)."""
    import content
    return hashlib.sha1("\n".join(content.public_paths()).encode()).hexdigest()


def _read_state() -> str:
    try:
        with open(_STATE_PATH, encoding="utf-8") as f:
            return f.read().strip()
    except OSError:
        return ""


def _write_state(sig: str) -> None:
    try:
        os.makedirs(_DATA_DIR, exist_ok=True)
        with open(_STATE_PATH, "w", encoding="utf-8") as f:
            f.write(sig)
    except OSError:
        log.warning("could not persist IndexNow state to %s", _STATE_PATH)


if __name__ == "__main__":
    import sys

    # Usage: indexnow.py [--if-changed] [BASE_URL]
    #   --if-changed  submit only when the URL set changed since the last run
    #                 (used by the deploy hook so code-only deploys don't resubmit).
    argv = sys.argv[1:]
    if_changed = "--if-changed" in argv
    positional = [a for a in argv if not a.startswith("-") and a.strip()]
    base = (positional[0] if positional else "") or os.environ.get("THERMOGRAPH_BASE_URL", "https://thermograph.org")
    logging.basicConfig(level=logging.INFO)

    sig = url_signature()
    if if_changed and sig == _read_state():
        print("IndexNow: URL set unchanged since last submission — skipping.")
        sys.exit(0)

    print(f"IndexNow key: {key()}  (served at {base.rstrip('/')}/{key()}.txt)")
    result = submit_all(base)
    print(f"Submitted {result['submitted']}/{result['total']} URLs "
          f"in {result['batches']} batch(es); statuses={result['statuses']}")
    if result["total"] and result["submitted"] == result["total"]:
        _write_state(sig)   # only remember a fully-successful submission