Redis очередь: /upload → RPUSH → 200 сразу, воркер → DB + парсинг, v1.38

2026-06-17 22:28:10 +04:00
parent 72a7ac4345
commit 65c0c80687
5 changed files with 121 additions and 33 deletions
@@ -0,0 +1,69 @@
+"""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)