From 79de42da96bd63430ac39e3226a01c80373b1268 Mon Sep 17 00:00:00 2001 From: Repinoid Date: Sat, 31 Oct 2026 19:06:34 +0300 Subject: [PATCH] fix: harden extraction batch handling --- site/config.py | 2 +- site/db/connection.py | 5 +++- site/services/engine.py | 48 +++++++++++++++++++++++++++++++++----- site/services/extractor.py | 30 +++++++++++++++++------- site/services/grouping.py | 6 +++-- site/static/app.js | 9 +++++-- 6 files changed, 79 insertions(+), 21 deletions(-) diff --git a/site/config.py b/site/config.py index 47851b5..03c5788 100644 --- a/site/config.py +++ b/site/config.py @@ -1,7 +1,7 @@ """Конфигурация приложения — все настройки в одном месте.""" import os -VERSION = "3.0.2" +VERSION = "3.0.3" def _load_env(): diff --git a/site/db/connection.py b/site/db/connection.py index 04b7928..a2abb63 100644 --- a/site/db/connection.py +++ b/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: diff --git a/site/services/engine.py b/site/services/engine.py index 1f22830..f892a52 100644 --- a/site/services/engine.py +++ b/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), - "total": len(documents), - "failed": extraction.get("failed", 0), + "done": done, + "total": total, + "failed": failed, + "failed_documents": failed_documents, "classes": classes, "ignored_accounting": classes.get("ignored_accounting", 0), "math_checks_passed": math_checks_passed, diff --git a/site/services/extractor.py b/site/services/extractor.py index db5463b..e0859de 100644 --- a/site/services/extractor.py +++ b/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"] - 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) + document_id = document.get("id") if isinstance(document, dict) else None 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, { - "doc_class": "unknown", "intent_mode": "no_spec", "counterparty": {}, - "referenced_contracts": [], "sections": [], - }, {"status": "FAILED", "error": str(error)}, "FAILED", raw_response) - return {"ok": False, "document_id": document_id, "error": str(error)} + result = {"ok": False, "document_id": document_id, "error": str(error)} + if document_id: + # Failure persistence is best-effort; a secondary write failure must + # not escape this handler and abort processing of remaining documents. + 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): diff --git a/site/services/grouping.py b/site/services/grouping.py index 69ef075..9617dec 100644 --- a/site/services/grouping.py +++ b/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: - passport = {"document_id": document["id"], "doc_class": "unknown"} + # A missing or failed passport says extraction produced no usable evidence; + # 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 = { diff --git a/site/static/app.js b/site/static/app.js index 9038454..713b67d 100644 --- a/site/static/app.js +++ b/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();