diff --git a/deploy/contracts.py b/deploy/contracts.py index 9916403..9e2bffb 100644 --- a/deploy/contracts.py +++ b/deploy/contracts.py @@ -15,10 +15,11 @@ class ParseResult: @classmethod def from_row(cls, row: dict): """Из строки БД (documents).""" + elements = row.get("elements_json", []) return cls( status=row.get("status", "error"), - element_count=0, # вычисляется отдельно - elements=row.get("elements_json", []), + element_count=len(elements) if isinstance(elements, list) else 0, + elements=elements, error=row.get("error_message"), ) @@ -44,22 +45,26 @@ class ClassifyResult: ) +@dataclass(frozen=True) +class BatchGroupingResult: + """Выход grouping.py — результат группировки всего батча.""" + groups: list = field(default_factory=list) # list[GroupingResult] + unresolved: list = field(default_factory=list) # документы без группы + + @dataclass(frozen=True) class GroupingResult: """Выход grouping.py — результат группировки по контрактам.""" contract_number: str # нормализованный номер договора counterparty: Optional[str] = None documents: list = field(default_factory=list) # [{doc_id, filename, doc_type, ...}] - unresolved: list = field(default_factory=list) # документы без группы @classmethod def from_row(cls, row: dict): - """Из строки БД или вычисленной группы.""" return cls( contract_number=row.get("contract_number", ""), counterparty=row.get("counterparty"), documents=row.get("documents", []), - unresolved=row.get("unresolved", []), ) diff --git a/deploy/repository.py b/deploy/repository.py index 4c77cae..c662271 100644 --- a/deploy/repository.py +++ b/deploy/repository.py @@ -63,6 +63,30 @@ class Repository(Protocol): """Текущая спецификация контракта.""" ... + def get_document(self, doc_id: str) -> dict | None: + """Получить документ по id.""" + ... + + def list_by_batch(self, batch_id: str) -> list[dict]: + """Все документы батча с полями классификации.""" + ... + + def count_by_status(self, batch_id: str) -> dict[str, int]: + """Количество документов по classify_status.""" + ... + + def reset_classify_status(self, batch_id: str) -> None: + """Сбросить classify_status на 'pending'.""" + ... + + def set_classify_processing(self, doc_id: str) -> None: + """Пометить документ как обрабатываемый.""" + ... + + def delete_document(self, doc_id: str) -> None: + """Удалить документ.""" + ... + # ── Продакшен: обёртка над db/*.py ────────────────────────────── @@ -76,8 +100,7 @@ class PgRepository: def set_document_parsed(self, doc_id, elements): from db import documents - import json as _json - documents.set_parsed(doc_id, _json.dumps(elements, ensure_ascii=False)) + documents.set_parsed(doc_id, elements) # documents.set_parsed уже делает json.dumps def set_document_error(self, doc_id, error): from db import documents @@ -118,6 +141,30 @@ class PgRepository: from db import spec_current return spec_current.list_by_contract(contract_id) + def get_document(self, doc_id): + from db import documents + return documents.get(doc_id) + + def list_by_batch(self, batch_id): + from db import documents + return documents.list_by_batch(batch_id) + + def count_by_status(self, batch_id): + from db import documents + return documents.count_by_status(batch_id) + + def reset_classify_status(self, batch_id): + from db import documents + documents.reset_classify_status(batch_id) + + def set_classify_processing(self, doc_id): + from db import documents + documents.set_classify_processing(doc_id) + + def delete_document(self, doc_id): + from db import documents + documents.delete(doc_id) + # ── Тестовый: in-memory заглушка ──────────────────────────────── @@ -191,3 +238,29 @@ class MemRepository: def get_spec_current(self, contract_id): return self.spec_current.get(contract_id, []) + + def get_document(self, doc_id): + return self.documents.get(doc_id) + + def list_by_batch(self, batch_id): + return [d for d in self.documents.values() if d.get("batch_id") == batch_id] + + def count_by_status(self, batch_id): + counts = {} + for d in self.documents.values(): + if d.get("batch_id") == batch_id: + s = d.get("classify_status", "unknown") + counts[s] = counts.get(s, 0) + 1 + return counts + + def reset_classify_status(self, batch_id): + for d in self.documents.values(): + if d.get("batch_id") == batch_id: + d["classify_status"] = "pending" + + def set_classify_processing(self, doc_id): + if doc_id in self.documents: + self.documents[doc_id]["classify_status"] = "processing" + + def delete_document(self, doc_id): + self.documents.pop(doc_id, None) diff --git a/deploy/services/classify.py b/deploy/services/classify.py index 227c3be..1edb347 100644 --- a/deploy/services/classify.py +++ b/deploy/services/classify.py @@ -79,13 +79,18 @@ def _call_llm_classify(header_text, llm_client=None): return parsed, raw_text, needed_fix -def classify_batch(batch_id): +def classify_batch(batch_id, llm_client=None, repo=None): """ Классифицировать все документы в batch. Сбрасывает статус на 'pending' для всех перед началом. Параллельно (ThreadPoolExecutor) обрабатывает до MAX_WORKERS документов. - Возвращает {ok, total, done, failed}. + Возвращает {ok, total, done, failed, garbage, json_fix_rate}. + llm_client: LLMClient (optional, default — HttpxLLMClient) + repo: Repository (optional, default — direct db.* calls) """ + _db = repo if repo else db_docs + _llm = llm_client if llm_client else _get_classify_client() + # Сбросить статус — allow re-classify after file changes db_docs.reset_classify_status(batch_id) pending = db_docs.list_pending(batch_id) @@ -118,7 +123,7 @@ def classify_batch(batch_id): # Stage 3: LLM classification (only for remaining) db_docs.set_classify_processing(doc["id"]) # crash recovery marker - result, raw, needed_fix = _call_llm_classify(text) + result, raw, needed_fix = _call_llm_classify(text, _llm) json_total += 1 if needed_fix: json_fixes += 1