SlabUploader/backend/prompts.py

"""Prompt revision REST (METADATA-106): append/activate/get for call1/call2.

Seed prompts live in inference (CALL1_SEED / CALL2_SEED); the shipped seed is
activated when nothing is active or the active revision is still seed-authored,
so deploys pick up seed updates without clobbering an owner revision.
"""
from __future__ import annotations

import hashlib
from datetime import datetime, timezone
from typing import Any

from fastapi import APIRouter
from fastapi.responses import JSONResponse

from .db import get_connection

router = APIRouter(prefix="/api/v1")

# Locked Call 1 seed (#86 / METADATA-40): expert helper + closed Woo lists only.
CALL1_SEED = (
    "You are a wood identification expert for retail lumber slabs. Judge only from the photos "
    "and the LISTS below. Use list ids only — never invent a name or id.\n"
    "\n"
    "Work in this order: (1) species, (2) wood categories, (3) edge type, (4) figure, "
    "(5) feature tags. Decide wood categories before edge type — edge rules depend on product form.\n"
    "\n"
    "For each field, pick from its list and give a confidence 0–1 (1 = certain, 0 = no idea). "
    "Prefer null or [] over a weak guess.\n"
    "\n"
    "Species — which tree species is this slab. Match grain, color, and figure to the species "
    "list. If more than one entry is plausible, return only the single highest-confidence match "
    "(one id). Null only when none are credible.\n"
    "\n"
    "Wood categories — product form / how it will be sold (live edge, dimensional, cookies & "
    "burl slices, turning blanks, etc.). Choose every wood_categories id that clearly fits; "
    "often one, sometimes more. [] if none fit. Finish this field before edge type.\n"
    "\n"
    "Edge type — the long-edge mill profile. One id from edge_terms, or null only if none fit.\n"
    "  Meanings (match the closest list name):\n"
    "  • Square / square edge — both long edges milled straight (2 milled long edges).\n"
    "  • Live one side / Live Edge 1 side — one long edge milled, the other natural bark or "
    "waney (1 milled long edge).\n"
    "  • Live both / Live edge 2 sides — neither long edge milled; both natural "
    "(0 milled long edges).\n"
    "  • Wild edge — use when the three mill-count cases do not fit, including cookies and burl "
    "cuts/slices (no conventional slab long-edge pair). Prefer Wild for those forms rather than null.\n"
    "\n"
    "Figure — visible figure in the wood (curl, quilt, spalt, plain, burl, …). Zero or more ids "
    "from figure_terms that clearly apply.\n"
    "\n"
    "Feature tags — defects and character marks the store tags for shoppers (knots, checks, "
    "inclusions, …). Zero or more ids from feature_tags. If that list is empty, return [].\n"
    "\n"
    "Return one JSON object with exactly these keys:\n"
    '{"species_id": <id|null>, "species_confidence": <0-1>, '
    '"wood_category_ids": [<id>,...], "wood_category_confidence": <0-1>, '
    '"edge_type_term_id": <id|null>, "edge_confidence": <0-1>, '
    '"figure_term_ids": [<id>,...], "figure_confidence": <0-1>, '
    '"feature_tag_ids": [<id>,...], "feature_tag_confidence": <0-1>}'
)
CALL2_SEED = (
    "Write the long product description for the slab from the supplied photos and listing "
    "facts. The name and short description are already built, so do not repeat them. Use the "
    "brand voice and geo context given. The dimensions in the fact block are exact: use those "
    "strings and invent no numbers. Write prose a premium lumber buyer would read."
)
_SEEDS = {"call1": CALL1_SEED, "call2": CALL2_SEED}
_SEEDS = {"call1": CALL1_SEED, "call2": CALL2_SEED}


def _error(code: str, message: str, detail: Any = None) -> dict:
    return {"error": {"code": code, "message": message, "detail": detail}}


def _now() -> str:
    return datetime.now(timezone.utc).replace(microsecond=0, tzinfo=None).isoformat()


