#!/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 чанками (лимит ingress ~30KB).""" 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: 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": 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, 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: 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()