scripts/audit_scrape_leftovers.py

#!/usr/bin/env python3
"""Catalog known scrape leftovers in kink text fields."""
from __future__ import annotations

import argparse
import json
import sqlite3
from collections import Counter, defaultdict
from pathlib import Path
from typing import Any

from backend.scrape_artifacts import (
    ARTIFACT_CATEGORIES,
    audit_scraped_notes,
    classify_scraped_note_line,
)


def _fetch_notes_rows(conn: sqlite3.Connection) -> list[tuple[str, str, str]]:
    rows = conn.execute(
        "select id, name, notes from kink where coalesce(notes, '') != '' order by id",
    ).fetchall()
    return [(str(row[0]), str(row[1]), str(row[2] or "")) for row in rows]


def _scan_text_rows(
    rows: list[tuple[str, str, str]],
    *,
    sample_limit: int,
) -> dict[str, Any]:
    line_counts: Counter[str] = Counter()
    row_counts: Counter[str] = Counter()
    samples: dict[str, list[dict[str, str]]] = defaultdict(list)
    for kink_id, kink_name, text in rows:
        row_categories = set()
        for raw_line in str(text or "").replace("\r", "").splitlines():
            category = classify_scraped_note_line("", kink_name, raw_line)
            if not category:
                continue
            line_counts[category] += 1
            row_categories.add(category)
            if len(samples[category]) < sample_limit:
                samples[category].append(
                    {
                        "id": kink_id,
                        "name": kink_name,
                        "line": " ".join(raw_line.split()),
                    },
                )
        for category in row_categories:
            row_counts[category] += 1
    return {
        "line_counts": {category: line_counts[category] for category in ARTIFACT_CATEGORIES},
        "row_counts": {category: row_counts[category] for category in ARTIFACT_CATEGORIES},
        "samples": dict(samples),
    }


def _other_text_fields(conn: sqlite3.Connection, *, sample_limit: int) -> dict[str, Any]:
    fields = {
        "kink.short_definition": """
            select id, name, short_definition
            from kink
            where coalesce(short_definition, '') != ''
            order by id
        """,
        "definition.text": """
            select k.id, k.name, d.text
            from definition d
            join kink k on k.id = d.kink_id
            where coalesce(d.text, '') != ''
            order by k.id, d.id
        """,
        "kinkexample.text": """
            select k.id, k.name, e.text
            from kinkexample e
            join kink k on k.id = e.kink_id
            where coalesce(e.text, '') != ''
            order by k.id, e.id
        """,
    }
    out = {}
    for field_name, sql in fields.items():
        rows = [(str(row[0]), str(row[1]), str(row[2] or "")) for row in conn.execute(sql).fetchall()]
        out[field_name] = _scan_text_rows(rows, sample_limit=sample_limit)
    return out


def audit_database(path: Path, *, sample_limit: int = 8) -> dict[str, Any]:
    with sqlite3.connect(path) as conn:
        total_kinks = int(conn.execute("select count(*) from kink").fetchone()[0])
        notes = audit_scraped_notes(_fetch_notes_rows(conn), sample_limit=sample_limit)
        other_fields = _other_text_fields(conn, sample_limit=sample_limit)
    return {
        "db": str(path),
        "total_kinks": total_kinks,
        "notes": notes,
        "other_text_fields": other_fields,
    }


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("db", nargs="*", type=Path, default=[Path("data/store_slim.db")])
    parser.add_argument("--json-out", type=Path, default=None)
    parser.add_argument("--sample-limit", type=int, default=8)
    args = parser.parse_args()

    payload = {
        "databases": [audit_database(path, sample_limit=args.sample_limit) for path in args.db],
    }
    rendered = json.dumps(payload, indent=2, sort_keys=True)
    if args.json_out:
        args.json_out.parent.mkdir(parents=True, exist_ok=True)
        args.json_out.write_text(rendered + "\n", encoding="utf-8")
    print(rendered)
    return 0


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