From 3d84883efc6015973395f3680e034051d55ff67c Mon Sep 17 00:00:00 2001 From: Repinoid Date: Sat, 31 Oct 2026 09:42:03 +0300 Subject: [PATCH] feat: add stateless document extraction --- site/db/documents.py | 15 ++ site/services/extractor.py | 328 +++++++++++++++++++++++++++++++++++++ 2 files changed, 343 insertions(+) create mode 100644 site/services/extractor.py diff --git a/site/db/documents.py b/site/db/documents.py index 6e26e9f..6450c33 100644 --- a/site/db/documents.py +++ b/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( diff --git a/site/services/extractor.py b/site/services/extractor.py new file mode 100644 index 0000000..7aa00db --- /dev/null +++ b/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, + } \ No newline at end of file