убрал LLM: только парсинг с интерактивным прогрессом в таблице
This commit is contained in:
+38
-69
@@ -1,7 +1,7 @@
|
||||
"""app.py — Точка входа и сборка слоёв."""
|
||||
|
||||
from dotenv import load_dotenv
|
||||
from flask import Flask, render_template, request, redirect, url_for, Response
|
||||
from flask import Flask, render_template, request, redirect, url_for, Response, jsonify
|
||||
import json
|
||||
|
||||
import schema
|
||||
@@ -26,7 +26,7 @@ class ContractsApp:
|
||||
|
||||
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=["GET"])
|
||||
self.app.add_url_rule("/parse/<cid>", "parse", self._parse, methods=["GET"])
|
||||
self.app.add_url_rule("/health", "health", self._health)
|
||||
self.app.register_blueprint(test_bp)
|
||||
self.app.register_blueprint(upload_bp)
|
||||
@@ -80,27 +80,13 @@ class ContractsApp:
|
||||
(filename,),
|
||||
)
|
||||
doc_id = doc["id"] if doc else None
|
||||
if not doc_id:
|
||||
continue
|
||||
if doc_id:
|
||||
db.execute(
|
||||
"INSERT INTO supplements (contract_id, document_id, type) VALUES (%s,%s,'initial')",
|
||||
(str(contract_id), str(doc_id)),
|
||||
)
|
||||
|
||||
# Парсинг (быстро, без 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))
|
||||
return jsonify({"contract_id": str(contract_id)})
|
||||
|
||||
def _show_contract(self, cid):
|
||||
"""Показать договор с кнопкой «Обработать»."""
|
||||
@@ -132,38 +118,35 @@ class ContractsApp:
|
||||
contract=contract, supplements=supp_list,
|
||||
all_rows=all_rows, unprocessed=unprocessed)
|
||||
|
||||
# ── Шаг 2: LLM-обработка (SSE) ──────────────────────────────
|
||||
# ── Шаг 2: Парсинг (SSE) ────────────────────────────────────
|
||||
|
||||
def _process(self, cid):
|
||||
"""SSE-поток: обработка допников с интерактивным прогрессом."""
|
||||
def _parse(self, cid):
|
||||
"""SSE-поток: парсинг файлов договора (без LLM)."""
|
||||
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, "
|
||||
"SELECT s.id, d.id as doc_id, d.filename, d.mime_type, "
|
||||
"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",
|
||||
"WHERE s.contract_id=%s",
|
||||
(cid,),
|
||||
)
|
||||
if not supps:
|
||||
yield f"data: {json.dumps({'type': 'error', 'message': 'Нет файлов для обработки'})}\n\n"
|
||||
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"],))
|
||||
# Пропустить уже распарсенные
|
||||
existing, _ = db.query(
|
||||
"SELECT 1 FROM documents WHERE id=%s AND status='parsed'", (s["doc_id"],)
|
||||
)
|
||||
if existing and existing["rows"]:
|
||||
continue
|
||||
|
||||
@@ -173,51 +156,37 @@ class ContractsApp:
|
||||
|
||||
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"
|
||||
# Получить байты
|
||||
doc, _ = db.query_one("SELECT original_bytes FROM documents WHERE id=%s", (s["doc_id"],))
|
||||
if not doc or not doc.get("original_bytes"):
|
||||
yield f"data: {json.dumps({'type': 'file_error', 'name': s['filename'], 'error': 'Нет данных'})}\n\n"
|
||||
continue
|
||||
|
||||
rows = ext.get("rows", [])
|
||||
total_rows += len(rows)
|
||||
raw = doc["original_bytes"]
|
||||
file_data = bytes(raw) if isinstance(raw, memoryview) else raw
|
||||
|
||||
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")),
|
||||
)
|
||||
# Парсинг
|
||||
pr = parser_mod.parse(file_data, s["mime_type"])
|
||||
elements = []
|
||||
if s["mime_type"] == "application/zip" and "files" in pr:
|
||||
for zf in pr["files"]:
|
||||
elements.extend(zf.get("elements", []))
|
||||
else:
|
||||
elements = pr.get("elements", [])
|
||||
|
||||
# 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"]),
|
||||
text = textify.to_text(elements) if elements else ""
|
||||
db.execute(
|
||||
"UPDATE documents SET parsed_text=%s, status='parsed' WHERE id=%s",
|
||||
(text, str(s["doc_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"
|
||||
yield f"data: {json.dumps({'type': 'file_done', 'name': s['filename'], 'time_s': elapsed, 'elements': len(elements)})}\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"
|
||||
yield f"data: {json.dumps({'type': 'summary', 'total_time_s': total_time, 'total_bytes': total_bytes, 'files_processed': files_processed})}\n\n"
|
||||
|
||||
return Response(generate(), mimetype="text/event-stream")
|
||||
|
||||
|
||||
Reference in New Issue
Block a user