"""Построение истории услуг из паспортов документов 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)