341 lines
15 KiB
Python
341 lines
15 KiB
Python
"""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", 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, progress_callback=None, documents=None):
|
||
"""Обработать документы batch и вернуть сводку без batch-wide failure.
|
||
|
||
Optional progress callback получает (completed_count, document_result) после
|
||
каждого документа; ошибки callback не должны останавливать обработку batch.
|
||
"""
|
||
if documents is None:
|
||
documents = db_docs.list_for_extraction(batch_id)
|
||
results = []
|
||
for completed, document in enumerate(documents, start=1):
|
||
result = extract_document(document, llm_client=llm_client, batch_id=batch_id)
|
||
results.append(result)
|
||
if progress_callback is not None:
|
||
try:
|
||
progress_callback(completed, result)
|
||
except Exception:
|
||
log.exception("Extraction progress callback failed for batch %s", batch_id)
|
||
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,
|
||
} |