From 56997cb64e072d0ed46f822c28bf19eadb662af4 Mon Sep 17 00:00:00 2001
From: Repinoid <tazet@narod.ru>
Date: Sat, 31 Oct 2026 10:02:55 +0300
Subject: [PATCH] feat: build dossier service timelines

---
 site/db/dossiers.py               |  14 ++
 site/db/timeline.py               |  42 ++++
 site/services/timeline_builder.py | 402 ++++++++++++++++++++++++++++++
 3 files changed, 458 insertions(+)
 create mode 100644 site/services/timeline_builder.py

diff --git a/site/db/dossiers.py b/site/db/dossiers.py
index 4d767c9..7cbf27e 100644
--- a/site/db/dossiers.py
+++ b/site/db/dossiers.py
@@ -107,6 +107,20 @@ def get_dossier(dossier_id):
     return _decode_dossier(rows[0]) if rows else None
 
 
+def list_dossier_documents(dossier_id):
+    """Вернуть документы досье в установленном порядке вместе с ролью связи."""
+    return query(
+        """SELECT dd.*, d.filename, d.status AS document_status, d.doc_type,
+                  d.own_number, d.parent_number, d.doc_date, d.counterparty,
+                  d.zip_source
+           FROM dossier_documents dd
+           JOIN documents d ON d.id = dd.document_id
+           WHERE dd.dossier_id = %s
+           ORDER BY dd.seq_order, d.created_at, d.id""",
+        (dossier_id,),
+    )
+
+
 def clear_batch_dossiers(batch_id):
     """Удалить все досье batch; таблицы с ON DELETE CASCADE очистятся вместе с ними."""
     with transaction():
diff --git a/site/db/timeline.py b/site/db/timeline.py
index 28c5e2d..d640032 100644
--- a/site/db/timeline.py
+++ b/site/db/timeline.py
@@ -30,6 +30,14 @@ def get_service_item(service_item_id):
     return rows[0] if rows else None
 
 
+def list_service_items(dossier_id):
+    """Получить все идентичности услуг досье в стабильном порядке."""
+    return query(
+        "SELECT * FROM service_items WHERE dossier_id = %s ORDER BY created_at, id",
+        (dossier_id,),
+    )
+
+
 def insert_interval(data):
     """Создать интервал, сохраняя точные цены/количества строками.
 
@@ -65,6 +73,40 @@ def close_interval(interval_id, valid_to, closed_by_document_id=None):
     )
 
 
+def update_interval(interval_id, values):
+    """Обновить разрешённый набор полей интервала без динамических имён от клиента."""
+    allowed = {
+        "valid_from", "valid_to", "date_status", "date_basis", "raw_name",
+        "config_params_json", "price", "qty", "sum", "vat_mode", "table_ref",
+        "quote", "status", "closed_by_document_id", "source_document_id",
+    }
+    selected = {field: value for field, value in values.items() if field in allowed}
+    if not selected:
+        return 0
+    assignments = ", ".join(f"{field} = %s" for field in selected)
+    return execute(
+        f"UPDATE timeline_intervals SET {assignments} WHERE id = %s",
+        [*selected.values(), interval_id],
+    )
+
+
+def list_open_intervals(dossier_id, service_item_id=None):
+    """Вернуть открытые интервалы, опционально отфильтрованные по услуге."""
+    if service_item_id is None:
+        return query(
+            """SELECT * FROM timeline_intervals
+               WHERE dossier_id = %s AND valid_to IS NULL
+               ORDER BY valid_from, service_item_id, id""",
+            (dossier_id,),
+        )
+    return query(
+        """SELECT * FROM timeline_intervals
+           WHERE dossier_id = %s AND service_item_id = %s AND valid_to IS NULL
+           ORDER BY valid_from, id""",
+        (dossier_id, service_item_id),
+    )
+
+
 def list_timeline(dossier_id, as_of_date=None):
     """Получить историю либо активный на дату срез по досье.
 
