diff --git a/requirements.txt b/requirements.txt
index 7a285f8..db54537 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -6,5 +6,6 @@ psycopg2-binary
python-dotenv
pdfplumber
camelot-py>=2.0
+redis
httpx
h2
diff --git a/site/app.py b/site/app.py
index 349a866..8b6710c 100644
--- a/site/app.py
+++ b/site/app.py
@@ -14,6 +14,8 @@ import mimeutil
from test_routes import test_bp
from upload import upload_bp
from api import api_bp
+import redis_client
+import redis_worker
load_dotenv()
@@ -74,6 +76,7 @@ class ContractsApp:
def __init__(self):
self.app = Flask(__name__)
schema.ensure_schema()
+ redis_worker.start_worker()
self._add_routes()
def _add_routes(self):
@@ -156,44 +159,27 @@ class ContractsApp:
mime = mimeutil.guess_mime(filename) or "application/octet-stream"
cid = data.get("cid")
- conn, err = db.connect()
- if err:
- return jsonify({"error": f"БД: {err}"}), 500
- try:
- cur = conn.cursor()
- if cid:
- contract_id = cid
- else:
+ # Контракт — синхронно (быстро, нужно для следующих загрузок)
+ if not cid:
+ conn, err = db.connect()
+ if err:
+ return jsonify({"error": f"БД: {err}"}), 500
+ try:
+ 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]
+ cid = cur.fetchone()[0]
+ conn.commit()
+ cur.close()
+ finally:
+ db.put_conn(conn)
- cur.execute(
- "INSERT INTO documents (filename, mime_type, original_b64, status) VALUES (%s,%s,%s,'uploaded')",
- (filename, mime, b64),
- )
+ # Файл — в Redis (мгновенно)
+ redis_client.push({"filename": filename, "b64": b64, "mime": mime, "cid": str(cid)})
- 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')",
- (str(contract_id), str(doc_id)),
- )
-
- conn.commit()
- cur.close()
- except Exception as e:
- try: conn.rollback()
- except: pass
- raise e
- finally:
- db.put_conn(conn)
-
- return jsonify({"contract_id": str(contract_id), "doc_id": str(doc_id) if doc_id else None})
+ return jsonify({"contract_id": str(cid)})
except Exception as e:
_log("upload_outer_error", str(e))
return jsonify({"error": f"Крах: {e}"}), 500
diff --git a/site/redis_client.py b/site/redis_client.py
new file mode 100644
index 0000000..a4c9e81
--- /dev/null
+++ b/site/redis_client.py
@@ -0,0 +1,32 @@
+"""redis_client.py — Очередь загрузок через Redis."""
+
+import os, json, redis
+
+UPLOAD_QUEUE = "contracts:upload_queue"
+
+_r = None
+
+def _get():
+ global _r
+ if _r is None:
+ _r = redis.Redis(
+ host=os.getenv("REDIS_HOST", "redisk8s.f6303a9d-4ab1-4b2f-8aa8-08f933d02f82.svc.cluster.local"),
+ port=int(os.getenv("REDIS_PORT", "6379")),
+ password=os.getenv("REDIS_PASS", "aLITloRefJEiCPqUc2xB"),
+ decode_responses=True,
+ socket_connect_timeout=5,
+ )
+ return _r
+
+
+def push(task: dict):
+ """Добавить задание в очередь."""
+ _get().rpush(UPLOAD_QUEUE, json.dumps(task, ensure_ascii=False))
+
+
+def pop(timeout=5):
+ """Взять задание из очереди (блокирующий)."""
+ result = _get().blpop(UPLOAD_QUEUE, timeout=timeout)
+ if result:
+ return json.loads(result[1])
+ return None
diff --git a/site/redis_worker.py b/site/redis_worker.py
new file mode 100644
index 0000000..cb7c881
--- /dev/null
+++ b/site/redis_worker.py
@@ -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)
diff --git a/site/templates/upload.html b/site/templates/upload.html
index eeea7e9..4659968 100644
--- a/site/templates/upload.html
+++ b/site/templates/upload.html
@@ -59,7 +59,7 @@
 }})
-
Сверка договоров v1.36
+
Сверка договоров v1.38