goon/scripts/fix_html_entity_titles.py

"""Dekoduje encje HTML w tytulach scen i filmow zapisanych przed poprawka normalizera.

Tuby (glownie xvideos, JSON-LD) oddawaly tytuly typu `we're [AI]`,
a my zapisywalismy je tak jak przyszly. `normalize` robil z tego slowa `apos`, `lpar`,
wiec tytul nie trafial ani w wyszukiwarke, ani w kanon. Od poprawki `normalize_scene`
przepuszcza tytul przez `unescape_html`, ale istniejace wiersze trzeba poprawic osobno:
scraper tytulu juz ustawionego nie nadpisuje.

Co zmienia na wierszu:
  - `title` na wersje po `unescape_html`,
  - `title_normalized` na `normalize(nowy tytul)`, tak jak liczy go ingest,
  - `slug` tylko gdy byl wyliczony ze starego tytulu (`slugify(stary) == slug`); slug
    ustawiony z innego zrodla zostaje.

Zapis jest warunkowy (`WHERE title = stary`), wiec wiersz zmieniony w miedzyczasie przez
ingest albo merge jest pomijany. Ponowne odpalenie jest bezpieczne: bierze tylko tytuly,
w ktorych nadal jest encja.

Uzycie:
    python3 -u /srv/scripts/fix_html_entity_titles.py --dry-run
    python3 -u /srv/scripts/fix_html_entity_titles.py
"""
from __future__ import annotations

import argparse
import logging
import random
import sys

import sqlalchemy as sa

from app.db import session_scope
from app.normalize.text import normalize, slugify, unescape_html

log = logging.getLogger("fix_html_entity_titles")

# Ten sam zestaw co `_HTML_ENTITY_RE` w app/normalize/text.py, w skladni Postgresa.
_PG_ENTITY_RE = r"&(#[0-9]{1,7}|#[xX][0-9a-fA-F]{1,6}|[A-Za-z][A-Za-z0-9]{1,31});"
_TABLES = ("scenes", "movies")


def _plan(table: str, *, limit: int | None) -> list[dict]:
    with session_scope() as session:
        rows = session.execute(
            sa.text(
                f"SELECT id, title, title_normalized, slug FROM {table} "
                "WHERE title ~ :re" + (" LIMIT :limit" if limit else "")
            ),
            {"re": _PG_ENTITY_RE, "limit": limit},
        ).all()

    plan: list[dict] = []
    for row_id, title, title_norm, slug in rows:
        new_title = unescape_html(title)
        if not new_title or new_title == title:
            continue
        new_slug = slug
        if slug is None or slug == slugify(title):
            new_slug = slugify(new_title) or slug
        plan.append(
            {
                "id": row_id,
                "old_title": title,
                "title": new_title,
                "old_title_normalized": title_norm,
                "title_normalized": normalize(new_title),
                "old_slug": slug,
                "slug": new_slug,
            }
        )
    log.info("%s: %d wierszy z encja, %d do zmiany", table, len(rows), len(plan))
    return plan


def _apply(table: str, plan: list[dict], *, batch_size: int) -> int:
    # Jedno UPDATE na partie przez unnest: rowcount jest wtedy wiarygodny.
    stmt = sa.text(
        f"UPDATE {table} t SET title = v.title, title_normalized = v.title_normalized, slug = v.slug "
        "FROM unnest(CAST(:ids AS uuid[]), CAST(:old_titles AS text[]), CAST(:titles AS text[]), "
        "CAST(:norms AS text[]), CAST(:slugs AS text[])) "
        "AS v(id, old_title, title, title_normalized, slug) "
        "WHERE t.id = v.id AND t.title = v.old_title"
    )
    updated = 0
    for start in range(0, len(plan), batch_size):
        batch = plan[start : start + batch_size]
        params = {
            "ids": [str(item["id"]) for item in batch],
            "old_titles": [item["old_title"] for item in batch],
            "titles": [item["title"] for item in batch],
            "norms": [item["title_normalized"] for item in batch],
            "slugs": [item["slug"] for item in batch],
        }
        with session_scope() as session:
            updated += session.execute(stmt, params).rowcount
        log.info("%s: %d/%d przejrzane, %d zapisane", table, start + len(batch), len(plan), updated)
    return updated


def main() -> int:
    p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    p.add_argument("--dry-run", action="store_true", help="tylko policz i pokaz probke")
    p.add_argument("--batch-size", type=int, default=500)
    p.add_argument("--sample", type=int, default=20, help="ile losowych zmian pokazac")
    p.add_argument("--limit", type=int, default=None, help="max wierszy na tabele (test)")
    args = p.parse_args()
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")

    total_planned = total_updated = 0
    for table in _TABLES:
        plan = _plan(table, limit=args.limit)
        if not plan:
            continue
        total_planned += len(plan)
        slug_changes = sum(1 for item in plan if item["slug"] != item["old_slug"])
        log.info("%s: slug zmieniony w %d z %d", table, slug_changes, len(plan))
        for item in random.sample(plan, min(args.sample, len(plan))):
            print(f"[{table}] {item['id']}")
            print(f"  title: {item['old_title']!r}")
            print(f"      -> {item['title']!r}")
            print(f"  norm:  {item['old_title_normalized']!r}")
            print(f"      -> {item['title_normalized']!r}")
            if item["slug"] != item["old_slug"]:
                print(f"  slug:  {item['old_slug']!r}")
                print(f"      -> {item['slug']!r}")
        sys.stdout.flush()
        if not args.dry_run:
            total_updated += _apply(table, plan, batch_size=args.batch_size)

    if args.dry_run:
        log.info("DRY RUN: %d wierszy do zmiany, nic nie zapisane", total_planned)
    else:
        log.info("zapisane %d z %d (reszta zmieniona w miedzyczasie)", total_updated, total_planned)
    return 0


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