goon/scripts/fix_html_entities.py

"""Czysci encje i znaczniki HTML w tytulach i opisach scen i filmow sprzed poprawki normalizera.

Tuby (glownie xvideos, JSON-LD) oddawaly tytuly typu `we're [AI]`, a opisy
z `"` i `<br>`. Zapisywalismy je tak jak przyszly, wiec `normalize` robil z tytulu
slowa `apos`, `lpar`, a w apce i na stronie SEO encje widac doslownie. Od poprawki
`normalize_scene` przepuszcza tytul przez `unescape_html`, a opis przez `clean_description`,
ale istniejace wiersze trzeba poprawic osobno: scraper pola juz ustawionego nie nadpisuje.

Przebieg `titles` 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`).
Przebieg `descriptions` zmienia `description` na wersje po `clean_description`.

Zapis jest warunkowy (`WHERE pole = stara wartosc`), wiec wiersz zmieniony w miedzyczasie
przez ingest albo merge jest pomijany. Ponowne odpalenie jest bezpieczne: bierze tylko
wiersze, w ktorych nadal jest encja albo znacznik.

Uzycie:
    python3 -u /srv/scripts/fix_html_entities.py --dry-run
    python3 -u /srv/scripts/fix_html_entities.py --only descriptions
"""
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 clean_description, normalize, slugify, unescape_html

log = logging.getLogger("fix_html_entities")

# 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});"
# Wstepny filtr na znaczniki z `_BREAK_TAG_RE`/`_INLINE_TAG_RE`; decyduje `clean_description`.
_PG_TAG_RE = (
    r"</?(a|b|i|u|em|strong|span|div|p|font|small|big|sup|sub|h[1-6]|ul|ol|li|center"
    r"|blockquote|br)([[:space:]/>])"
)
_TABLES = ("scenes", "movies")


def _plan_titles(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.title: %d wierszy z encja, %d do zmiany", table, len(rows), len(plan))
    return plan


def _plan_descriptions(table: str, *, limit: int | None) -> list[dict]:
    with session_scope() as session:
        rows = session.execute(
            sa.text(
                f"SELECT id, description FROM {table} "
                "WHERE description ~ :ent OR description ~* :tag" + (" LIMIT :limit" if limit else "")
            ),
            {"ent": _PG_ENTITY_RE, "tag": _PG_TAG_RE, "limit": limit},
        ).all()

    plan = [
        {"id": row_id, "old_description": desc, "description": new}
        for row_id, desc in rows
        if (new := clean_description(desc)) != desc
    ]
    log.info("%s.description: %d wierszy z encja lub znacznikiem, %d do zmiany", table, len(rows), len(plan))
    return plan


def _apply_titles(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"
    )
    return _run_batches(
        f"{table}.title",
        stmt,
        plan,
        batch_size=batch_size,
        columns={"ids": "id", "old_titles": "old_title", "titles": "title", "norms": "title_normalized", "slugs": "slug"},
    )


def _apply_descriptions(table: str, plan: list[dict], *, batch_size: int) -> int:
    stmt = sa.text(
        f"UPDATE {table} t SET description = v.description "
        "FROM unnest(CAST(:ids AS uuid[]), CAST(:old AS text[]), CAST(:new AS text[])) "
        "AS v(id, old_description, description) "
        "WHERE t.id = v.id AND t.description = v.old_description"
    )
    return _run_batches(
        f"{table}.description",
        stmt,
        plan,
        batch_size=batch_size,
        columns={"ids": "id", "old": "old_description", "new": "description"},
    )


def _run_batches(label: str, stmt, plan: list[dict], *, batch_size: int, columns: dict[str, str]) -> int:
    updated = 0
    for start in range(0, len(plan), batch_size):
        batch = plan[start : start + batch_size]
        params = {
            param: [str(item[key]) if key == "id" else item[key] for item in batch]
            for param, key in columns.items()
        }
        with session_scope() as session:
            updated += session.execute(stmt, params).rowcount
        log.info("%s: %d/%d przejrzane, %d zapisane", label, start + len(batch), len(plan), updated)
    return updated


def _diff_window(old: str, new: str | None, width: int = 90) -> tuple[str, str]:
    """Wycinek wokol pierwszej roznicy, bo opisy sa dlugie, a zmiana bywa w srodku."""
    new = new or ""
    i = next((k for k, (a, b) in enumerate(zip(old, new)) if a != b), min(len(old), len(new)))
    start = max(0, i - width // 3)
    return old[start : start + width], new[start : start + width]


def _print_title_sample(table: str, plan: list[dict], n: int) -> None:
    for item in random.sample(plan, min(n, len(plan))):
        print(f"[{table}.title] {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}")


def _print_description_sample(table: str, plan: list[dict], n: int) -> None:
    for item in random.sample(plan, min(n, len(plan))):
        old, new = _diff_window(item["old_description"], item["description"])
        print(f"[{table}.description] {item['id']}")
        print(f"  ...{old!r}")
        print(f"  ...{new!r}")


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("--only", choices=("titles", "descriptions"), default=None)
    p.add_argument("--batch-size", type=int, default=500)
    p.add_argument("--sample", type=int, default=20, help="ile losowych zmian pokazac na przebieg")
    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")

    passes = []
    if args.only in (None, "titles"):
        passes.append((_plan_titles, _print_title_sample, _apply_titles))
    if args.only in (None, "descriptions"):
        passes.append((_plan_descriptions, _print_description_sample, _apply_descriptions))

    total_planned = total_updated = 0
    for plan_fn, sample_fn, apply_fn in passes:
        for table in _TABLES:
            plan = plan_fn(table, limit=args.limit)
            if not plan:
                continue
            total_planned += len(plan)
            sample_fn(table, plan, args.sample)
            sys.stdout.flush()
            if not args.dry_run:
                total_updated += apply_fn(table, plan, batch_size=args.batch_size)

    if args.dry_run:
        log.info("DRY RUN: %d zmian w planie, 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())