From 8481044deaf0a3b643dec5dd71bc08895a66f656 Mon Sep 17 00:00:00 2001 From: Repinoid Date: Sat, 31 Oct 2026 10:08:26 +0300 Subject: [PATCH] feat: reconcile dossier timelines with billing --- site/db/connection.py | 11 + site/db/timeline.py | 4 +- site/llm_prompt.py | 3 + site/services/catalog_matcher.py | 5 +- site/services/reconciliation.py | 381 ++++++++++++++++++++++++++++++ site/services/timeline_builder.py | 2 + 6 files changed, 403 insertions(+), 3 deletions(-) create mode 100644 site/services/reconciliation.py diff --git a/site/db/connection.py b/site/db/connection.py index ce54343..c613ce6 100644 --- a/site/db/connection.py +++ b/site/db/connection.py @@ -208,6 +208,7 @@ def _create_schema(conn): price TEXT, qty TEXT, sum TEXT, + vat_rate TEXT, vat_mode TEXT NOT NULL DEFAULT 'with_vat', source_document_id TEXT REFERENCES documents(id) ON DELETE CASCADE, closed_by_document_id TEXT REFERENCES documents(id) ON DELETE SET NULL, @@ -250,6 +251,16 @@ def _create_schema(conn): CREATE INDEX IF NOT EXISTS idx_exceptions_dossier ON exceptions_queue(dossier_id); CREATE INDEX IF NOT EXISTS idx_exceptions_batch ON exceptions_queue(batch_id); """) + # Existing /tmp databases survive application restarts. Additive migrations + # therefore inspect the live schema before ALTER TABLE, while fresh databases + # receive the same column from the CREATE TABLE definition above. + timeline_columns = { + row["name"] for row in conn.execute("PRAGMA table_info(timeline_intervals)").fetchall() + } + if "vat_rate" not in timeline_columns: + conn.execute("ALTER TABLE timeline_intervals ADD COLUMN vat_rate TEXT") + if not getattr(_local, "in_transaction", False): + conn.commit() def cleanup_db(): diff --git a/site/db/timeline.py b/site/db/timeline.py index d640032..b685520 100644 --- a/site/db/timeline.py +++ b/site/db/timeline.py @@ -48,7 +48,7 @@ def insert_interval(data): fields = ( "dossier_id", "service_item_id", "charge_type", "valid_from", "valid_to", "date_status", "date_basis", "raw_name", "config_params_json", "price", - "qty", "sum", "vat_mode", "source_document_id", "closed_by_document_id", + "qty", "sum", "vat_mode", "vat_rate", "source_document_id", "closed_by_document_id", "table_ref", "quote", "status", ) values = [data[field] for field in fields if field in data] @@ -77,7 +77,7 @@ 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", + "config_params_json", "price", "qty", "sum", "vat_mode", "vat_rate", "table_ref", "quote", "status", "closed_by_document_id", "source_document_id", } selected = {field: value for field, value in values.items() if field in allowed} diff --git a/site/llm_prompt.py b/site/llm_prompt.py index 2201ec8..444716d 100644 --- a/site/llm_prompt.py +++ b/site/llm_prompt.py @@ -313,6 +313,7 @@ def build_stateless_extractor_prompt(structured_text): "header_stated_total": null, "table_footer_total": null, "vat_mode": "with_vat | without_vat | vat_unknown", + "vat_rate": null, "rows": [ { "row_ref": "table_1_row_1", @@ -352,6 +353,8 @@ def build_stateless_extractor_prompt(structured_text): date_status="inherited". Иначе ставь "explicit" или "unknown". 6. Не включай итоговые строки в rows. Их суммы указывай в table_footer_total; сумму из заголовка секции указывай в header_stated_total. + Если ставка НДС указана явно, верни числовой процент в vat_rate; не выводи её из года, + контрагента или внешних правил. 7. Числа верни JSON-числами с точкой как десятичным разделителем, без валюты и разделителей тысяч. Не вычисляй и не исправляй price, qty, sum или итоги. 8. Если документ является счетом, счетом-фактурой, УПД, актом сверки, актом услуг, diff --git a/site/services/catalog_matcher.py b/site/services/catalog_matcher.py index f01adfb..893a21c 100644 --- a/site/services/catalog_matcher.py +++ b/site/services/catalog_matcher.py @@ -44,7 +44,10 @@ def match_catalog_sku(service_item, catalog_entries): continue if entry_charge not in {"unknown", charge_type}: continue - exact_sku = bool(service_sku and service_sku == entry_sku) + expected_sku = service_sku + if charge_type == "payg" and service_sku and not service_sku.endswith("-M"): + expected_sku = service_sku + "-M" + exact_sku = bool(service_sku and entry_sku in {service_sku, expected_sku}) exact_name = bool( service_name and service_name in { diff --git a/site/services/reconciliation.py b/site/services/reconciliation.py new file mode 100644 index 0000000..af9d1aa --- /dev/null +++ b/site/services/reconciliation.py @@ -0,0 +1,381 @@ +"""Сверка внешней CRM/1С-выгрузки с утверждёнными интервалами dossier. + +Сверка выполняется по исходным суммам и датам на Decimal/date. Отсутствие +известного SKU не блокирует проверку договора; неизвестный VAT rate, напротив, +не позволяет объявить цены совпавшими, если стороны указали разные VAT modes. +""" +import re +from datetime import date, datetime, timedelta +from decimal import Decimal, InvalidOperation, ROUND_HALF_UP + +from db import dossiers as db_dossiers +from db import exceptions as db_exceptions +from db import timeline as db_timeline +from services.catalog_matcher import match_catalog_sku +from services.metrics import normalize_date +from services.service_identity import analyze_service_row, match_service_item + + +MONEY_TOLERANCE = Decimal("0.01") + + +def _decimal(value): + """Преобразовать внешнее или договорное число в конечный Decimal.""" + if value is None or value == "" or isinstance(value, bool): + return None + try: + result = Decimal(str(value).replace("\xa0", "").replace(" ", "").replace(",", ".")) + except (InvalidOperation, ValueError, TypeError): + return None + return result if result.is_finite() else None + + +def _date(value): + """Разобрать поддерживаемую ISO/DD.MM.YYYY дату без принятия мусорных значений.""" + if not value: + return None + try: + return datetime.strptime(normalize_date(str(value).strip()), "%Y-%m-%d").date() + except (TypeError, ValueError): + return None + + +def _window(target_date_or_period): + """Нормализовать точечную дату или полуоткрытый период [start, end).""" + if target_date_or_period is None: + point = date.today() + return {"kind": "point", "date": point, "start": point, "end": point + timedelta(days=1)} + if isinstance(target_date_or_period, (tuple, list)) and len(target_date_or_period) == 2: + start, end = map(_date, target_date_or_period) + elif isinstance(target_date_or_period, dict): + start = _date(target_date_or_period.get("start") or target_date_or_period.get("period_start")) + end = _date(target_date_or_period.get("end") or target_date_or_period.get("period_end")) + else: + point = _date(target_date_or_period) + if point is None: + raise ValueError("target date must be YYYY-MM-DD or DD.MM.YYYY") + return {"kind": "point", "date": point, "start": point, "end": point + timedelta(days=1)} + if start is None or end is None or end <= start: + raise ValueError("target period must have valid start < end") + return {"kind": "period", "date": None, "start": start, "end": end} + + +def _interval_overlaps(interval, window): + """Проверить активность интервала на дату или пересечение с period window.""" + start = _date(interval.get("valid_from")) + end = _date(interval.get("valid_to")) + if start is None: + return False + return start < window["end"] and (end is None or end > window["start"]) + + +def _external_end(row): + """Согласовать inclusive date_end CRM со stored exclusive valid_to.""" + exclusive = _date(row.get("date_end_exclusive")) + if exclusive: + return exclusive + inclusive = _date(row.get("date_end") or row.get("valid_to")) + if not inclusive: + return None + return inclusive + timedelta(days=1) if row.get("date_end_inclusive", True) else inclusive + + +def _amount(row, sum_key="sum", price_key="price", qty_key="qty"): + """Получить сумму напрямую либо вычислить её как точное price × qty.""" + total = _decimal(row.get(sum_key)) + if total is not None: + return total + price, qty = _decimal(row.get(price_key)), _decimal(row.get(qty_key)) + return price * qty if price is not None and qty is not None else None + + +def _vat_mode(value): + """Свести известные подписи режима НДС к gross/net/unknown.""" + mode = str(value or "unknown").strip().lower() + if mode in {"with_vat", "vat_included", "inclusive", "gross", "с ндс", "включен"}: + return "gross" + if mode in {"without_vat", "vat_excluded", "exclusive", "net", "без ндс", "не включен"}: + return "net" + return "unknown" + + +def _vat_rate(row): + """Прочитать только явно переданную ставку НДС; не выводить её по календарю.""" + value = _decimal(row.get("vat_rate")) + if value is None: + return None + if value > 1: + value /= Decimal("100") + if value < 0 or value > 1: + return None + return value + + +def _to_gross(amount, mode, rate): + """Нормализовать amount к gross, округляя по правилам денежных копеек.""" + normalized_mode = _vat_mode(mode) + if amount is None: + return None + if normalized_mode == "gross": + return amount.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) + if normalized_mode == "net" and rate is not None: + return (amount * (Decimal("1") + rate)).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) + return None + + +def _vat_comparable(contract, external): + """Разрешить денежную сверку при одинаковом режиме или двух известных ставках.""" + contract_mode, external_mode = _vat_mode(contract.get("vat_mode")), _vat_mode(external.get("vat_mode")) + if contract_mode == external_mode and contract_mode != "unknown": + return True + if contract_mode == "unknown" and external_mode == "unknown": + return True + return _vat_rate(contract) is not None and _vat_rate(external) is not None + + +def _money_comparison(contract, external): + """Сравнить amount, price и qty с точным VAT normalization.""" + if not _vat_comparable(contract, external): + return {"comparable": False, "matches": False, "reason": "VAT mode/rate is not comparable"} + contract_rate, external_rate = _vat_rate(contract), _vat_rate(external) + contract_amount = _to_gross(_amount(contract), contract.get("vat_mode"), contract_rate) + external_amount = _to_gross(_amount(external), external.get("vat_mode"), external_rate) + if contract_amount is None or external_amount is None: + return {"comparable": False, "matches": False, "reason": "amount is missing"} + contract_qty, external_qty = _decimal(contract.get("qty")), _decimal(external.get("qty")) + if contract_qty is not None and external_qty is not None and contract_qty != external_qty: + return {"comparable": True, "matches": False, "reason": "quantity differs"} + contract_price = _to_gross(_decimal(contract.get("price")), contract.get("vat_mode"), contract_rate) + external_price = _to_gross(_decimal(external.get("price")), external.get("vat_mode"), external_rate) + if contract_price is not None and external_price is not None: + if abs(contract_price - external_price) > MONEY_TOLERANCE: + return {"comparable": True, "matches": False, "reason": "unit price differs"} + if abs(contract_amount - external_amount) > MONEY_TOLERANCE: + return {"comparable": True, "matches": False, "reason": "sum differs"} + return {"comparable": True, "matches": True, "reason": ""} + + +def _find_service(external, items, catalog_entries): + """Сопоставить external row по явному ID, SKU или семейству/anchor.""" + explicit_id = external.get("service_item_id") + if explicit_id: + exact = [item for item in items if item.get("id") == explicit_id] + if exact: + return {"status": "MATCHED", "service_item": exact[0], "candidates": exact} + external_sku = re.sub(r"[^0-9A-ZА-ЯЁ-]", "", str(external.get("sku") or "").upper()) + if external_sku: + sku_matches = [ + item for item in items + if re.sub(r"[^0-9A-ZА-ЯЁ-]", "", str(item.get("catalog_sku") or "").upper()) == external_sku + ] + if len(sku_matches) == 1: + return {"status": "MATCHED", "service_item": sku_matches[0], "candidates": sku_matches} + raw_name = external.get("name") or external.get("service_name") or "" + if external_sku and catalog_entries: + sku_candidates = [] + for item in items: + catalog_result = match_catalog_sku( + {**item, "charge_type": external.get("charge_type")}, catalog_entries, + ) + if catalog_result.get("sku") and str(catalog_result["sku"]).upper() == external_sku: + sku_candidates.append(item) + if len(sku_candidates) == 1: + return {"status": "MATCHED", "service_item": sku_candidates[0], "candidates": sku_candidates} + if sku_candidates: + return {"status": "AMBIGUOUS", "service_item": None, "candidates": sku_candidates} + identity = analyze_service_row(raw_name, external.get("charge_type")) + return match_service_item(identity, items) + + +def _date_comparison(interval, external): + """Сверить точные границы external row с интервалом договора.""" + contract_start = _date(interval.get("valid_from")) + external_start = _date(external.get("date_start") or external.get("valid_from")) + contract_end = _date(interval.get("valid_to")) + external_end = _external_end(external) + if external_start != contract_start or external_end != contract_end: + return { + "matches": False, + "contract_start": contract_start.isoformat() if contract_start else None, + "external_start": external_start.isoformat() if external_start else None, + "contract_end_exclusive": contract_end.isoformat() if contract_end else None, + "external_end_exclusive": external_end.isoformat() if external_end else None, + } + return {"matches": True} + + +def _finding(status, external=None, service_item=None, interval=None, reason="", **details): + """Собрать отчётный finding только из JSON-совместимых значений.""" + return { + "status": status, + "external_row": _json_safe(external), + "service_item_id": service_item.get("id") if service_item else None, + "service_name": (interval or {}).get("raw_name") or (service_item or {}).get("canonical_name"), + "interval_id": (interval or {}).get("id"), + "reason": reason, + **details, + } + + +def _json_safe(value): + """Сериализовать Decimal/date в точные строки для Flask JSON response.""" + if isinstance(value, Decimal): + return format(value, "f") + if isinstance(value, (date, datetime)): + return value.isoformat() + if isinstance(value, dict): + return {key: _json_safe(item) for key, item in value.items()} + if isinstance(value, (list, tuple)): + return [_json_safe(item) for item in value] + return value + + +def reconcile_dossier(dossier_id, external_rows, target_date_or_period=None, + catalog_entries=None): + """Сверить CRM/1С строки с активными обязательствами dossier. + + Входные даты `date_end` считаются включительными; `date_end_exclusive` можно + передать для явной полуоткрытой границы. Неоднозначные service matches и + открытые исключения получают NEEDS_HUMAN_DECISION. SKU UNKNOWN не меняет + результат сверки дат, количеств и денежных сумм. + """ + dossier = db_dossiers.get_dossier(dossier_id) + if dossier is None: + return {"ok": False, "error": "dossier not found", "dossier_id": dossier_id} + window = _window(target_date_or_period) + items = db_timeline.list_service_items(dossier_id) + intervals = [ + interval for interval in db_timeline.list_timeline(dossier_id) + if _interval_overlaps(interval, window) + ] + open_exceptions = db_exceptions.list_exceptions(dossier_id=dossier_id, status="OPEN") + exceptions_by_item = {} + for issue in open_exceptions: + service_item_id = issue.get("service_item_id") + if service_item_id: + exceptions_by_item.setdefault(service_item_id, []).append(issue) + + findings = [] + observed_items = set() + uncertain_items = set() + for row_index, external in enumerate(external_rows or [], start=1): + if not isinstance(external, dict): + findings.append(_finding("NEEDS_HUMAN_DECISION", reason="external row must be an object", row_index=row_index)) + continue + match = _find_service(external, items, catalog_entries or []) + if match["status"] == "AMBIGUOUS": + candidate_ids = [item["id"] for item in match["candidates"]] + uncertain_items.update(candidate_ids) + findings.append(_finding( + "NEEDS_HUMAN_DECISION", external=external, reason="service match is ambiguous", + candidate_service_item_ids=candidate_ids, + )) + continue + service_item = match.get("service_item") + if service_item is None: + findings.append(_finding( + "UNCONTRACTED_BILLING", external=external, + reason="no matching service item exists in dossier", + )) + continue + observed_items.add(service_item["id"]) + catalog_status = service_item.get("sku_status") or "UNKNOWN" + if catalog_entries: + catalog_result = match_catalog_sku( + {**service_item, "charge_type": external.get("charge_type")}, + catalog_entries, + ) + catalog_status = catalog_result["status"] + if service_item["id"] in exceptions_by_item: + findings.append(_finding( + "NEEDS_HUMAN_DECISION", external=external, service_item=service_item, + reason="service item has open dossier exceptions", + sku_status=catalog_status, + exception_ids=[issue["id"] for issue in exceptions_by_item[service_item["id"]]], + )) + continue + + matching_intervals = [ + interval for interval in intervals + if interval.get("service_item_id") == service_item["id"] + and (not external.get("charge_type") or interval.get("charge_type") == external.get("charge_type")) + ] + if not matching_intervals: + findings.append(_finding( + "UNCONTRACTED_BILLING", external=external, service_item=service_item, + reason="service item has no contractual interval in the target window", + sku_status=catalog_status, + )) + continue + if len(matching_intervals) > 1: + exact = [interval for interval in matching_intervals if _date_comparison(interval, external)["matches"]] + if len(exact) == 1: + matching_intervals = exact + else: + uncertain_items.add(service_item["id"]) + findings.append(_finding( + "NEEDS_HUMAN_DECISION", external=external, service_item=service_item, + reason="external period overlaps multiple contractual intervals", + sku_status=catalog_status, + candidate_interval_ids=[interval["id"] for interval in matching_intervals], + )) + continue + + interval = matching_intervals[0] + dates = _date_comparison(interval, external) + if not dates["matches"]: + findings.append(_finding( + "DATE_MISMATCH", external=external, service_item=service_item, + interval=interval, reason="date range differs", sku_status=catalog_status, **dates, + )) + continue + money = _money_comparison(interval, external) + if not money["comparable"]: + findings.append(_finding( + "NEEDS_HUMAN_DECISION", external=external, service_item=service_item, + interval=interval, reason=money["reason"], sku_status=catalog_status, + )) + elif not money["matches"]: + findings.append(_finding( + "PRICE_OR_QTY_MISMATCH", external=external, service_item=service_item, + interval=interval, reason=money["reason"], sku_status=catalog_status, + )) + else: + findings.append(_finding( + "MATCH", external=external, service_item=service_item, interval=interval, + reason="dates, quantity and amount match", sku_status=catalog_status, + )) + + for interval in intervals: + service_item_id = interval.get("service_item_id") + if service_item_id in observed_items or service_item_id in uncertain_items: + continue + if interval.get("charge_type") == "payg": + continue + service_item = next((item for item in items if item.get("id") == service_item_id), None) + if service_item_id in exceptions_by_item: + findings.append(_finding( + "NEEDS_HUMAN_DECISION", service_item=service_item, interval=interval, + reason="service item has open dossier exceptions", + exception_ids=[issue["id"] for issue in exceptions_by_item[service_item_id]], + )) + else: + findings.append(_finding( + "UNBILLED_SERVICE", service_item=service_item, interval=interval, + reason="contractual service is active but absent from external rows", + )) + + summary = {} + for finding in findings: + summary[finding["status"]] = summary.get(finding["status"], 0) + 1 + return { + "ok": True, + "dossier_id": dossier_id, + "target": { + "kind": window["kind"], + "start": window["start"].isoformat(), + "end_exclusive": window["end"].isoformat(), + }, + "summary": summary, + "findings": findings, + } \ No newline at end of file diff --git a/site/services/timeline_builder.py b/site/services/timeline_builder.py index bd5c617..40f02b7 100644 --- a/site/services/timeline_builder.py +++ b/site/services/timeline_builder.py @@ -177,6 +177,7 @@ def _append_interval(dossier_id, service_item, row, section, document, "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), @@ -310,6 +311,7 @@ def _process_document(dossier, document, passport, items, counts): "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 "",