signal-bot/tool/gen_poems.py

#!/usr/bin/env python3
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Regenerate ``signal_bot/poems.json`` from Wolne Lektury.

Run this by hand; the output is committed, so the bot never searches the
catalogue at the moment somebody types ``!wiersz`` -- it fetches exactly one
known poem.

**The file holds slugs, not poems, and that is load-bearing.** The first
version of this generator committed the text, and
``scripts/check_personal_data.py`` refused the pack: Polish verse is full of
Polish given names, three of them collided with a real group member, and two
still did after a filter that had already discarded 31% of the corpus,
because the names sat at the start of a line where no filter can look. A slug
cannot carry a name the way a poem can, and everything a reader sees --
title, author, the poem itself -- is fetched at post time and never written
here. ``signal_bot/poem_command.py`` tells the longer version.

What is left to decide at harvest time is which poems are *worth* committing,
and that needs the text, so this reads all five thousand once:

* a work is dropped unless its own footer says *Ten utwór jest w domenie
  publicznej* -- Wolne Lektury hosts living authors too, and a third of the
  catalogue came back non-free;
* a translation is dropped -- the txt carries a ``tłum.`` credit under the
  title, and a poem translated into Polish is not a Polish poem;
* anything longer than the caps is dropped, whole poems only;
* a title carrying a proper noun is dropped. This one is belt-and-braces
  rather than load-bearing now: the slug is derived from the title, so
  filtering the title is what keeps a dedication like *Do Zosi* from
  committing ``do-zosi``. It costs about a tenth of the pool and removes the
  only free text left in the file.
"""

from __future__ import annotations

import argparse
import json
import re
import sys
import time
from collections import Counter
from concurrent.futures import ThreadPoolExecutor
from functools import partial
from pathlib import Path
from typing import Final

import httpx

from signal_bot.poem_command import MAX_CHARS, MAX_LINES
from signal_bot.wolnelektury import parse

API: Final = "https://wolnelektury.pl/api/genres/{genre}/books/?format=json"
TEXT: Final = "https://wolnelektury.pl/media/book/txt/{slug}.txt"

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

# Lyric forms only. "Poemat" is deliberately absent: a poemat is book-length
# by definition and every one would fail the caps anyway, costing a few
# hundred requests to learn nothing.
GENRES: Final = ("wiersz", "sonet", "fraszka", "wiersz-sylabotoniczny")

# A capitalised word that is not the title's first and does not open a
# bracket. Catches the dedications -- "Do Zosi", "Na śmierć Jana", "O
# Mickiewiczu" -- while keeping the very common "IV (Jak ptaki, kiedy
# odlatywać poczną...)" incipit titles, which are a large slice of the
# catalogue and carry no name at all.
TITLE_PROPER_NOUN: Final = re.compile(r"\w[\s,]+([A-ZĄĆĘŁŃÓŚŹŻ][a-ząćęłńóśźż]+)")

# A pool smaller than this is a truncated harvest, and committing it would
# quietly shrink the command's whole world.
MINIMUM: Final = 300

# Five thousand sequential fetches spends half an hour waiting on latency
# rather than on Wolne Lektury. Four workers against a static text endpoint
# is roughly ten requests a second -- polite for a volunteer-run library, and
# it turns the harvest from thirty-odd minutes into six. httpx.Client is
# thread-safe and they share its one connection pool.
WORKERS: Final = 4

TOO_MANY_REQUESTS: Final = 429
RETRIES: Final = 5
PAUSE: Final = 0.1


def _get(url: str, client: httpx.Client) -> str:
    """One request, retried with a widening pause when the site says slow down.

    The client is passed in rather than made here, for the reason app.py
    keeps one AsyncClient for the whole bot: a bare ``httpx.get`` is a fresh
    TCP connection and a fresh TLS handshake every time, and across five
    thousand poems that handshake is most of the harvest rather than most of
    the politeness. One reused connection is both faster and less to ask of a
    volunteer-run library.
    """
    for attempt in range(RETRIES):
        time.sleep(PAUSE)
        response = client.get(url, 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()
        return response.text
    msg = f"wolnelektury.pl kept answering {TOO_MANY_REQUESTS} for {url}"
    raise RuntimeError(msg)


def catalogue(client: httpx.Client) -> set[str]:
    """Every lyric slug in the catalogue.

    A set because the genres overlap heavily -- a sonnet is also a wiersz --
    and fetching the same poem four times would be four times the requests
    for the same answer.
    """
    found: set[str] = set()
    for genre in GENRES:
        books = json.loads(_get(API.format(genre=genre), client))
        found |= {str(book["slug"]) for book in books}
        sys.stderr.write(f"{genre}: {len(books)} books ({len(found)} distinct so far)\n")
        sys.stderr.flush()
    return found


def _fetch(slug: str, client: httpx.Client) -> tuple[str, str | None]:
    """One slug and its text, or None when the fetch failed.

    Failure comes back as a value rather than an exception because this runs
    in a worker thread: one dead slug out of five thousand should cost that
    slug, not the harvest.
    """
    try:
        return slug, _get(TEXT.format(slug=slug), client)
    except httpx.HTTPError:
        return slug, None


def consider(slug: str, raw: str) -> tuple[str | None, str]:
    """The slug if it is worth committing, or None and the reason it is not."""
    poem = parse(raw)
    if poem is None:
        return None, "not public domain, translated or malformed"
    if poem.lines > MAX_LINES or len(poem.text) > MAX_CHARS:
        return None, "too long"
    if TITLE_PROPER_NOUN.search(poem.title):
        return None, "title carries a proper noun"
    return slug, ""


def harvest(client: httpx.Client) -> tuple[list[str], Counter[str]]:
    """Every slug worth committing, and a tally of why the rest were not.

    ``map`` rather than ``as_completed``: it yields in submission order, so
    the progress line means what it says and the result does not depend on
    which worker happened to finish first.
    """
    slugs = sorted(catalogue(client))
    kept: list[str] = []
    why: Counter[str] = Counter()
    with ThreadPoolExecutor(max_workers=WORKERS) as pool:
        for number, (slug, raw) in enumerate(
            pool.map(partial(_fetch, client=client), slugs), start=1
        ):
            if number % 500 == 0:
                sys.stderr.write(f"[{number}/{len(slugs)}] kept {len(kept)}\n")
                sys.stderr.flush()
            if raw is None:
                why["no text"] += 1
                continue
            wanted, reason = consider(slug, raw)
            if wanted is None:
                why[reason] += 1
            else:
                kept.append(wanted)
    return kept, why


def main() -> int:
    """Write poems.json, refusing to write a harvest that failed its own checks."""
    parser = argparse.ArgumentParser(description="Regenerate signal_bot/poems.json.")
    parser.add_argument("--slugs", action="store_true", help="print every kept slug")
    arguments = parser.parse_args()

    with httpx.Client(follow_redirects=True) as client:
        kept, why = harvest(client)
    for reason, count in why.most_common():
        sys.stderr.write(f"rejected {count}: {reason}\n")
    if len(kept) < MINIMUM:
        sys.stderr.write(f"refusing to write: got {len(kept)}, expected at least {MINIMUM}\n")
        return 1

    if arguments.slugs:
        for slug in kept:
            sys.stdout.write(f"{slug}\n")

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


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