backend/scrape_artifacts.py

"""Scraped catalog text classifiers.

The FetLife imports historically stored UI chrome in ``kink.notes``: profile
bucket labels, popularity counters, repeated titles, and Kinktionary section
headings.  These are useful as audit evidence, but not as user-facing copy.
"""
from __future__ import annotations

import re
from collections import Counter, defaultdict
from typing import Any

PROFILE_BUCKET_KEYS = frozenset(
    {
        "into",
        "curious",
        "curious_about",
        "hard_limits",
        "soft_limits",
    },
)
PROFILE_BUCKET_VALUES = frozenset(
    {
        "everything to do with it",
        "giving",
        "none",
        "receiving",
        "watching",
        "watching others wear",
        "wearing",
    },
)
FETLIFE_METRIC_KEYS = frozenset(
    {
        "kinksters",
        "lists",
        "pictures",
        "related groups",
        "similar fetishes",
        "statuses",
        "videos",
        "writings",
    },
)
ARTIFACT_CATEGORIES = (
    "profile_bucket_line",
    "fetlife_metric_line",
    "fetlife_title_popularity_line",
    "kinktionary_toc_note",
    "title_echo_line",
)

_COUNT_VALUE_RE = re.compile(r"^[\d,.]+\s*[kmb]?$", re.IGNORECASE)
_KINKSTERS_ONLY_RE = re.compile(
    r"^[\d,.]+\s*[kmb]?\s+kinksters?(?:\s+into(?:\s+and\s+curious)?)?$",
    re.IGNORECASE,
)


def _collapse_spaces(text: str) -> str:
    return " ".join(str(text or "").split())


def _line_key(text: str) -> str:
    return _collapse_spaces(text).casefold()


def _split_label_value(text: str) -> tuple[str, str] | None:
    if ":" not in text:
        return None
    label, value = text.split(":", 1)
    return label.strip().casefold(), value.strip().casefold()


def _strip_title_echo(kink_name: str, line: str) -> str:
    title = _collapse_spaces(kink_name)
    text = _collapse_spaces(line)
    if not title or not text:
        return text
    stripped = re.sub(r"^" + re.escape(title) + r"\s*", "", text, flags=re.IGNORECASE)
    return stripped.strip(" \t-:|,.!?")


def classify_scraped_note_line(kink_id: str, kink_name: str, line: str) -> str:
    """Return the artifact category for a notes line, or ``""`` if it is real text."""
    text = _collapse_spaces(line)
    if not text:
        return ""

    if str(kink_id).startswith("fetlife_kinktionary_"):
        return "kinktionary_toc_note"

    key = text.casefold()
    if key == _line_key(kink_name):
        return "title_echo_line"
    if key in PROFILE_BUCKET_KEYS:
        return "profile_bucket_line"

    label_value = _split_label_value(text)
    if label_value is not None:
        label, value = label_value
        if label in PROFILE_BUCKET_KEYS and value in PROFILE_BUCKET_VALUES:
            return "profile_bucket_line"
        if label in FETLIFE_METRIC_KEYS and _COUNT_VALUE_RE.fullmatch(value):
            return "fetlife_metric_line"

    if _KINKSTERS_ONLY_RE.fullmatch(_strip_title_echo(kink_name, text)):
        return "fetlife_title_popularity_line"

    return ""


def clean_scraped_notes(kink_id: str, kink_name: str, notes: str) -> str:
    """Remove known scrape leftovers from ``kink.notes`` while preserving real future notes."""
    kept = []
    for raw_line in str(notes or "").replace("\r", "").splitlines():
        line = raw_line.strip()
        if not line:
            continue
        if classify_scraped_note_line(kink_id, kink_name, line):
            continue
        kept.append(line)
    return "\n".join(kept)


def clean_scraped_catalog_text(kink_name: str, text: str) -> str:
    """Remove cross-field scrape leftovers without treating Kinktionary prose as TOC notes."""
    kept = []
    for raw_line in str(text or "").replace("\r", "").splitlines():
        line = raw_line.strip()
        if not line:
            continue
        if classify_scraped_note_line("", kink_name, line):
            continue
        kept.append(line)
    return "\n".join(kept)


def audit_scraped_notes(rows: list[tuple[str, str, str]], *, sample_limit: int = 8) -> dict[str, Any]:
    """Classify notes rows for audit scripts and tests."""
    line_counts: Counter[str] = Counter()
    row_counts: Counter[str] = Counter()
    samples: dict[str, list[dict[str, str]]] = defaultdict(list)
    nonempty_rows = 0
    artifact_rows = 0
    remaining_rows_after_filter = 0

    for kink_id, kink_name, raw_notes in rows:
        notes = str(raw_notes or "")
        if not notes.strip():
            continue
        nonempty_rows += 1
        row_categories: set[str] = set()
        for raw_line in notes.replace("\r", "").splitlines():
            category = classify_scraped_note_line(kink_id, 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": str(kink_id),
                        "name": str(kink_name),
                        "line": _collapse_spaces(raw_line),
                    },
                )
        if row_categories:
            artifact_rows += 1
        for category in row_categories:
            row_counts[category] += 1
        if clean_scraped_notes(kink_id, kink_name, notes):
            remaining_rows_after_filter += 1

    return {
        "nonempty_rows": nonempty_rows,
        "artifact_rows": artifact_rows,
        "remaining_rows_after_filter": remaining_rows_after_filter,
        "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),
    }