v2: чанки 30KB + MD5, /v2/chunk + /v2/finalize, Redis, worker, v2.0

v2/worker.py

@@ -0,0 +1,51 @@
+"""v2/worker.py — Фоновый парсинг через site/parser.py."""
+
+import sys, os, json, logging, time
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'site'))
+import db, parser as parser_mod, textify
+sys.path.insert(0, os.path.dirname(__file__))
+import init as v2
+
+logging.basicConfig(level=logging.INFO, format="[worker] %(message)s")
+logger = logging.getLogger(__name__)
+
+def run():
+    logger.info("started")
+    while True:
+        try:
+            task = v2.pop_finalize(timeout=5)
+            if task:
+                _process(task)
+        except Exception as e:
+            logger.error("loop: %s", e)
+            time.sleep(1)
+
+def _process(task):
+    doc_id = task.get("doc_id")
+    filename = task.get("filename", "")
+    mime = task.get("mime_type", "")
+
+    doc, err = db.query_one("SELECT original_bytes, mime_type FROM documents WHERE id=%s", (doc_id,))
+    if not doc:
+        logger.info("%s: not found", doc_id)
+        return
+
+    raw = doc["original_bytes"]
+    file_bytes = bytes(raw) if isinstance(raw, memoryview) else raw
+    mime = doc["mime_type"] or mime
+
+    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 ""
+    db.execute(
+        "UPDATE documents SET parsed_text=%s, elements_json=%s, status='parsed' WHERE id=%s",
+        (text, json.dumps(elements, ensure_ascii=False), doc_id),
+    )
+    logger.info("%s: parsed, %d elements", filename, len(elements))
+
+if __name__ == "__main__":
+    run()