signal-bot/tool/gen_traps.py

#!/usr/bin/env python3
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Regenerate ``signal_bot/traps.json`` from Wikimedia Commons.

Run this by hand; the output is committed, so the bot never searches Commons
at the moment somebody types ``!pułapka``.

**The file holds page ids, not filenames, and that is load-bearing.** Commons
filenames are free text carrying people's names -- the harvest turns up things
like ``"Fishing-Boat with Fishermen" - Vasileios Chatzis.jpg``. Committing two
thousand of those would eventually collide with a label somebody minted from
the chat with ``!add_photo``, and ``scripts/check_personal_data.py`` would then
refuse every commit for good, over data nobody can edit out by hand. An
integer cannot carry a name. ``prop=imageinfo`` accepts ``pageids=`` and puts
``thumburl`` at the same place ``titles=`` does, so nothing downstream cares.

Depth is one, and that is also measured: recursing three levels from
*Category:Fish traps* reaches Alfred Sisley's net paintings and Hawaiian
fisherman watercolours. Immediate subcategories only.
"""

from __future__ import annotations

import argparse
import json
import sys
import time
from pathlib import Path
from typing import Any, Final

import httpx

CATEGORIES_FILE: Final = Path(__file__).with_name("trap_categories.json")

API: Final = "https://commons.wikimedia.org/w/api.php"
# Wikimedia's policy requires a descriptive agent with a contact URL; the
# generic client default is rate-limited hard and may be blocked outright.
USER_AGENT: Final = "signal-bot traps generator (https://github.com/kuhyx/signal-bot)"

# The extension is what tells Signal how to render the attachment, so an
# unknown one is dropped rather than guessed. Same set as portraits.SUFFIXES.
SUFFIXES: Final = frozenset({".jpg", ".jpeg", ".png", ".gif"})

# Commons hosts non-free content too. !gekon filters licences at its query and
# endpoints.py calls that filter load-bearing; this is the same idea, on the
# only field Commons exposes for it. Matched case-folded, by prefix.
LICENCES: Final = ("public domain", "cc0", "cc by", "cc-by", "no restrictions")

# imageinfo takes fifty page ids per request, joined with "|". A comma is
# rejected outright ("badinteger"), which is worth writing down because the
# error names the parameter rather than the separator.
BATCH: Final = 50
# A harvest smaller than this is a truncated one, and committing it would
# quietly shrink the command's whole world.
MINIMUM: Final = 500

TOO_MANY_REQUESTS: Final = 429
RETRIES: Final = 6
PAUSE: Final = 0.4


def categories() -> tuple[tuple[str, ...], frozenset[str], tuple[str, ...]]:
    """Return the seeds, the pruned set and the pruned patterns."""
    document = json.loads(CATEGORIES_FILE.read_text(encoding="utf-8"))
    return (
        tuple(document["seeds"]),
        frozenset(document["excluded"]),
        tuple(document["prune_patterns"]),
    )


def _get(params: dict[str, str]) -> dict[str, Any]:
    """One API call, retried with a widening pause when Commons says slow down.

    Commons answers 429 after a few hundred requests even at a steady 0.4s,
    and a harvest that silently loses a category is worse than a slow one.
    """
    query = {"action": "query", "format": "json", "formatversion": "2", **params}
    for attempt in range(RETRIES):
        time.sleep(PAUSE)
        response = httpx.get(API, params=query, headers={"User-Agent": USER_AGENT}, timeout=60.0)
        if response.status_code == TOO_MANY_REQUESTS:
            time.sleep(2.0**attempt)
            continue
        response.raise_for_status()
        document: dict[str, Any] = response.json()
        return document
    msg = f"Commons kept answering {TOO_MANY_REQUESTS} for {params}"
    raise RuntimeError(msg)


def _members(category: str, kind: str) -> list[dict[str, Any]]:
    """Every member of one category, following continuations."""
    found: list[dict[str, Any]] = []
    carry_on: str | None = None
    while True:
        params = {
            "list": "categorymembers",
            "cmtitle": category,
            "cmtype": kind,
            "cmlimit": "500",
        }
        if carry_on is not None:
            params["cmcontinue"] = carry_on
        document = _get(params)
        found += document.get("query", {}).get("categorymembers", [])
        carry_on = document.get("continue", {}).get("cmcontinue")
        if carry_on is None:
            return found


def verify_seeds(seeds: tuple[str, ...]) -> None:
    """Fail before harvesting if any seed category does not exist.

    A category that is not there is not an error to Commons -- it is an empty
    answer -- so a plausible-looking typo removes a whole subject silently.
    Five of the first draft's names were wrong this way.

    Raises:
        RuntimeError: If any seed is missing, naming all of them at once.
    """
    missing: list[str] = []
    for start in range(0, len(seeds), BATCH):
        document = _get({"titles": "|".join(seeds[start : start + BATCH])})
        missing += [
            str(page["title"])
            for page in document.get("query", {}).get("pages", [])
            if page.get("missing")
        ]
    if missing:
        msg = f"these seed categories do not exist on Commons: {sorted(missing)}"
        raise RuntimeError(msg)


def harvest() -> dict[int, str]:
    """Page id to title, for every usable file one step from the seeds."""
    seeds, excluded, patterns = categories()
    verify_seeds(seeds)
    files: dict[int, str] = {}
    seen: set[str] = set(excluded)
    for number, seed in enumerate(seeds, start=1):
        sys.stderr.write(f"[{number}/{len(seeds)}] {seed} ({len(files)} files so far)\n")
        sys.stderr.flush()
        branch = [seed, *(sub["title"] for sub in _members(seed, "subcat"))]
        for category in branch:
            if category in seen or any(pattern in category for pattern in patterns):
                continue
            seen.add(category)
            for entry in _members(category, "file"):
                title = str(entry["title"])
                if Path(title).suffix.lower() in SUFFIXES:
                    files[int(entry["pageid"])] = title
    return files


def licensed(files: dict[int, str]) -> dict[int, str]:
    """Drop everything whose licence does not permit reposting it."""
    kept: dict[int, str] = {}
    ids = sorted(files)
    sys.stderr.write(f"checking licences on {len(ids)} files\n")
    sys.stderr.flush()
    for start in range(0, len(ids), BATCH):
        chunk = ids[start : start + BATCH]
        document = _get(
            {
                "prop": "imageinfo",
                "iiprop": "extmetadata",
                "iiextmetadatafilter": "LicenseShortName",
                "pageids": "|".join(str(page) for page in chunk),
            }
        )
        for page in document.get("query", {}).get("pages", []):
            info = page.get("imageinfo") or [{}]
            short = info[0].get("extmetadata", {}).get("LicenseShortName", {}).get("value", "")
            name = str(short).casefold()
            if name.startswith(LICENCES):
                kept[int(page["pageid"])] = files[int(page["pageid"])]
    return kept


def _rejection(traps: dict[int, str]) -> str:
    """Return why this harvest must not be written, or an empty string."""
    if len(traps) < MINIMUM:
        return f"got {len(traps)} traps, expected at least {MINIMUM}"
    return ""


def main() -> int:
    """Write traps.json, refusing to write a harvest that failed its own checks."""
    parser = argparse.ArgumentParser(description="Regenerate signal_bot/traps.json.")
    parser.add_argument(
        "--titles",
        action="store_true",
        help="print every kept title; the committed file is integers, so this is the review",
    )
    arguments = parser.parse_args()

    traps = licensed(harvest())
    rejection = _rejection(traps)
    if rejection:
        sys.stderr.write(f"refusing to write: {rejection}\n")
        return 1

    if arguments.titles:
        for page in sorted(traps):
            sys.stdout.write(f"{page}\t{traps[page]}\n")

    out = Path(__file__).resolve().parent.parent / "signal_bot" / "traps.json"
    document = {"source": str(CATEGORIES_FILE.name), "traps": sorted(traps)}
    out.write_text(json.dumps(document, ensure_ascii=False, indent=1) + "\n", encoding="utf-8")
    sys.stdout.write(f"wrote {len(traps)} traps to {out}\n")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())