70 lines
2.3 KiB
Python
70 lines
2.3 KiB
Python
"""redis_worker.py — Фоновый обработчик очереди загрузок."""
|
|
|
|
import threading, time, base64, json
|
|
import db, mimeutil, parser as parser_mod, textify
|
|
|
|
def start_worker():
|
|
"""Запустить фоновый поток обработки очереди."""
|
|
t = threading.Thread(target=_worker_loop, daemon=True)
|
|
t.start()
|
|
|
|
def _worker_loop():
|
|
from redis_client import pop as queue_pop
|
|
while True:
|
|
try:
|
|
task = queue_pop(timeout=5)
|
|
if task:
|
|
_process(task)
|
|
except Exception as e:
|
|
time.sleep(1)
|
|
|
|
def _process(task):
|
|
filename = task.get("filename", "")
|
|
b64 = task.get("b64", "")
|
|
mime = task.get("mime", "application/octet-stream")
|
|
cid = task.get("cid", "")
|
|
|
|
if not filename or not b64:
|
|
return
|
|
|
|
# Декодировать и сохранить в БД
|
|
file_bytes = base64.b64decode(b64)
|
|
|
|
conn, err = db.connect()
|
|
if err:
|
|
return
|
|
try:
|
|
cur = conn.cursor()
|
|
cur.execute(
|
|
"INSERT INTO documents (filename, mime_type, original_bytes, original_b64, status) VALUES (%s,%s,%s,%s,'uploaded')",
|
|
(filename, mime, file_bytes, b64),
|
|
)
|
|
cur.execute("SELECT id FROM documents WHERE filename=%s AND status='uploaded' ORDER BY created_at DESC LIMIT 1", (filename,))
|
|
row = cur.fetchone()
|
|
doc_id = row[0] if row else None
|
|
if doc_id:
|
|
cur.execute(
|
|
"INSERT INTO supplements (contract_id, document_id, type) VALUES (%s,%s,'initial')",
|
|
(cid, str(doc_id)),
|
|
)
|
|
|
|
# Парсинг
|
|
pr = parser_mod.parse(file_bytes, mime)
|
|
elements = pr.get("elements", [])
|
|
if mime == "application/zip" and "files" in pr:
|
|
for zf in pr["files"]:
|
|
elements.extend(zf.get("elements", []))
|
|
text = textify.to_text(elements) if elements else ""
|
|
|
|
cur.execute(
|
|
"UPDATE documents SET parsed_text=%s, elements_json=%s, status='parsed' WHERE id=%s",
|
|
(text, json.dumps(elements, ensure_ascii=False), str(doc_id)),
|
|
)
|
|
conn.commit()
|
|
cur.close()
|
|
except Exception as e:
|
|
try: conn.rollback()
|
|
except: pass
|
|
finally:
|
|
db.put_conn(conn)
|