Nail/contractor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v1.0.155: VM PostgreSQL + modular db/ & services/ — Lucee stays SPA shell only

Browse Source
This commit is contained in:
naeel
2026-06-23 16:01:26 +04:00
parent 568758a23c
commit ee374171d8
16 changed files with 814 additions and 494 deletions
Download Patch File Download Diff File Expand all files Collapse all files
deploy/convert_server.py
+76 -494
View File
@@ -1,59 +1,31 @@
#!/usr/bin/env python3
"""HTTP-сервер: конвертация .doc → .docx + LLM-анализ ДС (/llm-ops) + SSE v2 (/process-v2)"""
"""Contracts VM server — thin HTTP router. All logic in db/ and services/."""
from http.server import HTTPServer, BaseHTTPRequestHandler
from socketserver import ThreadingMixIn
class ThreadingHTTPServer(ThreadingMixIn, HTTPServer):
"""Многопоточный HTTP-сервер."""
daemon_threads = True
from urllib.parse import urlparse, parse_qs
import subprocess, tempfile, os, sys, json, time, re
import json, re
import httpx
# ── DB auto-seed ──────────────────────────────────────────────────────────
from db import prompts as db_prompts
from db.connection import DB_CONFIG
db_prompts.seed_defaults()
# ── Services ──────────────────────────────────────────────────────────────
from services.upload import handle_upload
from services.unzip import handle_unzip
from services.process import run_pipeline
from llm_prompt import build_prompt
LLM_URL = "https://api.aillm.ru/v1/chat/completions"
LLM_KEY = os.environ.get("LLM_KEY", "")
LLM_MODEL = "gpt-oss-120b"
LUCEE_URL = "https://contractor.luceek8s.dev.nubes.ru"
def call_llm(current_spec, doc_text):
"""Вызов LLM API. Возвращает ({mode, ops}, prompt_id) или кидает исключение."""
prompt, prompt_id = build_prompt(current_spec, doc_text)
payload = {
"model": LLM_MODEL,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 8000,
"temperature": 0.1,
}
with httpx.Client(http2=True, timeout=120, verify=False) as client:
resp = client.post(LLM_URL, json=payload,
headers={"Authorization": f"Bearer {LLM_KEY}", "Content-Type": "application/json"})
resp.raise_for_status()
data = resp.json()
raw_text = data.get("choices", [{}])[0].get("message", {}).get("content", "")
json_text = raw_text
if "```json" in json_text:
json_text = json_text.split("```json")[1].split("```")[0]
elif "```" in json_text:
json_text = json_text.split("```")[1].split("```")[0]
return json.loads(json_text.strip()), prompt_id
class ThreadingHTTPServer(ThreadingMixIn, HTTPServer):
daemon_threads = True
class Handler(BaseHTTPRequestHandler):
def do_GET(self):
parsed = urlparse(self.path)
if parsed.path == "/process-v2":
self._handle_process_v2(parsed)
else:
self.send_error(404)
def do_OPTIONS(self):
self.send_response(200)
self.send_header("Access-Control-Allow-Origin", "*")
self.send_header("Access-Control-Allow-Methods", "GET, OPTIONS")
self._send_cors()
self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
self.send_header("Access-Control-Allow-Headers", "Content-Type")
self.end_headers()
@@ -61,50 +33,29 @@ class Handler(BaseHTTPRequestHandler):
parsed = urlparse(self.path)
if parsed.path == "/process-v2":
self._handle_process_v2(parsed)
elif parsed.path == "/health":
self._json({"ok": True, "db": DB_CONFIG["dbname"]})
else:
self.send_error(404)
def do_POST(self):
if self.path == "/llm-ops":
self._handle_llm_ops()
elif self.path == "/unzip-upload":
self._handle_unzip_upload()
elif self.path == "/upload":
if self.path == "/upload":
self._handle_upload()
elif self.path == "/parse-pdf":
self._handle_parse_pdf()
elif self.path == "/unzip-upload":
self._handle_unzip()
elif self.path == "/llm-ops":
self._handle_llm_ops()
else:
self._handle_doc_convert()
self.send_error(404)
def _sse(self, data):
"""Отправить SSE-событие."""
self.wfile.write(f"data: {json.dumps(data, ensure_ascii=False)}\n\n".encode())
self.wfile.flush()
def _lucee_query(self, sql):
"""Запрос к Lucee API."""
resp = httpx.get(f"{LUCEE_URL}/api.cfm", params={"action": "query", "sql": sql}, timeout=30)
resp.raise_for_status()
d = resp.json()
if not d.get("OK"):
raise Exception(d.get("ERROR", "API error"))
cols = d["COLUMNS"]
rows = []
for r in d["ROWS"]:
row = {}
for i, c in enumerate(cols):
row[c.lower()] = r[i] if isinstance(r, list) else r[c]
rows.append(row)
return rows
# ── /process-v2 (SSE) ─────────────────────────────────────────────────
def _handle_process_v2(self, parsed):
"""SSE-стриминг v2 пайплайна."""
params = parse_qs(parsed.query)
cid = params.get("contract_id", [None])[0]
if not cid:
self.send_error(400, "contract_id required")
return
# Validate UUID to prevent SQL injection
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}', cid, re.I):
self.send_error(400, "invalid contract_id format")
return
@@ -117,445 +68,76 @@ class Handler(BaseHTTPRequestHandler):
self.wfile.write(b": ok\n\n")
self.wfile.flush()
t0 = time.time()
order_ids = params.get("order", [None])[0] or ""
try:
# 0. Сброс предыдущих результатов сравнения
r = httpx.get(f"{LUCEE_URL}/reset_contract.cfm?contract_id={cid}", timeout=10)
if r.status_code != 200:
self._sse({"type": "error", "message": f"reset failed: HTTP {r.status_code}"})
return
# 1. Список supplements (с document_id для provenance)
supps = self._lucee_query(
f"SELECT s.id, s.type, d.filename, d.id as document_id FROM supplements s "
f"JOIN documents d ON s.document_id=d.id "
f"WHERE s.contract_id='{cid}' AND d.elements_json IS NOT NULL ORDER BY s.created_at"
)
# Reorder if client specified &order=...
order_ids = (params.get("order", [None])[0] or "").strip()
if order_ids:
order_list = [x.strip() for x in order_ids.split(",") if x.strip()]
order_map = {oid: i for i, oid in enumerate(order_list)}
supps.sort(key=lambda s: order_map.get(s["id"], 999999))
if not supps:
self._sse({"type": "error", "message": "Нет распарсенных файлов"})
return
for s in supps:
sid = s["id"]
# Текущая спецификация
cur = self._lucee_query(
f"SELECT name_hash, name, price, qty, sum, date_start "
f"FROM spec_current WHERE contract_id='{cid}' ORDER BY name"
)
current_spec = []
for r in cur:
current_spec.append({
"hash": r["name_hash"],
"name": r["name"],
"price": float(r["price"]) if r.get("price") else None,
"qty": float(r["qty"]) if r.get("qty") else None,
"sum": float(r["sum"]) if r.get("sum") else None,
"date_start": r["date_start"],
})
# elements_json → текст
docs = self._lucee_query(
f"SELECT elements_json FROM documents d "
f"JOIN supplements s ON s.document_id=d.id WHERE s.id='{sid}'"
)
if not docs:
continue
ej = docs[0]["elements_json"]
# Lucee JSONB может вернуть: строку, dict-обёртку {Value,Type}, или уже список
if isinstance(ej, dict) and "Value" in ej:
ej = ej["Value"]
if isinstance(ej, str):
elements = json.loads(ej)
elif isinstance(ej, list):
elements = ej
else:
elements = []
# Lucee serializeJSON → UPPERCASE keys, normalize to lowercase
elements = [{k.lower(): v for k, v in el.items()} for el in elements]
lines = []
for el in elements:
if el.get("type") == "paragraph":
prefix = f"[{el['style']}] " if el.get("style") else ""
lines.append(prefix + el.get("text", ""))
elif el.get("type") == "table":
rows = el.get("rows", [])
if rows:
ncols = len(rows[0])
lines.append(f"--- Таблица ({len(rows)}×{ncols}) ---")
for row in rows:
cells = [str(c).replace("\n", " ").replace("|", "\\|") for c in row[:ncols]]
lines.append("| " + " | ".join(cells) + " |")
lines.append("")
doc_text = "\n".join(lines)
self._sse({"type": "extract_start", "supplement_id": sid, "filename": s["filename"]})
# LLM (прямой вызов, не HTTP self-call)
import threading
t1 = time.time()
done = threading.Event()
result = [None]
error = [None]
def do_llm():
try:
result[0] = call_llm(current_spec, doc_text)
except Exception as e:
error[0] = str(e)
finally:
done.set()
t = threading.Thread(target=do_llm)
t.start()
# Keepalive каждые 15с пока LLM думает
while not done.wait(15):
self._sse({"type": "keepalive"})
t.join()
elapsed = round(time.time() - t1, 1)
if error[0]:
self._sse({"type": "extract_error", "supplement_id": sid, "filename": s["filename"],
"error": error[0], "time_s": elapsed})
continue
llm_result, prompt_id = result[0]
ops = llm_result.get("ops", [])
mode = llm_result.get("mode", "partial")
# Маппинг target_id (r1, r2...) → target_hash из current_spec
id_map = {f"r{i+1}": r["hash"] for i, r in enumerate(current_spec)}
# Обогащение ops: target_id → target_hash + имя из current_spec
spec_names = {r["hash"]: r["name"] for r in current_spec}
for op in ops:
tid = op.get("target_id")
if tid and tid in id_map:
op["target_hash"] = id_map[tid]
# Добавить имя для UPDATE/DELETE из spec_current
th = op.get("target_hash")
if th and th in spec_names and not op.get("new_row", {}).get("name"):
op.setdefault("new_row", {})["name"] = spec_names[th]
self._sse({"type": "llm_done", "supplement_id": sid, "filename": s["filename"],
"ops_count": len(ops), "mode": mode, "time_s": elapsed})
# Применить (с provenance: document_id, prompt_id, raw_llm_response)
apply_resp = httpx.post(
f"{LUCEE_URL}/apply_events.cfm?contract_id={cid}&mode={mode}",
json={
"supplement_id": sid,
"document_id": s.get("document_id", ""),
"ops": ops,
"prompt_id": prompt_id,
"raw_llm_response": llm_result
}, timeout=30
)
if apply_resp.status_code == 200:
ar = apply_resp.json()
# Lucee serializeJSON → UPPERCASE keys
summary = {k.lower(): v for k, v in ar.get("SUMMARY", {}).items()}
self._sse({"type": "applied", "supplement_id": sid, "summary": summary, "ops": ops})
else:
self._sse({"type": "apply_error", "supplement_id": sid,
"error": f"apply HTTP {apply_resp.status_code}"})
total_time = round(time.time() - t0, 1)
self._sse({"type": "done", "total_time_s": total_time})
run_pipeline(cid, order_ids, self._sse, build_prompt)
except Exception as e:
self._sse({"type": "error", "message": str(e)})
def _sse(self, data):
self.wfile.write(f"data: {json.dumps(data, ensure_ascii=False)}\n\n".encode())
self.wfile.flush()
# ── /upload ───────────────────────────────────────────────────────────
def _handle_upload(self):
try:
content_type = self.headers.get("Content-Type", "")
content_length = int(self.headers.get("Content-Length", 0))
result = handle_upload(self.rfile, content_type, content_length)
self._json(result)
except Exception as e:
self._json({"ok": False, "error": str(e)}, 500)
# ── /unzip-upload ─────────────────────────────────────────────────────
def _handle_unzip(self):
try:
content_length = int(self.headers.get("Content-Length", 0))
result = handle_unzip(self.rfile, content_length)
self._json(result)
except Exception as e:
self._json({"ok": False, "error": str(e)}, 500)
# ── /llm-ops (debug) ──────────────────────────────────────────────────
def _handle_llm_ops(self):
from services.llm import call_llm
length = int(self.headers.get("Content-Length", 0))
body = json.loads(self.rfile.read(length))
try:
result, prompt_id = call_llm(body.get("current_spec", []), body.get("doc_text", ""))
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps(result, ensure_ascii=False).encode())
result, prompt_id = call_llm(
body.get("current_spec", []),
body.get("doc_text", ""),
build_prompt,
)
self._json(result)
except Exception as e:
self.send_response(502)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps({"error": str(e)}, ensure_ascii=False).encode())
self._json({"error": str(e)}, 502)
def _handle_upload(self):
"""Принять файл и передать в Lucee чанками (лимит ingress ~30KB)."""
import uuid, base64
from email.parser import BytesParser
# ── Helpers ───────────────────────────────────────────────────────────
content_type = self.headers.get("Content-Type", "")
length = int(self.headers.get("Content-Length", 0))
body = self.rfile.read(length)
msg = BytesParser().parsebytes(
b"Content-Type: " + content_type.encode() + b"\r\n\r\n" + body
)
filename = None
file_data = None
contract_id = ""
if msg.is_multipart():
for part in msg.get_payload():
fn = part.get_filename()
if fn:
filename = fn
file_data = part.get_payload(decode=True)
else:
disp = part.get("Content-Disposition", "")
if "name=\"contract_id\"" in disp:
contract_id = part.get_payload(decode=True).decode("utf-8", errors="ignore").strip()
if not filename or not file_data:
self._send_error_cors(400, "no file in request")
return
CHUNK = 10000 # 10KB raw → 20KB hex → помещается
upload_id = str(uuid.uuid4())
raw = file_data
total = (len(raw) + CHUNK - 1) // CHUNK
for i in range(total):
chunk = raw[i*CHUNK:(i+1)*CHUNK]
with httpx.Client(http2=True, timeout=30, verify=False) as client:
r = client.post(
f"{LUCEE_URL}/chunk.cfm?action=put",
json={
"upload_id": upload_id,
"chunk_index": i,
"total_chunks": total,
"filename": filename,
"data": base64.b64encode(chunk).decode(),
},
)
if r.status_code != 200:
self._send_error_cors(502, f"chunk {i+1}/{total} failed: HTTP {r.status_code}")
return
with httpx.Client(http2=True, timeout=60, verify=False) as client:
resp = client.post(
f"{LUCEE_URL}/chunk.cfm?action=assemble",
json={"upload_id": upload_id, "contract_id": contract_id},
)
if resp.status_code == 200:
result = resp.json()
doc_id = result.get("DOC_ID", "")
# Парсинг на VM + сохранить в Lucee
parsed = self._parse_file(filename, file_data)
if parsed and parsed.get("status") == "parsed" and doc_id:
try:
escaped = json.dumps(parsed["elements"], ensure_ascii=False).replace("'", "''")
with httpx.Client(http2=True, timeout=15, verify=False) as c:
c.post(
f"{LUCEE_URL}/api.cfm",
data={
"action": "execute",
"sql": f"UPDATE documents SET elements_json='{escaped}'::jsonb, status='parsed' WHERE id='{doc_id}'"
}
)
except Exception:
pass
result["parsed"] = parsed
self.send_response(200)
def _json(self, data, status=200):
self.send_response(status)
self._send_cors()
self.send_header("Content-Type", "application/json; charset=utf-8")
self.end_headers()
self.wfile.write(json.dumps(result, ensure_ascii=False).encode())
else:
self._send_error_cors(502, f"assemble failed: HTTP {resp.status_code}")
def _parse_file(self, filename, data):
"""Распарсить файл на VM: PDF→PyPDF2, DOCX/DOC→python-docx, иначе plain text."""
import base64
ext = filename.rsplit('.', 1)[-1].lower() if '.' in filename else ''
try:
if ext == 'pdf':
from PyPDF2 import PdfReader
import io as io_mod
reader = PdfReader(io_mod.BytesIO(data))
elements = []
for page in reader.pages:
text = page.extract_text()
if text:
for line in text.split('\n'):
line = line.strip()
if line:
elements.append({"type": "paragraph", "text": line, "style": ""})
return {"status": "parsed", "element_count": len(elements), "elements": elements}
elif ext in ('docx', 'doc'):
# Для DOCX/DOC — оставляем Lucee (Apache POI работает)
return None
else:
# Plain text
text = data.decode('utf-8', errors='ignore')[:5000]
lines = [l.strip() for l in text.split('\n') if l.strip()]
return {"status": "parsed", "element_count": len(lines),
"elements": [{"type": "paragraph", "text": l, "style": ""} for l in lines]}
except Exception as e:
return {"status": "error", "error": str(e), "element_count": 0}
def _handle_parse_pdf(self):
"""Распарсить PDF через PyPDF2 (получает base64 тело от Lucee)."""
import base64
from PyPDF2 import PdfReader
import io as io_module
length = int(self.headers.get("Content-Length", 0))
body = self.rfile.read(length).decode("utf-8", errors="ignore").strip()
raw = base64.b64decode(body)
try:
reader = PdfReader(io_module.BytesIO(raw))
elements = []
for page in reader.pages:
text = page.extract_text()
if text:
for line in text.split('\n'):
line = line.strip()
if line:
elements.append({"type": "paragraph", "text": line, "style": ""})
result = {
"ok": True,
"status": "parsed",
"element_count": len(elements),
"elements": elements,
"tables": 0,
"paragraphs": len(elements),
}
except Exception as e:
result = {"ok": False, "status": "error", "error": str(e), "element_count": 0}
self.send_response(200)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Access-Control-Allow-Origin", "*")
self.end_headers()
self.wfile.write(json.dumps(result, ensure_ascii=False).encode())
def _handle_unzip_upload(self):
"""Распаковать ZIP и загрузить каждый файл в Lucee. Только файлы из корня архива."""
import zipfile, io
from email.parser import BytesParser
content_type = self.headers.get("Content-Type", "")
length = int(self.headers.get("Content-Length", 0))
body = self.rfile.read(length)
# Raw binary (браузер) или multipart (curl fallback)
if "multipart" in content_type:
msg = BytesParser().parsebytes(
b"Content-Type: " + content_type.encode() + b"\r\n\r\n" + body
)
zip_data = None
if msg.is_multipart():
for part in msg.get_payload():
if part.get_filename():
zip_data = part.get_payload(decode=True)
break
else:
zip_data = body
else:
zip_data = body
if not zip_data:
self._send_error_cors(400, "no file in request")
return
results = []
try:
with zipfile.ZipFile(io.BytesIO(zip_data)) as zf:
for name in zf.namelist():
if name.endswith('/') or '/' in name:
continue
content = zf.read(name)
ext = name.rsplit('.', 1)[-1].lower() if '.' in name else ''
if ext not in ('docx', 'doc', 'pdf'):
continue
import base64 as b64
content_b64 = b64.b64encode(content).decode()
resp = httpx.post(
f"{LUCEE_URL}/upload.cfm",
data={"filename": name, "data": content_b64},
timeout=30
)
if resp.status_code == 200:
rj = resp.json()
if rj.get("OK"):
entry = {
"filename": name,
"doc_id": rj.get("DOC_ID", ""),
"contract_id": rj.get("CONTRACT_ID", ""),
"size": len(content),
}
# Парсинг PDF на VM
parsed = self._parse_file(name, content)
if parsed:
entry["parsed"] = parsed
results.append(entry)
else:
results.append({"filename": name, "error": rj.get("ERROR", "upload failed")})
else:
results.append({"filename": name, "error": f"HTTP {resp.status_code}"})
except zipfile.BadZipFile:
self._send_error_cors(400, "not a valid ZIP file")
return
except Exception as e:
self._send_error_cors(500, str(e))
return
self.send_response(200)
self._send_cors()
self.send_header("Content-Type", "application/json; charset=utf-8")
self.end_headers()
self.wfile.write(json.dumps({"ok": True, "files": results}, ensure_ascii=False).encode())
self.wfile.write(json.dumps(data, ensure_ascii=False).encode())
def _send_cors(self):
self.send_header("Access-Control-Allow-Origin", "*")
self.send_header("Access-Control-Allow-Methods", "POST, OPTIONS")
self.send_header("Access-Control-Allow-Headers", "*")
def _send_error_cors(self, code, message):
self.send_response(code)
self._send_cors()
self.send_header("Content-Type", "application/json; charset=utf-8")
self.end_headers()
self.wfile.write(json.dumps({"ok": False, "error": message}, ensure_ascii=False).encode())
def log_message(self, format, *args):
pass
def _handle_doc_convert(self):
length = int(self.headers.get("Content-Length", 0))
data = self.rfile.read(length)
with tempfile.NamedTemporaryFile(suffix=".doc", delete=False) as f:
f.write(data); doc_path = f.name
tmpdir = tempfile.mkdtemp()
if __name__ == "__main__":
import os
port = int(os.environ.get("PORT", "8766"))
server = ThreadingHTTPServer(("0.0.0.0", port), Handler)
print(f"Contracts VM server on :{port}, db={DB_CONFIG['dbname']}")
try:
subprocess.run(["libreoffice", "--headless", "--convert-to", "docx", "--outdir", tmpdir, doc_path],
timeout=30, capture_output=True)
docx = [x for x in os.listdir(tmpdir) if x.endswith(".docx")]
if docx:
with open(os.path.join(tmpdir, docx[0]), "rb") as f:
out = f.read()
self.send_response(200)
self.send_header("Content-Type", "application/vnd.openxmlformats-officedocument.wordprocessingml.document")
self.send_header("Content-Length", len(out))
self.end_headers()
self.wfile.write(out)
else:
self.send_error(500, "Conversion failed")
finally:
os.unlink(doc_path)
for x in os.listdir(tmpdir): os.unlink(os.path.join(tmpdir, x))
os.rmdir(tmpdir)
def log_message(self, *a): pass
ThreadingHTTPServer(("127.0.0.1", 8766), Handler).serve_forever()
server.serve_forever()
except KeyboardInterrupt:
server.shutdown()
deploy/db/__init__.py
View File
deploy/db/connection.py
+70
View File
@@ -0,0 +1,70 @@
"""Database connection — psycopg2 connection pool."""
import os
import psycopg2
import psycopg2.pool
import psycopg2.extras
_pool = None
DB_CONFIG = {
"host": os.getenv("DB_HOST", "127.0.0.1"),
"port": int(os.getenv("DB_PORT", "5432")),
"dbname": os.getenv("DB_NAME", "baza"),
"user": os.getenv("DB_USER", "super"),
"password": os.getenv("DB_PASS", ""),
}
def get_pool():
global _pool
if _pool is None:
_pool = psycopg2.pool.ThreadedConnectionPool(
minconn=1, maxconn=10, **DB_CONFIG
)
return _pool
def query(sql, params=None):
"""SELECT → list of dicts (lowercase keys)."""
pool = get_pool()
conn = pool.getconn()
try:
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
cur.execute(sql, params)
rows = cur.fetchall()
return [{k.lower(): v for k, v in r.items()} for r in rows]
finally:
pool.putconn(conn)
def execute(sql, params=None):
"""INSERT/UPDATE/DELETE → rowcount."""
pool = get_pool()
conn = pool.getconn()
try:
with conn.cursor() as cur:
cur.execute(sql, params)
conn.commit()
return cur.rowcount
except Exception:
conn.rollback()
raise
finally:
pool.putconn(conn)
def execute_returning(sql, params=None):
"""INSERT/UPDATE/DELETE with RETURNING → first row dict."""
pool = get_pool()
conn = pool.getconn()
try:
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
cur.execute(sql, params)
conn.commit()
row = cur.fetchone()
return {k.lower(): v for k, v in row.items()} if row else None
except Exception:
conn.rollback()
raise
finally:
pool.putconn(conn)
deploy/db/contracts.py
+25
View File
@@ -0,0 +1,25 @@
"""Contracts CRUD."""
from .connection import query, execute, execute_returning
def insert(number, client=""):
return execute_returning(
"INSERT INTO contracts (number, client) VALUES (%s, %s) RETURNING *",
(number, client),
)
def get(contract_id):
rows = query("SELECT * FROM contracts WHERE id = %s", (contract_id,))
return rows[0] if rows else None
def delete(contract_id):
return execute("DELETE FROM contracts WHERE id = %s", (contract_id,))
def delete_orphaned():
"""Remove contracts with no supplements."""
execute(
"DELETE FROM contracts WHERE id NOT IN (SELECT DISTINCT contract_id FROM supplements)"
)
deploy/db/documents.py
+36
View File
@@ -0,0 +1,36 @@
"""Documents CRUD."""
from .connection import query, execute, execute_returning
def insert(filename, mime_type, original_bytes, status="uploaded"):
"""Insert document, return row dict."""
return execute_returning(
"""INSERT INTO documents (filename, mime_type, original_bytes, status)
VALUES (%s, %s, %s, %s) RETURNING *""",
(filename, mime_type, original_bytes, status),
)
def get(doc_id):
rows = query("SELECT * FROM documents WHERE id = %s", (doc_id,))
return rows[0] if rows else None
def set_parsed(doc_id, elements_json):
"""Update elements_json + status='parsed'."""
import json
return execute(
"UPDATE documents SET elements_json = %s::jsonb, status = 'parsed' WHERE id = %s",
(json.dumps(elements_json, ensure_ascii=False), doc_id),
)
def set_error(doc_id, error_message):
return execute(
"UPDATE documents SET status = 'error', error_message = %s WHERE id = %s",
(error_message, doc_id),
)
def delete(doc_id):
return execute("DELETE FROM documents WHERE id = %s", (doc_id,))
deploy/db/prompts.py
+37
View File
@@ -0,0 +1,37 @@
"""Prompts CRUD."""
from .connection import query, execute, execute_returning
def get_active(role):
rows = query(
"SELECT * FROM prompts WHERE role = %s AND is_active = true ORDER BY created_at DESC LIMIT 1",
(role,),
)
return rows[0] if rows else None
def get(prompt_id):
rows = query("SELECT * FROM prompts WHERE id = %s", (prompt_id,))
return rows[0] if rows else None
def seed_defaults():
"""Auto-seed default prompts if table is empty."""
count = query("SELECT COUNT(*) as cnt FROM prompts")
if count[0]["cnt"] > 0:
return
execute("""
INSERT INTO prompts (role, name, body, is_active, notes) VALUES (
'extract', 'default-v1',
'Ты — анализатор договоров облачного провайдера.\n\nНиже текст спецификации услуг из ПЕРВОГО документа (базовый договор).\nИзвлеки ВСЕ строки спецификации в JSON-массив.\n\nВерни СТРОГО JSON без пояснений. Не используй markdown-блоки.\n\nФОРМАТ:\n{\n "mode": "partial",\n "ops": [\n {"action": "ADD", "new_row": {"name": "полное наименование", "price": число, "qty": число, "sum": число, "date_start": "YYYY-MM-DD"}, "comment": ""}\n ]\n}\n\nПРАВИЛА:\n1. Извлеки КАЖДУЮ строку таблицы спецификации как отдельную ADD-операцию.\n2. Пропускай итоговые строки ("Итого...") и строки с подписями.\n3. Если ячейка пустая — ставь null (не пиши 0).\n4. price, qty, sum — ЧИСЛА, не строки.\n\nТЕКСТ ДОКУМЕНТА:\n---\n{doc_text}\n---',
true, 'Авто-создан из llm_prompt.py _build_initial'
)
""")
execute("""
INSERT INTO prompts (role, name, body, is_active, notes) VALUES (
'diff', 'default-v1',
'Ты — анализатор допсоглашений к договорам облачного провайдера.\n\nУ тебя есть текущая спецификация услуг и текст нового допсоглашения (ДС).\nТвоя задача — определить, какие изменения вносит ДС в текущую спецификацию.\n\nВерни СТРОГО JSON без пояснений. Не используй markdown-блоки.\n\nФОРМАТ:\n{\n "mode": "partial" | "full_replace",\n "ops": [\n {"action": "ADD", "new_row": {"name": "...", "price": число, "qty": число, "sum": число, "date_start": "YYYY-MM-DD"}, "comment": "..."},\n {"action": "UPDATE", "target_id": "rN", "new_values": {"price": число}, "comment": "..."},\n {"action": "DELETE", "target_id": "rN", "comment": "..."},\n {"action": "UNRESOLVED", "new_values": {"name": "...", "price": число, ...}, "reason": "почему не смог сопоставить"}\n ]\n}\n\nПРАВИЛА:\n1. mode = "full_replace" — если в тексте есть фразы: «в следующей редакции», «заменить приложение», «излагается в следующей редакции». При full_replace — опиши ВСЕ новые строки как ADD.\n2. mode = "partial" — если ДС меняет только отдельные строки.\n3. Для UPDATE/DELETE — укажи target_id (r1, r2...) из списка текущей спецификации. НЕ придумывай новые id.\n4. new_values в UPDATE — только ИЗМЕНЁННЫЕ поля (не все).\n5. Если не можешь однозначно сопоставить строку — action: UNRESOLVED с reason.\n6. Пропускай итоговые строки и подписи.\n7. price, qty, sum — ЧИСЛА (не строки).\n\nТЕКУЩАЯ СПЕЦИФИКАЦИЯ:\n{spec_current}\n\nТЕКСТ ДОПСОГЛАШЕНИЯ:\n---\n{doc_text}\n---',
true, 'Авто-создан из llm_prompt.py _build_diff'
)
""")
deploy/db/spec_current.py
+19
View File
@@ -0,0 +1,19 @@
"""spec_current — текущее состояние спецификации."""
from .connection import query
def list_by_contract(contract_id):
"""Return list of dicts with name_hash, name, price, qty, sum, date_start."""
return query(
"""SELECT name_hash, name, price, qty, sum, date_start
FROM spec_current WHERE contract_id = %s ORDER BY name""",
(contract_id,),
)
def get_elements_json(document_id):
"""Get elements_json for a document."""
rows = query(
"SELECT elements_json FROM documents WHERE id = %s", (document_id,)
)
return rows[0]["elements_json"] if rows and rows[0]["elements_json"] else None
deploy/db/spec_events.py
+145
View File
@@ -0,0 +1,145 @@
"""spec_events — event sourcing: apply ops, reset contract."""
import json, uuid
from .connection import query, execute
def reset(contract_id):
"""Clear spec_events + spec_current for contract."""
execute("DELETE FROM spec_current WHERE contract_id = %s", (contract_id,))
execute("DELETE FROM spec_events WHERE contract_id = %s", (contract_id,))
def get_next_seq(contract_id):
rows = query(
"SELECT COALESCE(MAX(seq), 0) + 1 AS n FROM spec_events WHERE contract_id = %s",
(contract_id,),
)
return rows[0]["n"] if rows else 1
def apply_ops(contract_id, supplement_id, document_id, ops, prompt_id, raw_llm_response):
"""Apply ADD/UPDATE/DELETE ops. Returns summary dict."""
added = 0
updated = 0
deleted = 0
seq = get_next_seq(contract_id)
for op in ops:
action = op.get("action", "").upper()
if action == "ADD":
nr = op.get("new_row", {})
name = nr.get("name", "")
name_hash = _hash(name)
execute(
"""INSERT INTO spec_events (contract_id, supplement_id, seq, action, target_hash,
new_values, comment, status, prompt_version, source_document_id, raw_llm_response)
VALUES (%s, %s, %s, 'ADD', %s, %s, %s, 'applied', %s, %s, %s)""",
(
contract_id, supplement_id, seq, name_hash,
json.dumps(nr, ensure_ascii=False),
op.get("comment", ""), prompt_id, document_id,
json.dumps(raw_llm_response, ensure_ascii=False),
),
)
_upsert_spec_current(contract_id, name_hash, nr)
seq += 1
added += 1
elif action == "UPDATE":
nv = op.get("new_values", {})
th = op.get("target_hash", "")
execute(
"""INSERT INTO spec_events (contract_id, supplement_id, seq, action, target_hash,
new_values, comment, status, prompt_version, source_document_id, raw_llm_response)
VALUES (%s, %s, %s, 'UPDATE', %s, %s, %s, 'applied', %s, %s, %s)""",
(
contract_id, supplement_id, seq, th,
json.dumps(nv, ensure_ascii=False),
op.get("comment", ""), prompt_id, document_id,
json.dumps(raw_llm_response, ensure_ascii=False),
),
)
_update_spec_current(contract_id, th, nv)
seq += 1
updated += 1
elif action == "DELETE":
th = op.get("target_hash", "")
execute(
"""INSERT INTO spec_events (contract_id, supplement_id, seq, action, target_hash,
new_values, comment, status, prompt_version, source_document_id, raw_llm_response)
VALUES (%s, %s, %s, 'DELETE', %s, %s, %s, 'applied', %s, %s, %s)""",
(
contract_id, supplement_id, seq, th,
json.dumps({}), op.get("comment", ""),
prompt_id, document_id,
json.dumps(raw_llm_response, ensure_ascii=False),
),
)
execute("DELETE FROM spec_current WHERE contract_id = %s AND name_hash = %s", (contract_id, th))
seq += 1
deleted += 1
elif action == "UNRESOLVED":
# Log but don't apply
execute(
"""INSERT INTO spec_events (contract_id, supplement_id, seq, action, target_hash,
new_values, comment, status, prompt_version, source_document_id, raw_llm_response)
VALUES (%s, %s, %s, 'UNRESOLVED', %s, %s, %s, 'unresolved', %s, %s, %s)""",
(
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),
),
)
seq += 1
return {"added": added, "updated": updated, "deleted": deleted}
def _hash(name):
import hashlib
return hashlib.sha256(name.strip().lower().encode()).hexdigest()[:16]
def _upsert_spec_current(contract_id, name_hash, row):
"""INSERT or UPDATE spec_current."""
existing = query(
"SELECT id FROM spec_current WHERE contract_id = %s AND name_hash = %s",
(contract_id, name_hash),
)
if existing:
execute(
"""UPDATE spec_current SET name=%s, price=%s, qty=%s, sum=%s, date_start=%s, updated_at=now()
WHERE contract_id=%s AND name_hash=%s""",
(row.get("name"), row.get("price"), row.get("qty"), row.get("sum"),
row.get("date_start"), contract_id, name_hash),
)
else:
execute(
"""INSERT INTO spec_current (contract_id, name_hash, name, price, qty, sum, date_start)
VALUES (%s, %s, %s, %s, %s, %s, %s)""",
(contract_id, name_hash, row.get("name"), row.get("price"),
row.get("qty"), row.get("sum"), row.get("date_start")),
)
def _update_spec_current(contract_id, name_hash, new_values):
"""Update specific fields in spec_current."""
sets = []
params = []
for field in ("name", "price", "qty", "sum", "date_start"):
if field in new_values:
sets.append(f"{field} = %s")
params.append(new_values[field])
if sets:
sets.append("updated_at = now()")
params.extend([contract_id, name_hash])
execute(
f"UPDATE spec_current SET {', '.join(sets)} WHERE contract_id = %s AND name_hash = %s",
params,
)
deploy/db/supplements.py
+47
View File
@@ -0,0 +1,47 @@
"""Supplements CRUD."""
from .connection import query, execute, execute_returning
def insert(contract_id, document_id, supp_type="additional"):
return execute_returning(
"""INSERT INTO supplements (contract_id, document_id, type)
VALUES (%s, %s, %s) RETURNING *""",
(contract_id, document_id, supp_type),
)
def list_by_contract(contract_id):
"""Supplements with parsed documents, ordered by created_at."""
return query(
"""SELECT s.id, s.type, s.document_id, d.filename
FROM supplements s
JOIN documents d ON s.document_id = d.id
WHERE s.contract_id = %s AND d.elements_json IS NOT NULL
ORDER BY s.created_at""",
(contract_id,),
)
def get(supp_id):
rows = query("SELECT * FROM supplements WHERE id = %s", (supp_id,))
return rows[0] if rows else None
def delete_by_document(contract_id, filename):
"""Delete supplement+document by contract+filename."""
row = query(
"""SELECT s.id as sid, s.document_id FROM supplements s
JOIN documents d ON d.id = s.document_id
WHERE s.contract_id = %s AND d.filename = %s""",
(contract_id, filename),
)
if row:
r = row[0]
execute("DELETE FROM supplements WHERE id = %s", (r["sid"],))
execute("DELETE FROM documents WHERE id = %s", (r["document_id"],))
return True
return False
def delete(supp_id):
return execute("DELETE FROM supplements WHERE id = %s", (supp_id,))
deploy/services/__init__.py
View File
deploy/services/llm.py
+37
View File
@@ -0,0 +1,37 @@
"""LLM service — call LLM API."""
import os, json
import httpx
LLM_URL = "https://api.aillm.ru/v1/chat/completions"
LLM_KEY = os.environ.get("LLM_KEY", "")
LLM_MODEL = "gpt-oss-120b"
def call_llm(current_spec, doc_text, build_prompt_fn):
"""Call LLM. Returns (parsed_result, prompt_id)."""
prompt, prompt_id = build_prompt_fn(current_spec, doc_text)
payload = {
"model": LLM_MODEL,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 8000,
"temperature": 0.1,
}
with httpx.Client(http2=True, timeout=120, verify=False) as client:
resp = client.post(
LLM_URL,
json=payload,
headers={
"Authorization": f"Bearer {LLM_KEY}",
"Content-Type": "application/json",
},
)
resp.raise_for_status()
data = resp.json()
raw_text = data.get("choices", [{}])[0].get("message", {}).get("content", "")
json_text = raw_text
if "```json" in json_text:
json_text = json_text.split("```json")[1].split("```")[0]
elif "```" in json_text:
json_text = json_text.split("```")[1].split("```")[0]
return json.loads(json_text.strip()), prompt_id
deploy/services/parse.py
+73
View File
@@ -0,0 +1,73 @@
"""Parse service — PDF (PyPDF2), DOCX (python-docx), plain text fallback."""
import io
from PyPDF2 import PdfReader
def parse_file(filename, data):
"""Parse file bytes → {status, elements, element_count}."""
ext = filename.rsplit(".", 1)[-1].lower() if "." in filename else ""
try:
if ext == "pdf":
return _parse_pdf(data)
elif ext == "docx":
return _parse_docx(data)
elif ext == "doc":
return _parse_docx(data) # python-docx handles most .doc
else:
return _parse_text(data)
except Exception as e:
return {"status": "error", "error": str(e), "element_count": 0}
def _parse_pdf(data):
reader = PdfReader(io.BytesIO(data))
elements = []
for page in reader.pages:
text = page.extract_text()
if text:
for line in text.split("\n"):
line = line.strip()
if line:
elements.append({"type": "paragraph", "text": line, "style": ""})
return {"status": "parsed", "element_count": len(elements), "elements": elements}
def _parse_docx(data):
import docx
doc = docx.Document(io.BytesIO(data))
elements = []
for block in doc.element.body:
tag = block.tag.split("}")[-1] if "}" in block.tag else block.tag
if tag == "p":
text = _extract_paragraph_text(block)
if text:
elements.append({"type": "paragraph", "text": text, "style": ""})
elif tag == "tbl":
rows = []
for tr in block.findall(".//{http://schemas.openxmlformats.org/wordprocessingml/2006/main}tr"):
cells = []
for tc in tr.findall(".//{http://schemas.openxmlformats.org/wordprocessingml/2006/main}tc"):
cell_text = "".join(t.text or "" for t in tc.findall(".//{http://schemas.openxmlformats.org/wordprocessingml/2006/main}t"))
cells.append(cell_text.strip())
if cells:
rows.append(cells)
if rows:
elements.append({"type": "table", "rows": rows})
return {"status": "parsed", "element_count": len(elements), "elements": elements}
def _extract_paragraph_text(p_elem):
texts = []
for t in p_elem.findall(".//{http://schemas.openxmlformats.org/wordprocessingml/2006/main}t"):
if t.text:
texts.append(t.text)
return "".join(texts).strip()
def _parse_text(data):
text = data.decode("utf-8", errors="ignore")[:5000]
lines = [l.strip() for l in text.split("\n") if l.strip()]
return {"status": "parsed", "element_count": len(lines),
"elements": [{"type": "paragraph", "text": l, "style": ""} for l in lines]}
deploy/services/process.py
+132
View File
@@ -0,0 +1,132 @@
"""Process service — SSE pipeline: reset → supplements → LLM → apply."""
import json, time, threading
from db import supplements, spec_current, spec_events
def run_pipeline(contract_id, order_ids, sse_send, build_prompt_fn):
"""Run full SSE comparison pipeline. sse_send is a callback(dict)."""
t0 = time.time()
# 0. Reset
spec_events.reset(contract_id)
# 1. Get supplements with parsed documents
supps = supplements.list_by_contract(contract_id)
if order_ids:
order_list = [x.strip() for x in order_ids.split(",") if x.strip()]
order_map = {oid: i for i, oid in enumerate(order_list)}
supps.sort(key=lambda s: order_map.get(s["id"], 999999))
if not supps:
sse_send({"type": "error", "message": "Нет распарсенных файлов"})
return
from .llm import call_llm
for s in supps:
sid = s["id"]
# Current spec
cur = spec_current.list_by_contract(contract_id)
current_spec = []
for r in cur:
current_spec.append({
"hash": r["name_hash"],
"name": r["name"],
"price": float(r["price"]) if r.get("price") is not None else None,
"qty": float(r["qty"]) if r.get("qty") is not None else None,
"sum": float(r["sum"]) if r.get("sum") is not None else None,
"date_start": r["date_start"],
})
# Get elements_json
ej = spec_current.get_elements_json(s["document_id"])
if not ej:
continue
if isinstance(ej, dict) and "Value" in ej:
ej = ej["Value"]
if isinstance(ej, str):
elements = json.loads(ej)
elif isinstance(ej, list):
elements = ej
else:
elements = []
elements = [{k.lower(): v for k, v in el.items()} for el in elements]
doc_text = _elements_to_text(elements)
sse_send({"type": "extract_start", "supplement_id": sid, "filename": s["filename"]})
# LLM call in thread with keepalive
t1 = time.time()
done = threading.Event()
result = [None]
error = [None]
def do_llm():
try:
result[0] = call_llm(current_spec, doc_text, build_prompt_fn)
except Exception as e:
error[0] = str(e)
finally:
done.set()
t = threading.Thread(target=do_llm)
t.start()
while not done.wait(15):
sse_send({"type": "keepalive"})
t.join()
elapsed = round(time.time() - t1, 1)
if error[0]:
sse_send({"type": "extract_error", "supplement_id": sid,
"filename": s["filename"], "error": error[0], "time_s": elapsed})
continue
llm_result, prompt_id = result[0]
ops = llm_result.get("ops", [])
mode = llm_result.get("mode", "partial")
# Enrich ops
id_map = {f"r{i+1}": r["hash"] for i, r in enumerate(current_spec)}
spec_names = {r["hash"]: r["name"] for r in current_spec}
for op in ops:
tid = op.get("target_id")
if tid and tid in id_map:
op["target_hash"] = id_map[tid]
th = op.get("target_hash")
if th and th in spec_names and not op.get("new_row", {}).get("name"):
op.setdefault("new_row", {})["name"] = spec_names[th]
sse_send({"type": "llm_done", "supplement_id": sid, "filename": s["filename"],
"ops_count": len(ops), "mode": mode, "time_s": elapsed})
# Apply events
summary = spec_events.apply_ops(
contract_id, sid, s.get("document_id", ""), ops, prompt_id, llm_result
)
sse_send({"type": "applied", "supplement_id": sid, "summary": summary, "ops": ops})
total_time = round(time.time() - t0, 1)
sse_send({"type": "done", "total_time_s": total_time})
def _elements_to_text(elements):
lines = []
for el in elements:
if el.get("type") == "paragraph":
prefix = f"[{el['style']}] " if el.get("style") else ""
lines.append(prefix + el.get("text", ""))
elif el.get("type") == "table":
rows = el.get("rows", [])
if rows:
ncols = len(rows[0])
lines.append(f"--- Таблица ({len(rows)}×{ncols}) ---")
for row in rows:
cells = [str(c).replace("\n", " ").replace("|", "\\|") for c in row[:ncols]]
lines.append("| " + " | ".join(cells) + " |")
lines.append("")
return "\n".join(lines)
deploy/services/unzip.py
+38
View File
@@ -0,0 +1,38 @@
"""Unzip service — extract files from ZIP archive."""
import zipfile, io, base64, json
def handle_unzip(rfile, content_length):
"""Parse ZIP upload, return list of extracted files."""
body = rfile.read(content_length)
# Find file data in raw multipart
# Simple approach: find ZIP bytes after header
idx = body.find(b"\r\n\r\n")
if idx < 0:
return {"ok": False, "error": "invalid multipart"}
# Find first boundary end
raw = body[idx + 4:]
# Find the actual ZIP data (after Content-Disposition etc)
zip_start = raw.find(b"PK\x03\x04")
if zip_start < 0:
return {"ok": False, "error": "not a ZIP file"}
zip_data = raw[zip_start:]
files = []
with zipfile.ZipFile(io.BytesIO(zip_data)) as zf:
for name in zf.namelist():
if name.endswith("/"):
continue
data = zf.read(name)
ext = name.rsplit(".", 1)[-1].lower() if "." in name else ""
files.append({
"filename": name,
"data_b64": base64.b64encode(data).decode(),
"ext": ext,
"size": len(data),
})
return {"ok": True, "files": files}
deploy/services/upload.py
+79
View File
@@ -0,0 +1,79 @@
"""Upload service — accept file, store to DB, parse."""
import uuid, base64, json
from email.parser import BytesParser
from db import documents, contracts, supplements
from .parse import parse_file
def handle_upload(rfile, content_type, content_length):
"""Parse multipart upload, store in DB, return result dict."""
body = rfile.read(content_length)
msg = BytesParser().parsebytes(
b"Content-Type: " + content_type.encode() + b"\r\n\r\n" + body
)
filename = None
file_data = None
contract_id = ""
if msg.is_multipart():
for part in msg.get_payload():
fn = part.get_filename()
if fn:
filename = fn
file_data = part.get_payload(decode=True)
else:
disp = part.get("Content-Disposition", "")
if 'name="contract_id"' in disp:
contract_id = part.get_payload(decode=True).decode("utf-8", errors="ignore").strip()
if not filename or not file_data:
return {"ok": False, "error": "no file in request"}
mime = _mime_for(filename)
b64 = base64.b64encode(file_data).decode()
# Удалить старый файл с тем же именем в этом контракте
if contract_id:
supplements.delete_by_document(contract_id, filename)
doc = documents.insert(filename, mime, b64)
# Создать контракт если нет
if not contract_id:
from datetime import datetime
now = datetime.now()
c = contracts.insert(f"б/н {now.strftime('%Y%m%d')}-{now.strftime('%H%M')}")
contract_id = c["id"]
supp_type = "initial"
else:
supp_type = "additional"
supplements.insert(contract_id, doc["id"], supp_type)
# Парсинг на VM
parsed = parse_file(filename, file_data)
if parsed and parsed.get("status") == "parsed":
documents.set_parsed(doc["id"], parsed["elements"])
elif parsed and parsed.get("status") == "error":
documents.set_error(doc["id"], parsed.get("error", "parse failed"))
return {
"ok": True,
"doc_id": doc["id"],
"contract_id": contract_id,
"filename": filename,
"size": len(file_data),
"parsed": parsed,
}
def _mime_for(filename):
ext = filename.rsplit(".", 1)[-1].lower() if "." in filename else ""
mime_map = {
"pdf": "application/pdf",
"docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
"doc": "application/msword",
"zip": "application/zip",
}
return mime_map.get(ext, "application/octet-stream")
index.cfm
+1 -1
View File
@@ -75,7 +75,7 @@
<body>
<div class="topbar">
<img src="/nubes-logo.svg" alt="Nubes">
<span class="title">Сверка договоров — LLM AI-driven Event Sourcing <span style="font-weight:400;color:var(--muted);font-size:12px;">v1.0.154 — Lucee</span></span>
<span class="title">Сверка договоров — LLM AI-driven Event Sourcing <span style="font-weight:400;color:var(--muted);font-size:12px;">v1.0.155 — Lucee</span></span>
</div>
<div class="content">