feat: add authenticated engine processing boundary

site/routes/pipeline_bp.py

@@ -1,18 +1,25 @@
 """Pipeline blueprint — SSE-сравнение + classify."""
-import json, re, os, threading
+import hashlib, json, re, os, threading, time
 from flask import Blueprint, request, jsonify, Response, stream_with_context
 from services.process import run_pipeline
 from services import extractor as stateless_extractor
+from services.engine import analyze_batch
 from services.grouping import build_dossiers_for_batch
 from services.timeline_builder import build_dossier_timeline
 from llm_prompt import build_prompt
-from db import documents
+from db import documents, extraction_jobs
+from db.connection import transaction
 
 pipeline_bp = Blueprint("pipeline", __name__)
 
-# In-memory lock для classify (замена файлового lock)
-_classify_locks: dict[str, threading.Thread] = {}
-_extraction_jobs: dict[str, dict] = {}
+_engine_rate_lock = threading.Lock()
+_engine_rate_hits: dict[str, list[float]] = {}
+_ENGINE_MAX_BODY_BYTES = 10 * 1024 * 1024
+_ENGINE_MAX_DOCUMENTS = 100
+_ENGINE_MAX_ELEMENTS = 20_000
+_ENGINE_MAX_DOCUMENT_BYTES = 2 * 1024 * 1024
+_ENGINE_RATE_WINDOW_SECONDS = 60
+_ENGINE_RATE_LIMIT = 10
 
 
 @pipeline_bp.route("/process-v2", methods=["GET"])
@@ -110,85 +117,209 @@ def classify_batch_route():
     if not isinstance(batch_id, str) or not batch_id.strip():
         return jsonify(ok=False, error="batch_id required"), 400
 
-    # Guard: уже запущена?
-    if batch_id in _classify_locks:
-        t = _classify_locks[batch_id]
-        if t.is_alive():
-            return jsonify(ok=False, error="classify already running"), 409
-        else:
-            del _classify_locks[batch_id]
-
     pending = documents.list_for_extraction(batch_id)
     total = len(pending)
     if total == 0:
         return jsonify(ok=False, error="no parsed documents"), 400
+    owner = extraction_jobs.claim(batch_id, total)
+    if not owner:
+        return jsonify(ok=False, error="classify already running"), 409
 
     # Keep small batches synchronous for the existing browser workflow.
     if total <= 10:
-        result = _run_v3_extraction(batch_id, total)
+        result = _run_extraction_job(batch_id, pending, owner)
         return jsonify(result)
 
-    _extraction_jobs[batch_id] = {
-        "state": "running", "total": total, "done": 0,
-        "failed": 0, "classes": {},
-    }
-
-    def _run():
-        try:
-            _run_v3_extraction(batch_id, total, _extraction_progress)
-            _extraction_jobs[batch_id]["state"] = "done"
-        except Exception as error:
-            _extraction_jobs[batch_id].update(state="failed", error=str(error))
-        finally:
-            _classify_locks.pop(batch_id, None)
-
-    t = threading.Thread(target=_run, daemon=True)
-    _classify_locks[batch_id] = t
-    t.start()
-
+    _start_extraction_job(batch_id, pending, owner)
     return jsonify(ok=True, total=total, v3=True), 202
 
 
