463 lines
20 KiB
Python
Executable File
463 lines
20 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""HTTP-сервер: конвертация .doc → .docx + LLM-анализ ДС (/llm-ops) + SSE v2 (/process-v2)"""
|
||
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 httpx
|
||
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 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_header("Access-Control-Allow-Headers", "Content-Type")
|
||
self.end_headers()
|
||
|
||
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":
|
||
self._handle_upload()
|
||
else:
|
||
self._handle_doc_convert()
|
||
|
||
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
|
||
|
||
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
|
||
|
||
self.send_response(200)
|
||
self.send_header("Content-Type", "text/event-stream; charset=utf-8")
|
||
self.send_header("Cache-Control", "no-cache")
|
||
self.send_header("Connection", "keep-alive")
|
||
self.end_headers()
|
||
self.wfile.write(b": ok\n\n")
|
||
self.wfile.flush()
|
||
|
||
t0 = time.time()
|
||
|
||
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})
|
||
|
||
except Exception as e:
|
||
self._sse({"type": "error", "message": str(e)})
|
||
|
||
def _handle_llm_ops(self):
|
||
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())
|
||
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())
|
||
|
||
def _handle_upload(self):
|
||
"""Принять файл и передать в Lucee чанками (обходит лимит тела запроса)."""
|
||
import uuid
|
||
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)
|
||
|
||
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:
|
||
# Может быть поле contract_id
|
||
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
|
||
|
||
# Режем на чанки (Lucee сбрасывает HTTP/2 стрим — новый клиент на каждый)
|
||
CHUNK = 12000
|
||
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) 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": chunk.hex(),
|
||
},
|
||
)
|
||
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) 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:
|
||
self.send_response(200)
|
||
self._send_cors()
|
||
self.send_header("Content-Type", "application/json; charset=utf-8")
|
||
self.end_headers()
|
||
self.wfile.write(resp.content)
|
||
else:
|
||
self._send_error_cors(502, f"assemble failed: HTTP {resp.status_code}")
|
||
|
||
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
|
||
|
||
content_hex = "\\x" + content.hex()
|
||
resp = httpx.post(
|
||
f"{LUCEE_URL}/upload.cfm",
|
||
data={"filename": name, "data": content_hex},
|
||
timeout=30
|
||
)
|
||
if resp.status_code == 200:
|
||
rj = resp.json()
|
||
if rj.get("OK"):
|
||
results.append({
|
||
"filename": name,
|
||
"doc_id": rj.get("DOC_ID", ""),
|
||
"contract_id": rj.get("CONTRACT_ID", ""),
|
||
"size": len(content),
|
||
})
|
||
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())
|
||
|
||
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 _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()
|
||
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()
|