diff --git a/site/services/timeline_builder.py b/site/services/timeline_builder.py
new file mode 100644
index 0000000..bd5c617
--- /dev/null
+++ b/site/services/timeline_builder.py
@@ -0,0 +1,402 @@
+"""Построение истории услуг из паспортов документов dossier.
+
+Сервис преобразует извлечённые утверждения в стабильные ServiceItem и интервалы
+обязательств. Полный snapshot закрывает только отсутствующие recurring-услуги;
+разовые и PAYG-позиции не удаляются из истории. Дата окончания в документе
+считается включительной и переводится в исключительную границу следующего дня,
+поскольку DB-срез использует интервалы [valid_from, valid_to).
+"""
+import json
+import re
+from datetime import date, datetime, timedelta
+from decimal import Decimal, InvalidOperation
+
+from db import dossiers as db_dossiers
+from db import exceptions as db_exceptions
+from db import extractions as db_extractions
+from db import timeline as db_timeline
+from db.connection import execute, query, transaction
+from services.metrics import normalize_date
+from services.service_identity import analyze_service_row, match_service_item
+
+
+def _date(value):
+    """Нормализовать дату в ISO и отклонить несуществующие календарные даты."""
+    if not value:
+        return None
+    normalized = normalize_date(str(value).strip())
+    try:
+        return datetime.strptime(normalized, "%Y-%m-%d").date()
+    except (TypeError, ValueError):
+        return None
+
+
+def _date_text(value):
+    """Сериализовать дату для SQLite либо оставить SQL NULL."""
+    return value.isoformat() if value else None
+
+
+def _decimal_text(value):
+    """Сохранить числовое значение без float-преобразований и лишних нулей."""
+    if value is None or value == "":
+        return None
+    try:
+        number = Decimal(str(value).replace("\xa0", "").replace(" ", "").replace(",", "."))
+    except (InvalidOperation, ValueError, TypeError):
+        return None
+    if not number.is_finite():
+        return None
+    return format(number.normalize(), "f")
+
+
+def _effective_dates(row, section, passport, document):
+    """Получить valid_from/valid_to и provenance дат для одной строки.
+
+    Явная дата строки приоритетнее периода секции. При унаследованном конце
+    периода дата интерпретируется как включительная и получает границу +1 день.
+    Дата документа используется только как последнее известное начало, но не
+    маскируется под явно указанную дату услуги.
+    """
+    row_start = _date(row.get("date_start"))
+    section_start = _date(section.get("period_start"))
+    document_date = _date(passport.get("own_date") or document.get("doc_date"))
+    start = row_start or section_start or document_date
+    row_end = _date(row.get("date_end"))
+    section_end = _date(section.get("period_end"))
+    inclusive_end = row_end or section_end
+    end = inclusive_end + timedelta(days=1) if inclusive_end else None
+    if row_start:
+        date_status = row.get("date_status") or "explicit"
+        date_basis = "row.date_start"
+    elif section_start:
+        date_status = "inherited" if row.get("date_status") != "unknown" else "unknown"
+        date_basis = "section.period_start"
+    else:
+        date_status = "unknown"
+        date_basis = "document.own_date" if document_date else "missing"
+    return start, end, date_status, date_basis
+
+
+def _config_json(config_params):
+    """Детерминированно сериализовать параметры конфигурации для сравнения."""
+    return json.dumps(config_params or {}, ensure_ascii=False, sort_keys=True, default=str)
+
+
+def _row_quote(row):
+    """Собрать короткую аудируемую цитату исходной строки."""
+    return str(row.get("raw_name") or "")[:1000]
+
+
+def _add_lineage_exception(dossier_id, batch_id, document_id, title,
+                            quote="", service_item_id=None, options=None):
+    """Создать исключение сопоставления один раз на документ и заголовок."""
+    existing = db_exceptions.list_exceptions(
+        dossier_id=dossier_id, status="OPEN",
+    )
+    if any(
+        item.get("document_id") == document_id
+        and item.get("stage") == "lineage_matching"
+        and item.get("title") == title
+        for item in existing
+    ):
+        return
+    db_exceptions.add_exception(
+        batch_id=batch_id, dossier_id=dossier_id, document_id=document_id,
+        service_item_id=service_item_id, severity="AMBIGUOUS",
+        stage="lineage_matching", title=title, source_quote=quote,
+        options=options or [],
+    )
+
+
+def _passport_for(document):
+    """Вернуть паспорт документа или пустую форму для неполных legacy-досье."""
+    passport = db_extractions.get_by_document(document["document_id"])
+    return passport or {
+        "document_id": document["document_id"], "doc_class": "unknown",
+        "intent_mode": "no_spec", "own_date": None, "sections": [],
+        "math_validation": {},
+    }
+
+
+def _ordered_documents(dossier_id):
+    """Упорядочить документы по роли, затем по дате и номеру допсоглашения."""
+    documents = db_dossiers.list_dossier_documents(dossier_id)
+    role_order = {"frame_contract": 0, "base_spec": 1, "specification": 2, "supplement": 3}
+
+    def key(document):
+        passport = _passport_for(document)
+        role = document.get("role") or "document"
+        own_number = str(passport.get("own_number") or "")
+        digits = re.sub(r"\D", "", own_number)
+        return (
+            role_order.get(role, 4),
+            int(digits) if digits else 0,
+            passport.get("own_date") or document.get("doc_date") or "9999",
+            document["document_id"],
+        )
+
+    return sorted(documents, key=key)
+
+
+def _interval_content_matches(interval, row, identity, section):
+    """Сравнить изменяемые утверждения, не используя их для entity identity."""
+    return (
+        interval.get("charge_type") == (row.get("charge_type") or section.get("charge_type") or "unknown")
+        and _decimal_text(interval.get("price")) == _decimal_text(row.get("price"))
+        and _decimal_text(interval.get("qty")) == _decimal_text(row.get("qty"))
+        and _decimal_text(interval.get("sum")) == _decimal_text(row.get("sum"))
+        and interval.get("config_params_json") == _config_json(identity.get("config_params"))
+        and interval.get("vat_mode", "vat_unknown") == (section.get("vat_mode") or "vat_unknown")
+    )
+
+
+def _close_open_interval(interval, close_date, document_id):
+    """Закрыть открытый интервал только если новая граница после его начала."""
+    start = _date(interval.get("valid_from"))
+    if close_date and start and close_date > start:
+        db_timeline.close_interval(interval["id"], close_date.isoformat(), document_id)
+        return True
+    return False
+
+
+def _append_interval(dossier_id, service_item, row, section, document,
+                     passport, start, end, date_status, date_basis, status):
+    """Записать один интервал с точными десятичными строками и provenance."""
+    identity = analyze_service_row(row.get("raw_name"), row.get("charge_type") or section.get("charge_type"))
+    return db_timeline.insert_interval({
+        "dossier_id": dossier_id,
+        "service_item_id": service_item["id"],
+        "charge_type": row.get("charge_type") or section.get("charge_type") or "unknown",
+        "valid_from": _date_text(start),
+        "valid_to": _date_text(end),
+        "date_status": date_status,
+        "date_basis": date_basis,
+        "raw_name": row.get("raw_name") or identity["canonical_name"] or "",
+        "config_params_json": _config_json(identity["config_params"]),
+        "price": _decimal_text(row.get("price")),
+        "qty": _decimal_text(row.get("qty")),
+        "sum": _decimal_text(row.get("sum")),
+        "vat_mode": section.get("vat_mode") or "vat_unknown",
+        "source_document_id": document["document_id"],
+        "table_ref": section.get("table_ref") or "",
+        "quote": _row_quote(row),
+        "status": status,
+    })
+
+
+def _process_document(dossier, document, passport, items, counts):
+    """Reconcile every section of one passport against current dossier services."""
+    doc_id = document["document_id"]
+    mode = passport.get("intent_mode")
+    if mode == "no_spec" or passport.get("doc_class") not in {"specification", "supplement"}:
+        return
+    sections = [section for section in passport.get("sections") or [] if isinstance(section, dict)]
+    sections.sort(key=lambda section: _date(section.get("period_start")) or date.max)
+    seen_recurring = set()
+    effective_dates = []
+    math_data = passport.get("math_validation") or {}
+    extraction_status = passport.get("status") or "UNKNOWN"
+    snapshot_is_complete = mode == "full_snapshot" and extraction_status not in {"CONFLICT", "FAILED"}
+
+    for section in sections:
+        rows = [row for row in section.get("rows") or [] if isinstance(row, dict)]
+        for row in rows:
+            action = (row.get("delta_action") or ("snapshot" if mode == "full_snapshot" else "unknown")).lower()
+            charge_type = row.get("charge_type") or section.get("charge_type") or "unknown"
+            if mode == "delta_amendment" and action in {"snapshot", "unknown", "unresolved"}:
+                _add_lineage_exception(
+                    dossier["id"], dossier.get("batch_id"), doc_id,
+                    "UNRESOLVED_DELTA_ACTION", _row_quote(row),
+                )
+                counts["ambiguous"] += 1
+                continue
+            start, end, date_status, date_basis = _effective_dates(row, section, passport, document)
+            if start:
+                effective_dates.append(start)
+            identity = analyze_service_row(row.get("raw_name"), charge_type)
+            match = match_service_item(identity, items)
+            if match["status"] == "AMBIGUOUS":
+                _add_lineage_exception(
+                    dossier["id"], dossier.get("batch_id"), doc_id,
+                    "AMBIGUOUS_SERVICE_ITEM", _row_quote(row),
+                    options=[{"service_item_id": item["id"], "canonical_name": item.get("canonical_name")}
+                             for item in match["candidates"]],
+                )
+                counts["ambiguous"] += 1
+                continue
+            service_item = match["service_item"]
+            if service_item is None:
+                service_item = db_timeline.insert_service_item(
+                    dossier_id=dossier["id"],
+                    service_family=identity["service_family"],
+                    canonical_name=identity["canonical_name"] or "(empty)",
+                    instance_anchor=identity["instance_anchor"],
+                )
+                items.append(service_item)
+
+            open_intervals = [
+                interval for interval in db_timeline.list_open_intervals(dossier["id"], service_item["id"])
+                if interval.get("charge_type") == charge_type
+            ]
+            history = [
+                interval for interval in db_timeline.list_timeline(dossier["id"])
+                if interval.get("service_item_id") == service_item["id"]
+                and interval.get("charge_type") == charge_type
+            ]
+            if charge_type == "recurring":
+                seen_recurring.add(service_item["id"])
+
+            if action in {"delete", "terminate"}:
+                if not start:
+                    _add_lineage_exception(
+                        dossier["id"], dossier.get("batch_id"), doc_id,
+                        "TERMINATION_DATE_UNKNOWN", _row_quote(row), service_item["id"],
+                    )
+                    counts["ambiguous"] += 1
+                elif not open_intervals:
+                    _add_lineage_exception(
+                        dossier["id"], dossier.get("batch_id"), doc_id,
+                        "TERMINATION_TARGET_NOT_OPEN", _row_quote(row), service_item["id"],
+                    )
+                    counts["ambiguous"] += 1
+                else:
+                    for interval in open_intervals:
+                        if _close_open_interval(interval, start, doc_id):
+                            counts["closed"] += 1
+                continue
+
+            if not start:
+                _add_lineage_exception(
+                    dossier["id"], dossier.get("batch_id"), doc_id,
+                    "SERVICE_START_DATE_UNKNOWN", _row_quote(row), service_item["id"],
+                )
+                date_status = "unknown"
+                interval_status = "WARNING"
+                counts["ambiguous"] += 1
+            else:
+                interval_status = "CONFLICT" if math_data.get("status") == "CONFLICT" else "VERIFIED"
+
+            previous = open_intervals[-1] if open_intervals else None
+            if previous and _interval_content_matches(previous, row, identity, section):
+                previous_start = _date(previous.get("valid_from"))
+                if previous_start == start or start is None:
+                    new_end = end or previous.get("valid_to")
+                    db_timeline.update_interval(previous["id"], {
+                        "valid_to": _date_text(end) if end else new_end,
+                        "date_status": date_status,
+                    })
+                    counts["unchanged"] += 1
+                    continue
+
+            if not previous and history and start:
+                latest = history[-1]
+                if (
+                    latest.get("valid_to") == start.isoformat()
+                    and _interval_content_matches(latest, row, identity, section)
+                ):
+                    db_timeline.update_interval(latest["id"], {
+                        "valid_to": _date_text(end),
+                        "closed_by_document_id": None,
+                    })
+                    counts["unchanged"] += 1
+                    continue
+
+            if previous and start and _date(previous.get("valid_from")) == start:
+                db_timeline.update_interval(previous["id"], {
+                    "valid_to": _date_text(end),
+                    "raw_name": row.get("raw_name") or "",
+                    "config_params_json": _config_json(identity["config_params"]),
+                    "price": _decimal_text(row.get("price")),
+                    "qty": _decimal_text(row.get("qty")),
+                    "sum": _decimal_text(row.get("sum")),
+                    "vat_mode": section.get("vat_mode") or "vat_unknown",
+                    "source_document_id": doc_id,
+                    "closed_by_document_id": None,
+                    "table_ref": section.get("table_ref") or "",
+                    "quote": _row_quote(row),
+                    "status": interval_status,
+                })
+                counts["updated"] += 1
+                continue
+
+            for interval in open_intervals:
+                if start and _close_open_interval(interval, start, doc_id):
+                    counts["closed"] += 1
+            _append_interval(
+                dossier["id"], service_item, row, section, document, passport,
+                start, end, date_status, date_basis, interval_status,
+            )
+            counts["created"] += 1
+
+    if snapshot_is_complete and sections and effective_dates:
+        snapshot_date = min(effective_dates)
+        for service_item in items:
+            if service_item["id"] in seen_recurring:
+                continue
+            for interval in db_timeline.list_open_intervals(dossier["id"], service_item["id"]):
+                if interval.get("charge_type") == "recurring" and _close_open_interval(interval, snapshot_date, doc_id):
+                    counts["closed"] += 1
+
+
+def build_dossier_timeline(dossier_id, contract_id=None):
+    """Пересобрать таймлайн dossier из сохранённых паспортов без повторного вызова LLM.
+
+    Для legacy-синхронизации caller должен передать `contract_id`: в схеме нет
+    автоматической связи `contract_dossiers.id` с `contracts.id`, поэтому сервис
+    не выводит её из похожих номеров или имён контрагентов.
+    """
+    dossier = db_dossiers.get_dossier(dossier_id)
+    if dossier is None:
+        return {"ok": False, "error": "dossier not found", "dossier_id": dossier_id}
+    documents = _ordered_documents(dossier_id)
+    items = []
+    counts = {"created": 0, "updated": 0, "unchanged": 0, "closed": 0, "ambiguous": 0}
+
+    with transaction():
+        execute(
+            """DELETE FROM exceptions_queue
+               WHERE dossier_id = %s AND stage = 'lineage_matching' AND status = 'OPEN'""",
+            (dossier_id,),
+        )
+        execute("DELETE FROM service_items WHERE dossier_id = %s", (dossier_id,))
+        for document in documents:
+            passport = _passport_for(document)
+            _process_document(dossier, document, passport, items, counts)
+
+    if contract_id:
+        sync_legacy_spec_current(dossier_id, contract_id)
+    return {
+        "ok": True,
+        "dossier_id": dossier_id,
+        "counts": counts,
+        "timeline": db_timeline.list_timeline(dossier_id),
+        "exceptions": db_exceptions.list_exceptions(dossier_id=dossier_id, status="OPEN"),
+    }
+
+
+def sync_legacy_spec_current(dossier_id, contract_id, as_of_date=None):
+    """Материализовать срез таймлайна в legacy spec_current по явному contract_id."""
+    current_date = _date(as_of_date) or date.today()
+    active_intervals = db_timeline.list_timeline(dossier_id, current_date.isoformat())
+    with transaction():
+        execute("DELETE FROM spec_current WHERE contract_id = %s", (contract_id,))
+        for interval in active_intervals:
+            if interval.get("valid_from") is None:
+                continue
+            item = db_timeline.get_service_item(interval["service_item_id"])
+            if not item:
+                continue
+            name_hash = __import__("hashlib").sha256(
+                f"{item['canonical_name'].strip().lower()}|{interval['valid_from']}".encode(),
+            ).hexdigest()[:16]
+            execute(
+                """INSERT INTO spec_current
+                   (id, contract_id, name_hash, name, price, qty, sum, date_start)
+                   VALUES (%s, %s, %s, %s, %s, %s, %s, %s)""",
+                (
+                    interval["id"], contract_id, name_hash, interval["raw_name"],
+                    interval.get("price"), interval.get("qty"), interval.get("sum"),
+                    interval["valid_from"],
+                ),
+            )
+    return len(active_intervals)
\ No newline at end of file