shira-hermes/api/services/kb/labels.py

"""kb_label CRUD + many-to-many helpers for kb_source_label.

Phase 6 (Task #20). Backs the v0.8.0 review-screen label chips and the
new 'תוויות' admin tab in the KnowledgeBase EspoCRM extension. A label is
a free-form sub-topic — typical names are "ילד נכה / אוטיזם" or
"נפגעי עבודה / שמיעה" — that helps end users filter the KB beyond the
coarse-grained `topic`. Labels are global; `topic_id` is an optional
hint that scopes a label to one topic for the autocomplete dropdown.

Design choices documented in scripts/migrations/006_labels.sql and
~/.claude/plans/hidden-tickling-ullman.md:
  • Flat schema (no parent_id). Slash separators in names are a UI
    display convention only.
  • ASCII slug is the unique key. `slugify` transliterates Hebrew
    letter-by-letter so the slug is deterministic without external deps.
  • `usage_count` denormalized — the autocomplete dropdown is hot, and
    we don't want a `COUNT(*)` join on every keystroke.
  • Auto-creation happens at commit time with a clean
    INSERT...ON CONFLICT DO NOTHING pattern, race-safe across concurrent
    uploads.
"""

from __future__ import annotations

import logging
import re
from typing import Optional

from api.services.kb.db import get_pool

logger = logging.getLogger("shira.kb.labels")


# Letter-by-letter Hebrew → Latin map. We're not aiming for phonetic
# accuracy (that's a research project) — the goal is a deterministic,
# reversible-enough mapping that produces unique ASCII slugs and reads
# OK in URLs. Final/medial letter forms map identically to their
# canonical form. Niqqud is dropped (handled below in slugify).
_HEB_LATIN = {
    "א": "a", "ב": "b", "ג": "g", "ד": "d", "ה": "h",
    "ו": "v", "ז": "z", "ח": "ch", "ט": "t", "י": "y",
    "כ": "k", "ך": "k", "ל": "l", "מ": "m", "ם": "m",
    "נ": "n", "ן": "n", "ס": "s", "ע": "a", "פ": "p",
    "ף": "f", "צ": "tz", "ץ": "tz", "ק": "q", "ר": "r",
    "ש": "sh", "ת": "t",
}

# Niqqud + cantillation marks (U+0591..U+05C7) — strip before mapping.
_NIQQUD_RE = re.compile(r"[֑-ׇ]")


def slugify(name: str) -> str:
    """Hebrew/Latin name → URL-safe slug matching `^[a-z0-9][a-z0-9-]*$`.

    Examples (verified):
      "ילד נכה / אוטיזם"            → "yld-nkh-aoytyzm"
      "נפגעי עבודה / שמיעה"        → "nfgay-aabvdh-shmyah"
      "Employment-Law (2024)"      → "employment-law-2024"
      ""                            → ValueError

    The transliteration isn't pretty Hebrew but it's deterministic.
    Two callers naming the same concept will collide on the same slug,
    which is what we want.
    """
    if not name:
        raise ValueError("name is required")
    # Strip niqqud / cantillation, lowercase ASCII, then transliterate
    # any remaining Hebrew letter to its Latin equivalent.
    cleaned = _NIQQUD_RE.sub("", name).lower()
    out_chars: list[str] = []
    for ch in cleaned:
        if ch in _HEB_LATIN:
            out_chars.append(_HEB_LATIN[ch])
        elif "a" <= ch <= "z" or "0" <= ch <= "9":
            out_chars.append(ch)
        else:
            # Spaces, punctuation, slashes, emoji — all become a single
            # hyphen. Multiple consecutive hyphens collapse below.
            out_chars.append("-")
    raw = "".join(out_chars)
    # Collapse runs of hyphens, trim ends.
    raw = re.sub(r"-+", "-", raw).strip("-")
    if not raw:
        raise ValueError(f"name {name!r} transliterates to an empty slug")
    if len(raw) > 200:
        raw = raw[:200].rstrip("-")
    # The CHECK constraint forbids leading hyphen / digit-then-hyphen
    # gymnastics — start char must be [a-z0-9].
    if not re.match(r"^[a-z0-9]", raw):
        raise ValueError(f"slug must start with [a-z0-9], got {raw!r}")
    return raw


# ── Read paths ──────────────────────────────────────────────────────────────

