scripts/prune_noise.py

#!/usr/bin/env python3
"""Remove zero-popularity noise kinks that have no edges, no preferences, and unclean names."""
from __future__ import annotations

import re
import sqlite3
import sys
from pathlib import Path

DB_PATH = Path(__file__).resolve().parent.parent / "data" / "store.db"

MAX_NAME_WORDS = 6
MAX_NAME_CHARS = 50
PUNCT_REGEX = re.compile(r"[^\w\s/&-]", re.UNICODE)


def name_is_clean(name: str) -> bool:
    stripped = name.strip()
    if not stripped:
        return False
    word_count = len(stripped.split())
    if word_count > MAX_NAME_WORDS or len(stripped) > MAX_NAME_CHARS:
        return False
    if PUNCT_REGEX.search(stripped):
        return False
    if any(char.isdigit() for char in stripped):
        return False
    return True


def main():
    if not DB_PATH.exists():
        print(f"Database not found: {DB_PATH}")
        sys.exit(1)

    conn = sqlite3.connect(DB_PATH, timeout=30)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA busy_timeout=30000")
    conn.execute("PRAGMA foreign_keys=ON")

    before = conn.execute("SELECT COUNT(*) AS n FROM kink").fetchone()["n"]
    print(f"Kinks before: {before}")

    zero_pop = conn.execute("""
        SELECT k.id, k.name FROM kink k
        LEFT JOIN fetlifekinkmeta m ON k.id = m.kink_id
        WHERE COALESCE(m.popularity, 0) = 0
          AND k.id NOT IN (SELECT DISTINCT left_kink_id FROM similarityedge)
          AND k.id NOT IN (SELECT DISTINCT right_kink_id FROM similarityedge)
          AND k.id NOT IN (SELECT DISTINCT kink_id FROM playpreference)
    """).fetchall()

    ids_to_delete = []
    for row in zero_pop:
        if not name_is_clean(row["name"]):
            ids_to_delete.append(row["id"])

    print(f"Zero-pop candidates: {len(zero_pop)}")
    print(f"Failing name_is_clean: {len(ids_to_delete)}")

    if not ids_to_delete:
        print("Nothing to prune.")
        conn.close()
        return

    batch_size = 500
    for i in range(0, len(ids_to_delete), batch_size):
        batch = ids_to_delete[i : i + batch_size]
        placeholders = ",".join("?" * len(batch))
        conn.execute(f"DELETE FROM alias WHERE kink_id IN ({placeholders})", batch)
        conn.execute(f"DELETE FROM definition WHERE kink_id IN ({placeholders})", batch)
        conn.execute(f"DELETE FROM kinkexample WHERE kink_id IN ({placeholders})", batch)
        conn.execute(f"DELETE FROM asset WHERE kink_id IN ({placeholders})", batch)
        conn.execute(f"DELETE FROM kinkfacet WHERE kink_id IN ({placeholders})", batch)
        conn.execute(f"DELETE FROM kinkcontenttype WHERE kink_id IN ({placeholders})", batch)
        conn.execute(f"DELETE FROM fetlifepictureref WHERE kink_id IN ({placeholders})", batch)
        conn.execute(f"DELETE FROM fetlifekinkmeta WHERE kink_id IN ({placeholders})", batch)
        conn.execute(f"DELETE FROM kink WHERE id IN ({placeholders})", batch)
        if (i + batch_size) % 5000 == 0:
            conn.commit()
            print(f"  deleted {min(i + batch_size, len(ids_to_delete))} / {len(ids_to_delete)}")

    conn.commit()

    after = conn.execute("SELECT COUNT(*) AS n FROM kink").fetchone()["n"]
    print(f"Kinks after: {after} (removed {before - after})")

    print("Rebuilding FTS index...")
    conn.execute("DELETE FROM play_fts")
    conn.commit()
    print("FTS cleared (will rebuild on next catalog load).")

    conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
    conn.close()
    print("Done.")


if __name__ == "__main__":
    main()