sqs/contracts-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: build dossier service timelines

Browse Source
This commit is contained in:
Repinoid
2026-10-31 10:02:55 +03:00
parent 78f4a052e9
commit 56997cb64e
3 changed files with 458 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
site/services/timeline_builder.py
+402
View File
@@ -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)