diff --git a/api/routes/admin_kb.py b/api/routes/admin_kb.py index 616f81a..9dba0da 100644 --- a/api/routes/admin_kb.py +++ b/api/routes/admin_kb.py @@ -2,6 +2,7 @@ from __future__ import annotations +import asyncio import logging import os from typing import Literal @@ -9,7 +10,9 @@ from typing import Literal from fastapi import APIRouter, File, Form, HTTPException, Request, UploadFile from api.services.kb import ingest as kb_ingest +from api.services.kb import ingest_jobs as kb_jobs 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 # Arbitrary non-zero constant used as a pg_try_advisory_lock key. Any int fits; @@ -212,3 +215,104 @@ async def _do_scan() -> dict: return {"processed": sum(1 for r in results if r["status"] == "processed"), "failed": sum(1 for r in results if r["status"] == "failed"), "items": results} + + +# ── Phase 2: async upload + job tracking (Task #13) ──────────────────────── + +# Bytes; cap upload size so a stray 500MB scan doesn't OOM the API container. +_MAX_UPLOAD_BYTES = 50 * 1024 * 1024 + + +@router.post("/upload") +async def upload( + request: Request, + kind: Literal["law", "regulation", "circular"] = Form(...), + topic_id: int = Form(...), + title: str | None = Form(None), + identifier: str | None = Form(None), + published_at: str | None = Form(None), + effective_at: str | None = Form(None), + source_url: str | None = Form(None), + file: UploadFile = File(...), +): + """Async upload: queue a kb_ingest_job, return immediately with job_id. + + The browser polls `/admin/kb/jobs/{id}` until it sees a terminal state. + Errors during parse/embed are recorded on the job row, not bubbled here. + """ + _verify_admin(request) + + data = await file.read() + if not data: + raise HTTPException(status_code=400, detail="Empty file") + if len(data) > _MAX_UPLOAD_BYTES: + raise HTTPException( + status_code=413, + detail=f"File too large ({len(data)} bytes; max {_MAX_UPLOAD_BYTES})", + ) + + # Validate the topic exists and is active. Using resolve_topic_id + # mirrors what /kb/ask and /kb/search already do. + 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 + ) + + metadata = { + "title": (title or "").strip() or None, + "identifier": (identifier or "").strip() or None, + "published_at": published_at or None, + "effective_at": effective_at or None, + "source_url": source_url or None, + } + + try: + job_id = await kb_jobs.create_job( + topic_id=topic_id, + topic_slug=topic["slug"], + kind=kind, + original_filename=file.filename or "upload", + file_bytes=data, + metadata=metadata, + requested_by_user=requested_by, + ) + except Exception as e: + logger.exception("[admin.kb.upload] failed to create job") + raise HTTPException(status_code=500, detail=f"Failed to queue job: {e}") + + # Fire-and-forget: don't await. The HTTP response returns now; the + # background task transitions the job through processing → done|failed. + asyncio.create_task(kb_jobs.process_job(job_id)) + + return {"job_id": job_id, "status": "queued"} + + +@router.get("/jobs") +async def list_ingest_jobs( + request: Request, + topic_id: int | None = None, + status: str | None = None, + limit: int = 50, +): + _verify_admin(request) + if status and status not in ("queued", "processing", "done", "failed"): + raise HTTPException(status_code=400, detail=f"Invalid status: {status}") + items = await kb_jobs.list_jobs(topic_id=topic_id, status=status, limit=limit) + return {"items": items} + + +@router.get("/jobs/{job_id}") +async def get_ingest_job(request: Request, job_id: int): + _verify_admin(request) + job = await kb_jobs.get_job(job_id) + if not job: + raise HTTPException(status_code=404, detail="Job not found") + return job diff --git a/api/services/kb/ingest.py b/api/services/kb/ingest.py index 2314c81..f4f3351 100644 --- a/api/services/kb/ingest.py +++ b/api/services/kb/ingest.py @@ -145,6 +145,7 @@ async def ingest_source( effective_at: str | None = None, source_url: str | None = None, original_path: str | None = None, + topic_id: int | None = None, ) -> dict: """Parse, chunk, embed, and upsert a source document. @@ -154,6 +155,12 @@ async def ingest_source( if not text: raise IngestError(f"no text extracted from {filename}") + # Phase 1 made topic_id NOT NULL on kb_source. Legacy callers (admin + # /ingest, /scan-inbox) don't pass it yet — fall back to topic_id=1 + # (ביטוח לאומי) so the existing inbox cron keeps working unchanged. + if topic_id is None: + topic_id = 1 + checksum = hashlib.sha256(text.encode("utf-8")).hexdigest() chunks = chunk(text, kind) if not chunks: @@ -205,13 +212,13 @@ async def ingest_source( """ INSERT INTO kb_source (kind, title, identifier, published_at, effective_at, - source_url, original_path, checksum) - VALUES ($1,$2,$3,$4,$5,$6,$7,$8) + source_url, original_path, checksum, topic_id) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9) RETURNING id """, kind, title, identifier, _coerce_date(published_at), _coerce_date(effective_at), - source_url, original_path, checksum, + source_url, original_path, checksum, topic_id, ) source_id = row["id"] diff --git a/api/services/kb/ingest_jobs.py b/api/services/kb/ingest_jobs.py new file mode 100644 index 0000000..211f333 --- /dev/null +++ b/api/services/kb/ingest_jobs.py @@ -0,0 +1,254 @@ +"""Async ingest-job tracking for the KB upload UI (Phase 2 / Task #13). + +When a user uploads a PDF/DOCX/TXT through the KB management panel, we don't +want to make the browser wait while we parse → chunk → embed (that's tens +of seconds for a large PDF). Instead: + + 1. The HTTP handler writes the bytes to MinIO under + `inbox///-`. + 2. Inserts a row in `kb_ingest_job` with `status='queued'`. + 3. Schedules `asyncio.create_task(process_job(job_id))` and returns + immediately with `{job_id, status:'queued'}`. + 4. The background task transitions queued → processing → done|failed, + calling `kb_ingest.ingest_source` with the user-supplied metadata. + 5. The UI polls `/admin/kb/jobs/` every 2s until it sees a terminal + status. + +The legacy `/admin/kb/scan-inbox` cron path is unchanged and unaware of +`kb_ingest_job` — it still reads from the flat `inbox//` layout. The +new upload path uses a topic-scoped layout (`inbox///`) +so the cron can't accidentally pick up an upload-form file. Single-source +of truth: when an item is in `kb_ingest_job`, only `process_job` ever +touches it. +""" + +from __future__ import annotations + +import asyncio +import datetime as _dt +import json +import logging + +from api.services.kb import ingest as kb_ingest +from api.services.kb import s3 as kb_s3 +from api.services.kb.db import get_pool + +logger = logging.getLogger("shira.kb.jobs") + + +async def create_job( + *, + topic_id: int, + topic_slug: str, + kind: str, + original_filename: str, + file_bytes: bytes, + metadata: dict, + requested_by_user: str | 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. + """ + import uuid as _uuid + + # UUID prefix prevents collisions when two users upload the same filename. + safe_filename = original_filename.replace("/", "_") + s3_key = f"inbox/{topic_slug}/{kind}/{_uuid.uuid4().hex[:8]}-{safe_filename}" + + # Upload to MinIO BEFORE the DB insert, so we never have a queued row + # pointing at a non-existent S3 object. If the upload fails the user + # gets a 5xx and no job is created. + s3 = kb_s3._client() + s3.put_object( + Bucket=kb_s3._bucket(), + Key=s3_key, + Body=file_bytes, + ContentType=_guess_content_type(original_filename), + ) + + pool = await get_pool() + async with pool.acquire() as conn: + row = await conn.fetchrow( + """ + 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) + RETURNING id + """, + original_filename, s3_key, kind, topic_id, + json.dumps(metadata, ensure_ascii=False), + requested_by_user, + ) + 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, + ) + return job_id + + +async def get_job(job_id: int) -> dict | None: + pool = await get_pool() + async with pool.acquire() as conn: + row = await conn.fetchrow( + "SELECT * FROM kb_ingest_job WHERE id = $1", + job_id, + ) + return _row_to_dict(row) if row else None + + +async def list_jobs( + *, + topic_id: int | None = None, + status: str | None = None, + limit: int = 50, +) -> list[dict]: + limit = max(1, min(200, limit)) + pool = await get_pool() + where: list[str] = [] + args: list = [] + if topic_id is not None: + args.append(topic_id) + where.append(f"topic_id = ${len(args)}") + if status: + args.append(status) + where.append(f"status = ${len(args)}") + where_sql = ("WHERE " + " AND ".join(where)) if where else "" + args.append(limit) + sql = f""" + SELECT * FROM kb_ingest_job + {where_sql} + ORDER BY created_at DESC + LIMIT ${len(args)} + """ + async with pool.acquire() as conn: + rows = await conn.fetch(sql, *args) + return [_row_to_dict(r) for r in rows] + + +async def process_job(job_id: int) -> None: + """Run a single job through ingest_source, recording status transitions. + + Called via `asyncio.create_task(...)` from the upload route. Any + exception here is caught and recorded as `status='failed'` so the + polling UI never gets stuck on `processing`. + """ + pool = await get_pool() + + # 1) queued → processing + async with pool.acquire() as conn: + # Guard against double-processing if process_job is somehow scheduled + # twice for the same id (e.g. retry on container restart later). + updated = await conn.fetchrow( + """ + UPDATE kb_ingest_job + SET status = 'processing', started_at = now() + WHERE id = $1 AND status = 'queued' + RETURNING * + """, + job_id, + ) + if not updated: + logger.warning("[kb.jobs] process_job %s: not in queued state, skipping", job_id) + return + + job = _row_to_dict(updated) + logger.info("[kb.jobs] processing job_id=%s s3=%s", job_id, job["s3_key"]) + + try: + data = kb_s3.fetch(job["s3_key"]) + if not data: + raise kb_ingest.IngestError("S3 object is empty") + meta = job.get("metadata_json") or {} + # Title is required by ingest_source; fall back to filename stem. + title = meta.get("title") or _filename_stem(job["original_filename"]) + + result = await kb_ingest.ingest_source( + kind=job["kind"], + title=title, + identifier=meta.get("identifier"), + content=data, + filename=job["original_filename"], + published_at=meta.get("published_at"), + effective_at=meta.get("effective_at"), + source_url=meta.get("source_url"), + original_path=f"s3://{kb_s3._bucket()}/{job['s3_key']}", + topic_id=job["topic_id"], + ) + + async with pool.acquire() as conn: + await conn.execute( + """ + UPDATE kb_ingest_job + SET status = 'done', + source_id = $2, + chunks_created = $3, + completed_at = now() + WHERE id = $1 + """, + job_id, + result["source_id"], + result["chunks_created"], + ) + logger.info( + "[kb.jobs] done job_id=%s source_id=%s chunks=%d", + job_id, result["source_id"], result["chunks_created"], + ) + + except Exception as exc: + logger.exception("[kb.jobs] failed job_id=%s", job_id) + 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, + # asyncpg can't store >1GB; clamp the message defensively. + str(exc)[:4000], + ) + + +# ── helpers ───────────────────────────────────────────────────────────────── + +def _row_to_dict(row) -> dict: + """Convert an asyncpg.Record to a plain dict the API can return as JSON.""" + if row is None: + return None # type: ignore[return-value] + out = {} + for k, v in dict(row).items(): + if isinstance(v, _dt.datetime): + out[k] = v.isoformat() + elif k == "metadata_json" and isinstance(v, str): + try: + out[k] = json.loads(v) + except Exception: + out[k] = {} + else: + out[k] = v + return out + + +def _filename_stem(filename: str) -> str: + stem = filename.rsplit("/", 1)[-1] + for ext in (".pdf", ".docx", ".txt", ".md"): + if stem.lower().endswith(ext): + return stem[: -len(ext)] + return stem + + +def _guess_content_type(filename: str) -> str: + lower = filename.lower() + if lower.endswith(".pdf"): + return "application/pdf" + if lower.endswith(".docx"): + return "application/vnd.openxmlformats-officedocument.wordprocessingml.document" + if lower.endswith(".txt"): + return "text/plain; charset=utf-8" + return "application/octet-stream" diff --git a/scripts/migrations/002_ingest_jobs.sql b/scripts/migrations/002_ingest_jobs.sql new file mode 100644 index 0000000..6d46229 --- /dev/null +++ b/scripts/migrations/002_ingest_jobs.sql @@ -0,0 +1,53 @@ +-- Migration 002: track async ingest jobs so the KB UI can show upload status. +-- +-- Phase 2 of Task Master #13 (espocrm-extensions/KnowledgeBase). Today the +-- only path into the KB is dropping a file in MinIO and waiting for the +-- /admin/kb/scan-inbox cron. Phase 2 adds an upload form in the EspoCRM UI +-- that POSTs the file to shira-hermes; we need a row to track the async +-- processing so the browser can poll for "queued → processing → done". +-- +-- Idempotent — safe to run twice. Wrapped in a single transaction. +-- +-- Run as postgres superuser: +-- docker exec -i psql -U postgres -d insurance_kb \ +-- -f /tmp/002_ingest_jobs.sql + +BEGIN; + +CREATE TABLE IF NOT EXISTS kb_ingest_job ( + id SERIAL PRIMARY KEY, + -- Set after success; null while queued/processing and on failure. + source_id INTEGER REFERENCES kb_source(id) ON DELETE SET NULL, + original_filename TEXT NOT NULL, + s3_key TEXT NOT NULL, + kind TEXT NOT NULL CHECK (kind IN ('law','regulation','circular')), + topic_id INTEGER NOT NULL REFERENCES kb_topic(id), + -- User-supplied metadata at upload time (title/identifier/dates/source_url). + -- Stored as JSONB so the schema doesn't have to grow when we add fields. + metadata_json JSONB NOT NULL DEFAULT '{}'::jsonb, + status TEXT NOT NULL DEFAULT 'queued' + CHECK (status IN ('queued','processing','done','failed')), + error_message TEXT, + chunks_created INTEGER, + -- Auth identity from EspoCRM proxy header. Free-form so renames don't + -- break old jobs; UI shows it as informational only. + requested_by_user TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + started_at TIMESTAMPTZ, + completed_at TIMESTAMPTZ +); + +-- Index for the "recent jobs" listing — newest first, optionally filtered +-- by status. (status, created_at DESC) covers `WHERE status = $1 ORDER BY ...` +-- and the unfiltered `ORDER BY created_at DESC` (PG can scan it backwards). +CREATE INDEX IF NOT EXISTS kb_ingest_job_status_created_idx + ON kb_ingest_job (status, created_at DESC); + +-- Topic scope is the most common second filter from the UI. +CREATE INDEX IF NOT EXISTS kb_ingest_job_topic_idx + ON kb_ingest_job (topic_id, created_at DESC); + +GRANT SELECT, INSERT, UPDATE, DELETE ON kb_ingest_job TO shira_kb; +GRANT USAGE, SELECT, UPDATE ON SEQUENCE kb_ingest_job_id_seq TO shira_kb; + +COMMIT;