backend/content_types.py

"""Backend content types service."""
from __future__ import annotations

from collections import defaultdict
from datetime import UTC, datetime
from sqlmodel import Session, select
from models import Kink, KinkContentType, KinkTypeOverride
from backend.constants import (DEFAULT_TYPE_OVERRIDES, DISCOVERABLE_CLUSTERS, DYNAMIC_CLUSTERS,
    EXACT_DYNAMIC_NAMES, EXACT_ROLE_NAMES, FETLIFE_DYNAMIC_KEYWORDS, META_CLUSTERS,
    PLAY_OVERRIDE_NAMES, ROLE_CLUSTERS, STARTER_CLUSTERS)



def _seed_type_overrides(self) -> None:
    with Session(self.engine) as session:
        for normalized_name, cluster, content_kind, note in DEFAULT_TYPE_OVERRIDES:
            override_id = f"type_override_{cluster}_{self.slugify(normalized_name)}"
            if session.get(KinkTypeOverride, override_id):
                continue
            session.add(
                KinkTypeOverride(
                    id=override_id,
                    normalized_name=normalized_name,
                    cluster=cluster,
                    content_kind=content_kind,
                    note=note,
                )
            )
        session.commit()

def _derive_content_kind(self, name: str, cluster: str, override_map: dict[tuple[str, str], str] | None = None) -> tuple[str, str]:
    lowered_name = name.strip().lower()
    if override_map:
        if (lowered_name, cluster) in override_map:
            return override_map[(lowered_name, cluster)], "override:name+cluster"
        if (lowered_name, "*") in override_map:
            return override_map[(lowered_name, "*")], "override:name"
    if cluster in ROLE_CLUSTERS:
        return "role", "cluster:role"
    if cluster in DYNAMIC_CLUSTERS:
        return "dynamic", "cluster:dynamic"
    if cluster == "fetlife_fetish":
        if lowered_name in EXACT_ROLE_NAMES:
            return "dynamic", "fetlife:name:roleish_dynamic"
        if lowered_name in EXACT_DYNAMIC_NAMES:
            return "dynamic", "fetlife:name:dynamic"
        if any(keyword in lowered_name for keyword in FETLIFE_DYNAMIC_KEYWORDS):
            return "dynamic", "fetlife:name:dynamic_keyword"
        return "play", "cluster:fetlife_fetish"
    if cluster in META_CLUSTERS:
        return "meta", "cluster:meta"
    if lowered_name in PLAY_OVERRIDE_NAMES:
        return "play", "allowlist:name"
    if lowered_name in EXACT_ROLE_NAMES:
        return "role", "allowlist:role_name"
    if cluster in DISCOVERABLE_CLUSTERS or cluster in STARTER_CLUSTERS:
        return "play", "cluster:discoverable"
    return "meta", "fallback:meta"

def _sync_content_types(self) -> dict[str, dict[str, str]]:
    with Session(self.engine) as session:
        overrides = session.exec(select(KinkTypeOverride)).all()
        override_map = {
            (row.normalized_name.strip().lower(), row.cluster.strip() or "*"): row.content_kind
            for row in overrides
        }
        kinks = session.exec(select(Kink)).all()
    rows_to_write = []
    for kink in kinks:
        content_kind, evidence = self._derive_content_kind(kink.name, kink.cluster, override_map)
        rows_to_write.append((kink.id, content_kind, evidence, self._now_iso()))
    with self._sqlite() as conn:
        conn.executemany(
            """
            INSERT INTO kinkcontenttype (kink_id, content_kind, evidence, updated_at)
            VALUES (?, ?, ?, ?)
            ON CONFLICT(kink_id) DO UPDATE SET
                content_kind = excluded.content_kind,
                evidence = excluded.evidence,
                updated_at = excluded.updated_at
            """,
            rows_to_write,
        )
        conn.commit()
        stored = conn.execute(
            "SELECT kink_id, content_kind, evidence FROM kinkcontenttype"
        ).fetchall()
    return {
        row["kink_id"]: {
            "content_kind": row["content_kind"],
            "evidence": row["evidence"],
        }
        for row in stored
    }

def _content_type_map(self, *, sync: bool = False) -> dict[str, dict[str, str]]:
    if sync:
        return self._sync_content_types()
    with Session(self.engine) as session:
        rows = session.exec(select(KinkContentType)).all()
        if rows:
            return {
                row.kink_id: {
                    "content_kind": row.content_kind,
                    "evidence": row.evidence,
                }
                for row in rows
            }
    return {}

def _content_kind(self, item: dict[str, Any]) -> str:
    if item.get("content_kind"):
        return str(item["content_kind"])
    cluster = item.get("cluster", "")
    lowered_name = str(item.get("name", "")).strip().lower()
    return self._derive_content_kind(lowered_name, cluster)[0]