fix: ревью Opus — баги #1-2 + Protocol расширен + BatchGroupingResult

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,  # вычисляется отдельно
+            element_count=len(elements) if isinstance(elements, list) else 0,
-            elements=row.get("elements_json", []),
+            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", []),
         )
 
 

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, elements)  # documents.set_parsed уже делает json.dumps
-        documents.set_parsed(doc_id, _json.dumps(elements, ensure_ascii=False))
 
     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)

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