467 lines
22 KiB
Python
467 lines
22 KiB
Python
"""Построение истории услуг из паспортов документов 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",
|
||
"vat_rate": _decimal_text(section.get("vat_rate")),
|
||
"source_document_id": document["document_id"],
|
||
"table_ref": section.get("table_ref") or "",
|
||
"quote": _row_quote(row),
|
||
"status": status,
|
||
})
|
||
|
||
|
||
def _resolved_option_for(document_id, row, resolved_exceptions):
|
||
"""Найти сохранённый выбор для точной строки и документа lineage исключения."""
|
||
quote = _row_quote(row)
|
||
for issue in resolved_exceptions:
|
||
if (
|
||
issue.get("stage") == "lineage_matching"
|
||
and issue.get("document_id") == document_id
|
||
and issue.get("source_quote") == quote
|
||
):
|
||
resolution = issue.get("resolution")
|
||
if isinstance(resolution, dict):
|
||
option = resolution.get("selected_option")
|
||
return option if isinstance(option, dict) else None
|
||
return None
|
||
|
||
|
||
def _reusable_service_item(identity, service_items):
|
||
"""Найти прежнюю устойчивую identity, не включая будущие строки в matching."""
|
||
family = identity.get("service_family")
|
||
anchor = identity.get("instance_anchor") or ""
|
||
candidates = [item for item in service_items if item.get("service_family") == family]
|
||
if anchor:
|
||
candidates = [item for item in candidates if item.get("instance_anchor", "") == anchor]
|
||
else:
|
||
candidates = [
|
||
item for item in candidates
|
||
if item.get("canonical_name") == identity.get("canonical_name")
|
||
]
|
||
return candidates[0] if len(candidates) == 1 else None
|
||
|
||
|
||
def _process_document(dossier, document, passport, items, counts,
|
||
resolved_exceptions=None, reusable_items=None):
|
||
"""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":
|
||
selected_option = _resolved_option_for(
|
||
doc_id, row, resolved_exceptions or [],
|
||
)
|
||
if selected_option:
|
||
selected_id = selected_option.get("service_item_id")
|
||
candidates = [item for item in match["candidates"] if item.get("id") == selected_id]
|
||
if not candidates:
|
||
candidates = [
|
||
item for item in match["candidates"]
|
||
if item.get("service_family") == selected_option.get("service_family")
|
||
and item.get("instance_anchor", "") == selected_option.get("instance_anchor", "")
|
||
]
|
||
if len(candidates) == 1:
|
||
match = {
|
||
"status": "MATCHED", "service_item": candidates[0],
|
||
"candidates": candidates,
|
||
}
|
||
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"],
|
||
"service_family": item.get("service_family"),
|
||
"instance_anchor": item.get("instance_anchor", ""),
|
||
"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 = _reusable_service_item(identity, reusable_items or [])
|
||
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",
|
||
"vat_rate": _decimal_text(section.get("vat_rate")),
|
||
"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)
|
||
resolved_exceptions = db_exceptions.list_exceptions(
|
||
dossier_id=dossier_id, status="RESOLVED",
|
||
)
|
||
items = []
|
||
reusable_items = db_timeline.list_service_items(dossier_id)
|
||
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 timeline_intervals WHERE dossier_id = %s", (dossier_id,))
|
||
for document in documents:
|
||
passport = _passport_for(document)
|
||
_process_document(
|
||
dossier, document, passport, items, counts,
|
||
resolved_exceptions, reusable_items,
|
||
)
|
||
|
||
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) |