From 811be85efe0cf66f8fb4df98998d3cc3c00cede0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E2=80=9CNaeel=E2=80=9D?= Date: Thu, 27 Aug 2026 11:23:14 +0300 Subject: [PATCH] Fix comparison pipeline event and transaction handling --- site/config.py | 2 +- site/db/connection.py | 20 ++++++++++++++++- site/db/spec_events.py | 45 ++++++++++++++++++++++++++------------- site/services/process.py | 14 ++++++++++-- site/templates/index.html | 14 ++++++------ tests/test_spec_events.py | 4 ++-- 6 files changed, 71 insertions(+), 28 deletions(-) diff --git a/site/config.py b/site/config.py index efca7bc..7ded77f 100644 --- a/site/config.py +++ b/site/config.py @@ -1,7 +1,7 @@ """Конфигурация приложения — все настройки в одном месте.""" import os -VERSION = "2.0.20" +VERSION = "2.0.21" LLM_URL = os.getenv("LLM_API_URL", "https://api.aillm.ru/v1/chat/completions") LLM_KEY = os.getenv("LLM_API_KEY", "") diff --git a/site/db/connection.py b/site/db/connection.py index fadf220..cae8207 100644 --- a/site/db/connection.py +++ b/site/db/connection.py @@ -10,12 +10,29 @@ import sqlite3 import threading import time import os +from contextlib import contextmanager DB_PATH = "/tmp/contracts.db" _db_session_key = None _local = threading.local() +@contextmanager +def transaction(): + """Run DB writes atomically on the current thread connection.""" + conn = get_conn() + conn.execute("BEGIN IMMEDIATE") + _local.in_transaction = True + try: + yield conn + conn.commit() + except Exception: + conn.rollback() + raise + finally: + _local.in_transaction = False + + def init_db(): """Создать/пересоздать БД + схему. Вызывается при старте и после cleanup.""" global _db_session_key @@ -171,7 +188,8 @@ def execute(sql, params=None): conn = get_conn() sql = _pg_to_sqlite(sql) cur = conn.execute(sql, params or []) - conn.commit() + if not getattr(_local, "in_transaction", False): + conn.commit() return cur.rowcount diff --git a/site/db/spec_events.py b/site/db/spec_events.py index 1a82fbe..fb102e3 100644 --- a/site/db/spec_events.py +++ b/site/db/spec_events.py @@ -1,6 +1,6 @@ """spec_events — event sourcing: apply ops, reset contract.""" import json, uuid -from db.connection import query, execute, get_conn +from db.connection import query, execute, get_conn, transaction def reset(contract_id): @@ -27,9 +27,15 @@ def get_next_seq(contract_id): def apply_ops(contract_id, supplement_id, document_id, ops, prompt_id, raw_llm_response): """Apply ADD/UPDATE/DELETE ops. Returns summary dict.""" + with transaction(): + return _apply_ops(contract_id, supplement_id, document_id, ops, prompt_id, raw_llm_response) + + +def _apply_ops(contract_id, supplement_id, document_id, ops, prompt_id, raw_llm_response): added = 0 updated = 0 deleted = 0 + unresolved = 0 seq = get_next_seq(contract_id) for op in ops: @@ -42,6 +48,7 @@ def apply_ops(contract_id, supplement_id, document_id, ops, prompt_id, raw_llm_r # ADD without name → UNRESOLVED _log_unresolved(contract_id, supplement_id, seq, op, prompt_id, document_id, raw_llm_response, "ADD with empty name") seq += 1 + unresolved += 1 continue name_hash = _hash(name, nr.get("date_start")) execute( @@ -65,6 +72,7 @@ def apply_ops(contract_id, supplement_id, document_id, ops, prompt_id, raw_llm_r if not th: _log_unresolved(contract_id, supplement_id, seq, op, prompt_id, document_id, raw_llm_response, "UPDATE with empty target_hash") seq += 1 + unresolved += 1 continue execute( """INSERT INTO spec_events (id, contract_id, supplement_id, seq, action, target_hash, @@ -86,6 +94,7 @@ def apply_ops(contract_id, supplement_id, document_id, ops, prompt_id, raw_llm_r if not th: _log_unresolved(contract_id, supplement_id, seq, op, prompt_id, document_id, raw_llm_response, "DELETE with empty target_hash") seq += 1 + unresolved += 1 continue execute( """INSERT INTO spec_events (id, contract_id, supplement_id, seq, action, target_hash, @@ -104,27 +113,19 @@ def apply_ops(contract_id, supplement_id, document_id, ops, prompt_id, raw_llm_r elif action == "UNRESOLVED": # Log but don't apply - execute( - """INSERT INTO spec_events (id, contract_id, supplement_id, seq, action, target_hash, - new_values, comment, status, prompt_version, source_document_id, raw_llm_response) - VALUES (%s, %s, %s, %s, 'UNRESOLVED', %s, %s, %s, 'unresolved', %s, %s, %s)""", - ( - str(uuid.uuid4()), contract_id, supplement_id, seq, - op.get("target_hash", ""), - json.dumps(op.get("new_values", {}), ensure_ascii=False), - op.get("reason", op.get("comment", "")), - prompt_id, document_id, - json.dumps(raw_llm_response, ensure_ascii=False), - ), - ) + _log_unresolved(contract_id, supplement_id, seq, op, prompt_id, document_id, raw_llm_response, + op.get("reason", op.get("comment", ""))) seq += 1 + unresolved += 1 else: # Unknown action — log as UNRESOLVED _log_unresolved(contract_id, supplement_id, seq, op, prompt_id, document_id, raw_llm_response, f"unknown action: {action}") + seq += 1 + unresolved += 1 - return {"added": added, "updated": updated, "deleted": deleted} + return {"added": added, "updated": updated, "deleted": deleted, "unresolved": unresolved} def _log_unresolved(contract_id, supplement_id, seq, op, prompt_id, document_id, raw_llm_response, reason): @@ -187,6 +188,20 @@ def _update_spec_current(contract_id, name_hash, new_values): sets.append(f"{field} = %s") params.append(new_values[field]) if sets: + if "name" in new_values or "date_start" in new_values: + updated_name = new_values.get("name") + updated_date = new_values.get("date_start") + if updated_name is None or updated_date is None: + current = query( + "SELECT name, date_start FROM spec_current WHERE contract_id = %s AND name_hash = %s", + (contract_id, name_hash), + ) + if current: + updated_name = updated_name if updated_name is not None else current[0]["name"] + updated_date = updated_date if updated_date is not None else current[0]["date_start"] + if updated_name is not None: + sets.append("name_hash = %s") + params.append(_hash(updated_name, updated_date)) sets.append("updated_at = datetime('now')") params.extend([contract_id, name_hash]) execute( diff --git a/site/services/process.py b/site/services/process.py index 760d0ad..90279c6 100644 --- a/site/services/process.py +++ b/site/services/process.py @@ -77,6 +77,15 @@ def run_pipeline(contract_id, order_ids, build_prompt_fn): ops = result.get("ops", []) mode = result.get("mode", "llm") + if mode == "full_replace" and not ops: + yield { + "type": "extract_error", + "supplement_id": sid, + "filename": filename, + "error": "full_replace returned empty ops", + } + continue + # Трансляция target_id ("r1","r2"...) → target_hash (name_hash из current_spec). # LLM возвращает target_id, а apply_ops() читает target_hash — без этого UPDATE/DELETE уходят в UNRESOLVED. for _op in ops: @@ -120,7 +129,7 @@ def run_pipeline(contract_id, order_ids, build_prompt_fn): } # Arithmetic check - check_arithmetic(ops) + arithmetic_warnings = check_arithmetic(ops) yield { "type": "applied", @@ -128,6 +137,7 @@ def run_pipeline(contract_id, order_ids, build_prompt_fn): "filename": filename, "summary": summary, "ops": applied_ops, + "arithmetic_warnings": arithmetic_warnings, } except Exception as e: @@ -139,7 +149,7 @@ def run_pipeline(contract_id, order_ids, build_prompt_fn): } total_time = round(time.time() - t0, 1) - yield {"type": "complete", "total_time_s": total_time} + yield {"type": "done", "total_time_s": total_time} def _elements_to_text(ej): diff --git a/site/templates/index.html b/site/templates/index.html index 843f736..859d3c4 100644 --- a/site/templates/index.html +++ b/site/templates/index.html @@ -73,7 +73,7 @@
Nubes - Сверка договоров — LLM AI-driven Event Sourcing v2.0.20 + Сверка договоров — LLM AI-driven Event Sourcing v2.0.21
○ Загрузка→ ○ Классификация→ @@ -224,11 +224,11 @@ import { listZipFiles } from '/upload/zip/list_zip_files.js'; window.listZipFiles = listZipFiles; - - - - - - + + + + + + diff --git a/tests/test_spec_events.py b/tests/test_spec_events.py index 1de1ff9..2760462 100644 --- a/tests/test_spec_events.py +++ b/tests/test_spec_events.py @@ -17,7 +17,7 @@ class TestApplyOpsAdd: def test_add(self, db): ops = [{"action": "ADD", "new_row": {"name": "Аренда стойко-места", "price": 50000, "qty": 1, "sum": 50000, "date_start": "2025-01-01"}}] s = spec_events.apply_ops(CID, SID, DID, ops, "pid", {}) - assert s == {"added": 1, "updated": 0, "deleted": 0} + assert s == {"added": 1, "updated": 0, "deleted": 0, "unresolved": 0} rows = spec_current.list_by_contract(CID) assert len(rows) == 1 assert rows[0]["name"] == "Аренда стойко-места" @@ -27,7 +27,7 @@ class TestApplyOpsAdd: def test_add_empty_name_unresolved(self, db): ops = [{"action": "ADD", "new_row": {"name": ""}}] s = spec_events.apply_ops(CID, SID, DID, ops, "pid", {}) - assert s == {"added": 0, "updated": 0, "deleted": 0} + assert s == {"added": 0, "updated": 0, "deleted": 0, "unresolved": 1} assert spec_current.list_by_contract(CID) == [] def test_add_multiple(self, db):