feat: build contract dossier graph

site/services/grouping.py

@@ -12,6 +12,15 @@ import re
 from db import documents as db_docs
 from db import contracts as db_contracts
 from db import supplements as db_supplements
+from db import dossiers as db_dossiers
+from db import exceptions as db_exceptions
+from db import extractions as db_extractions
+
+
+_HOMOGLYPHS = str.maketrans({
+    "A": "А", "B": "В", "C": "С", "E": "Е", "H": "Н", "K": "К",
+    "M": "М", "O": "О", "P": "Р", "T": "Т", "X": "Х", "Y": "У",
+})
 
 
 def normalize_number(num):
@@ -25,6 +34,288 @@ def normalize_number(num):
     return re.sub(r"[^A-Z0-9А-Я]", "", num.upper())
 
 
+def canonicalize_contract_number(number):
+    """Нормализовать полный номер, включая визуальные латинско-кириллические пары.
+
+    Функция намеренно не отбрасывает префиксы и не строит ключ только по
+    суффиксу: одинаковый короткий хвост у разных контрагентов не является
+    достаточным доказательством тождества договоров.
+    """
+    if not number:
+        return ""
+    normalized = str(number).upper().translate(_HOMOGLYPHS)
+    return re.sub(r"[^0-9А-ЯЁ]", "", normalized)
+
+
+def canonicalize_counterparty(name, inn):
+    """Вернуть ключи контрагента для сравнения с жёстким ограничением по ИНН."""
+    inn_key = re.sub(r"\D", "", str(inn or ""))
+    if inn_key and len(inn_key) != 10 and len(inn_key) != 12:
+        inn_key = ""
+    normalized_name = str(name or "").upper().translate(_HOMOGLYPHS)
+    normalized_name = re.sub(r"[«»\"'`.,()\s]", "", normalized_name)
+    normalized_name = re.sub(r"^(ООО|ЗАО|АО|ПАО|ИП)", "", normalized_name)
+    return {"name": normalized_name, "inn": inn_key}
+
+
+def _counterparties_compatible(left, right):
+    """Разрешить связь при совпадении идентичности или отсутствующих данных."""
+    left_id = canonicalize_counterparty(left.get("name"), left.get("inn"))
+    right_id = canonicalize_counterparty(right.get("name"), right.get("inn"))
+    if left_id["inn"] and right_id["inn"]:
+        return left_id["inn"] == right_id["inn"]
+    left_has_identity = bool(left_id["inn"] or left_id["name"])
+    right_has_identity = bool(right_id["inn"] or right_id["name"])
+    if not left_has_identity or not right_has_identity:
+        return True
+    return bool(left_id["name"] and left_id["name"] == right_id["name"])
+
+
+def _passport_keys(passport):
+    """Собрать полные ключи договора, исключая номер самого допсоглашения."""
+    doc_class = passport.get("doc_class")
+    values = []
+    if doc_class in {"contract_frame", "specification"} and passport.get("own_number"):
+        values.append((passport["own_number"], "own_number"))
+    for reference in passport.get("referenced_numbers") or []:
+        if isinstance(reference, dict) and reference.get("number"):
+            values.append((reference["number"], reference.get("source") or "referenced_contract"))
+    unique = {}
+    for raw, source in values:
+        key = canonicalize_contract_number(raw)
+        if key:
+            unique.setdefault(key, {"raw": str(raw), "source": str(source)})
+    return unique
+
+
+def _number_sort_key(number):
+    """Разбить номер на текстовые и числовые части для сортировки ДС по номеру."""
+    return tuple((0, int(part)) if part.isdigit() else (1, part.casefold())
+                 for part in re.split(r"(\d+)", str(number or "")))
+
+
+def _dossier_exception(batch_id, document_id, title, source_quote):
+    """Создать устойчивую запись исключения, не дублируя её при пересборке batch."""
+    existing = db_exceptions.list_exceptions(batch_id=batch_id)
+    if any(
+        item.get("document_id") == document_id
+        and item.get("stage") == "dossier_linking"
+        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="AMBIGUOUS", stage="dossier_linking",
+        title=title, source_quote=source_quote,
+    )
+
+
+def build_dossiers_for_batch(batch_id):
+    """Построить и сохранить dossier graph из паспортов документов одного batch.
+
+    Полные номера и ссылки образуют рёбра только при совместимом контрагенте.
+    Компоненты дополнительно защищены от транзитивного слияния двух разных ИНН;
+    документ, который остаётся кандидатом сразу для нескольких таких компонент,
+    исключается из автоматических связей и передаётся в очередь исключений.
+    """
+    documents = db_docs.list_by_batch(batch_id)
+    passports = db_extractions.list_by_batch(batch_id)
+    passport_by_id = {row["document_id"]: row for row in passports}
+    records = []
+    for document in documents:
+        passport = passport_by_id.get(document["id"])
+        if not passport:
+            passport = {"document_id": document["id"], "doc_class": "unknown"}
+        if passport.get("doc_class") == "ignored_accounting":
+            continue
+        counterparty = {
+            "name": passport.get("counterparty_name"),
+            "inn": passport.get("counterparty_inn"),
+        }
+        records.append({
+            "document": document,
+            "passport": passport,
+            "counterparty": counterparty,
+            "keys": _passport_keys(passport),
+        })
+
+    parent = list(range(len(records)))
+    component_inns = []
+    for record in records:
+        counterparty_id = canonicalize_counterparty(
+            record["counterparty"].get("name"), record["counterparty"].get("inn"),
+        )
+        component_inns.append({counterparty_id["inn"]} if counterparty_id["inn"] else set())
+
+    def find(index):
+        while parent[index] != index:
+            parent[index] = parent[parent[index]]
+            index = parent[index]
+        return index
+
+    candidates = [[] for _ in records]
+    for left_index, left in enumerate(records):
+        for right_index in range(left_index + 1, len(records)):
+            right = records[right_index]
+            shared = set(left["keys"]) & set(right["keys"])
+            if not shared or not _counterparties_compatible(left["counterparty"], right["counterparty"]):
+                continue
+            shared_key = sorted(shared)[0]
+            candidates[left_index].append((right_index, shared_key))
+            candidates[right_index].append((left_index, shared_key))
+            left_root, right_root = find(left_index), find(right_index)
+            if left_root == right_root:
+                continue
+            merged_inns = component_inns[left_root] | component_inns[right_root]
+            if len(merged_inns) > 1:
+                continue
+            parent[right_root] = left_root
+            component_inns[left_root] = merged_inns
+
+    ambiguous_indices = set()
+    for index, edges in enumerate(candidates):
+        key_targets = {}
+        for other, shared_key in edges:
+            key_targets.setdefault(shared_key, set()).add(find(other))
+        candidate_roots = set().union(*key_targets.values()) if key_targets else set()
+        if len(candidate_roots) > 1:
+            ambiguous_indices.add(index)
+            matching_numbers = sorted({
+                records[index]["keys"].get(key, {}).get("raw", key)
+                for _other, key in edges
+            })
+            _dossier_exception(
+                batch_id, records[index]["document"]["id"],
+                "AMBIGUOUS_DOSSIER_LINK", ", ".join(matching_numbers),
+            )
+
+    component_members = {}
+    for index, record in enumerate(records):
+        if index in ambiguous_indices:
+            continue
+        if not record["keys"]:
+            _dossier_exception(
+                batch_id, record["document"]["id"], "NO_RELIABLE_CONTRACT_NUMBER",
+                record["document"].get("filename") or "",
+            )
+            continue
+        component_members.setdefault(find(index), []).append(record)
+
+    db_dossiers.clear_batch_dossiers(batch_id)
+    dossier_results = []
+    for members in component_members.values():
+        frames = [r for r in members if r["passport"].get("doc_class") == "contract_frame"]
+        specifications = [r for r in members if r["passport"].get("doc_class") == "specification"]
+        specifications.sort(key=lambda r: (
+            r["passport"].get("own_date") or r["document"].get("doc_date") or "9999",
+            r["document"].get("id", ""),
+        ))
+        identifiers = []
+        for record in members:
+            for key, value in record["keys"].items():
+                identifiers.append((key, value["raw"], value["source"]))
+        canonical_candidates = [
+            (r["passport"].get("own_number"), "frame") for r in frames
+            if r["passport"].get("own_number")
+        ] + [
+            (r["passport"].get("own_number"), "specification") for r in specifications
+            if r["passport"].get("own_number")
+        ] + [(raw, source) for _key, raw, source in identifiers]
+        canonical_number = canonical_candidates[0][0] if canonical_candidates else ""
+        canonical_key = canonicalize_contract_number(canonical_number)
+        aliases = list(dict.fromkeys(
+            raw for key, raw, _source in identifiers
+            if key != canonical_key and raw != canonical_number
+        ))
+        counterparty_record = next(
+            (r["counterparty"] for r in members
+             if r["counterparty"].get("inn") or r["counterparty"].get("name")),
+            {},
+        )
+        dossier = db_dossiers.create_dossier(
+            batch_id=batch_id,
+            canonical_number=canonical_number,
+            alias_numbers=aliases,
+            counterparty_name=counterparty_record.get("name") or "",
+            counterparty_inn=counterparty_record.get("inn") or "",
+            has_frame_contract=bool(frames),
+        )
+        base_spec_id = specifications[0]["document"]["id"] if specifications else None
+        members.sort(key=lambda r: _dossier_order_key(r, base_spec_id))
+        linked_documents = []
+        for sequence, record in enumerate(members, start=1):
+            document = record["document"]
+            passport = record["passport"]
+            doc_class = passport.get("doc_class")
+            if doc_class == "contract_frame":
+                role = "frame_contract"
+            elif doc_class == "specification" and document["id"] == base_spec_id:
+                role = "base_spec"
+            elif doc_class == "specification":
+                role = "specification"
+            elif doc_class == "supplement":
+                role = "supplement"
+            else:
+                role = "document"
+            reason = _link_reason(record, canonical_key, canonical_number)
+            link = db_dossiers.link_document(
+                dossier_id=dossier["id"], document_id=document["id"], role=role,
+                link_status="CONFIRMED", link_reason=reason, seq_order=sequence,
+            )
+            linked_documents.append({**document, "role": role, "link_reason": reason, "link": link})
+        dossier_results.append({**dossier, "documents": linked_documents})
+
+    return {
+        "ok": True,
+        "batch_id": batch_id,
+        "total_docs": len(documents),
+        "dossiers": dossier_results,
+        "unlinked": sorted(records[i]["document"]["id"] for i in ambiguous_indices)
+        + [r["document"]["id"] for r in records if not r["keys"]],
+        "ambiguous": len(ambiguous_indices),
+    }
+
+
+def _dossier_order_key(record, base_spec_id):
+    """Упорядочить документы так, чтобы рамка и базовая спека шли первыми."""
+    document = record["document"]
+    passport = record["passport"]
+    doc_class = passport.get("doc_class")
+    if doc_class == "contract_frame":
+        rank = 0
+    elif doc_class == "specification" and document["id"] == base_spec_id:
+        rank = 1
+    elif doc_class == "specification":
+        rank = 2
+    elif doc_class == "supplement":
+        rank = 3
+    else:
+        rank = 4
+    if doc_class == "supplement":
+        secondary = (_number_sort_key(passport.get("own_number")),
+                     passport.get("own_date") or document.get("doc_date") or "9999")
+    else:
+        secondary = (passport.get("own_date") or document.get("doc_date") or "9999",)
+    return rank, secondary, document.get("id", "")
+
+
+def _link_reason(record, canonical_key, canonical_number):
+    """Сформировать аудируемое основание связи из совпавшей ссылки документа."""
+    for reference in record["passport"].get("referenced_numbers") or []:
+        if not isinstance(reference, dict) or not reference.get("number"):
+            continue
+        if canonicalize_contract_number(reference["number"]) == canonical_key:
+            source = reference.get("source")
+            if source in {"header", "preamble", "appendix"}:
+                return f"Связано по номеру {reference['number']} из {source}"
+            return f"Связано по номеру {reference['number']}"
+    own_number = record["passport"].get("own_number")
+    if own_number and canonicalize_contract_number(own_number) == canonical_key:
+        return f"Собственный номер документа совпадает с {canonical_number}"
+    return f"Связано в dossier по номеру {canonical_number}"
+
+
 def group_documents(batch_id):
     """
     Сгруппировать классифицированные документы по контрактам.