-def _extraction_progress(batch_id, completed, result):
-    """Обновить лёгкий in-memory прогресс для крупного batch."""
-    job = _extraction_jobs[batch_id]
-    job["done"] = completed
-    if not result.get("ok"):
-        job["failed"] += 1
-    else:
-        extraction = result.get("extraction") or {}
-        doc_class = extraction.get("doc_class", "unknown")
-        job["classes"][doc_class] = job["classes"].get(doc_class, 0) + 1
+@pipeline_bp.route("/api/engine/analyze", methods=["POST"])
+def engine_analyze_route():
+    """Accept parsed documents over a versioned, authenticated JSON boundary.
+
+    Validation is completed for the whole payload before opening the write
+    transaction, so malformed input cannot leave a partial batch. The caller
+    supplies parsed elements only; extraction passports and dossier decisions
+    are always produced by the in-process Engine and persisted by its adapters.
+    """
+    if _engine_rate_limited():
+        return jsonify(ok=False, error="rate limit exceeded"), 429, {"Retry-After": "60"}
+
+    raw_body = request.stream.read(_ENGINE_MAX_BODY_BYTES + 1)
+    if len(raw_body) > _ENGINE_MAX_BODY_BYTES:
+        return jsonify(ok=False, error="request body too large"), 413
+    try:
+        body = json.loads(raw_body) if request.is_json else None
+    except (ValueError, UnicodeDecodeError):
+        body = None
+    if not isinstance(body, dict):
+        return jsonify(ok=False, error="JSON object required"), 400
+    if type(body.get("schema_version")) is not int or body["schema_version"] != 1:
+        return jsonify(ok=False, error="unsupported schema_version"), 400
+
+    batch_id = body.get("batch_id")
+    if not isinstance(batch_id, str) or not re.fullmatch(r"[A-Za-z0-9._:-]{1,128}", batch_id):
+        return jsonify(ok=False, error="invalid batch_id"), 400
+    source_documents = body.get("documents")
+    if not isinstance(source_documents, list) or not source_documents:
+        return jsonify(ok=False, error="documents must be a non-empty array"), 400
+    if len(source_documents) > _ENGINE_MAX_DOCUMENTS:
+        return jsonify(ok=False, error="too many documents"), 413
+    validated = []
+    for index, source_document in enumerate(source_documents):
+        if not isinstance(source_document, dict):
+            return jsonify(ok=False, error=f"documents[{index}] must be an object"), 400
+        filename = source_document.get("filename")
+        elements = source_document.get("elements")
+        if not isinstance(filename, str) or not filename.strip() or len(filename) > 512:
+            return jsonify(ok=False, error=f"documents[{index}].filename is invalid"), 400
+        if not isinstance(elements, list) or len(elements) > _ENGINE_MAX_ELEMENTS:
+            return jsonify(ok=False, error=f"documents[{index}].elements is invalid"), 400
+        for element_index, element in enumerate(elements):
+            if not _valid_engine_element(element):
+                return jsonify(
+                    ok=False, error=f"documents[{index}].elements[{element_index}] is invalid",
+                ), 400
+        try:
+            elements_json = json.dumps(
+                elements, ensure_ascii=False, sort_keys=True, separators=(",", ":"),
+                allow_nan=False,
+            )
+        except (TypeError, ValueError):
+            return jsonify(ok=False, error=f"documents[{index}].elements is not valid JSON"), 400
+        encoded_elements = elements_json.encode("utf-8")
+        if len(encoded_elements) > _ENGINE_MAX_DOCUMENT_BYTES:
+            return jsonify(ok=False, error=f"documents[{index}] is too large"), 413
+        validated.append((filename.strip(), elements, elements_json, encoded_elements))
+
+    inserted = []
+    with transaction():
+        # Engine API owns a fresh batch per request. Preventing reuse inside
+        # BEGIN IMMEDIATE keeps old parsed rows out of this run and closes the
+        # race between two concurrent requests using the same batch ID.
+        if documents.list_by_batch(batch_id):
+            return jsonify(ok=False, error="batch_id already exists"), 409
+        owner = extraction_jobs.claim(batch_id, len(validated))
+        if not owner:
+            return jsonify(ok=False, error="classify already running"), 409
+        for filename, elements, elements_json, encoded_elements in validated:
+            content_hash = hashlib.sha256(encoded_elements).hexdigest()
+            document = documents.insert(
+                filename, "application/json", None, status="uploaded",
+                batch_id=batch_id, content_hash=content_hash,
+            )
+            documents.set_parsed(document["id"], elements)
+            inserted.append({"id": document["id"], "filename": filename})
+
+    pending = documents.list_for_extraction(batch_id)
+    total = len(pending)
+    if total > 10:
+        _start_extraction_job(batch_id, pending, owner)
+        return jsonify(ok=True, v3=True, total=total, documents=inserted), 202
+
+    result = _run_extraction_job(batch_id, pending, owner)
+    result["documents"] = inserted
+    return jsonify(result)
 
 
-def _run_v3_extraction(batch_id, total, progress_callback=None):
+def _valid_engine_element(element):
+    """Validate the parser fields consumed by structured-text formatting."""
+    if not isinstance(element, dict):
+        return False
+    if element.get("type") == "paragraph":
+        return isinstance(element.get("text"), str) and isinstance(element.get("style", ""), str)
+    if element.get("type") != "table":
+        return False
+    rows = element.get("rows")
+    context = element.get("preceding_paragraphs", [])
+    table_index = element.get("table_index")
+    return (
+        isinstance(rows, list)
+        and all(isinstance(row, list) and all(isinstance(cell, str) for cell in row) for row in rows)
+        and isinstance(context, list)
+        and all(isinstance(paragraph, str) for paragraph in context)
+        and (table_index is None or (type(table_index) is int and table_index > 0))
+    )
+
+
+def _start_extraction_job(batch_id, pending, owner):
+    """Run a claimed batch in the background with persistent progress."""
+    def _run():
+        try:
+            _run_extraction_job(batch_id, pending, owner)
+        except Exception:
+            pass
+
+    thread = threading.Thread(target=_run, daemon=True)
+    try:
+        thread.start()
+    except Exception as error:
+        extraction_jobs.finish(batch_id, owner, error=str(error))
+        raise
+
+
+def _engine_rate_limited():
+    """Apply a small per-source fixed-window limit before accepting LLM work."""
+    now = time.monotonic()
+    source = request.remote_addr or "unknown"
+    cutoff = now - _ENGINE_RATE_WINDOW_SECONDS
+    with _engine_rate_lock:
+        for key in list(_engine_rate_hits):
+            recent = [stamp for stamp in _engine_rate_hits[key] if stamp > cutoff]
+            if recent:
+                _engine_rate_hits[key] = recent
+            else:
+                del _engine_rate_hits[key]
+        recent = _engine_rate_hits.get(source, [])
+        if len(recent) >= _ENGINE_RATE_LIMIT:
+            return True
+        recent.append(now)
+        _engine_rate_hits[source] = recent
+    return False
+
+
+def _run_extraction_job(batch_id, pending, owner):
+    """Keep the lease alive during LLM calls and persist every final outcome."""
+    stop = threading.Event()
+
+    def renew_lease():
+        while not stop.wait(15):
+            if not extraction_jobs.heartbeat(batch_id, owner):
+                return
+
+    heartbeat = threading.Thread(target=renew_lease, daemon=True)
+    heartbeat.start()
+    try:
+        result = _run_v3_extraction(
+            batch_id, pending,
+            progress_callback=lambda completed, item: extraction_jobs.progress(batch_id, owner, completed, item),
+        )
+        extraction_jobs.finish(batch_id, owner, result=result)
+        return result
+    except Exception as error:
+        extraction_jobs.finish(batch_id, owner, error=str(error))
+        raise
+    finally:
+        stop.set()
+        heartbeat.join()
+
+
+def _run_v3_extraction(batch_id, pending, progress_callback=None, llm_client=None):
     """Собрать extraction и dossier summaries в один ответ API."""
