From e2ba9ab4518390874862983a052cd249a2930282 Mon Sep 17 00:00:00 2001 From: Repinoid Date: Sat, 31 Oct 2026 17:36:55 +0300 Subject: [PATCH] feat: add authenticated engine processing boundary --- site/app.py | 25 +++- site/db/connection.py | 15 +- site/db/dossiers.py | 5 +- site/db/extraction_jobs.py | 79 +++++++++++ site/routes/pipeline_bp.py | 273 +++++++++++++++++++++++++++---------- site/services/engine.py | 46 +++++++ site/services/extractor.py | 7 +- site/services/grouping.py | 7 + site/static/app.js | 2 +- 9 files changed, 378 insertions(+), 81 deletions(-) create mode 100644 site/db/extraction_jobs.py create mode 100644 site/services/engine.py diff --git a/site/app.py b/site/app.py index f9cfb4d..c458a94 100644 --- a/site/app.py +++ b/site/app.py @@ -1,7 +1,7 @@ """contracts-flask v2.0 — полный перенос с ВМ на Flask. Больше никаких прокси на contracts.kube5s.ru — всё локально. """ -import sys, os +import sys, os, hmac # site/ в sys.path — импортируем модули напрямую, без префиксов sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) # корень репо — для переиспользуемого модуля upload/ (как в drhider) @@ -9,8 +9,8 @@ _REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) if _REPO_ROOT not in sys.path: sys.path.insert(0, _REPO_ROOT) -from flask import Flask -from config import VERSION, MAX_CONTENT_LENGTH +from flask import Flask, jsonify, request +from config import VERSION, MAX_CONTENT_LENGTH, API_KEY from routes import register_routes @@ -18,10 +18,29 @@ def create_app(): app = Flask(__name__) app.config["VERSION"] = VERSION app.config["MAX_CONTENT_LENGTH"] = MAX_CONTENT_LENGTH + app.config["API_KEY"] = API_KEY _init_db() register_routes(app) + @app.before_request + def authorize_request(): + if request.endpoint == "health.health": + return None + key = app.config.get("API_KEY", "") + if not key: + return jsonify(ok=False, error="application API_KEY is not configured"), 503 + if "X-API-Key" in request.headers: + supplied = request.headers.get("X-API-Key", "") + else: + authorization = request.authorization + supplied = authorization.password if authorization and authorization.type.lower() == "basic" else "" + if not hmac.compare_digest((supplied or "").encode("utf-8"), key.encode("utf-8")): + return jsonify(ok=False, error="unauthorized"), 401, { + "WWW-Authenticate": 'Basic realm="Contracts", charset="UTF-8"', + } + return None + @app.after_request def no_cache(response): response.headers["Cache-Control"] = "no-cache, no-store, must-revalidate" diff --git a/site/db/connection.py b/site/db/connection.py index c613ce6..04b7928 100644 --- a/site/db/connection.py +++ b/site/db/connection.py @@ -12,7 +12,7 @@ import time import os from contextlib import contextmanager -DB_PATH = "/tmp/contracts.db" +DB_PATH = os.getenv("CONTRACTS_DB_PATH", "/tmp/contracts.db") _db_session_key = None _local = threading.local() @@ -67,6 +67,19 @@ def _create_schema(conn): created_at TEXT DEFAULT (datetime('now')) ); + CREATE TABLE IF NOT EXISTS extraction_jobs ( + batch_id TEXT PRIMARY KEY, + owner TEXT NOT NULL, + state TEXT NOT NULL, + total INTEGER NOT NULL, + done INTEGER NOT NULL DEFAULT 0, + failed INTEGER NOT NULL DEFAULT 0, + classes_json TEXT NOT NULL DEFAULT '{}', + result_json TEXT, + error TEXT, + updated_at REAL NOT NULL + ); + CREATE TABLE IF NOT EXISTS contracts ( id TEXT PRIMARY KEY, number TEXT NOT NULL, diff --git a/site/db/dossiers.py b/site/db/dossiers.py index 7cbf27e..5597e51 100644 --- a/site/db/dossiers.py +++ b/site/db/dossiers.py @@ -7,7 +7,7 @@ import json import uuid -from db.connection import execute, query, transaction +from db.connection import execute, query def _json_value(value, default): @@ -123,5 +123,4 @@ def list_dossier_documents(dossier_id): def clear_batch_dossiers(batch_id): """Удалить все досье batch; таблицы с ON DELETE CASCADE очистятся вместе с ними.""" - with transaction(): - return execute("DELETE FROM contract_dossiers WHERE batch_id = %s", (batch_id,)) \ No newline at end of file + return execute("DELETE FROM contract_dossiers WHERE batch_id = %s", (batch_id,)) \ No newline at end of file diff --git a/site/db/extraction_jobs.py b/site/db/extraction_jobs.py new file mode 100644 index 0000000..1f84982 --- /dev/null +++ b/site/db/extraction_jobs.py @@ -0,0 +1,79 @@ +"""Persistent extraction progress and per-batch worker ownership.""" +import json +import time +import uuid + +from db.connection import execute, query + +LEASE_SECONDS = 300 + + +def claim(batch_id, total): + """Claim a batch atomically, allowing retry after failure or expired lease.""" + owner = str(uuid.uuid4()) + now = time.time() + changed = execute( + """INSERT INTO extraction_jobs (batch_id, owner, state, total, updated_at) + VALUES (%s, %s, 'running', %s, %s) + ON CONFLICT(batch_id) DO UPDATE SET owner=excluded.owner, state='running', + total=excluded.total, done=0, failed=0, classes_json='{}', + result_json=NULL, error=NULL, updated_at=excluded.updated_at + WHERE extraction_jobs.state != 'running' OR extraction_jobs.updated_at < %s""", + (batch_id, owner, total, now, now - LEASE_SECONDS), + ) + return owner if changed else None + + +def heartbeat(batch_id, owner): + return execute( + "UPDATE extraction_jobs SET updated_at=%s WHERE batch_id=%s AND owner=%s AND state='running'", + (time.time(), batch_id, owner), + ) + + +def progress(batch_id, owner, completed, result): + rows = query( + "SELECT failed, classes_json FROM extraction_jobs WHERE batch_id=%s AND owner=%s AND state='running'", + (batch_id, owner), + ) + if not rows: + return + failed = rows[0]["failed"] + classes = json.loads(rows[0]["classes_json"]) + if not result.get("ok"): + failed += 1 + else: + doc_class = (result.get("extraction") or {}).get("doc_class", "unknown") + classes[doc_class] = classes.get(doc_class, 0) + 1 + execute( + """UPDATE extraction_jobs SET done=%s, failed=%s, classes_json=%s, updated_at=%s + WHERE batch_id=%s AND owner=%s AND state='running'""", + (completed, failed, json.dumps(classes), time.time(), batch_id, owner), + ) + + +def finish(batch_id, owner, result=None, error=None): + state = "done" if result and result.get("ok") and not error else "failed" + return execute( + """UPDATE extraction_jobs SET state=%s, result_json=%s, error=%s, updated_at=%s + WHERE batch_id=%s AND owner=%s AND state='running'""", + (state, json.dumps(result, ensure_ascii=False) if result is not None else None, + error or ("extraction failed" if state == "failed" else None), time.time(), batch_id, owner), + ) + + +def get(batch_id): + """Mark an abandoned lease interrupted; callers may explicitly retry it.""" + execute( + """UPDATE extraction_jobs SET state='interrupted', error='worker stopped; retry the batch' + WHERE batch_id=%s AND state='running' AND updated_at < %s""", + (batch_id, time.time() - LEASE_SECONDS), + ) + rows = query("SELECT * FROM extraction_jobs WHERE batch_id=%s", (batch_id,)) + if not rows: + return None + job = rows[0] + job["classes"] = json.loads(job.pop("classes_json")) + result_json = job.pop("result_json") + job["result"] = json.loads(result_json) if result_json else None + return job \ No newline at end of file diff --git a/site/routes/pipeline_bp.py b/site/routes/pipeline_bp.py index ce367af..415d514 100644 --- a/site/routes/pipeline_bp.py +++ b/site/routes/pipeline_bp.py @@ -1,18 +1,25 @@ """Pipeline blueprint — SSE-сравнение + classify.""" -import json, re, os, threading +import hashlib, json, re, os, threading, time from flask import Blueprint, request, jsonify, Response, stream_with_context from services.process import run_pipeline from services import extractor as stateless_extractor +from services.engine import analyze_batch 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 +from db import documents, extraction_jobs +from db.connection import transaction pipeline_bp = Blueprint("pipeline", __name__) -# In-memory lock для classify (замена файлового lock) -_classify_locks: dict[str, threading.Thread] = {} -_extraction_jobs: dict[str, dict] = {} +_engine_rate_lock = threading.Lock() +_engine_rate_hits: dict[str, list[float]] = {} +_ENGINE_MAX_BODY_BYTES = 10 * 1024 * 1024 +_ENGINE_MAX_DOCUMENTS = 100 +_ENGINE_MAX_ELEMENTS = 20_000 +_ENGINE_MAX_DOCUMENT_BYTES = 2 * 1024 * 1024 +_ENGINE_RATE_WINDOW_SECONDS = 60 +_ENGINE_RATE_LIMIT = 10 @pipeline_bp.route("/process-v2", methods=["GET"]) @@ -110,85 +117,209 @@ def classify_batch_route(): if not isinstance(batch_id, str) or not batch_id.strip(): return jsonify(ok=False, error="batch_id required"), 400 - # Guard: уже запущена? - if batch_id in _classify_locks: - t = _classify_locks[batch_id] - if t.is_alive(): - return jsonify(ok=False, error="classify already running"), 409 - else: - del _classify_locks[batch_id] - pending = documents.list_for_extraction(batch_id) total = len(pending) if total == 0: return jsonify(ok=False, error="no parsed documents"), 400 + owner = extraction_jobs.claim(batch_id, total) + if not owner: + return jsonify(ok=False, error="classify already running"), 409 # Keep small batches synchronous for the existing browser workflow. if total <= 10: - result = _run_v3_extraction(batch_id, total) + result = _run_extraction_job(batch_id, pending, owner) return jsonify(result) - _extraction_jobs[batch_id] = { - "state": "running", "total": total, "done": 0, - "failed": 0, "classes": {}, - } - - def _run(): - try: - _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) - - t = threading.Thread(target=_run, daemon=True) - _classify_locks[batch_id] = t - t.start() - + _start_extraction_job(batch_id, pending, owner) 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 +@pipeline_bp.route("/api/engine/analyze", methods=["POST"]) +def engine_analyze_route(): + """Accept parsed documents over a versioned, authenticated JSON boundary. + + Validation is completed for the whole payload before opening the write + transaction, so malformed input cannot leave a partial batch. The caller + supplies parsed elements only; extraction passports and dossier decisions + are always produced by the in-process Engine and persisted by its adapters. + """ + if _engine_rate_limited(): + return jsonify(ok=False, error="rate limit exceeded"), 429, {"Retry-After": "60"} + + raw_body = request.stream.read(_ENGINE_MAX_BODY_BYTES + 1) + if len(raw_body) > _ENGINE_MAX_BODY_BYTES: + return jsonify(ok=False, error="request body too large"), 413 + try: + body = json.loads(raw_body) if request.is_json else None + except (ValueError, UnicodeDecodeError): + body = None + if not isinstance(body, dict): + return jsonify(ok=False, error="JSON object required"), 400 + if type(body.get("schema_version")) is not int or body["schema_version"] != 1: + return jsonify(ok=False, error="unsupported schema_version"), 400 + + batch_id = body.get("batch_id") + if not isinstance(batch_id, str) or not re.fullmatch(r"[A-Za-z0-9._:-]{1,128}", batch_id): + return jsonify(ok=False, error="invalid batch_id"), 400 + source_documents = body.get("documents") + if not isinstance(source_documents, list) or not source_documents: + return jsonify(ok=False, error="documents must be a non-empty array"), 400 + if len(source_documents) > _ENGINE_MAX_DOCUMENTS: + return jsonify(ok=False, error="too many documents"), 413 + validated = [] + for index, source_document in enumerate(source_documents): + if not isinstance(source_document, dict): + return jsonify(ok=False, error=f"documents[{index}] must be an object"), 400 + filename = source_document.get("filename") + elements = source_document.get("elements") + if not isinstance(filename, str) or not filename.strip() or len(filename) > 512: + return jsonify(ok=False, error=f"documents[{index}].filename is invalid"), 400 + if not isinstance(elements, list) or len(elements) > _ENGINE_MAX_ELEMENTS: + return jsonify(ok=False, error=f"documents[{index}].elements is invalid"), 400 + for element_index, element in enumerate(elements): + if not _valid_engine_element(element): + return jsonify( + ok=False, error=f"documents[{index}].elements[{element_index}] is invalid", + ), 400 + try: + elements_json = json.dumps( + elements, ensure_ascii=False, sort_keys=True, separators=(",", ":"), + allow_nan=False, + ) + except (TypeError, ValueError): + return jsonify(ok=False, error=f"documents[{index}].elements is not valid JSON"), 400 + encoded_elements = elements_json.encode("utf-8") + if len(encoded_elements) > _ENGINE_MAX_DOCUMENT_BYTES: + return jsonify(ok=False, error=f"documents[{index}] is too large"), 413 + validated.append((filename.strip(), elements, elements_json, encoded_elements)) + + inserted = [] + with transaction(): + # Engine API owns a fresh batch per request. Preventing reuse inside + # BEGIN IMMEDIATE keeps old parsed rows out of this run and closes the + # race between two concurrent requests using the same batch ID. + if documents.list_by_batch(batch_id): + return jsonify(ok=False, error="batch_id already exists"), 409 + owner = extraction_jobs.claim(batch_id, len(validated)) + if not owner: + return jsonify(ok=False, error="classify already running"), 409 + for filename, elements, elements_json, encoded_elements in validated: + content_hash = hashlib.sha256(encoded_elements).hexdigest() + document = documents.insert( + filename, "application/json", None, status="uploaded", + batch_id=batch_id, content_hash=content_hash, + ) + documents.set_parsed(document["id"], elements) + inserted.append({"id": document["id"], "filename": filename}) + + pending = documents.list_for_extraction(batch_id) + total = len(pending) + if total > 10: + _start_extraction_job(batch_id, pending, owner) + return jsonify(ok=True, v3=True, total=total, documents=inserted), 202 + + result = _run_extraction_job(batch_id, pending, owner) + result["documents"] = inserted + return jsonify(result) -def _run_v3_extraction(batch_id, total, progress_callback=None): +def _valid_engine_element(element): + """Validate the parser fields consumed by structured-text formatting.""" + if not isinstance(element, dict): + return False + if element.get("type") == "paragraph": + return isinstance(element.get("text"), str) and isinstance(element.get("style", ""), str) + if element.get("type") != "table": + return False + rows = element.get("rows") + context = element.get("preceding_paragraphs", []) + table_index = element.get("table_index") + return ( + isinstance(rows, list) + and all(isinstance(row, list) and all(isinstance(cell, str) for cell in row) for row in rows) + and isinstance(context, list) + and all(isinstance(paragraph, str) for paragraph in context) + and (table_index is None or (type(table_index) is int and table_index > 0)) + ) + + +def _start_extraction_job(batch_id, pending, owner): + """Run a claimed batch in the background with persistent progress.""" + def _run(): + try: + _run_extraction_job(batch_id, pending, owner) + except Exception: + pass + + thread = threading.Thread(target=_run, daemon=True) + try: + thread.start() + except Exception as error: + extraction_jobs.finish(batch_id, owner, error=str(error)) + raise + + +def _engine_rate_limited(): + """Apply a small per-source fixed-window limit before accepting LLM work.""" + now = time.monotonic() + source = request.remote_addr or "unknown" + cutoff = now - _ENGINE_RATE_WINDOW_SECONDS + with _engine_rate_lock: + for key in list(_engine_rate_hits): + recent = [stamp for stamp in _engine_rate_hits[key] if stamp > cutoff] + if recent: + _engine_rate_hits[key] = recent + else: + del _engine_rate_hits[key] + recent = _engine_rate_hits.get(source, []) + if len(recent) >= _ENGINE_RATE_LIMIT: + return True + recent.append(now) + _engine_rate_hits[source] = recent + return False + + +def _run_extraction_job(batch_id, pending, owner): + """Keep the lease alive during LLM calls and persist every final outcome.""" + stop = threading.Event() + + def renew_lease(): + while not stop.wait(15): + if not extraction_jobs.heartbeat(batch_id, owner): + return + + heartbeat = threading.Thread(target=renew_lease, daemon=True) + heartbeat.start() + try: + result = _run_v3_extraction( + batch_id, pending, + progress_callback=lambda completed, item: extraction_jobs.progress(batch_id, owner, completed, item), + ) + extraction_jobs.finish(batch_id, owner, result=result) + return result + except Exception as error: + extraction_jobs.finish(batch_id, owner, error=str(error)) + raise + finally: + stop.set() + heartbeat.join() + + +def _run_v3_extraction(batch_id, pending, progress_callback=None, llm_client=None): """Собрать extraction и dossier summaries в один ответ API.""" - extraction = stateless_extractor.extract_batch( + parsed_documents = [{ + "id": document["id"], + "filename": document["filename"], + "elements": stateless_extractor._decode_elements(document.get("elements_json")), + } for document in pending] + return analyze_batch( batch_id, - progress_callback=( - (lambda completed, result: progress_callback(batch_id, completed, result)) - if progress_callback else None - ), + parsed_documents, + extractor=stateless_extractor.extract_batch, + dossier_builder=build_dossiers_for_batch, + llm_client=llm_client, + progress_callback=progress_callback, ) - 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") @@ -197,13 +328,13 @@ def classify_progress_route(): batch_id = request.args.get("batch") if not batch_id: return jsonify(ok=False, error="batch required"), 400 - job = _extraction_jobs.get(batch_id) + job = extraction_jobs.get(batch_id) if not job: return jsonify(ok=False, error="batch extraction not found"), 404 response = { - "ok": job["state"] != "failed", + "ok": job["state"] not in {"failed", "interrupted"}, "v3": True, - "finished": job["state"] in {"done", "failed"}, + "finished": job["state"] in {"done", "failed", "interrupted"}, "state": job["state"], "done": job["done"], "total": job["total"], @@ -212,4 +343,6 @@ def classify_progress_route(): } if job.get("error"): response["error"] = job["error"] + if job.get("result") is not None: + response["result"] = job["result"] return jsonify(response) diff --git a/site/services/engine.py b/site/services/engine.py new file mode 100644 index 0000000..1f22830 --- /dev/null +++ b/site/services/engine.py @@ -0,0 +1,46 @@ +"""Shared orchestration for the contract reconstruction engine. + +This module deliberately accepts the batch's persistence-backed extractor and +dossier builder as dependencies. Both the browser workflow and the JSON +integration boundary therefore execute the same business flow in-process, +while tests can replace either dependency without constructing HTTP calls or +making network requests. +""" + + +def analyze_batch(batch_id, documents, extractor, dossier_builder, + progress_callback=None, llm_client=None): + """Extract document passports, assemble dossiers, and shape the API result. + + `extractor` owns LLM interaction and persistence of each extraction; + `dossier_builder` consumes those saved passports and persists the resulting + graph. Keeping the calls in this order is an invariant: grouping must never + observe a partially extracted batch. The callback remains optional because + small batches run synchronously, while the existing large-batch route uses + it to update persistent progress. + """ + extraction = extractor( + batch_id, + llm_client=llm_client, + progress_callback=progress_callback, + documents=documents, + ) + dossier_result = dossier_builder(batch_id) if extraction.get("ok") else {"ok": False} + 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("ok")) and bool(dossier_result.get("ok")), + "v3": True, + "done": extraction.get("done", 0), + "total": len(documents), + "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", []), + } \ No newline at end of file diff --git a/site/services/extractor.py b/site/services/extractor.py index f898720..db5463b 100644 --- a/site/services/extractor.py +++ b/site/services/extractor.py @@ -252,7 +252,7 @@ def extract_document(document, llm_client=None, batch_id=None): """Извлечь паспорт одного документа, изолируя ошибки на уровне документа.""" document_id = document["id"] batch_id = batch_id if batch_id is not None else document.get("batch_id") - elements = _decode_elements(document.get("elements_json")) + elements = _decode_elements(document.get("elements", document.get("elements_json"))) structured_text = format_elements_with_structure(elements) raw_response = "{}" try: @@ -306,13 +306,14 @@ 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, progress_callback=None): +def extract_batch(batch_id, llm_client=None, progress_callback=None, documents=None): """Обработать документы batch и вернуть сводку без batch-wide failure. Optional progress callback получает (completed_count, document_result) после каждого документа; ошибки callback не должны останавливать обработку batch. """ - documents = db_docs.list_for_extraction(batch_id) + if documents is None: + documents = db_docs.list_for_extraction(batch_id) results = [] for completed, document in enumerate(documents, start=1): result = extract_document(document, llm_client=llm_client, batch_id=batch_id) diff --git a/site/services/grouping.py b/site/services/grouping.py index ad24062..69ef075 100644 --- a/site/services/grouping.py +++ b/site/services/grouping.py @@ -15,6 +15,7 @@ from db import supplements as db_supplements from db import dossiers as db_dossiers from db import exceptions as db_exceptions from db import extractions as db_extractions +from db.connection import transaction _HOMOGLYPHS = str.maketrans({ @@ -164,6 +165,12 @@ def _dossier_exception(batch_id, document_id, title, source_quote): def build_dossiers_for_batch(batch_id): + """Atomically rebuild the dossier graph, retaining the old graph on failure.""" + with transaction(): + return _build_dossiers_for_batch(batch_id) + + +def _build_dossiers_for_batch(batch_id): """Построить и сохранить dossier graph из паспортов документов одного batch. Полные номера и ссылки образуют рёбра только при совместимом контрагенте. diff --git a/site/static/app.js b/site/static/app.js index 82899fe..c480962 100644 --- a/site/static/app.js +++ b/site/static/app.js @@ -157,7 +157,7 @@ async function runClassify() { clearInterval(pollTimer); clearInterval(timer); classifyDone = true; - if (d.state === 'failed') { + if (d.state === 'failed' || d.state === 'interrupted') { btn.innerHTML = '✗ ' + escHtml(d.error || 'ошибка извлечения'); btn.disabled = false; return;