async def list_labels(
    *,
    topic_id: Optional[int] = None,
    query: Optional[str] = None,
    limit: int = 50,
) -> list[dict]:
    """Autocomplete-friendly listing. ORDER BY usage_count DESC.

    Filters:
      • topic_id: include labels with that topic_id OR NULL (cross-topic
        labels are visible everywhere).
      • query: case-insensitive substring on name or slug.
    """
    limit = max(1, min(200, limit))
    where: list[str] = []
    args: list = []
    if topic_id is not None:
        args.append(topic_id)
        where.append(f"(topic_id = ${len(args)} OR topic_id IS NULL)")
    if query:
        args.append(f"%{query}%")
        where.append(f"(name ILIKE ${len(args)} OR slug ILIKE ${len(args)})")
    where_sql = ("WHERE " + " AND ".join(where)) if where else ""
    args.append(limit)
    sql = f"""
        SELECT id, slug, name, topic_id, usage_count, created_at
        FROM kb_label
        {where_sql}
        ORDER BY usage_count DESC, name
        LIMIT ${len(args)}
    """
    pool = await get_pool()
    async with pool.acquire() as conn:
        rows = await conn.fetch(sql, *args)
    return [_row_to_dict(r) for r in rows]


async def get_label(label_id: int) -> Optional[dict]:
    pool = await get_pool()
    async with pool.acquire() as conn:
        row = await conn.fetchrow(
            "SELECT * FROM kb_label WHERE id = $1", label_id
        )
    return _row_to_dict(row) if row else None


async def get_labels_for_source(source_id: int) -> list[dict]:
    """Labels attached to one source, ordered by name."""
    pool = await get_pool()
    async with pool.acquire() as conn:
        rows = await conn.fetch(
            """
            SELECT l.id, l.slug, l.name, l.topic_id, l.usage_count
            FROM kb_label l
            JOIN kb_source_label sl ON sl.label_id = l.id
            WHERE sl.source_id = $1
            ORDER BY l.name
            """,
            source_id,
        )
    return [_row_to_dict(r) for r in rows]


# ── Write paths ─────────────────────────────────────────────────────────────

async def create_label(
    *,
    slug: str,
    name: str,
    topic_id: Optional[int] = None,
    created_by: Optional[str] = None,
) -> dict:
    """Insert one label. Returns the row. Raises ValueError on dup slug."""
    if not slug:
        raise ValueError("slug is required")
    if not name:
        raise ValueError("name is required")
    pool = await get_pool()
    async with pool.acquire() as conn:
        try:
            row = await conn.fetchrow(
                """
                INSERT INTO kb_label (slug, name, topic_id, created_by)
                VALUES ($1, $2, $3, $4)
                RETURNING id, slug, name, topic_id, usage_count, created_at
                """,
                slug, name, topic_id, created_by,
            )
        except Exception as e:
            # asyncpg raises UniqueViolationError or CheckViolationError;
            # we surface them as ValueError so the route can return 400.
            msg = str(e).lower()
            if "unique" in msg or "duplicate" in msg:
                raise ValueError(f"slug {slug!r} already exists")
            if "kb_label_slug_check" in msg:
                raise ValueError(f"slug {slug!r} fails shape check (lowercase ASCII + hyphens)")
            raise
    logger.info("[labels] created id=%s slug=%s", row["id"], row["slug"])
    return _row_to_dict(row)


async def lookup_or_create_batch(
    items: list[dict],
    *,
    created_by: Optional[str] = None,
) -> dict[str, int]:
    """For a list of {slug, name, topic_id?} dicts, return {slug → label_id}.

    Race-safe: each row is INSERT...ON CONFLICT DO NOTHING + a follow-up
    SELECT for the loser of the race. Used by the commit path to resolve
    user-confirmed labels (some existing, some new) in a single call.
    """
    if not items:
        return {}
    pool = await get_pool()
    out: dict[str, int] = {}
    async with pool.acquire() as conn:
        for item in items:
            slug = item.get("slug") or ""
            name = item.get("name") or ""
            topic_id = item.get("topic_id")
            if not slug or not name:
                continue
            row = await conn.fetchrow(
                """
                INSERT INTO kb_label (slug, name, topic_id, created_by)
                VALUES ($1, $2, $3, $4)
                ON CONFLICT (slug) DO NOTHING
                RETURNING id
                """,
                slug, name, topic_id, created_by,
            )
            if row is None:
                # Lost the race; the row already exists. Re-fetch.
                row = await conn.fetchrow(
                    "SELECT id FROM kb_label WHERE slug = $1", slug
                )
            if row is None:
                # Genuinely impossible — slug check failed somewhere.
                logger.error("[labels] failed to resolve slug=%s", slug)
                continue
            out[slug] = row["id"]
    return out


async def apply_to_source(source_id: int, label_ids: list[int]) -> int:
    """Attach labels to a source. Increments usage_count for each NEW
    assignment (skips already-assigned). Returns count of new attachments.
    """
    if not label_ids:
        return 0
    pool = await get_pool()
    new_count = 0
    async with pool.acquire() as conn:
        async with conn.transaction():
            for lid in label_ids:
                inserted = await conn.fetchrow(
                    """
                    INSERT INTO kb_source_label (source_id, label_id)
                    VALUES ($1, $2)
                    ON CONFLICT DO NOTHING
                    RETURNING source_id
                    """,
                    source_id, lid,
                )
                if inserted:
                    new_count += 1
                    await conn.execute(
                        "UPDATE kb_label SET usage_count = usage_count + 1 WHERE id = $1",
                        lid,
                    )
    return new_count