-    extraction = stateless_extractor.extract_batch(
+    parsed_documents = [{
+        "id": document["id"],
+        "filename": document["filename"],
+        "elements": stateless_extractor._decode_elements(document.get("elements_json")),
+    } for document in pending]
+    return analyze_batch(
         batch_id,
-        progress_callback=(
-            (lambda completed, result: progress_callback(batch_id, completed, result))
-            if progress_callback else None
-        ),
+        parsed_documents,
+        extractor=stateless_extractor.extract_batch,
+        dossier_builder=build_dossiers_for_batch,
+        llm_client=llm_client,
+        progress_callback=progress_callback,
     )
-    dossier_result = build_dossiers_for_batch(batch_id)
-    classes = extraction.get("classes", {})
-    math_checks_passed = sum(
-        1 for result in extraction.get("results", [])
-        if (result.get("extraction") or {}).get("math_validation", {}).get("status") == "VERIFIED"
-    )
-    return {
-        "ok": bool(extraction.get("results")) and dossier_result.get("ok", False),
-        "v3": True,
-        "done": total,
-        "total": total,
-        "failed": extraction.get("failed", 0),
-        "classes": classes,
-        "ignored_accounting": classes.get("ignored_accounting", 0),
-        "math_checks_passed": math_checks_passed,
-        "dossier_count": len(dossier_result.get("dossiers", [])),
-        "ambiguous": dossier_result.get("ambiguous", 0),
-        "unlinked": dossier_result.get("unlinked", []),
-    }
 
 
 @pipeline_bp.route("/api/classify-progress")
@@ -197,13 +328,13 @@ def classify_progress_route():
     batch_id = request.args.get("batch")
     if not batch_id:
         return jsonify(ok=False, error="batch required"), 400
-    job = _extraction_jobs.get(batch_id)
+    job = extraction_jobs.get(batch_id)
     if not job:
         return jsonify(ok=False, error="batch extraction not found"), 404
     response = {
-        "ok": job["state"] != "failed",
+        "ok": job["state"] not in {"failed", "interrupted"},
         "v3": True,
-        "finished": job["state"] in {"done", "failed"},
+        "finished": job["state"] in {"done", "failed", "interrupted"},
         "state": job["state"],
         "done": job["done"],
         "total": job["total"],
@@ -212,4 +343,6 @@ def classify_progress_route():
     }
     if job.get("error"):
         response["error"] = job["error"]
+    if job.get("result") is not None:
+        response["result"] = job["result"]
     return jsonify(response)