scripts/audit_data_quality.py

from __future__ import annotations

import json
import os
import sqlite3
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
DB_PATH = Path(os.environ["KINK_STORE_PATH"]) if os.environ.get("KINK_STORE_PATH") else ROOT / "data" / "store.db"


def scalar(conn: sqlite3.Connection, query: str, params: tuple = ()):
    row = conn.execute(query, params).fetchone()
    return row[0] if row else 0


def rows(conn: sqlite3.Connection, query: str, params: tuple = ()):
    cols = [d[0] for d in conn.execute(query, params).description]
    return [dict(zip(cols, row)) for row in conn.execute(query, params).fetchall()]


def main() -> int:
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row

    report: dict[str, object] = {
        "database": str(DB_PATH),
        "counts": {
            "kinks": scalar(conn, "select count(*) from kink"),
            "assets": scalar(conn, "select count(*) from asset"),
            "picture_refs": scalar(conn, "select count(*) from fetlifepictureref"),
            "similarity_edges": scalar(conn, "select count(*) from similarityedge"),
            "play_preferences": scalar(conn, "select count(*) from playpreference"),
        },
        "issues": [],
    }

    orphan_assets = scalar(conn, """
        select count(*)
        from asset a
        left join kink k on k.id = a.kink_id
        where k.id is null
    """)
    orphan_picture_refs = scalar(conn, """
        select count(*)
        from fetlifepictureref f
        left join kink k on k.id = f.kink_id
        where k.id is null
    """)
    broken_similarity_edges = scalar(conn, """
        select count(*)
        from similarityedge s
        left join kink lk on lk.id = s.left_kink_id
        left join kink rk on rk.id = s.right_kink_id
        where lk.id is null or rk.id is null
    """)
    role_content_mismatch = rows(conn, """
        select k.id as kink_id, k.name, k.cluster, kc.content_kind
        from kink k
        left join kinkcontenttype kc on kc.kink_id = k.id
        where k.cluster in ('roles', 'dynamics', 'positions')
          and (kc.content_kind is null or kc.content_kind != 'role')
        order by k.name asc
        limit 10
    """)
    top_popular_with_assets = rows(conn, """
        select
          m.kink_id,
          k.name,
          m.popularity,
          m.has_real_images,
          (select count(*) from asset a where a.kink_id = m.kink_id) as asset_count,
          (select count(*) from fetlifepictureref f where f.kink_id = m.kink_id) as picture_ref_count
        from fetlifekinkmeta m
        left join kink k on k.id = m.kink_id
        order by m.popularity desc
        limit 15
    """)

    if orphan_assets:
        report["issues"].append({"kind": "orphan_assets", "count": orphan_assets})
    if orphan_picture_refs:
        report["issues"].append({"kind": "orphan_picture_refs", "count": orphan_picture_refs})
    if broken_similarity_edges:
        report["issues"].append({"kind": "broken_similarity_edges", "count": broken_similarity_edges})
    if role_content_mismatch:
        report["issues"].append({"kind": "role_content_mismatch", "count": len(role_content_mismatch), "sample": role_content_mismatch})

    report["top_popular_with_assets"] = top_popular_with_assets
    print(json.dumps(report, indent=2))
    return 0


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