def _hash(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def seed_prompts(conn) -> None:
    """Ensure each call has a prompt row and the shipped seed revision."""
    for name, seed in _SEEDS.items():
        prow = conn.execute("SELECT id, active_revision_id FROM prompts WHERE name=?", (name,)).fetchone()
        if prow is None:
            cur = conn.execute(
                "INSERT INTO prompts(name, active_revision_id, created_at, updated_at) VALUES (?,?,?,?)",
                (name, None, _now(), _now()),
            )
            pid, active_id = cur.lastrowid, None
        else:
            pid, active_id = prow[0], prow[1]
        h = _hash(seed)
        existing = conn.execute(
            "SELECT id FROM prompt_revisions WHERE prompt_id=? AND content_hash=?", (pid, h)
        ).fetchone()
        seed_rev_id = existing[0] if existing else conn.execute(
            "INSERT INTO prompt_revisions(prompt_id, content, content_hash, created_at, created_by) VALUES (?,?,?,?,?)",
            (pid, seed, h, _now(), "seed"),
        ).lastrowid
        activate = active_id is None
        if not activate and active_id != seed_rev_id:
            act = conn.execute("SELECT created_by FROM prompt_revisions WHERE id=?", (active_id,)).fetchone()
            activate = act is None or act[0] == "seed"
        if activate and active_id != seed_rev_id:
            conn.execute(
                "UPDATE prompts SET active_revision_id=?, updated_at=? WHERE id=?",
                (seed_rev_id, _now(), pid),
            )


def active_prompt(conn, call: str) -> tuple[int, str] | None:
    """(revision_id, content) for the call's active prompt revision, or None."""
    row = conn.execute(
        "SELECT p.id, pr.id, pr.content FROM prompts p "
        "LEFT JOIN prompt_revisions pr ON p.active_revision_id=pr.id WHERE p.name=?",
        (call,),
    ).fetchone()
    if row is None or row[2] is None:
        return None
    return row[1], row[2]


@router.post("/prompts/{call}/revisions", status_code=201)
async def append_revision(call: str, payload: dict[str, Any]) -> dict:
    if call not in _SEEDS:
        return JSONResponse(status_code=404, content=_error("not_found", "unknown call"))
    content = payload.get("content")
    if not content:
        return JSONResponse(status_code=400, content=_error("validation", "content required"))
    conn = get_connection()
    seed_prompts(conn)
    prow = conn.execute("SELECT id FROM prompts WHERE name=?", (call,)).fetchone()
    cur = conn.execute(
        "INSERT INTO prompt_revisions(prompt_id, content, content_hash, created_at, created_by) VALUES (?,?,?,?,?)",
        (prow[0], content, _hash(content), _now(), "api"),
    )
    conn.execute("UPDATE prompts SET active_revision_id=?, updated_at=? WHERE id=?", (cur.lastrowid, _now(), prow[0]))
    conn.commit()
    return {"id": cur.lastrowid, "prompt": call, "content_hash": _hash(content)}


@router.post("/prompts/{call}/activate")
async def activate_revision(call: str, payload: dict[str, Any]) -> dict:
    if call not in _SEEDS:
        return JSONResponse(status_code=404, content=_error("not_found", "unknown call"))
    rev_id = payload.get("revision_id")
    conn = get_connection()
    seed_prompts(conn)
    prow = conn.execute("SELECT id FROM prompts WHERE name=?", (call,)).fetchone()
    if conn.execute("SELECT id FROM prompt_revisions WHERE id=? AND prompt_id=?", (rev_id, prow[0])).fetchone() is None:
        return JSONResponse(status_code=404, content=_error("not_found", "revision not found"))
    conn.execute("UPDATE prompts SET active_revision_id=?, updated_at=? WHERE id=?", (rev_id, _now(), prow[0]))
    conn.commit()
    return {"ok": True, "active_revision_id": rev_id}


@router.get("/prompts/{call}")
async def get_prompt(call: str) -> dict:
    if call not in _SEEDS:
        return JSONResponse(status_code=404, content=_error("not_found", "unknown call"))
    conn = get_connection()
    seed_prompts(conn)
    prow = conn.execute("SELECT * FROM prompts WHERE name=?", (call,)).fetchone()
    revs = conn.execute(
        "SELECT id, content_hash, created_at, created_by FROM prompt_revisions WHERE prompt_id=? ORDER BY id",
        (prow[0],),
    ).fetchall()
    active = (
        conn.execute("SELECT content FROM prompt_revisions WHERE id=?", (prow["active_revision_id"],)).fetchone()
        if prow["active_revision_id"] is not None
        else None
    )
    return {
        "name": call,
        "active_revision_id": prow["active_revision_id"],
        "active_content": active[0] if active else None,
        "revisions": [dict(r) for r in revs],
    }