async def replace_for_source(source_id: int, label_ids: list[int]) -> dict:
    """Set the label set of a source to exactly `label_ids`. Adjusts
    usage_count for both adds and removes. Used by the source edit path
    in admin_sources.update_source.
    """
    pool = await get_pool()
    async with pool.acquire() as conn:
        async with conn.transaction():
            existing = await conn.fetch(
                "SELECT label_id FROM kb_source_label WHERE source_id = $1",
                source_id,
            )
            current = {r["label_id"] for r in existing}
            target = set(label_ids)
            to_add = target - current
            to_remove = current - target
            for lid in to_remove:
                await conn.execute(
                    "DELETE FROM kb_source_label WHERE source_id = $1 AND label_id = $2",
                    source_id, lid,
                )
                await conn.execute(
                    "UPDATE kb_label SET usage_count = GREATEST(0, usage_count - 1) WHERE id = $1",
                    lid,
                )
            for lid in to_add:
                inserted = await conn.fetchrow(
                    """
                    INSERT INTO kb_source_label (source_id, label_id)
                    VALUES ($1, $2)
                    ON CONFLICT DO NOTHING
                    RETURNING source_id
                    """,
                    source_id, lid,
                )
                if inserted:
                    await conn.execute(
                        "UPDATE kb_label SET usage_count = usage_count + 1 WHERE id = $1",
                        lid,
                    )
    return {"added": len(to_add), "removed": len(to_remove)}


async def merge_labels(from_id: int, into_id: int) -> dict:
    """Redirect every assignment of `from_id` to `into_id`, then delete
    the source label. Used to clean up duplicates created by typos.
    Race-safe: ON CONFLICT DO NOTHING handles assignments that already
    exist on `into_id`.
    """
    if from_id == into_id:
        raise ValueError("cannot merge a label into itself")
    pool = await get_pool()
    async with pool.acquire() as conn:
        async with conn.transaction():
            from_label = await conn.fetchrow(
                "SELECT id FROM kb_label WHERE id = $1", from_id
            )
            into_label = await conn.fetchrow(
                "SELECT id FROM kb_label WHERE id = $1", into_id
            )
            if not from_label or not into_label:
                raise LookupError("one or both labels not found")
            # Move assignments. ON CONFLICT skips sources already tagged
            # with `into_id` — they end up tagged once (correct).
            await conn.execute(
                """
                INSERT INTO kb_source_label (source_id, label_id)
                SELECT source_id, $2 FROM kb_source_label WHERE label_id = $1
                ON CONFLICT DO NOTHING
                """,
                from_id, into_id,
            )
            await conn.execute(
                "DELETE FROM kb_source_label WHERE label_id = $1", from_id
            )
            # Recompute usage_count for `into_id` (the joiner won the
            # union; some assignments may have been deduped).
            await conn.execute(
                """
                UPDATE kb_label SET usage_count = (
                    SELECT COUNT(*) FROM kb_source_label WHERE label_id = $1
                ) WHERE id = $1
                """,
                into_id,
            )
            await conn.execute("DELETE FROM kb_label WHERE id = $1", from_id)
    logger.info("[labels] merged id=%s into id=%s", from_id, into_id)
    return {"merged": True, "from_id": from_id, "into_id": into_id}


async def delete_label(label_id: int) -> dict:
    """Hard-delete. Refuses if any source still references the label.

    Soft-removal (usage_count → 0 via the source-edit UI) is the path
    callers should use to retire a label without affecting historical
    sources. This DELETE is for cleaning out abandoned labels.
    """
    pool = await get_pool()
    async with pool.acquire() as conn:
        cnt = await conn.fetchval(
            "SELECT COUNT(*) FROM kb_source_label WHERE label_id = $1", label_id
        )
        if cnt and cnt > 0:
            raise ValueError(f"label still has {cnt} source(s); merge or detach first")
        result = await conn.execute(
            "DELETE FROM kb_label WHERE id = $1", label_id
        )
    if not result.endswith(" 1"):
        raise LookupError(f"label {label_id} not found")
    return {"deleted": True, "label_id": label_id}


# ── helpers ─────────────────────────────────────────────────────────────────

def _row_to_dict(row) -> Optional[dict]:
    if row is None:
        return None
    out: dict = {}
    for k, v in dict(row).items():
        # asyncpg gives us datetime objects; the API serializer expects ISO.
        try:
            from datetime import datetime
            if isinstance(v, datetime):
                out[k] = v.isoformat()
                continue
        except Exception:
            pass
        out[k] = v
    return out