"""app.py — Точка входа и сборка слоёв.""" from dotenv import load_dotenv from flask import Flask, render_template, request, redirect, url_for, Response import json import schema import db import parser as parser_mod import textify import extractor import differ import mimeutil from test_routes import test_bp from upload import upload_bp from api import api_bp load_dotenv() class ContractsApp: def __init__(self): self.app = Flask(__name__) schema.ensure_schema() self._add_routes() def _add_routes(self): self.app.add_url_rule("/", "index", self._index, methods=["GET", "POST"]) self.app.add_url_rule("/process/", "process", self._process, methods=["GET"]) self.app.add_url_rule("/health", "health", self._health) self.app.register_blueprint(test_bp) self.app.register_blueprint(upload_bp) self.app.register_blueprint(api_bp) def _health(self): return "OK", 200, {"Content-Type": "text/plain"} # ── Шаг 1: загрузка файлов ────────────────────────────────── def _index(self): if request.method == "POST": return self._upload_files() cid = request.args.get("id") if cid: return self._show_contract(cid) return render_template("upload.html") def _upload_files(self): """Только сохранить файлы, без LLM.""" files = request.files.getlist("files") if not files or not any(f.filename for f in files): return render_template("upload.html", error="Нет файлов") conn, err = db.connect() if err: return render_template("upload.html", error=f"БД: {err}") cur = conn.cursor() cur.execute( "INSERT INTO contracts (number, client) VALUES (%s, %s) RETURNING id", ("б/н " + __import__("datetime").datetime.now().strftime("%Y%m%d-%H%M"), ""), ) contract_id = cur.fetchone()[0] conn.commit() cur.close() conn.close() for f in files: if not f.filename: continue filename = f.filename mime = mimeutil.guess_mime(filename) or f.content_type or "application/octet-stream" file_bytes = f.read() db.execute( "INSERT INTO documents (filename, mime_type, original_bytes, status) VALUES (%s,%s,%s,'uploaded')", (filename, mime, file_bytes), ) doc, _ = db.query_one( "SELECT id FROM documents WHERE filename=%s AND status='uploaded' ORDER BY created_at DESC LIMIT 1", (filename,), ) doc_id = doc["id"] if doc else None if not doc_id: continue # Парсинг (быстро, без LLM) pr = parser_mod.parse(file_bytes, mime) elements = [] if mime == "application/zip" and "files" in pr: for zf in pr["files"]: elements.extend(zf.get("elements", [])) else: elements = pr.get("elements", []) text = textify.to_text(elements) if elements else "" db.execute("UPDATE documents SET parsed_text=%s, status='parsed' WHERE id=%s", (text, str(doc_id))) db.execute( "INSERT INTO supplements (contract_id, document_id, type) VALUES (%s,%s,'initial')", (str(contract_id), str(doc_id)), ) return redirect("/?id=" + str(contract_id)) def _show_contract(self, cid): """Показать договор с кнопкой «Обработать».""" contract, _ = db.query_one("SELECT id,number,created_at FROM contracts WHERE id=%s", (cid,)) if not contract: return render_template("upload.html", error="Договор не найден") supps, _ = db.query( "SELECT s.id, s.created_at, d.filename, d.status, d.parsed_text IS NOT NULL as has_text " "FROM supplements s JOIN documents d ON s.document_id=d.id " "WHERE s.contract_id=%s ORDER BY s.created_at", (cid,), ) supp_list = [dict(zip(supps["columns"], r)) for r in supps["rows"]] if supps else [] # Есть ли уже результаты? rows_res, _ = db.query( "SELECT sr.row_num,sr.name,sr.price,sr.qty,sr.sum,sr.date_start " "FROM spec_rows sr JOIN supplements s ON sr.supplement_id=s.id " "WHERE s.contract_id=%s ORDER BY sr.row_num", (cid,), ) all_rows = [dict(zip(rows_res["columns"], r)) for r in rows_res["rows"]] if rows_res else [] # Есть необработанные допники? unprocessed = [s for s in supp_list if s["status"] in ("uploaded", "parsed")] return render_template("upload.html", contract=contract, supplements=supp_list, all_rows=all_rows, unprocessed=unprocessed) # ── Шаг 2: LLM-обработка (SSE) ────────────────────────────── def _process(self, cid): """SSE-поток: обработка допников с интерактивным прогрессом.""" import time as time_mod def generate(): start_time = time_mod.time() total_bytes = 0 files_processed = 0 total_rows = 0 supps, _ = db.query( "SELECT s.id, d.id as doc_id, d.parsed_text, d.filename, " "LENGTH(d.original_bytes) as file_size " "FROM supplements s JOIN documents d ON s.document_id=d.id " "WHERE s.contract_id=%s AND d.parsed_text IS NOT NULL", (cid,), ) if not supps: yield f"data: {json.dumps({'type': 'error', 'message': 'Нет файлов для обработки'})}\n\n" return supp_rows = [dict(zip(supps["columns"], r)) for r in supps["rows"]] for s in supp_rows: text = s["parsed_text"] if not text: continue # Проверим, не обработан ли уже existing, _ = db.query("SELECT 1 FROM spec_rows WHERE supplement_id=%s LIMIT 1", (s["id"],)) if existing and existing["rows"]: continue file_start = time_mod.time() file_bytes = s.get("file_size", 0) or 0 total_bytes += file_bytes yield f"data: {json.dumps({'type': 'file_start', 'name': s['filename'], 'bytes': file_bytes})}\n\n" ext = extractor.extract(text) if "error" in ext: db.execute("UPDATE documents SET status='extract_error', error_message=%s WHERE id=%s", (ext["error"], s["doc_id"])) yield f"data: {json.dumps({'type': 'file_error', 'name': s['filename'], 'error': ext['error']})}\n\n" continue rows = ext.get("rows", []) total_rows += len(rows) for row in rows: db.execute( "INSERT INTO spec_rows (supplement_id,row_num,name,price,qty,sum,date_start) VALUES (%s,%s,%s,%s,%s,%s,%s)", (s["id"], row.get("row_num"), row.get("name"), row.get("price"), row.get("qty"), row.get("sum"), row.get("date_start")), ) # Diff prev_res, _ = db.query( "SELECT sr.row_num,sr.name,sr.price,sr.qty,sr.sum,sr.date_start " "FROM spec_rows sr JOIN supplements sup ON sr.supplement_id=sup.id " "WHERE sup.contract_id=%s AND sup.id!=%s ORDER BY sr.row_num", (cid, s["id"]), ) prev_rows = [dict(zip(prev_res["columns"], r)) for r in prev_res["rows"]] if prev_res else [] diff_r = differ.diff(prev_rows, rows) for ch in diff_r.get("changes", []): db.execute( "INSERT INTO spec_history (contract_id,supplement_id,row_num,change_type,old_values,new_values) " "VALUES (%s,%s,%s,%s,%s,%s)", (cid, s["id"], ch["row_num"], ch["change_type"], json.dumps(ch.get("old_values"), ensure_ascii=False, default=str) if ch.get("old_values") else None, json.dumps(ch.get("new_values"), ensure_ascii=False, default=str) if ch.get("new_values") else None), ) db.execute("UPDATE documents SET status='extracted' WHERE id=%s", (s["doc_id"],)) elapsed = round(time_mod.time() - file_start, 1) files_processed += 1 yield f"data: {json.dumps({'type': 'file_done', 'name': s['filename'], 'rows': len(rows), 'time_s': elapsed})}\n\n" total_time = round(time_mod.time() - start_time, 1) yield f"data: {json.dumps({'type': 'summary', 'total_time_s': total_time, 'total_bytes': total_bytes, 'total_rows': total_rows, 'files_processed': files_processed})}\n\n" return Response(generate(), mimetype="text/event-stream") def run(self): self.app.run(host="0.0.0.0", port=5000) if __name__ == "__main__": ContractsApp().run()