signal-bot/tool/gen_art.py

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

Run this by hand; the output is committed, so the bot never searches Commons
at the moment somebody types ``!sztuka``. Everything ``gen_traps.py`` says
about why the file holds **page ids rather than filenames** applies here with
more force, not less: a Commons painting filename is very often literally a
person's name -- ``Portret Stanisława Potockiego.jpg`` -- and committing
tens of thousands of those would collide with a label somebody minted from
the chat, at which point ``scripts/check_personal_data.py`` refuses every
commit for good. An integer cannot carry a name.

Two pools, written as two lists. The draw is weighted rather than merged
(see ``art_command.POLISH_SHARE``), so they must stay separate here, and
they are walked with one shared ``seen`` set so that no file lands in both.
"""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path
from typing import Final

import commons_api
import commons_cache
import commons_harvest
import wikidata

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

USER_AGENT: Final = "signal-bot art generator (https://github.com/kuhyx/signal-bot)"

# Per pool, not in total. A world pool of twenty thousand next to a Polish
# pool of eleven would still satisfy a combined floor while making the Polish
# 60% a draw from almost nothing, which is the failure this catches.
MINIMUM: Final = 300

# Where an interrupted harvest leaves what it already fetched. Disposable and
# gitignored: delete it to force a full re-read of Commons.
CACHE_FILE: Final = Path(__file__).with_name(".commons_cache.json")


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


def harvest() -> tuple[dict[int, str], dict[int, str]]:
    """Both pools, walked disjointly and licence-filtered in one pass.

    Polish first, and that ordering is the whole point of sharing ``seen``:
    *Category:Oil paintings* reaches Matejko too, and whichever pool walks a
    category first owns it. Walking Polish first means a Polish painting
    stays Polish rather than being counted as world art.
    """
    polish_seeds, world_seeds, excluded, patterns = categories()
    commons_api.verify_seeds(polish_seeds + world_seeds, USER_AGENT)
    seen: set[str] = set()
    cache = commons_cache.load_cache(CACHE_FILE)
    if cache:
        sys.stderr.write(f"resuming: {len(cache)} categories already fetched\n")
    sys.stderr.write("== polish pool\n")
    polish = commons_harvest.walk(
        polish_seeds, excluded, patterns, USER_AGENT, seen, with_licences=True, cache=cache
    )
    commons_cache.save_cache(CACHE_FILE, cache)
    # Wikidata knows which works were made by Poles, which is the question
    # the categories only approximate. Additive: it reaches works nobody
    # filed under a Polish category, the categories reach works nobody made
    # a Wikidata item for.
    sys.stderr.write("== polish pool, via wikidata\n")
    titles = wikidata.polish_artwork_titles(USER_AGENT)
    sys.stderr.write(f"wikidata named {len(titles)} artworks by Polish creators\n")
    before = len(polish)
    polish.update(commons_harvest.licensed_titles(titles, USER_AGENT))
    sys.stderr.write(f"polish pool {before} -> {len(polish)} after wikidata\n")
    sys.stderr.write("== world pool\n")
    world = commons_harvest.walk(
        world_seeds, excluded, patterns, USER_AGENT, seen, with_licences=True, cache=cache
    )
    commons_cache.save_cache(CACHE_FILE, cache)
    # A file can still be in two categories that were walked by different
    # pools; seen only makes the *categories* disjoint. Subtract explicitly.
    for page in polish:
        world.pop(page, None)
    # No second licence pass: walk(with_licences=True) already filtered, in
    # the same requests that listed the files. That second pass was 1,500
    # requests at 2.3s each -- forty-six minutes -- on the previous run.
    return polish, world


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


def main() -> int:
    """Write art.json, refusing to write a harvest that failed its own checks."""
    parser = argparse.ArgumentParser(description="Regenerate signal_bot/art.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()

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

    if arguments.titles:
        for label, pool in (("polish", polish), ("world", world)):
            for page in sorted(pool):
                sys.stdout.write(f"{label}\t{page}\t{pool[page]}\n")

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


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