fix: harden extraction batch handling

site/app.py

@@ -1,7 +1,7 @@
 """contracts-flask v2.0 — полный перенос с ВМ на Flask.
 Больше никаких прокси на contracts.kube5s.ru — всё локально.
 """
-import sys, os, hmac
+import sys, os
 # site/ в sys.path — импортируем модули напрямую, без префиксов
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
 # корень репо — для переиспользуемого модуля upload/ (как в drhider)
@@ -9,8 +9,8 @@ _REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 if _REPO_ROOT not in sys.path:
     sys.path.insert(0, _REPO_ROOT)
 
-from flask import Flask, jsonify, request
+from flask import Flask
-from config import VERSION, MAX_CONTENT_LENGTH, API_KEY
+from config import VERSION, MAX_CONTENT_LENGTH
 from routes import register_routes
 
 
@@ -18,29 +18,10 @@ def create_app():
     app = Flask(__name__)
     app.config["VERSION"] = VERSION
     app.config["MAX_CONTENT_LENGTH"] = MAX_CONTENT_LENGTH
-    app.config["API_KEY"] = API_KEY
 
     _init_db()
     register_routes(app)
 
-    @app.before_request
-    def authorize_request():
-        if request.endpoint == "health.health":
-            return None
-        key = app.config.get("API_KEY", "")
-        if not key:
-            return jsonify(ok=False, error="application API_KEY is not configured"), 503
-        if "X-API-Key" in request.headers:
-            supplied = request.headers.get("X-API-Key", "")
-        else:
-            authorization = request.authorization
-            supplied = authorization.password if authorization and authorization.type.lower() == "basic" else ""
-        if not hmac.compare_digest((supplied or "").encode("utf-8"), key.encode("utf-8")):
-            return jsonify(ok=False, error="unauthorized"), 401, {
-                "WWW-Authenticate": 'Basic realm="Contracts", charset="UTF-8"',
-            }
-        return None
-
     @app.after_request
     def no_cache(response):
         response.headers["Cache-Control"] = "no-cache, no-store, must-revalidate"

site/config.py

@@ -1,7 +1,7 @@
 """Конфигурация приложения — все настройки в одном месте."""
 import os
 
-VERSION = "3.0.0"
+VERSION = "3.0.3"
 
 
 def _load_env():

site/db/connection.py

@@ -345,7 +345,10 @@ def execute_returning(sql, params=None):
         sql = sql[:sql.upper().rfind(" RETURNING ")]
 
     cur = conn.execute(sql, params or [])
-    conn.commit()
+    # The outer transaction owns commit and rollback; committing here would make
+    # its rollback ineffective and could expose only part of a multi-write update.
+    if not getattr(_local, "in_transaction", False):
+        conn.commit()
     rowid = cur.lastrowid
 
     if table and rowid:

site/routes/pipeline_bp.py

@@ -1,6 +1,7 @@
 """Pipeline blueprint — SSE-сравнение + classify."""
-import hashlib, json, re, os, threading, time
+import hashlib, hmac, json, re, os, threading, time
 from flask import Blueprint, request, jsonify, Response, stream_with_context
+from config import API_KEY
 from services.process import run_pipeline
 from services import extractor as stateless_extractor
 from services.engine import analyze_batch
