diff --git a/site/routes/api_bp.py b/site/routes/api_bp.py index 68c3415..baba76f 100644 --- a/site/routes/api_bp.py +++ b/site/routes/api_bp.py @@ -1,4 +1,6 @@ """API blueprint — groups, documents, supplements, sync, cleanup, spec-current, chat.""" +import csv +import io import json from datetime import date from flask import Blueprint, request, jsonify @@ -6,8 +8,11 @@ from db import documents, supplements, spec_current from db import dossiers as db_dossiers from db import exceptions as db_exceptions from db import timeline as db_timeline +from db import extractions as db_extractions from db.connection import execute, query, transaction from services.grouping import group_documents, apply_groups +from services.reconciliation import reconcile_dossier +from services.timeline_builder import build_dossier_timeline from config import LLM_URL, LLM_KEY, LLM_MODEL import httpx @@ -46,6 +51,28 @@ def api_dossier_timeline(dossier_id): ) +@api_bp.route("/api/dossiers") +def api_dossiers(): + """Вернуть dossier cards batch вместе с alias и связанными документами.""" + batch_id = request.args.get("batch_id") + if not batch_id: + return jsonify(ok=False, error="batch_id required"), 400 + dossiers = db_dossiers.list_dossiers_by_batch(batch_id) + extractions_by_document = { + extraction["document_id"]: extraction + for extraction in db_extractions.list_by_batch(batch_id) + } + for dossier in dossiers: + for document in dossier.get("documents", []): + extraction = extractions_by_document.get(document.get("document_id")) + validation = (extraction or {}).get("math_validation") or {} + document["math_status"] = validation.get("status", "UNKNOWN") + return jsonify( + ok=True, + dossiers=dossiers, + ) + + @api_bp.route("/api/exceptions") def api_exceptions(): """Вернуть исключения только по явно заданной области batch или dossier.""" @@ -64,6 +91,93 @@ def api_exceptions(): ) +@api_bp.route("/api/reconcile", methods=["POST"]) +def api_reconcile(): + """Сверить JSON или CSV строки внешнего биллинга с эталоном dossier.""" + content_type = request.mimetype or "" + if content_type == "application/json": + body = request.get_json(silent=True) or {} + dossier_id = body.get("dossier_id") + rows = body.get("external_rows") + target = body.get("target_date_or_period") + catalog_entries = body.get("catalog_entries", []) + elif content_type == "multipart/form-data": + dossier_id = request.form.get("dossier_id") + upload = request.files.get("file") + if upload is None: + return jsonify(ok=False, error="CSV file required"), 400 + try: + rows = list(csv.DictReader(io.StringIO(upload.stream.read().decode("utf-8-sig")))) + except (UnicodeDecodeError, csv.Error): + return jsonify(ok=False, error="CSV must be UTF-8 with a header row"), 400 + target = request.form.get("target_date_or_period") + if target: + try: + target = json.loads(target) + except json.JSONDecodeError: + return jsonify(ok=False, error="target_date_or_period must be JSON"), 400 + catalog_entries = request.form.get("catalog_entries", "[]") + try: + catalog_entries = json.loads(catalog_entries) + except json.JSONDecodeError: + return jsonify(ok=False, error="catalog_entries must be JSON"), 400 + else: + return jsonify(ok=False, error="use application/json or multipart/form-data"), 415 + + if not isinstance(dossier_id, str) or not dossier_id.strip(): + return jsonify(ok=False, error="dossier_id required"), 400 + if not isinstance(rows, list) or any(not isinstance(row, dict) for row in rows): + return jsonify(ok=False, error="external_rows must be an array of objects"), 400 + if not isinstance(catalog_entries, list) or any(not isinstance(row, dict) for row in catalog_entries): + return jsonify(ok=False, error="catalog_entries must be an array of objects"), 400 + try: + result = reconcile_dossier(dossier_id, rows, target, catalog_entries) + except ValueError as error: + return jsonify(ok=False, error=str(error)), 400 + if not result.get("ok"): + return jsonify(result), 404 + return jsonify(result) + + +@api_bp.route("/api/exceptions//resolve", methods=["POST"]) +def api_resolve_exception(exception_id): + """Сохранить выбор оператора и без LLM пересобрать dossier timeline.""" + issue = db_exceptions.get_exception(exception_id) + if not issue: + return jsonify(ok=False, error="exception not found"), 404 + if issue.get("status") != "OPEN": + return jsonify(ok=False, error="exception is not open"), 409 + body = request.get_json(silent=True) or {} + resolution = body.get("resolution") + if not isinstance(resolution, dict) or not resolution: + return jsonify(ok=False, error="resolution object required"), 400 + + options = issue.get("options") or [] + if "option_index" in resolution: + option_index = resolution["option_index"] + if type(option_index) is not int or not 0 <= option_index < len(options): + return jsonify(ok=False, error="option_index is invalid"), 400 + resolution = { + "option_index": option_index, + "selected_option": options[option_index], + } + elif issue.get("stage") == "lineage_matching" and options: + return jsonify(ok=False, error="choose an option_index"), 400 + + if issue.get("dossier_id") and not db_dossiers.get_dossier(issue["dossier_id"]): + return jsonify(ok=False, error="dossier not found"), 404 + resolved = db_exceptions.resolve_exception(exception_id, resolution) + if not resolved: + return jsonify(ok=False, error="exception is not open"), 409 + + rebuild = None + if issue.get("dossier_id"): + rebuild = build_dossier_timeline(issue["dossier_id"]) + if not rebuild.get("ok"): + return jsonify(ok=False, error="timeline rebuild failed", rebuild=rebuild), 500 + return jsonify(ok=True, exception=resolved, rebuild=rebuild) + + # ── Supplements ────────────────────────────────────────────────── @api_bp.route("/api/supplements") diff --git a/site/routes/pipeline_bp.py b/site/routes/pipeline_bp.py index 409c1e3..ce367af 100644 --- a/site/routes/pipeline_bp.py +++ b/site/routes/pipeline_bp.py @@ -2,7 +2,9 @@ import json, re, os, threading from flask import Blueprint, request, jsonify, Response, stream_with_context from services.process import run_pipeline -from services.classify import classify_batch +from services import extractor as stateless_extractor +from services.grouping import build_dossiers_for_batch +from services.timeline_builder import build_dossier_timeline from llm_prompt import build_prompt from db import documents @@ -10,11 +12,58 @@ pipeline_bp = Blueprint("pipeline", __name__) # In-memory lock для classify (замена файлового lock) _classify_locks: dict[str, threading.Thread] = {} +_extraction_jobs: dict[str, dict] = {} @pipeline_bp.route("/process-v2", methods=["GET"]) def process_v2(): """SSE-стриминг сравнения договоров.""" + dossier_id = request.args.get("dossier_id") + if dossier_id: + if not re.fullmatch( + r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}', + dossier_id, re.I, + ): + return jsonify(ok=False, error="invalid dossier_id"), 400 + + def generate_timeline(): + import time as _time + started = _time.time() + yield ": ok\n\n" + yield "data: " + json.dumps({ + "type": "extract_start", "supplement_id": dossier_id, + "filename": "Построение таймлайна", + }, ensure_ascii=False) + "\n\n" + try: + result = build_dossier_timeline(dossier_id) + if not result.get("ok"): + yield "data: " + json.dumps({ + "type": "error", "message": result.get("error", "timeline build failed"), + }, ensure_ascii=False) + "\n\n" + return + yield "data: " + json.dumps({ + "type": "timeline", "dossier_id": dossier_id, + "timeline": result.get("timeline", []), + "exceptions": result.get("exceptions", []), + "counts": result.get("counts", {}), + }, ensure_ascii=False, default=str) + "\n\n" + yield "data: " + json.dumps({ + "type": "done", "dossier_id": dossier_id, + "total_time_s": round(_time.time() - started, 2), + }, ensure_ascii=False) + "\n\n" + except GeneratorExit: + return + except Exception as error: + yield "data: " + json.dumps({ + "type": "error", "message": str(error), + }, ensure_ascii=False) + "\n\n" + + return Response( + stream_with_context(generate_timeline()), + content_type="text/event-stream; charset=utf-8", + headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, + ) + cid = request.args.get("contract_id") if not cid or not re.fullmatch( r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}', cid, re.I @@ -53,10 +102,12 @@ def process_v2(): @pipeline_bp.route("/api/classify-batch", methods=["POST"]) def classify_batch_route(): - """Запустить классификацию. ≤10 файлов — sync, >10 — async (Thread).""" - body = request.get_json() + """Выполнить stateless extraction и построить contract dossiers.""" + body = request.get_json(silent=True) or {} + if not isinstance(body, dict): + return jsonify(ok=False, error="JSON object required"), 400 batch_id = body.get("batch_id") - if not batch_id: + if not isinstance(batch_id, str) or not batch_id.strip(): return jsonify(ok=False, error="batch_id required"), 400 # Guard: уже запущена? @@ -67,20 +118,27 @@ def classify_batch_route(): else: del _classify_locks[batch_id] - pending = documents.list_pending(batch_id) + pending = documents.list_for_extraction(batch_id) total = len(pending) if total == 0: - return jsonify(ok=False, error="no pending documents"), 400 + return jsonify(ok=False, error="no parsed documents"), 400 - # Sync для малых батчей + # Keep small batches synchronous for the existing browser workflow. if total <= 10: - result = classify_batch(batch_id) + result = _run_v3_extraction(batch_id, total) return jsonify(result) - # Async для больших + _extraction_jobs[batch_id] = { + "state": "running", "total": total, "done": 0, + "failed": 0, "classes": {}, + } + def _run(): try: - classify_batch(batch_id) + _run_v3_extraction(batch_id, total, _extraction_progress) + _extraction_jobs[batch_id]["state"] = "done" + except Exception as error: + _extraction_jobs[batch_id].update(state="failed", error=str(error)) finally: _classify_locks.pop(batch_id, None) @@ -88,4 +146,70 @@ def classify_batch_route(): _classify_locks[batch_id] = t t.start() - return jsonify(ok=True, total=total), 202 + return jsonify(ok=True, total=total, v3=True), 202 + + +def _extraction_progress(batch_id, completed, result): + """Обновить лёгкий in-memory прогресс для крупного batch.""" + job = _extraction_jobs[batch_id] + job["done"] = completed + if not result.get("ok"): + job["failed"] += 1 + else: + extraction = result.get("extraction") or {} + doc_class = extraction.get("doc_class", "unknown") + job["classes"][doc_class] = job["classes"].get(doc_class, 0) + 1 + + +def _run_v3_extraction(batch_id, total, progress_callback=None): + """Собрать extraction и dossier summaries в один ответ API.""" + extraction = stateless_extractor.extract_batch( + batch_id, + progress_callback=( + (lambda completed, result: progress_callback(batch_id, completed, result)) + if progress_callback else None + ), + ) + dossier_result = build_dossiers_for_batch(batch_id) + classes = extraction.get("classes", {}) + math_checks_passed = sum( + 1 for result in extraction.get("results", []) + if (result.get("extraction") or {}).get("math_validation", {}).get("status") == "VERIFIED" + ) + return { + "ok": bool(extraction.get("results")) and dossier_result.get("ok", False), + "v3": True, + "done": total, + "total": total, + "failed": extraction.get("failed", 0), + "classes": classes, + "ignored_accounting": classes.get("ignored_accounting", 0), + "math_checks_passed": math_checks_passed, + "dossier_count": len(dossier_result.get("dossiers", [])), + "ambiguous": dossier_result.get("ambiguous", 0), + "unlinked": dossier_result.get("unlinked", []), + } + + +@pipeline_bp.route("/api/classify-progress") +def classify_progress_route(): + """Вернуть состояние асинхронного stateless extraction batch.""" + batch_id = request.args.get("batch") + if not batch_id: + return jsonify(ok=False, error="batch required"), 400 + job = _extraction_jobs.get(batch_id) + if not job: + return jsonify(ok=False, error="batch extraction not found"), 404 + response = { + "ok": job["state"] != "failed", + "v3": True, + "finished": job["state"] in {"done", "failed"}, + "state": job["state"], + "done": job["done"], + "total": job["total"], + "failed": job["failed"], + "classes": job["classes"], + } + if job.get("error"): + response["error"] = job["error"] + return jsonify(response) diff --git a/site/services/extractor.py b/site/services/extractor.py index 7aa00db..f898720 100644 --- a/site/services/extractor.py +++ b/site/services/extractor.py @@ -306,10 +306,22 @@ def extract_document(document, llm_client=None, batch_id=None): return {"ok": False, "document_id": document_id, "error": str(error)} -def extract_batch(batch_id, llm_client=None): - """Обработать все документы batch и вернуть сводку без batch-wide failure.""" +def extract_batch(batch_id, llm_client=None, progress_callback=None): + """Обработать документы batch и вернуть сводку без batch-wide failure. + + Optional progress callback получает (completed_count, document_result) после + каждого документа; ошибки callback не должны останавливать обработку batch. + """ documents = db_docs.list_for_extraction(batch_id) - results = [extract_document(document, llm_client=llm_client, batch_id=batch_id) for document in documents] + results = [] + for completed, document in enumerate(documents, start=1): + result = extract_document(document, llm_client=llm_client, batch_id=batch_id) + results.append(result) + if progress_callback is not None: + try: + progress_callback(completed, result) + except Exception: + log.exception("Extraction progress callback failed for batch %s", batch_id) done = sum(1 for result in results if result["ok"]) failed = len(results) - done classes = {} diff --git a/site/services/timeline_builder.py b/site/services/timeline_builder.py index 40f02b7..b425502 100644 --- a/site/services/timeline_builder.py +++ b/site/services/timeline_builder.py @@ -185,7 +185,39 @@ def _append_interval(dossier_id, service_item, row, section, document, }) -def _process_document(dossier, document, passport, items, counts): +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") @@ -216,23 +248,47 @@ def _process_document(dossier, document, passport, items, counts): 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"], "canonical_name": item.get("canonical_name")} - for item in match["candidates"]], + 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 = 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"], - ) + 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 = [ @@ -351,7 +407,11 @@ def build_dossier_timeline(dossier_id, contract_id=None): 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(): @@ -360,10 +420,13 @@ def build_dossier_timeline(dossier_id, contract_id=None): WHERE dossier_id = %s AND stage = 'lineage_matching' AND status = 'OPEN'""", (dossier_id,), ) - execute("DELETE FROM service_items WHERE dossier_id = %s", (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) + _process_document( + dossier, document, passport, items, counts, + resolved_exceptions, reusable_items, + ) if contract_id: sync_legacy_spec_current(dossier_id, contract_id)