feat: add stateless document extraction

site/db/documents.py

@@ -109,6 +109,21 @@ def list_by_batch(batch_id):
     )
 
 
+def list_for_extraction(batch_id):
+    """Получить исходный текст парсинга для stateless-обработки batch.
+
+    Отдельный метод не расширяет legacy-ответ `list_by_batch()`, которым уже
+    пользуются старые API и UI. Extractor читает только поля документа и не
+    меняет состояние загрузки или классификации.
+    """
+    return query(
+          """SELECT id, filename, status, elements_json
+              FROM documents WHERE batch_id = %s AND status = 'parsed'
+              ORDER BY created_at, id""",
+        (batch_id,),
+    )
+
+
 def count_by_status(batch_id):
     """Count documents by classify_status."""
     rows = query(

site/services/extractor.py

@@ -0,0 +1,328 @@
+"""Stateless-извлечение паспорта и проверка числовых утверждений документа.
+
+Каждый вызов LLM ограничен одним документом. Результат сохраняется отдельно
+от загруженных байтов и legacy classify-полей; ошибка одного документа не
+останавливает остальные документы batch.
+"""
+import json
+import re
+from decimal import Decimal, InvalidOperation
+
+from config import LLM_KEY, LLM_MODEL, LLM_URL
+from db import documents as db_docs
+from db import exceptions as db_exceptions
+from db import extractions as db_extractions
+from llm_prompt import build_stateless_extractor_prompt
+from services.classify import _is_garbage_by_filename, _is_garbage_by_header
+from services.parse import format_elements_with_structure
+
+log = __import__("logging").getLogger(__name__)
+
+PROMPT_VERSION = "stateless-extractor-v1"
+MONEY_TOLERANCE = Decimal("0.05")
+DOC_CLASSES = {"contract_frame", "specification", "supplement", "ignored_accounting"}
+INTENT_MODES = {"no_spec", "full_snapshot", "delta_amendment", "termination"}
+_HOMOGLYPHS = str.maketrans({
+    "A": "А", "B": "В", "C": "С", "E": "Е", "H": "Н", "K": "К",
+    "M": "М", "O": "О", "P": "Р", "T": "Т", "X": "Х", "Y": "У",
+})
+
+_extractor_llm = None
+
+
+def _get_extractor_client():
+    """Создать production-клиент лениво, сохраняя инъекцию FakeLLMClient в тестах."""
+    global _extractor_llm
+    if _extractor_llm is None:
+        from services.llm_client import HttpxLLMClient
+        _extractor_llm = HttpxLLMClient(
+            url=LLM_URL, key=LLM_KEY, model=LLM_MODEL, max_tokens=12000, timeout=120,
+        )
+    return _extractor_llm
+
+
+def _decode_elements(value):
+    """Вернуть parsed elements как список при хранении JSON-текстом или объектом."""
+    if isinstance(value, str):
+        try:
+            value = json.loads(value)
+        except json.JSONDecodeError:
+            return []
+    return value if isinstance(value, list) else []
+
+
+def _decode_llm_json(raw_text):
+    """Декодировать JSON-объект модели, сохраняя десятичные числа как Decimal."""
+    if not isinstance(raw_text, str) or not raw_text.strip():
+        raise ValueError("empty LLM response")
+    text = raw_text.strip()
+    if text.startswith("```"):
+        text = re.sub(r"^```(?:json)?\s*|\s*```$", "", text, flags=re.IGNORECASE).strip()
+    start, end = text.find("{"), text.rfind("}")
+    if start < 0 or end < start:
+        raise ValueError("LLM response does not contain a JSON object")
+    result = json.loads(text[start:end + 1], parse_float=Decimal)
+    if not isinstance(result, dict):
+        raise ValueError("LLM response must be a JSON object")
+    return result
+
+
+def _normalize_number(value):
+    """Схлопнуть регистр, гомоглифы и разделители для проверки коллизий номеров."""
+    if value is None:
+        return ""
+    return re.sub(r"[^0-9А-ЯЁ]", "", str(value).upper().translate(_HOMOGLYPHS))
+
+
+def _as_decimal(value):
+    """Преобразовать JSON-число или строку в конечный Decimal без float-арифметики."""
+    if value is None or isinstance(value, bool):
+        return None
+    try:
+        number = Decimal(str(value))
+    except (InvalidOperation, ValueError, TypeError):
+        return None
+    return number if number.is_finite() else None
+
+
+def _counterparty_without_nubes(counterparty):
+    """Удалить исполнителя из поля контрагента по имени или официальному ИНН."""
+    if not isinstance(counterparty, dict):
+        return {"name": None, "inn": None}
+    name = counterparty.get("name")
+    inn = counterparty.get("inn")
+    if inn is not None:
+        digits = re.sub(r"\D", "", str(inn))
+        if digits == "9706005293":
+            return {"name": None, "inn": None}
+    if name and re.search(r"НУБЕС|NUBES", str(name), re.IGNORECASE):
+        return {"name": None, "inn": None}
+    return {"name": name, "inn": inn}
+
+
+def _ensure_exception(batch_id, document_id, severity, title, source_quote="",
+                      stage="extraction_math"):
+    """Добавить исключение только один раз, включая уже разрешённые записи."""
+    existing = db_exceptions.list_exceptions(batch_id=batch_id)
+    if any(
+        item.get("document_id") == document_id
+        and item.get("stage") == stage
+        and item.get("title") == title
+        for item in existing
+    ):
+        return
+    db_exceptions.add_exception(
+        batch_id=batch_id,
+        dossier_id=None,
+        document_id=document_id,
+        service_item_id=None,
+        severity=severity,
+        stage=stage,
+        title=title,
+        source_quote=source_quote,
+    )
+
+
+def validate_extraction(extracted_dict, doc_id, batch_id):
+    """Очистить контрагента, проверить суммы и вернуть паспорт с отчётом.
+
+    Пропущенные значения не трактуются как нули: сверка итогов выполняется
+    только когда у каждой строки таблицы есть корректно извлечённая сумма.
+    Денежная арифметика остаётся в Decimal от разбора JSON до сравнения.
+    """
+    extracted = dict(extracted_dict)
+    extracted["counterparty"] = _counterparty_without_nubes(extracted.get("counterparty"))
+    validation = {
+        "row_mismatches": [],
+        "footer_mismatches": [],
+        "header_mismatches": [],
+        "internal_number_conflict": False,
+    }
+    references = extracted.get("referenced_contracts") or []
+    numbers = {
+        normalized for reference in references
+        if isinstance(reference, dict)
+        if (normalized := _normalize_number(reference.get("number")))
+    }
+    validation["internal_number_conflict"] = len(numbers) > 1
+
+    for section_index, section in enumerate(extracted.get("sections") or [], start=1):
+        if not isinstance(section, dict):
+            continue
+        rows = section.get("rows") or []
+        row_sums = []
+        all_rows_have_sum = bool(rows)
+        for row_index, row in enumerate(rows, start=1):
+            if not isinstance(row, dict):
+                all_rows_have_sum = False
+                continue
+            price = _as_decimal(row.get("price"))
+            qty = _as_decimal(row.get("qty"))
+            amount = _as_decimal(row.get("sum"))
+            if amount is None:
+                all_rows_have_sum = False
+            else:
+                row_sums.append(amount)
+            if price is None or qty is None or amount is None:
+                continue
+            difference = abs(price * qty - amount)
+            if difference > MONEY_TOLERANCE:
+                row_ref = row.get("row_ref") or f"section_{section_index}_row_{row_index}"
+                mismatch = {
+                    "section_ref": section.get("table_ref") or f"section_{section_index}",
+                    "row_ref": row_ref,
+                    "price": price,
+                    "qty": qty,
+                    "reported_sum": amount,
+                    "difference": difference,
+                }
+                validation["row_mismatches"].append(mismatch)
+                _ensure_exception(
+                    batch_id, doc_id, "CONFLICT", f"MATH_MISMATCH:{row_ref}",
+                    str(row.get("raw_name") or ""),
+                )
+
+        section_total = sum(row_sums, Decimal("0"))
+        table_ref = section.get("table_ref") or f"section_{section_index}"
+        if all_rows_have_sum:
+            for field, mismatch_key, label in (
+                ("table_footer_total", "footer_mismatches", "FOOTER_MISMATCH"),
+                ("header_stated_total", "header_mismatches", "HEADER_MISMATCH"),
+            ):
+                stated_total = _as_decimal(section.get(field))
+                if stated_total is None:
+                    continue
+                difference = abs(section_total - stated_total)
+                if difference > MONEY_TOLERANCE:
+                    mismatch = {
+                        "section_ref": table_ref,
+                        "row_sum_total": section_total,
+                        "stated_total": stated_total,
+                        "difference": difference,
+                    }
+                    validation[mismatch_key].append(mismatch)
+                    _ensure_exception(
+                        batch_id, doc_id, "CONFLICT", f"{label}:{table_ref}",
+                        str(section.get("section_title") or ""),
+                    )
+
+    not_applicable = (
+        extracted.get("intent_mode") == "no_spec"
+        or extracted.get("doc_class") == "ignored_accounting"
+    )
+    has_math_conflict = any(validation[key] for key in (
+        "row_mismatches", "footer_mismatches", "header_mismatches",
+    ))
+    if not_applicable:
+        validation["status"] = "NOT_APPLICABLE"
+        status = "WARNING" if validation["internal_number_conflict"] else "VERIFIED"
+    elif has_math_conflict:
+        validation["status"] = "CONFLICT"
+        status = "CONFLICT"
+    elif validation["internal_number_conflict"]:
+        validation["status"] = "WARNING"
+        status = "WARNING"
+    else:
+        validation["status"] = "VERIFIED"
+        status = "VERIFIED"
+
+    return {"extracted": extracted, "math_validation": validation, "status": status}
+
+
+def _save_extraction(document_id, extracted, validation, status, raw_response="{}"):
+    """Map the public passport shape to normalized extraction-table columns."""
+    counterparty = extracted.get("counterparty") or {}
+    return db_extractions.upsert_extraction(document_id, {
+        "doc_class": extracted.get("doc_class", "unknown"),
+        "intent_mode": extracted.get("intent_mode", "no_spec"),
+        "own_number": extracted.get("own_number"),
+        "own_date": extracted.get("own_date"),
+        "counterparty_name": counterparty.get("name"),
+        "counterparty_inn": counterparty.get("inn"),
+        "referenced_numbers_json": extracted.get("referenced_contracts") or [],
+        "sections_json": extracted.get("sections") or [],
+        "math_validation_json": validation,
+        "status": status,
+        "prompt_version": PROMPT_VERSION,
+        "raw_llm_response": 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_json"))
+    structured_text = format_elements_with_structure(elements)
+    raw_response = "{}"
+    try:
+        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",
+                "own_number": None, "own_date": None,
+                "counterparty": {"name": None, "inn": None},
+                "referenced_contracts": [], "sections": [],
+            }
+        elif not elements:
+            extracted = {
+                "doc_class": "unreadable_scan", "intent_mode": "no_spec",
+                "own_number": None, "own_date": None,
+                "counterparty": {"name": None, "inn": None},
+                "referenced_contracts": [], "sections": [],
+            }
+            _ensure_exception(
+                batch_id, document_id, "WARNING", "UNREADABLE_SCAN",
+                document.get("filename") or "", stage="extraction",
+            )
+        else:
+            client = llm_client if llm_client is not None else _get_extractor_client()
+            raw_response = client.complete(build_stateless_extractor_prompt(structured_text))
+            extracted = _decode_llm_json(raw_response)
+
+        if extracted.get("doc_class") not in DOC_CLASSES and extracted.get("doc_class") != "unreadable_scan":
+            raise ValueError(f"unsupported doc_class: {extracted.get('doc_class')!r}")
+        if extracted.get("intent_mode") not in INTENT_MODES:
+            raise ValueError(f"unsupported intent_mode: {extracted.get('intent_mode')!r}")
+        if not isinstance(extracted.get("referenced_contracts", []), list):
+            raise ValueError("referenced_contracts must be an array")
+        if not isinstance(extracted.get("sections", []), list):
+            raise ValueError("sections must be an array")
+
+        result = validate_extraction(extracted, document_id, batch_id)
+        if extracted.get("doc_class") == "unreadable_scan":
+            result["math_validation"]["status"] = "NOT_APPLICABLE"
+            result["status"] = "WARNING"
+        saved = _save_extraction(
+            document_id, result["extracted"], result["math_validation"],
+            result["status"], raw_response,
+        )
+        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)}
+
+
+def extract_batch(batch_id, llm_client=None):
+    """Обработать все документы batch и вернуть сводку без batch-wide failure."""
+    documents = db_docs.list_for_extraction(batch_id)
+    results = [extract_document(document, llm_client=llm_client, batch_id=batch_id) for document in documents]
+    done = sum(1 for result in results if result["ok"])
+    failed = len(results) - done
+    classes = {}
+    for result in results:
+        if result["ok"]:
+            extraction = result.get("extraction") or {}
+            doc_class = extraction.get("doc_class", "unknown")
+            classes[doc_class] = classes.get(doc_class, 0) + 1
+    return {
+        "ok": failed == 0 and bool(results),
+        "total": len(results),
+        "done": done,
+        "failed": failed,
+        "classes": classes,
+        "results": results,
+    }