From 0d678da25f6f989adaefddc357f2c32b66f71324 Mon Sep 17 00:00:00 2001 From: Chaim Date: Sat, 25 Apr 2026 18:42:15 +0000 Subject: [PATCH] =?UTF-8?q?feat(kb):=20backend=20for=20v0.8.0=20=E2=80=94?= =?UTF-8?q?=20tool=20kind=20+=20labels=20+=20AI=20classifier=20+=20bulk=20?= =?UTF-8?q?upload?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 6 of the multi-topic KB refactor (backend half). Subsumes the original Task #19 design discussion (subject vs labels). EspoCRM extension UI ships separately. Migrations (4 new, all idempotent, transaction-wrapped): 004_kind_tool — adds 'tool' to kb_source/kb_ingest_job CHECK 005_source_description — kb_source.description + ai_classified_at 006_labels — kb_label + kb_source_label (M:N), GRANTs 007_ingest_classifier — awaiting_review status + processing_stage + ai_suggestions JSONB + batch_id Two-phase ingest (api/services/kb/ingest_jobs.py): queued → processing(stage=classifying) → awaiting_review ← user reviews AI output → processing(stage=embedding) ← after user commits → done | failed process_classify_stage() pulls the file, runs parse_first_pages (3-page text extract), calls classifier.classify (Claude tool-call via ai-gateway, 45s timeout, 5-concurrent semaphore, fail-soft on any error → empty suggestions), writes ai_suggestions JSONB, transitions to awaiting_review. commit_job() resolves user-confirmed labels (existing by slug, new ones via slugify+ON CONFLICT DO NOTHING), transitions to processing(embedding), schedules process_embed_stage. process_embed_stage() runs the legacy ingest_source path with user-edited metadata + summary as kb_source.description, then apply_to_source for labels. discard_job() removes a file from a batch (status=failed, S3 deleted). list_jobs_by_batch() — single-roundtrip review-screen load. Labels (NEW api/services/kb/labels.py): Hebrew → ASCII slugify (letter-by-letter map, no external dep); CRUD; lookup_or_create_batch (race-safe); apply_to_source + replace_for_source maintain usage_count; merge race-safely (UPDATE assignments + ON CONFLICT DO NOTHING). Classifier (NEW api/services/kb/classifier.py): AsyncOpenAI to ai-gateway, Sonnet, OpenAI-style tool calling for guaranteed JSON shape, Hebrew system prompt with 5 kind values, citation format examples, "do not invent" rule, summary length bounds 80-200 chars. Routes (api/routes/admin_kb.py — 8 new on top of existing 11): POST /admin/kb/upload-batch (multi-file, AI classify) GET /admin/kb/batch/{batch_id} (review-screen load) POST /admin/kb/jobs/{id}/commit (user confirms metadata) POST /admin/kb/jobs/{id}/discard (user removes from batch) GET /admin/kb/labels (typeahead) POST /admin/kb/labels (explicit create) POST /admin/kb/labels/{id}/merge (admin cleanup) DELETE /admin/kb/labels/{id} (only if usage=0) Public /kb/search gains optional label_id filter. admin_sources.list_sources LEFT JOINs labels into each row's output. update_source accepts label_ids to replace the set. End-to-end smoke test passed: synthetic Hebrew circular → upload-batch → background classify → awaiting_review with kind, title, subject, summary, identifier, published_at all populated correctly. Refs Task Master #20 (espocrm-extensions/KnowledgeBase v0.8.0) Subsumes: Task #19 (labels for sub-topics) Co-Authored-By: Claude Opus 4.6 (1M context) --- api/routes/admin_kb.py | 284 ++++++++++++- api/routes/kb_public.py | 5 +- api/services/kb/admin_sources.py | 111 +++-- api/services/kb/chunker.py | 13 +- api/services/kb/classifier.py | 231 ++++++++++ api/services/kb/ingest.py | 47 ++- api/services/kb/ingest_jobs.py | 286 ++++++++++++- api/services/kb/labels.py | 398 ++++++++++++++++++ api/services/kb/s3.py | 8 +- api/services/kb/search.py | 26 +- scripts/migrations/004_kind_tool.sql | 27 ++ scripts/migrations/005_source_description.sql | 21 + scripts/migrations/006_labels.sql | 60 +++ scripts/migrations/007_ingest_classifier.sql | 43 ++ 14 files changed, 1496 insertions(+), 64 deletions(-) create mode 100644 api/services/kb/classifier.py create mode 100644 api/services/kb/labels.py create mode 100644 scripts/migrations/004_kind_tool.sql create mode 100644 scripts/migrations/005_source_description.sql create mode 100644 scripts/migrations/006_labels.sql create mode 100644 scripts/migrations/007_ingest_classifier.sql diff --git a/api/routes/admin_kb.py b/api/routes/admin_kb.py index 99a186e..2331579 100644 --- a/api/routes/admin_kb.py +++ b/api/routes/admin_kb.py @@ -13,6 +13,7 @@ from pydantic import BaseModel, Field from api.services.kb import admin_sources as kb_admin_sources from api.services.kb import ingest as kb_ingest from api.services.kb import ingest_jobs as kb_jobs +from api.services.kb import labels as kb_labels from api.services.kb import s3 as kb_s3 from api.services.kb import topics as kb_topics from api.services.kb.db import get_pool @@ -39,7 +40,7 @@ def _verify_admin(request: Request) -> None: @router.post("/ingest") async def ingest( request: Request, - kind: Literal["law", "regulation", "circular", "caselaw"] = Form(...), + kind: Literal["law", "regulation", "circular", "caselaw", "tool"] = Form(...), title: str = Form(...), identifier: str | None = Form(None), published_at: str | None = Form(None), @@ -228,7 +229,7 @@ _MAX_UPLOAD_BYTES = 50 * 1024 * 1024 @router.post("/upload") async def upload( request: Request, - kind: Literal["law", "regulation", "circular", "caselaw"] = Form(...), + kind: Literal["law", "regulation", "circular", "caselaw", "tool"] = Form(...), topic_id: int = Form(...), title: str | None = Form(None), identifier: str | None = Form(None), @@ -329,6 +330,9 @@ class SourceUpdate(BaseModel): source_url: str | None = Field(None, max_length=1000) published_at: str | None = None effective_at: str | None = None + description: str | None = Field(None, max_length=500) + # Optional. None = leave labels unchanged; [] = clear all; [3,7] = exact set. + label_ids: list[int] | None = None @router.get("/sources") @@ -336,13 +340,14 @@ async def list_admin_sources( request: Request, topic_id: int | None = None, kind: str | None = None, + label_id: int | None = None, limit: int = 200, ): _verify_admin(request) - if kind and kind not in ("law", "regulation", "circular", "caselaw"): + if kind and kind not in ("law", "regulation", "circular", "caselaw", "tool"): raise HTTPException(status_code=400, detail=f"Invalid kind: {kind}") items = await kb_admin_sources.list_sources( - topic_id=topic_id, kind=kind, limit=limit + topic_id=topic_id, kind=kind, label_id=label_id, limit=limit, ) return {"items": items, "count": len(items)} @@ -350,9 +355,12 @@ async def list_admin_sources( @router.put("/sources/{source_id}") async def update_admin_source(request: Request, source_id: int, body: SourceUpdate): _verify_admin(request) - fields = body.model_dump(exclude_unset=True) + raw = body.model_dump(exclude_unset=True) + label_ids = raw.pop("label_ids", None) try: - updated = await kb_admin_sources.update_source(source_id, fields) + updated = await kb_admin_sources.update_source( + source_id, raw, label_ids=label_ids, + ) except LookupError: raise HTTPException(status_code=404, detail="Source not found") except ValueError as e: @@ -449,3 +457,267 @@ async def delete_admin_topic(request: Request, topic_id: int): raise HTTPException(status_code=404, detail="Topic not found") except kb_topics.ConflictError as e: raise HTTPException(status_code=409, detail=str(e)) + + +# ── Phase 6 (Task #20): bulk upload + AI classification + labels ─────────── + +@router.post("/upload-batch") +async def upload_batch( + request: Request, + kind: Literal["law", "regulation", "circular", "caselaw", "tool"] = Form(...), + topic_id: int = Form(...), + # FastAPI accepts a list when the form field name is repeated; the + # browser sends `files=...&files=...` for a multi-file + # or for FormData.append('files', f) called N times. + files: list[UploadFile] = File(...), +): + """Multi-file upload. Each file gets its own kb_ingest_job and a shared + batch_id (uuid). After insert, asyncio.create_task fires the + classify stage in the background. The browser opens the review screen + on /admin/kb/batch/{batch_id} and polls until all jobs are + 'awaiting_review' or 'failed'. + """ + _verify_admin(request) + + if not files: + raise HTTPException(status_code=400, detail="No files in batch") + if len(files) > 50: + raise HTTPException(status_code=400, detail="Max 50 files per batch") + + try: + topic = await kb_topics.get_topic(topic_id) + except Exception: + topic = None + if not topic or not topic.get("is_active"): + raise HTTPException(status_code=400, detail=f"Unknown or inactive topic_id={topic_id}") + + requested_by = ( + request.headers.get("X-User-Name") + or request.headers.get("X-Forwarded-User") + or None + ) + + import uuid as _uuid + batch_id = _uuid.uuid4().hex + + out_jobs: list[dict] = [] + for f in files: + data = await f.read() + if not data: + out_jobs.append({"original_filename": f.filename or "upload", + "error": "empty file"}) + continue + if len(data) > _MAX_UPLOAD_BYTES: + out_jobs.append({"original_filename": f.filename or "upload", + "error": f"too large ({len(data)} bytes)"}) + continue + try: + job_id = await kb_jobs.create_job( + topic_id=topic_id, + topic_slug=topic["slug"], + kind=kind, + original_filename=f.filename or "upload", + file_bytes=data, + # Empty user metadata at this stage — AI will populate + # ai_suggestions, then user confirms via /commit. + metadata={}, + requested_by_user=requested_by, + batch_id=batch_id, + ) + except Exception as e: + logger.exception("[upload-batch] create_job failed") + out_jobs.append({"original_filename": f.filename or "upload", + "error": f"queue failed: {e}"}) + continue + # Fire-and-forget classify stage. The classifier is fail-soft so + # the user always reaches the review screen. + asyncio.create_task(kb_jobs.process_classify_stage(job_id)) + out_jobs.append({ + "job_id": job_id, + "original_filename": f.filename or "upload", + "status": "queued", + }) + + return {"batch_id": batch_id, "jobs": out_jobs} + + +@router.get("/batch/{batch_id}") +async def get_batch(request: Request, batch_id: str): + _verify_admin(request) + items = await kb_jobs.list_jobs_by_batch(batch_id) + return {"batch_id": batch_id, "items": items, "count": len(items)} + + +class JobCommit(BaseModel): + """User-confirmed metadata from the review screen.""" + kind: Literal["law", "regulation", "circular", "caselaw", "tool"] + title: str = Field(..., min_length=1, max_length=500) + identifier: str | None = Field(None, max_length=200) + source_url: str | None = Field(None, max_length=1000) + published_at: str | None = None + effective_at: str | None = None + summary: str | None = Field(None, max_length=500) + # Existing labels — by slug. Lookup is exact match on kb_label.slug. + label_slugs: list[str] = Field(default_factory=list, max_length=10) + # New labels — display names that the user accepted as "create on commit". + # The route slugifies them and creates rows. + new_labels: list[str] = Field(default_factory=list, max_length=10) + + +@router.post("/jobs/{job_id}/commit") +async def commit_job(request: Request, job_id: int, body: JobCommit): + _verify_admin(request) + requested_by = ( + request.headers.get("X-User-Name") + or request.headers.get("X-Forwarded-User") + or None + ) + + # Resolve label_specs. Existing labels: pass slug + name (looked up + # by slug, name only used if INSERT happens via the lookup_or_create + # helper). New labels: slugify display names; reject empties. + label_specs: list[dict] = [] + seen_slugs: set[str] = set() + + # Existing slugs: look up names from DB so the helper can re-insert + # them safely (idempotent on conflict). + if body.label_slugs: + for slug in body.label_slugs: + slug = (slug or "").strip() + if not slug or slug in seen_slugs: + continue + existing = await kb_labels.list_labels(query=slug, limit=1) + existing = [l for l in existing if l["slug"] == slug] + if existing: + label_specs.append({ + "slug": slug, + "name": existing[0]["name"], + "topic_id": existing[0].get("topic_id"), + }) + seen_slugs.add(slug) + + # New labels: derive slug from name, dedupe. + if body.new_labels: + # Find topic for the job so new labels are scoped to it. + async with (await get_pool()).acquire() as conn: + row = await conn.fetchrow( + "SELECT topic_id FROM kb_ingest_job WHERE id = $1", job_id, + ) + topic_id = (row or {}).get("topic_id") if row else None + for name in body.new_labels: + name = (name or "").strip() + if not name: + continue + try: + slug = kb_labels.slugify(name) + except ValueError as e: + logger.warning("[commit] bad new label name=%r: %s", name, e) + continue + if slug in seen_slugs: + continue + label_specs.append({"slug": slug, "name": name, "topic_id": topic_id}) + seen_slugs.add(slug) + + fields = { + "kind": body.kind, + "title": body.title.strip(), + "identifier": (body.identifier or "").strip() or None, + "source_url": (body.source_url or "").strip() or None, + "published_at": body.published_at or None, + "effective_at": body.effective_at or None, + "summary": (body.summary or "").strip() or None, + } + + try: + result = await kb_jobs.commit_job( + job_id=job_id, + fields=fields, + label_specs=label_specs, + requested_by=requested_by, + ) + except LookupError: + raise HTTPException(status_code=404, detail="Job not found") + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + return result + + +@router.post("/jobs/{job_id}/discard") +async def discard_job(request: Request, job_id: int): + _verify_admin(request) + requested_by = ( + request.headers.get("X-User-Name") + or request.headers.get("X-Forwarded-User") + or None + ) + try: + return await kb_jobs.discard_job(job_id, requested_by) + except LookupError: + raise HTTPException(status_code=404, detail="Job not found") + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + + +# ── Labels CRUD ──────────────────────────────────────────────────────────── + +class LabelCreate(BaseModel): + slug: str = Field(..., min_length=1, max_length=200) + name: str = Field(..., min_length=1, max_length=300) + topic_id: int | None = None + + +class LabelMerge(BaseModel): + into_label_id: int + + +@router.get("/labels") +async def list_admin_labels( + request: Request, + topic_id: int | None = None, + q: str | None = None, + limit: int = 50, +): + _verify_admin(request) + items = await kb_labels.list_labels(topic_id=topic_id, query=q, limit=limit) + return {"items": items, "count": len(items)} + + +@router.post("/labels") +async def create_admin_label(request: Request, body: LabelCreate): + _verify_admin(request) + requested_by = ( + request.headers.get("X-User-Name") + or request.headers.get("X-Forwarded-User") + or None + ) + try: + return await kb_labels.create_label( + slug=body.slug, + name=body.name, + topic_id=body.topic_id, + created_by=requested_by, + ) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + + +@router.post("/labels/{label_id}/merge") +async def merge_admin_labels(request: Request, label_id: int, body: LabelMerge): + _verify_admin(request) + try: + return await kb_labels.merge_labels(label_id, body.into_label_id) + except LookupError as e: + raise HTTPException(status_code=404, detail=str(e)) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + + +@router.delete("/labels/{label_id}") +async def delete_admin_label(request: Request, label_id: int): + _verify_admin(request) + try: + return await kb_labels.delete_label(label_id) + except LookupError: + raise HTTPException(status_code=404, detail="Label not found") + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) diff --git a/api/routes/kb_public.py b/api/routes/kb_public.py index 35f0ae9..c842da8 100644 --- a/api/routes/kb_public.py +++ b/api/routes/kb_public.py @@ -36,7 +36,7 @@ def _verify_auth(request: Request) -> None: class SearchRequest(BaseModel): query: str = Field(..., min_length=1, max_length=500) - kind: Literal["any", "law", "regulation", "circular", "caselaw"] = "any" + kind: Literal["any", "law", "regulation", "circular", "caselaw", "tool"] = "any" top_k: int = Field(8, ge=1, le=15) # Multi-query expansion: ask the LLM for alternative phrasings, retrieve # for each, merge, then rerank against the original query. Recovers @@ -45,6 +45,8 @@ class SearchRequest(BaseModel): # Domain scope. None means "lowest active topic" (back-compat — equivalent # to topic_id=1 'national-insurance' until additional topics are added). topic_id: int | None = None + # Optional sub-topic filter (Phase 6 / kb_label). + label_id: int | None = None @router.get("/topics") @@ -71,6 +73,7 @@ async def search(body: SearchRequest, request: Request): expand=body.expand, topic_id=topic_id, topic_name=(topic or {}).get("name"), + label_id=body.label_id, ) # Dates come back as datetime.date; serialize as ISO strings for the UI. serialized = [] diff --git a/api/services/kb/admin_sources.py b/api/services/kb/admin_sources.py index cd07d64..18e3699 100644 --- a/api/services/kb/admin_sources.py +++ b/api/services/kb/admin_sources.py @@ -41,7 +41,8 @@ logger = logging.getLogger("shira.kb.admin_sources") # Whitelist of fields the user can edit through PUT /admin/kb/sources/{id}. # Anything else (kind, topic_id, checksum, source_id, ...) is structural # and must not be flipped from a metadata edit. -_EDITABLE_FIELDS = ("title", "identifier", "source_url", "published_at", "effective_at") +_EDITABLE_FIELDS = ("title", "identifier", "source_url", "published_at", + "effective_at", "description") # ── Listing ───────────────────────────────────────────────────────────────── @@ -50,13 +51,17 @@ async def list_sources( *, topic_id: Optional[int] = None, kind: Optional[str] = None, + label_id: Optional[int] = None, limit: int = 200, ) -> list[dict]: - """Source rows enriched with chunk_count and last_ingest_status. + """Source rows enriched with chunk_count, last_ingest_status, and labels. The last-ingest column is a LATERAL join against kb_ingest_job — picks the most recent job row for the source (regardless of status), so the UI can flag sources whose last re-ingest failed. + + Labels are aggregated as a JSON array via a correlated subquery — + keeps the row count clean (no N×labels Cartesian product). """ limit = max(1, min(500, limit)) where: list[str] = ["s.superseded_by IS NULL"] @@ -67,6 +72,10 @@ async def list_sources( if kind: args.append(kind) where.append(f"s.kind = ${len(args)}") + if label_id is not None: + args.append(label_id) + where.append(f"EXISTS (SELECT 1 FROM kb_source_label sl " + f"WHERE sl.source_id = s.id AND sl.label_id = ${len(args)})") args.append(limit) sql = f""" SELECT @@ -80,6 +89,8 @@ async def list_sources( s.original_path, s.checksum, s.topic_id, + s.description, + s.ai_classified_at, s.created_at, s.updated_at, (SELECT COUNT(*) FROM kb_chunk c WHERE c.source_id = s.id) AS chunk_count, @@ -95,7 +106,19 @@ async def list_sources( WHERE j.source_id = s.id ORDER BY j.created_at DESC LIMIT 1 - ) AS last_ingest + ) AS last_ingest, + COALESCE( + ( + SELECT json_agg(json_build_object( + 'id', l.id, 'slug', l.slug, 'name', l.name, + 'topic_id', l.topic_id + ) ORDER BY l.name) + FROM kb_label l + JOIN kb_source_label sl ON sl.label_id = l.id + WHERE sl.source_id = s.id + ), + '[]'::json + ) AS labels FROM kb_source s WHERE {' AND '.join(where)} ORDER BY s.kind, s.title @@ -119,8 +142,21 @@ async def get_source(source_id: int) -> Optional[dict]: # ── Editing ───────────────────────────────────────────────────────────────── -async def update_source(source_id: int, fields: dict) -> dict: - """UPDATE only whitelisted columns. Returns the post-update row.""" +async def update_source( + source_id: int, + fields: dict, + *, + label_ids: Optional[list[int]] = None, +) -> dict: + """UPDATE only whitelisted columns. Returns the post-update row. + + If `label_ids` is provided (not None), the source's label set is + REPLACED to exactly those ids — usage_count adjusted both ways. + Pass an empty list to clear all labels; pass None to leave labels + untouched. + """ + from api.services.kb import labels as kb_labels # local import: cycle + clean: dict = {} for k in _EDITABLE_FIELDS: if k not in fields: @@ -135,31 +171,43 @@ async def update_source(source_id: int, fields: dict) -> dict: if "title" in clean and not clean["title"]: raise ValueError("title cannot be empty") - if not clean: - # Caller passed nothing editable — just return the current row so - # the UI can move on. - row = await get_source(source_id) - if not row: - raise LookupError(f"source {source_id} not found") - return row - - set_pieces = [] - args: list = [source_id] - for k, v in clean.items(): - args.append(v) - set_pieces.append(f"{k} = ${len(args)}") - sql = f""" - UPDATE kb_source - SET {', '.join(set_pieces)} - WHERE id = $1 - RETURNING * - """ pool = await get_pool() - async with pool.acquire() as conn: - row = await conn.fetchrow(sql, *args) - if not row: - raise LookupError(f"source {source_id} not found") - return _row_to_dict(row) + + if clean: + set_pieces = [] + args: list = [source_id] + for k, v in clean.items(): + args.append(v) + set_pieces.append(f"{k} = ${len(args)}") + sql = f""" + UPDATE kb_source + SET {', '.join(set_pieces)} + WHERE id = $1 + RETURNING id + """ + async with pool.acquire() as conn: + updated = await conn.fetchrow(sql, *args) + if not updated: + raise LookupError(f"source {source_id} not found") + else: + # Caller passed nothing editable; verify the source still exists + # so we can return a sensible 404 vs blindly returning empty. + async with pool.acquire() as conn: + existing = await conn.fetchrow( + "SELECT id FROM kb_source WHERE id = $1", source_id + ) + if not existing: + raise LookupError(f"source {source_id} not found") + + if label_ids is not None: + await kb_labels.replace_for_source(source_id, label_ids) + + # Re-read with labels via list_sources([id])-style query — reuse + # get_source() for simplicity, then fetch labels separately. + row = await get_source(source_id) + if row is not None: + row["labels"] = await kb_labels.get_labels_for_source(source_id) + return row # ── Deletion ──────────────────────────────────────────────────────────────── @@ -400,6 +448,11 @@ def _row_to_dict(row) -> Optional[dict]: out[k] = json.loads(v) except Exception: out[k] = None + elif k == "labels" and isinstance(v, str): + try: + out[k] = json.loads(v) + except Exception: + out[k] = [] else: out[k] = v return out diff --git a/api/services/kb/chunker.py b/api/services/kb/chunker.py index df263f8..c3420c1 100644 --- a/api/services/kb/chunker.py +++ b/api/services/kb/chunker.py @@ -396,16 +396,15 @@ def chunk_circular( return chunks -def chunk(text: str, kind: Literal["law", "regulation", "circular", "caselaw"]) -> list[dict]: +def chunk(text: str, kind: Literal["law", "regulation", "circular", "caselaw", "tool"]) -> list[dict]: if kind in ("law", "regulation"): chunks = chunk_statute(text) else: - # circulars and caselaw share the same fallback: no enumerated - # statutory hierarchy, so we split by paragraph + size. The - # caselaw section_ref ends up empty (the chunker can't infer a - # case identifier from body text reliably) — the user supplies - # the case identifier via the upload form's `identifier` field - # and it's stored on the parent kb_source row instead. + # circulars, caselaw, and tools share the same fallback: no + # enumerated statutory hierarchy, so we split by paragraph + + # size. The case identifier (caselaw) or publishing organization + # (tool) is captured at upload time on the kb_source row, not + # inferred from chunk content. chunks = chunk_circular(text) # _split_oversized uses page markers still embedded in content to derive # per-piece page numbers; _as_dicts then strips markers and emits the diff --git a/api/services/kb/classifier.py b/api/services/kb/classifier.py new file mode 100644 index 0000000..d462345 --- /dev/null +++ b/api/services/kb/classifier.py @@ -0,0 +1,231 @@ +"""AI metadata extraction for uploaded KB documents. + +Phase 6 (Task #20). When a user uploads a PDF/DOCX/TXT through the +batch UI, the background classifier reads the first ~3 pages of text, +calls Claude (via the existing ai-gateway proxy), and returns a draft +metadata payload — kind, title, identifier, subject, labels, dates, +and a short summary. The user reviews these on the new bulk-upload +review screen and edits before committing. + +Design notes (see ~/.claude/plans/hidden-tickling-ullman.md): + • Tool-calling guarantees JSON shape — no prompt-engineering + fragility around "please return only JSON". + • Sonnet model (env CLAUDE_MODEL). Hebrew legal text is hard; + Haiku gets citations wrong too often. + • 15s timeout, semaphore-bounded to 5 concurrent calls so a 30-file + drop doesn't trip ai-gateway rate limits. + • Fail-soft: any error → empty dict. Caller transitions the job to + `awaiting_review` with `ai_suggestions = {}`, and the UI shows + "אין הצעות AI — מלא ידנית". + +The classifier is an aid, not a gate. Wrong suggestions are an +annoyance; the user always confirms before chunks land in the index. +""" + +from __future__ import annotations + +import asyncio +import json +import logging +import os +from typing import Optional + +from openai import AsyncOpenAI + +logger = logging.getLogger("shira.kb.classifier") + +# Bound concurrency so a 10–30 file batch doesn't fan out 30 parallel +# Claude calls. Five at a time gives reasonable wall-clock latency +# (~6–10s for 30 files) without overwhelming ai-gateway. +_CONCURRENCY = int(os.environ.get("KB_CLASSIFIER_CONCURRENCY", "5")) +# Empirically the ai-gateway → Claude tool-call round-trip with a ~3KB +# Hebrew system prompt + 5KB user content runs 18-25s. Pad to 45s so +# we don't fail-soft on the long-tail latency. +_TIMEOUT_S = float(os.environ.get("KB_CLASSIFIER_TIMEOUT_S", "45")) +_MAX_INPUT_CHARS = 20_000 # ~3 pages of Hebrew legal text + +_semaphore = asyncio.Semaphore(_CONCURRENCY) + + +_SYSTEM_PROMPT = """אתה מסווג מסמכים משפטיים בעברית עבור מאגר ידע משפטי. +תקבל את הטקסט של 3 העמודים הראשונים של מסמך, ועליך לחלץ ממנו מטא-דאטה. + +הסיווג חייב לכבד את החוקים הבאים: + +‏1. **kind** — אחד מהבאים בלבד: + • law — חוק (הצעת חוק, ספר חוקים, תיקון לחוק) + • regulation — תקנות + • circular — חוזר רשות / משרד / מוסד (חוזר נכות, חוזר מנכ"ל, חוזר לשכה רפואית) + • caselaw — פסק דין / החלטה. כותרת מתחילה לרוב ב-"בבית המשפט", "ע"א", "בג"ץ", "ב"ל", + "תיק", או שם תיק עם מספר. + • tool — כלי הערכה / מאמר / מדריך אקדמי. אינו מסמך משפטי פורמלי. + דוגמאות: GMFCS, MACS, CARS, Mini-MACS, מדריך הערכת תפקוד מאוניברסיטה. + +‏2. **title** — כותרת המסמך כפי שהיא מופיעה בעמוד הראשון. אל תוסיף, אל תקצר. + ללא מרכאות סוגרות. + +‏3. **identifier** — זיהוי קצר ומדויק. דוגמאות לפי kind: + • law/regulation: "חוק הביטוח הלאומי, התשנ\"ה–1995", "תקנה 37(א)" + • circular: "חוזר נכות 1990/02", "חוזר מנכ\"ל 14/2018" + • caselaw: "ע\"א 1234/24", "בג\"ץ 5678/22", "ב\"ל 9876-12-23" + • tool: שם הגוף המוציא ("CanChild — McMaster University") + אם אין זיהוי ברור — null. אל תמציא. + +‏4. **subject** — תיאור תת-נושא קצר (≤80 תווים). למשל: "ילד נכה / אוטיזם", + "ילד נכה / שיתוק מוחין", "נפגעי עבודה / מחלת מקצוע". + +‏5. **labels** — עד 3 תוויות לסיווג מעמיק. כל תווית בפורמט "קטגוריה / תת-קטגוריה" אם ניתן. + דוגמאות: + • "ילד נכה / אוטיזם" + • "ילד נכה / שיתוק מוחין" + • "ילד נכה / תלות בזולת" + • "ילד נכה / עיכוב התפתחותי" + • "ילד נכה / מומים בגפיים" + • "נפגעי עבודה / שמיעה" + +‏6. **published_at / effective_at** — בפורמט YYYY-MM-DD בלבד. + אם רק שנה ידועה — YYYY-01-01. אם לא ידוע — null. + effective_at הוא תאריך התחילה לתוקף (אם מצוין במפורש בנפרד מתאריך הפרסום). + +‏7. **summary** — משפט אחד באורך 80–200 תווים שמסכם על מה המסמך. ללא קלישאות + ("מסמך חשוב"...). תאר את התוכן הספציפי בקצרה. + +‏**אסור: לא להמציא נתונים שלא מופיעים בטקסט. עדיף null על ניחוש.** +""" + + +_TOOL_SCHEMA = { + "type": "function", + "function": { + "name": "extract_kb_metadata", + "description": "Return extracted metadata for the document.", + "parameters": { + "type": "object", + "required": ["kind", "title", "summary"], + "properties": { + "kind": { + "type": "string", + "enum": ["law", "regulation", "circular", "caselaw", "tool"], + }, + "title": {"type": "string", "maxLength": 500}, + "identifier": {"type": ["string", "null"], "maxLength": 200}, + "subject": {"type": ["string", "null"], "maxLength": 80}, + "labels": { + "type": "array", + "maxItems": 3, + "items": {"type": "string", "maxLength": 80}, + }, + "published_at": { + "type": ["string", "null"], + "pattern": r"^\d{4}-\d{2}-\d{2}$", + }, + "effective_at": { + "type": ["string", "null"], + "pattern": r"^\d{4}-\d{2}-\d{2}$", + }, + "summary": {"type": "string", "minLength": 40, "maxLength": 250}, + }, + "additionalProperties": False, + }, + }, +} + + +async def classify( + *, + text: str, + filename: str, + topic_name: Optional[str] = None, +) -> dict: + """Best-effort extraction. Returns {} on any error. + + `text` is the first ~3 pages of the document (caller decides — we + truncate to _MAX_INPUT_CHARS just in case). `topic_name` lets the + LLM bias label suggestions toward the active legal domain + ("ביטוח לאומי" vs "דיני עבודה"). + """ + if not text or not text.strip(): + logger.info("[classifier] empty text for %s — skipping", filename) + return {} + + base = (os.environ.get("AI_GATEWAY_URL") or "http://localhost:3000").rstrip("/") + api_key = os.environ.get("AI_GATEWAY_API_KEY", "") + if not api_key: + logger.warning("[classifier] AI_GATEWAY_API_KEY not set — skipping") + return {} + + truncated = text[:_MAX_INPUT_CHARS] + user_msg = ( + f"topic_hint: {topic_name or 'כללי'}\n" + f"filename: {filename}\n\n" + f"--- מסמך ---\n{truncated}\n--- סוף ---" + ) + + client = AsyncOpenAI(base_url=f"{base}/v1", api_key=api_key) + async with _semaphore: + try: + resp = await asyncio.wait_for( + client.chat.completions.create( + model=os.environ.get("CLAUDE_MODEL", "sonnet"), + messages=[ + {"role": "system", "content": _SYSTEM_PROMPT}, + {"role": "user", "content": user_msg}, + ], + tools=[_TOOL_SCHEMA], + tool_choice={ + "type": "function", + "function": {"name": "extract_kb_metadata"}, + }, + max_tokens=600, + temperature=0.1, + ), + timeout=_TIMEOUT_S, + ) + except asyncio.TimeoutError: + logger.warning("[classifier] timeout on %s after %ss", filename, _TIMEOUT_S) + return {} + except Exception as e: + logger.warning("[classifier] error on %s: %s", filename, e) + return {} + + # Extract the tool call. We forced the model to call exactly one tool; + # if it didn't, we got nothing useful. + try: + choice = resp.choices[0] + tool_calls = choice.message.tool_calls or [] + if not tool_calls: + logger.warning("[classifier] no tool call returned for %s", filename) + return {} + args_str = tool_calls[0].function.arguments + parsed = json.loads(args_str) + except (KeyError, IndexError, AttributeError, json.JSONDecodeError) as e: + logger.warning("[classifier] failed to parse tool call for %s: %s", filename, e) + return {} + + return _normalize(parsed) + + +def _normalize(d: dict) -> dict: + """Coerce types and drop empty values from the LLM output.""" + out: dict = {} + if isinstance(d.get("kind"), str) and d["kind"] in ( + "law", "regulation", "circular", "caselaw", "tool" + ): + out["kind"] = d["kind"] + for key in ("title", "identifier", "subject", "summary"): + v = d.get(key) + if isinstance(v, str) and v.strip(): + out[key] = v.strip() + for key in ("published_at", "effective_at"): + v = d.get(key) + if isinstance(v, str) and len(v) == 10 and v[4] == "-" and v[7] == "-": + out[key] = v + labels = d.get("labels") or [] + if isinstance(labels, list): + clean: list[str] = [] + for raw in labels[:3]: + if isinstance(raw, str) and raw.strip(): + clean.append(raw.strip()) + if clean: + out["labels"] = clean + return out diff --git a/api/services/kb/ingest.py b/api/services/kb/ingest.py index 28e640f..a3dd2c5 100644 --- a/api/services/kb/ingest.py +++ b/api/services/kb/ingest.py @@ -119,6 +119,40 @@ def _parse_text(data: bytes) -> str: return data.decode("utf-8", errors="replace").strip() +def parse_first_pages(data: bytes, filename: str, n: int = 3) -> str: + """Plain-text extraction of the first ~N pages, used by the + AI classifier on upload (api/services/kb/classifier.py). + + Faster and lighter than `parse(...)` because: + • Uses `get_text("text")` (line-by-line, no block sorting / RTL + merging). The classifier is robust to interleaved columns — + it just needs the document title, citation header, and a + paragraph or two of body to decide kind/subject/dates. + • Caps at N pages so a 600-page ספר הליקויים doesn't burn + seconds of parsing every time it's re-uploaded. + + Returns empty string if the file isn't a PDF/DOCX/TXT or has no + extractable text. Caller (classifier) treats empty as fail-soft. + """ + name = (filename or "").lower() + try: + if name.endswith(".pdf"): + with fitz.open(stream=data, filetype="pdf") as doc: + pages = [] + for i in range(min(n, len(doc))): + pages.append(doc[i].get_text("text") or "") + return "\n\n".join(p.strip() for p in pages if p.strip()) + if name.endswith(".docx"): + # DOCX has no real "page" concept; first 1500 chars of body + # is a reasonable proxy for "first 3 pages worth". + return _parse_docx(data)[:6000] + if name.endswith(".txt") or name.endswith(".md"): + return _parse_text(data)[:6000] + except Exception as e: + logger.warning("[kb.ingest] parse_first_pages failed for %s: %s", filename, e) + return "" + + def parse(data: bytes, filename: str) -> str: name = (filename or "").lower() if name.endswith(".pdf"): @@ -136,7 +170,7 @@ def parse(data: bytes, filename: str) -> str: async def ingest_source( *, - kind: Literal["law", "regulation", "circular", "caselaw"], + kind: Literal["law", "regulation", "circular", "caselaw", "tool"], title: str, identifier: str | None, content: bytes, @@ -146,6 +180,8 @@ async def ingest_source( source_url: str | None = None, original_path: str | None = None, topic_id: int | None = None, + description: str | None = None, + ai_classified: bool = False, ) -> dict: """Parse, chunk, embed, and upsert a source document. @@ -208,17 +244,20 @@ async def ingest_source( } # Upsert source (new row per checksum — supersession is explicit). + ai_ts = "now()" if ai_classified else "NULL" row = await conn.fetchrow( - """ + f""" INSERT INTO kb_source (kind, title, identifier, published_at, effective_at, - source_url, original_path, checksum, topic_id) - VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9) + source_url, original_path, checksum, topic_id, + description, ai_classified_at) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,{ai_ts}) RETURNING id """, kind, title, identifier, _coerce_date(published_at), _coerce_date(effective_at), source_url, original_path, checksum, topic_id, + description, ) source_id = row["id"] diff --git a/api/services/kb/ingest_jobs.py b/api/services/kb/ingest_jobs.py index 211f333..1beb33f 100644 --- a/api/services/kb/ingest_jobs.py +++ b/api/services/kb/ingest_jobs.py @@ -29,8 +29,11 @@ import datetime as _dt import json import logging +from api.services.kb import classifier as kb_classifier from api.services.kb import ingest as kb_ingest +from api.services.kb import labels as kb_labels from api.services.kb import s3 as kb_s3 +from api.services.kb import topics as kb_topics from api.services.kb.db import get_pool logger = logging.getLogger("shira.kb.jobs") @@ -45,11 +48,17 @@ async def create_job( file_bytes: bytes, metadata: dict, requested_by_user: str | None, + batch_id: str | None = None, ) -> int: """Write the file to S3 and insert a queued kb_ingest_job row. Returns the new job id. The caller is expected to schedule - `asyncio.create_task(process_job(job_id))` after this returns. + `asyncio.create_task(process_job(job_id))` (legacy single-stage) or + `asyncio.create_task(process_classify_stage(job_id, topic_name))` + (Phase 6 two-phase) after this returns. + + `batch_id` groups multiple files uploaded in one drag-drop drop. + The review screen uses it to load the whole batch in one round trip. """ import uuid as _uuid @@ -74,18 +83,18 @@ async def create_job( """ INSERT INTO kb_ingest_job (original_filename, s3_key, kind, topic_id, metadata_json, - status, requested_by_user) - VALUES ($1, $2, $3, $4, $5::jsonb, 'queued', $6) + status, requested_by_user, batch_id) + VALUES ($1, $2, $3, $4, $5::jsonb, 'queued', $6, $7) RETURNING id """, original_filename, s3_key, kind, topic_id, json.dumps(metadata, ensure_ascii=False), - requested_by_user, + requested_by_user, batch_id, ) job_id = row["id"] logger.info( - "[kb.jobs] queued job_id=%s kind=%s topic_id=%s file=%r s3=%s", - job_id, kind, topic_id, original_filename[:80], s3_key, + "[kb.jobs] queued job_id=%s kind=%s topic_id=%s file=%r s3=%s batch=%s", + job_id, kind, topic_id, original_filename[:80], s3_key, batch_id, ) return job_id @@ -100,6 +109,23 @@ async def get_job(job_id: int) -> dict | None: return _row_to_dict(row) if row else None +async def list_jobs_by_batch(batch_id: str) -> list[dict]: + """All jobs in one drag-drop batch, oldest first. Used by the + review screen for a single round-trip load. + """ + pool = await get_pool() + async with pool.acquire() as conn: + rows = await conn.fetch( + """ + SELECT * FROM kb_ingest_job + WHERE batch_id = $1 + ORDER BY created_at, id + """, + batch_id, + ) + return [_row_to_dict(r) for r in rows] + + async def list_jobs( *, topic_id: int | None = None, @@ -215,6 +241,254 @@ async def process_job(job_id: int) -> None: ) +# ── Phase 6 (Task #20): two-phase classify → await_review → embed ───────── + +async def process_classify_stage(job_id: int) -> None: + """First half of the new two-phase pipeline. + + queued → processing(stage=classifying) + → awaiting_review (with ai_suggestions populated, possibly {}) + + Called via asyncio.create_task from the new /admin/kb/upload-batch + route. Fail-soft: the AI call timing out, returning malformed + output, or the file being unparseable all transition to + awaiting_review with empty suggestions — the user fills metadata + manually and proceeds normally. Hard fail only on S3-unreachable. + """ + pool = await get_pool() + + # queued → processing(classifying), guarded against double-entry + async with pool.acquire() as conn: + updated = await conn.fetchrow( + """ + UPDATE kb_ingest_job + SET status = 'processing', + processing_stage = 'classifying', + started_at = now() + WHERE id = $1 AND status = 'queued' + RETURNING * + """, + job_id, + ) + if not updated: + logger.warning("[classify] job %s not in queued state — skipping", job_id) + return + + job = _row_to_dict(updated) + logger.info("[classify] job_id=%s starting", job_id) + + suggestions: dict = {} + try: + # Pull just enough of the file to classify. Hard-fail here if S3 + # is unreachable — that's an infra issue, not a classifier issue. + data = kb_s3.fetch(job["s3_key"]) + if data: + text = kb_ingest.parse_first_pages(data, job["original_filename"], n=3) + topic = await kb_topics.get_topic(job["topic_id"]) + topic_name = (topic or {}).get("name") + suggestions = await kb_classifier.classify( + text=text, + filename=job["original_filename"], + topic_name=topic_name, + ) + except Exception as e: + # Note we don't `raise` — empty suggestions are an acceptable + # outcome. Any unexpected exception means the user fills the + # form manually instead of getting a head start. + logger.warning("[classify] job %s suggestions failed: %s", job_id, e) + suggestions = {} + + # Whatever the outcome, transition to awaiting_review so the UI + # never gets stuck on `processing`. + async with pool.acquire() as conn: + await conn.execute( + """ + UPDATE kb_ingest_job + SET status = 'awaiting_review', + processing_stage = NULL, + ai_suggestions = $2::jsonb + WHERE id = $1 + """, + job_id, + json.dumps(suggestions, ensure_ascii=False), + ) + logger.info( + "[classify] job_id=%s awaiting_review (suggestions=%d fields)", + job_id, len(suggestions), + ) + + +async def commit_job( + *, + job_id: int, + fields: dict, + label_specs: list[dict], + requested_by: str | None, +) -> dict: + """User confirmed metadata on the review screen. Resolve labels, + then transition awaiting_review → processing(embedding) and + schedule the embed task. + + `fields` is whitelisted in the route handler — kind, title, + identifier, source_url, published_at, effective_at, summary. + `label_specs` is a list of {slug, name, topic_id?} — existing labels + by slug, new ones by display name. lookup_or_create_batch resolves + them all to ids race-safely. + """ + pool = await get_pool() + + async with pool.acquire() as conn: + job_row = await conn.fetchrow( + "SELECT * FROM kb_ingest_job WHERE id = $1", job_id + ) + if not job_row: + raise LookupError(f"job {job_id} not found") + job = _row_to_dict(job_row) + if job["status"] != "awaiting_review": + raise ValueError( + f"job {job_id} is in status {job['status']!r}, expected 'awaiting_review'" + ) + + # Resolve labels — both existing and new — to ids before we kick off + # the embed task. If label resolution fails the user can re-commit. + label_id_map = await kb_labels.lookup_or_create_batch( + label_specs, created_by=requested_by, + ) + label_ids = list(label_id_map.values()) + + # Transition awaiting_review → processing(embedding) + async with pool.acquire() as conn: + await conn.execute( + """ + UPDATE kb_ingest_job + SET status = 'processing', + processing_stage = 'embedding', + started_at = COALESCE(started_at, now()), + metadata_json = $2::jsonb + WHERE id = $1 AND status = 'awaiting_review' + """, + job_id, + json.dumps(fields, ensure_ascii=False), + ) + + asyncio.create_task(process_embed_stage(job_id, fields, label_ids)) + return {"job_id": job_id, "status": "processing", "label_ids": label_ids} + + +async def process_embed_stage( + job_id: int, fields: dict, label_ids: list[int], +) -> None: + """Second half: ingest_source + apply_labels. + + Failures here are real failures — the file's been confirmed by the + user but parse/embed broke. Surface as `failed` with error_message. + """ + pool = await get_pool() + async with pool.acquire() as conn: + job_row = await conn.fetchrow( + "SELECT * FROM kb_ingest_job WHERE id = $1", job_id + ) + if not job_row: + logger.warning("[embed] job %s vanished", job_id) + return + job = _row_to_dict(job_row) + + try: + data = kb_s3.fetch(job["s3_key"]) + if not data: + raise kb_ingest.IngestError("S3 object is empty") + + title = (fields.get("title") or "").strip() or _filename_stem(job["original_filename"]) + result = await kb_ingest.ingest_source( + kind=fields.get("kind") or job["kind"], + title=title, + identifier=fields.get("identifier"), + content=data, + filename=job["original_filename"], + published_at=fields.get("published_at"), + effective_at=fields.get("effective_at"), + source_url=fields.get("source_url"), + description=fields.get("summary"), + original_path=f"s3://{kb_s3._bucket()}/{job['s3_key']}", + topic_id=job["topic_id"], + ai_classified=bool(job.get("ai_suggestions")), + ) + source_id = result["source_id"] + + if label_ids: + await kb_labels.apply_to_source(source_id, label_ids) + + async with pool.acquire() as conn: + await conn.execute( + """ + UPDATE kb_ingest_job + SET status = 'done', + processing_stage = NULL, + source_id = $2, + chunks_created = $3, + completed_at = now() + WHERE id = $1 + """, + job_id, source_id, result["chunks_created"], + ) + logger.info( + "[embed] done job_id=%s source_id=%s chunks=%d labels=%d", + job_id, source_id, result["chunks_created"], len(label_ids), + ) + except Exception as exc: + logger.exception("[embed] failed job_id=%s", job_id) + async with pool.acquire() as conn: + await conn.execute( + """ + UPDATE kb_ingest_job + SET status = 'failed', + processing_stage = NULL, + error_message = $2, + completed_at = now() + WHERE id = $1 + """, + job_id, str(exc)[:4000], + ) + + +async def discard_job(job_id: int, requested_by: str | None) -> dict: + """User removed a job from the review screen. Status → failed, + S3 file deleted (best-effort), label assignments aren't an issue + yet (commit hasn't happened). + """ + pool = await get_pool() + async with pool.acquire() as conn: + job_row = await conn.fetchrow( + "SELECT * FROM kb_ingest_job WHERE id = $1", job_id + ) + if not job_row: + raise LookupError(f"job {job_id} not found") + job = _row_to_dict(job_row) + if job["status"] in ("done",): + raise ValueError(f"job {job_id} already done — cannot discard") + + # Best-effort S3 cleanup; the row stays as audit trail. + try: + kb_s3._client().delete_object(Bucket=kb_s3._bucket(), Key=job["s3_key"]) + except Exception as e: + logger.warning("[discard] S3 delete failed for %s: %s", job["s3_key"], e) + + reason = f"discarded by {requested_by or 'user'}" + async with pool.acquire() as conn: + await conn.execute( + """ + UPDATE kb_ingest_job + SET status = 'failed', + error_message = $2, + completed_at = now() + WHERE id = $1 + """, + job_id, reason, + ) + logger.info("[discard] job_id=%s by=%s", job_id, requested_by) + return {"job_id": job_id, "status": "failed", "reason": reason} + + # ── helpers ───────────────────────────────────────────────────────────────── def _row_to_dict(row) -> dict: diff --git a/api/services/kb/labels.py b/api/services/kb/labels.py new file mode 100644 index 0000000..41bde28 --- /dev/null +++ b/api/services/kb/labels.py @@ -0,0 +1,398 @@ +"""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 diff --git a/api/services/kb/s3.py b/api/services/kb/s3.py index 137e640..720d363 100644 --- a/api/services/kb/s3.py +++ b/api/services/kb/s3.py @@ -1,9 +1,9 @@ """MinIO/S3 helpers for the insurance-kb bucket. Layout in the bucket: - inbox/{law,regulation,circular,caselaw}/ - processed/{law,regulation,circular,caselaw}/ - failed/{law,regulation,circular,caselaw}/ + inbox/{law,regulation,circular,caselaw,tool}/ + processed/{law,regulation,circular,caselaw,tool}/ + failed/{law,regulation,circular,caselaw,tool}/ After a successful ingest, the file is moved from inbox/* to processed/*. On failure it is moved to failed/*. @@ -20,7 +20,7 @@ from botocore.client import Config as BotoConfig logger = logging.getLogger("shira.kb.s3") -_KINDS = ("law", "regulation", "circular", "caselaw") +_KINDS = ("law", "regulation", "circular", "caselaw", "tool") def _client(): diff --git a/api/services/kb/search.py b/api/services/kb/search.py index 4cf3cc3..bcf63ee 100644 --- a/api/services/kb/search.py +++ b/api/services/kb/search.py @@ -34,8 +34,9 @@ _EXPANSION_TIMEOUT_S = 12.0 async def _retrieve_rrf( query: str, - kind: Literal["law", "regulation", "circular", "caselaw", "any"], + kind: Literal["law", "regulation", "circular", "caselaw", "tool", "any"], topic_id: int | None = None, + label_id: int | None = None, ) -> list[dict]: """Run vector + lexical retrieval, fuse with RRF. No rerank, no top_k cut. @@ -45,6 +46,9 @@ async def _retrieve_rrf( topic_id, when provided, restricts retrieval to a single domain (kb_source.topic_id). When None, all topics are searched — kept for transitional callers; new code should always pass a resolved id. + + label_id, when provided, further restricts retrieval to sources + tagged with that label (kb_source_label). """ [vec] = await voyage.embed([query], input_type="query") @@ -56,6 +60,12 @@ async def _retrieve_rrf( if topic_id is not None: params.append(topic_id) filters.append(f"AND s.topic_id = ${len(params)}") + if label_id is not None: + params.append(label_id) + filters.append( + f"AND EXISTS (SELECT 1 FROM kb_source_label sl " + f"WHERE sl.source_id = s.id AND sl.label_id = ${len(params)})" + ) extra = "\n ".join(filters) sql = f""" @@ -181,21 +191,22 @@ async def _rerank_or_truncate( async def search( query: str, - kind: Literal["law", "regulation", "circular", "any"] = "any", + kind: Literal["law", "regulation", "circular", "caselaw", "tool", "any"] = "any", top_k: int = 8, expand: bool = False, topic_id: int | None = None, topic_name: str | None = None, + label_id: int | None = None, ) -> list[dict]: query = (query or "").strip() if not query: return [] if not expand: - candidates = await _retrieve_rrf(query, kind, topic_id=topic_id) + candidates = await _retrieve_rrf(query, kind, topic_id=topic_id, label_id=label_id) logger.info( - "[kb.search] query=%r kind=%s topic_id=%s RRF_candidates=%d", - query[:80], kind, topic_id, len(candidates), + "[kb.search] query=%r kind=%s topic_id=%s label_id=%s RRF_candidates=%d", + query[:80], kind, topic_id, label_id, len(candidates), ) if not candidates: return [] @@ -204,10 +215,11 @@ async def search( # Multi-query expansion: variants run in parallel with the original. variants = await _expand_query(query, topic_name=topic_name) queries = [query] + variants - logger.info("[kb.search] expand=true topic_id=%s variants=%d", topic_id, len(variants)) + logger.info("[kb.search] expand=true topic_id=%s label_id=%s variants=%d", + topic_id, label_id, len(variants)) results = await asyncio.gather( - *[_retrieve_rrf(q, kind, topic_id=topic_id) for q in queries], + *[_retrieve_rrf(q, kind, topic_id=topic_id, label_id=label_id) for q in queries], return_exceptions=True, ) merged: dict[int, dict] = {} diff --git a/scripts/migrations/004_kind_tool.sql b/scripts/migrations/004_kind_tool.sql new file mode 100644 index 0000000..a4bf296 --- /dev/null +++ b/scripts/migrations/004_kind_tool.sql @@ -0,0 +1,27 @@ +-- Migration 004: add 'tool' as a fifth content kind. +-- +-- Task Master #20 (espocrm-extensions/KnowledgeBase v0.8.0). Academic +-- assessment instruments (GMFCS, MACS, Mini-MACS, etc.) published by +-- universities and professional bodies are not laws / regulations / +-- circulars / caselaw — they're reference tools that get cited by +-- circulars and rulings. They deserve their own kind. +-- +-- Idempotent — wrapped in a single transaction. +-- +-- Run as postgres superuser: +-- docker exec -i psql -U postgres -d insurance_kb \ +-- -f /tmp/004_kind_tool.sql + +BEGIN; + +ALTER TABLE kb_source DROP CONSTRAINT IF EXISTS kb_source_kind_check; +ALTER TABLE kb_source + ADD CONSTRAINT kb_source_kind_check + CHECK (kind IN ('law','regulation','circular','caselaw','tool')); + +ALTER TABLE kb_ingest_job DROP CONSTRAINT IF EXISTS kb_ingest_job_kind_check; +ALTER TABLE kb_ingest_job + ADD CONSTRAINT kb_ingest_job_kind_check + CHECK (kind IN ('law','regulation','circular','caselaw','tool')); + +COMMIT; diff --git a/scripts/migrations/005_source_description.sql b/scripts/migrations/005_source_description.sql new file mode 100644 index 0000000..96ad185 --- /dev/null +++ b/scripts/migrations/005_source_description.sql @@ -0,0 +1,21 @@ +-- Migration 005: kb_source gains a one-line description + AI-classification timestamp. +-- +-- Task Master #20 (v0.8.0). The classifier (api/services/kb/classifier.py) +-- returns a 40–200 char Hebrew summary for each uploaded document. It's +-- worth surfacing on search hit cards and in the management table, so we +-- store it as a permanent column on kb_source rather than only on the +-- ephemeral kb_ingest_job row. ai_classified_at marks AI-touched sources +-- for future audits / metrics; null for legacy rows. +-- +-- Idempotent — wrapped in a single transaction. +-- +-- Run as postgres superuser: +-- docker exec -i psql -U postgres -d insurance_kb \ +-- -f /tmp/005_source_description.sql + +BEGIN; + +ALTER TABLE kb_source ADD COLUMN IF NOT EXISTS description TEXT; +ALTER TABLE kb_source ADD COLUMN IF NOT EXISTS ai_classified_at TIMESTAMPTZ; + +COMMIT; diff --git a/scripts/migrations/006_labels.sql b/scripts/migrations/006_labels.sql new file mode 100644 index 0000000..746ec2c --- /dev/null +++ b/scripts/migrations/006_labels.sql @@ -0,0 +1,60 @@ +-- Migration 006: label-based sub-topic grouping (kb_label many-to-many kb_source). +-- +-- Task Master #20 (v0.8.0); subsumes Task #19 (the original "subject vs +-- tags" design discussion). Many legal documents touch multiple subjects +-- ("ילד נכה / אוטיזם" AND "ילד נכה / תלות בזולת"), which makes a M:N +-- relationship the right shape — a single subject column would force one +-- bucket per source and lose information. +-- +-- Schema choices: +-- • Flat table (no parent_id). Slash-separated names like +-- "ילד נכה / אוטיזם" are a UI display convention only — slug is +-- the unique key. +-- • Slug constrained to ASCII (lowercase letters, digits, hyphens). +-- Hebrew name → ASCII slug via transliteration in Python (unidecode). +-- • topic_id is nullable so a label can be cross-topic (rare but real: +-- "פסיקה כללית" might apply across multiple topics). +-- • usage_count denormalized so the autocomplete dropdown doesn't +-- COUNT(*) on every keystroke. Maintained by application code on +-- INSERT/DELETE of kb_source_label (no trigger — explicit is better). +-- • ON DELETE CASCADE on both FKs in kb_source_label: deleting a +-- source nukes its assignments; deleting a label nukes all +-- assignments. We never want orphans. +-- +-- Idempotent — wrapped in a single transaction. +-- +-- Run as postgres superuser: +-- docker exec -i psql -U postgres -d insurance_kb \ +-- -f /tmp/006_labels.sql + +BEGIN; + +CREATE TABLE IF NOT EXISTS kb_label ( + id SERIAL PRIMARY KEY, + slug TEXT NOT NULL UNIQUE + CHECK (slug ~ '^[a-z0-9][a-z0-9-]*$' AND length(slug) BETWEEN 1 AND 200), + name TEXT NOT NULL, + topic_id INTEGER NULL REFERENCES kb_topic(id) ON DELETE SET NULL, + -- Free-form userName from EspoCRM's X-User-Name header. Audit only. + created_by TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + -- Maintained by app code on every kb_source_label INSERT/DELETE so the + -- autocomplete can ORDER BY usage_count DESC without a join. + usage_count INTEGER NOT NULL DEFAULT 0 +); + +CREATE INDEX IF NOT EXISTS kb_label_topic_idx ON kb_label (topic_id); +CREATE INDEX IF NOT EXISTS kb_label_usage_idx ON kb_label (usage_count DESC); + +CREATE TABLE IF NOT EXISTS kb_source_label ( + source_id INTEGER NOT NULL REFERENCES kb_source(id) ON DELETE CASCADE, + label_id INTEGER NOT NULL REFERENCES kb_label(id) ON DELETE CASCADE, + PRIMARY KEY (source_id, label_id) +); +CREATE INDEX IF NOT EXISTS kb_source_label_label_idx ON kb_source_label (label_id); + +GRANT SELECT, INSERT, UPDATE, DELETE ON kb_label TO shira_kb; +GRANT SELECT, INSERT, UPDATE, DELETE ON kb_source_label TO shira_kb; +GRANT USAGE, SELECT, UPDATE ON SEQUENCE kb_label_id_seq TO shira_kb; + +COMMIT; diff --git a/scripts/migrations/007_ingest_classifier.sql b/scripts/migrations/007_ingest_classifier.sql new file mode 100644 index 0000000..5212122 --- /dev/null +++ b/scripts/migrations/007_ingest_classifier.sql @@ -0,0 +1,43 @@ +-- Migration 007: kb_ingest_job gains AI-classification staging. +-- +-- Task Master #20 (v0.8.0). The two-phase ingest pipeline introduced in +-- this release runs: +-- queued → processing(stage=classifying) +-- → awaiting_review (user reviews AI suggestions) +-- → processing(stage=embedding) (after user commits) +-- → done | failed +-- +-- New columns: +-- • status gains 'awaiting_review' value +-- • processing_stage: 'classifying' | 'embedding' | 'upserting' (audit) +-- • ai_suggestions: JSONB written by the classifier; consumed by +-- the review screen as default form values +-- • batch_id: uuid string assigned by the upload route to all +-- files in one drag-drop drop. Lets the review +-- screen group N files even after a tab refresh. +-- +-- Idempotent — wrapped in a single transaction. +-- +-- Run as postgres superuser: +-- docker exec -i psql -U postgres -d insurance_kb \ +-- -f /tmp/007_ingest_classifier.sql + +BEGIN; + +ALTER TABLE kb_ingest_job DROP CONSTRAINT IF EXISTS kb_ingest_job_status_check; +ALTER TABLE kb_ingest_job + ADD CONSTRAINT kb_ingest_job_status_check + CHECK (status IN ('queued','processing','awaiting_review','done','failed')); + +ALTER TABLE kb_ingest_job + ADD COLUMN IF NOT EXISTS processing_stage TEXT, + ADD COLUMN IF NOT EXISTS ai_suggestions JSONB, + ADD COLUMN IF NOT EXISTS batch_id TEXT; + +CREATE INDEX IF NOT EXISTS kb_ingest_job_batch_idx ON kb_ingest_job (batch_id); +-- Partial index keeps it tiny: production should have very few rows in +-- 'awaiting_review' at any moment (it's a transient state). +CREATE INDEX IF NOT EXISTS kb_ingest_job_awaiting_idx + ON kb_ingest_job (created_at DESC) WHERE status = 'awaiting_review'; + +COMMIT;