@@ -143,6 +144,11 @@ def engine_analyze_route():
     supplies parsed elements only; extraction passports and dossier decisions
     are always produced by the in-process Engine and persisted by its adapters.
     """
+    if not API_KEY:
+        return jsonify(ok=False, error="engine API is not configured"), 503
+    supplied_key = request.headers.get("X-API-Key", "")
+    if not hmac.compare_digest(supplied_key.encode("utf-8"), API_KEY.encode("utf-8")):
+        return jsonify(ok=False, error="unauthorized"), 401
     if _engine_rate_limited():
         return jsonify(ok=False, error="rate limit exceeded"), 429, {"Retry-After": "60"}
 

site/services/engine.py

@@ -25,18 +25,54 @@ def analyze_batch(batch_id, documents, extractor, dossier_builder,
         progress_callback=progress_callback,
         documents=documents,
     )
-    dossier_result = dossier_builder(batch_id) if extraction.get("ok") else {"ok": False}
+    # Extractor-level `ok` means every input succeeded; dossier construction only
+    # requires at least one persisted passport and can proceed for that subset.
+    extraction_results = extraction.get("results") or []
+    total = extraction.get("total", len(documents))
+    done = extraction.get("done")
+    if done is None:
+        done = sum(1 for result in extraction_results if result.get("ok"))
+    failed = extraction.get("failed")
+    if failed is None:
+        failed = max(total - done, 0)
+    dossier_result = dossier_builder(batch_id) if done > 0 else {"ok": False}
+    dossier_ok = bool(dossier_result.get("ok"))
+    # Keep result usability separate from completeness so clients can continue
+    # with valid dossiers while still reporting failures explicitly.
+    status = (
+        "failed" if done == 0 or not dossier_ok
+        else "partial" if failed > 0
+        else "completed"
+    )
+    documents_by_id = {document.get("id"): document for document in documents}
+    failed_documents = []
+    for result in extraction_results:
+        if result.get("ok"):
+            continue
+        document_id = result.get("document_id")
+        source_document = documents_by_id.get(document_id, {})
+        failure = {
+            "document_id": document_id,
+            "filename": source_document.get("filename"),
+            "error": result.get("error") or "extraction failed",
+        }
+        if result.get("persistence_error"):
+            failure["persistence_error"] = result["persistence_error"]
+        failed_documents.append(failure)
     classes = extraction.get("classes", {})
     math_checks_passed = sum(
-        1 for result in extraction.get("results", [])
+        1 for result in extraction_results
         if (result.get("extraction") or {}).get("math_validation", {}).get("status") == "VERIFIED"
     )
     return {
-        "ok": bool(extraction.get("ok")) and bool(dossier_result.get("ok")),
+        "ok": status != "failed",
+        "status": status,
+        "partial": status == "partial",
         "v3": True,
-        "done": extraction.get("done", 0),
+        "done": done,
-        "total": len(documents),
+        "total": total,
-        "failed": extraction.get("failed", 0),
+        "failed": failed,
+        "failed_documents": failed_documents,
         "classes": classes,
         "ignored_accounting": classes.get("ignored_accounting", 0),
         "math_checks_passed": math_checks_passed,

site/services/extractor.py

@@ -250,12 +250,16 @@ def _save_extraction(document_id, extracted, validation, status, raw_response="{
 
 def extract_document(document, llm_client=None, batch_id=None):
     """Извлечь паспорт одного документа, изолируя ошибки на уровне документа."""
-    document_id = document["id"]
+    document_id = document.get("id") if isinstance(document, dict) else None
-    batch_id = batch_id if batch_id is not None else document.get("batch_id")
-    elements = _decode_elements(document.get("elements", document.get("elements_json")))
-    structured_text = format_elements_with_structure(elements)
     raw_response = "{}"
     try:
+        if not document_id:
+            raise ValueError("document id is required")
+        batch_id = batch_id if batch_id is not None else document.get("batch_id")
+        elements = _decode_elements(document.get("elements", document.get("elements_json")))
+        # Formatting is part of this document's failure boundary: corrupt stored
+        # elements must not escape and stop extraction of later batch members.
+        structured_text = format_elements_with_structure(elements)
         if _is_garbage_by_filename(document.get("filename") or "") or _is_garbage_by_header(structured_text[:2000]):
             extracted = {
                 "doc_class": "ignored_accounting", "intent_mode": "no_spec",
@@ -299,11 +303,19 @@ def extract_document(document, llm_client=None, batch_id=None):
         return {"ok": True, "document_id": document_id, "status": result["status"], "extraction": saved}
     except Exception as error:
         log.exception("Stateless extraction failed for document %s", document_id)
-        _save_extraction(document_id, {
+        result = {"ok": False, "document_id": document_id, "error": str(error)}
-            "doc_class": "unknown", "intent_mode": "no_spec", "counterparty": {},
+        if document_id:
-            "referenced_contracts": [], "sections": [],
+            # Failure persistence is best-effort; a secondary write failure must
-        }, {"status": "FAILED", "error": str(error)}, "FAILED", raw_response)
+            # not escape this handler and abort processing of remaining documents.
-        return {"ok": False, "document_id": document_id, "error": str(error)}
+            try:
+                _save_extraction(document_id, {
+                    "doc_class": "unknown", "intent_mode": "no_spec", "counterparty": {},
+                    "referenced_contracts": [], "sections": [],
+                }, {"status": "FAILED", "error": str(error)}, "FAILED", raw_response)
+            except Exception as persistence_error:
+                log.exception("Failed to persist extraction failure for document %s", document_id)
+                result["persistence_error"] = str(persistence_error)
+        return result
 
 
 def extract_batch(batch_id, llm_client=None, progress_callback=None, documents=None):

site/services/grouping.py

@@ -184,8 +184,10 @@ def _build_dossiers_for_batch(batch_id):
     records = []
     for document in documents:
         passport = passport_by_id.get(document["id"])
-        if not passport:
+        # A missing or failed passport says extraction produced no usable evidence;
-            passport = {"document_id": document["id"], "doc_class": "unknown"}
+        # do not misreport that technical failure as a missing contract number.
+        if not passport or passport.get("status") == "FAILED":
+            continue
         if passport.get("doc_class") == "ignored_accounting":
             continue
         counterparty = {

site/static/app.js

@@ -162,7 +162,10 @@ async function runClassify() {
             btn.disabled = false;
             return;
           }
-          btn.innerHTML = '✓ ' + completed + '/' + total + ' документов извлечено';
+          // Partial jobs are terminal and usable, but must not look like full success.
+          var failed = d.failed || 0;
+          btn.innerHTML = (failed ? '⚠ ' : '✓ ') + completed + '/' + total + ' документов извлечено'
+            + (failed ? ' | ошибок: ' + failed : '');
           stepDone('stepClassify');
           stepActive('stepGroups');
           loadGroupsAction();
@@ -212,7 +215,9 @@ async function runClassify() {
       if (data.v3 && data.failed) summary += ' | ошибок: ' + data.failed;
       if (data.garbage) summary += ' | 🗑️ ' + data.garbage + ' мусор';
       if (data.types) summary += ' | ' + Object.entries(data.types).map(function (e) { return e[1] + '×' + e[0]; }).join(', ');
-      btn.innerHTML = '✓ ' + summary + ' (' + Math.round((Date.now() - start) / 1000) + 'с)';
+      // Continue to dossier rendering for partial success while keeping its warning visible.
+      btn.innerHTML = (data.v3 && data.partial ? '⚠ ' : '✓ ') + summary
+        + ' (' + Math.round((Date.now() - start) / 1000) + 'с)';
       stepDone('stepClassify');
       stepActive('stepGroups');
       loadGroupsAction();