naeel
fdd7d89ce5
Шаг 1: POST / → сохранить файлы, парсинг → редирект на /?id=X Шаг 2: POST /process/X → LLM extract + diff → редирект на /?id=X Нет таймаута — загрузка мгновенная, LLM по отдельной кнопке.
206 lines
8.1 KiB
Python
206 lines
8.1 KiB
Python
"""app.py — Точка входа и сборка слоёв."""
|
||
|
||
from dotenv import load_dotenv
|
||
from flask import Flask, render_template, request, redirect
|
||
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/<cid>", "process", self._process, methods=["POST"])
|
||
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-обработка ────────────────────────────────────
|
||
|
||
def _process(self, cid):
|
||
"""Запустить LLM-обработку для всех необработанных допников договора."""
|
||
supps, _ = db.query(
|
||
"SELECT s.id, d.id as doc_id, d.parsed_text, d.filename "
|
||
"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:
|
||
return redirect("/?id=" + cid)
|
||
|
||
supp_rows = [dict(zip(supps["columns"], r)) for r in supps["rows"]]
|
||
total_rows = 0
|
||
|
||
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
|
||
|
||
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"]))
|
||
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"],))
|
||
|
||
return redirect("/?id=" + cid)
|
||
|
||
def run(self):
|
||
self.app.run(host="0.0.0.0", port=5000)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
ContractsApp().run()
|