# План миграции: ВМ → Flask (полный перенос, без ВМ) **Дата:** 2026-07-14 **Цель:** Убрать `deploy/convert_server.py` и весь ВМ-слой. Вся логика — во Flask. **Принцип:** НИКАКОГО монолита. Каждый слой — отдельный модуль/blueprint. --- ## 0. Аудит: насколько код уже decoupled ### ✅ УЖЕ ГОТОВО (можно брать как есть) | Модуль | Строк | Статус | Почему | |--------|-------|--------|--------| | `db/connection.py` | ~60 | ✅ Идеально | Connection pool, ноль зависимостей | | `db/documents.py` | ~100 | ✅ Идеально | Чистый CRUD, только `query()/execute()` | | `db/contracts.py` | ~25 | ✅ Идеально | Чистый CRUD | | `db/supplements.py` | ~60 | ✅ Идеально | Чистый CRUD | | `db/spec_current.py` | ~20 | ✅ Идеально | Чистые запросы | | `db/spec_events.py` | ~30 | ✅ Идеально | Чистый CRUD | | `db/prompts.py` | ~80 | ✅ Идеально | Чистый CRUD + seed | | `compare/parse.py` | ~100 | ✅ Идеально | Чистая функция: `(filename, bytes) → dict` | | `compare/llm_client.py` | ~80 | ✅ Идеально | Protocol + Httpx + Fake, DI-ready | | `compare/grouping.py` | ~160 | ✅ Идеально | Чистая логика: `batch_id → groups`, нет HTTP | | `compare/llm.py` | ~40 | ✅ Хорошо | Уже принимает `llm_client` опционально (DI) | | `llm_prompt.py` | ~80 | ✅ Идеально | Чистые функции сборки промптов | | `repository.py` | ~120 | ✅ Хорошо | Protocol есть, PgRepository частично реализован | ### 🔧 НУЖНА АДАПТАЦИЯ (логика готова, интерфейс — нет) | Модуль | Проблема | Что сделать | |--------|----------|-------------| | `compare/upload.py` | Использует `cgi.FieldStorage` | Заменить на `request.files` из Flask | | `compare/unzip.py` | Читает `rfile.read()` сырые байты | Заменить на `request.files` / `request.data` | | `compare/classify.py` | `ThreadPoolExecutor` ок, но запускается из HTTP-метода | Обернуть в Flask background task (см. ниже) | | `compare/process.py` | SSE через `self.wfile.write()` | Заменить на `Response(stream_with_context(...))` | ### ❌ НУЖНО ПЕРЕПИСАТЬ | Модуль | Проблема | Что сделать | |--------|----------|-------------| | `convert_server.py` | Монолит ~500 строк, все роуты в одном классе | Разобрать на Flask blueprints | | `classify_worker.py` | subprocess.Popen — не нужно во Flask | Убрать, classify в отдельном потоке/процессе через `@app.route` | --- ## 1. Целевая архитектура Flask ``` contracts-flask/ ├── site/ │ ├── app.py # create_app(), регистрация blueprints │ ├── config.py # Настройки (DB, LLM, лимиты) │ ├── routes/ │ │ ├── __init__.py # register_routes(app) │ │ ├── upload_bp.py # POST /upload, /convert-doc, /unzip-upload │ │ ├── pipeline_bp.py # GET /process-v2 (SSE), POST /classify-batch │ │ ├── api_bp.py # GET/POST /api/* (groups, documents, supplements, sync, cleanup) │ │ ├── prompts_bp.py # GET/POST /api/prompts/* │ │ ├── health_bp.py # GET /health │ │ └── pages_bp.py # GET /, /architect (HTML-страницы) │ ├── services/ # Бизнес-логика (перенос из deploy/compare/) │ │ ├── __init__.py │ │ ├── parse.py # ← deploy/compare/parse.py │ │ ├── classify.py # ← deploy/compare/classify.py │ │ ├── grouping.py # ← deploy/compare/grouping.py │ │ ├── llm.py # ← deploy/compare/llm.py │ │ ├── llm_client.py # ← deploy/compare/llm_client.py │ │ └── metrics.py # ← deploy/compare/metrics.py │ ├── db/ # Перенос из deploy/db/ │ │ ├── __init__.py │ │ ├── connection.py # ← deploy/db/connection.py │ │ ├── documents.py # ← deploy/db/documents.py │ │ ├── contracts.py # ← deploy/db/contracts.py │ │ ├── supplements.py # ← deploy/db/supplements.py │ │ ├── spec_current.py # ← deploy/db/spec_current.py │ │ ├── spec_events.py # ← deploy/db/spec_events.py │ │ └── prompts.py # ← deploy/db/prompts.py │ ├── repository.py # ← deploy/repository.py (расширить) │ ├── llm_prompt.py # ← deploy/llm_prompt.py │ ├── templates/ │ │ └── index.html # УЖЕ ЕСТЬ, почти не меняется │ └── static/ # JS-файлы: state.js, app_utils.js, files.js, groups.js, compare.js, app.js ├── requirements.txt # Flask + psycopg2 + httpx + pdfplumber + python-docx ├── Dockerfile └── deploy/ # ОСТАЁТСЯ только: ├── nginx-contracts.conf ├── sync.sh └── convert_doc.py # libreoffice-конвертер (stdin→stdout, НЕ HTTP) ``` ### Ключевое правило: ZERO монолита - **Blueprints** — каждый на ≤100 строк, одна ответственность - **services/** — чистые функции, никакого `request`/`Response` - **db/** — чистый CRUD, никакого Flask - **repository.py** — единая точка доступа к данным (DI во все services) --- ## 2. Пошаговый план (8 шагов) ### Шаг 1: Перенести `db/` как есть **Файлы:** `deploy/db/*.py` → `site/db/*.py` **Что делать:** Копировать. Менять НИЧЕГО не надо. - `connection.py` уже использует `psycopg2.pool.ThreadedConnectionPool` — идеально для Flask (каждый request — своё соединение из пула) - Все модули зависят только от `connection.query()` и `connection.execute()` **Проверка:** `python3 -c "from site.db import documents; print(documents.list_by_batch('test'))"` --- ### Шаг 2: Перенести `services/` (бывший `compare/`) **Файлы:** `deploy/compare/{parse,classify,grouping,llm,llm_client,metrics}.py` → `site/services/` **Что делать:** Копировать, исправить импорты: - `from db import ...` → `from site.db import ...` - `from .parse import ...` → `from site.services.parse import ...` - `from llm_prompt import ...` → `from site.llm_prompt import ...` **НЕ переносить:** `upload.py`, `unzip.py`, `process.py` — они привязаны к HTTP и будут переписаны в blueprints. **Проверка:** `python3 -c "from site.services.classify import classify_batch; print('ok')"` --- ### Шаг 3: Создать `config.py` ```python # site/config.py import os DB_CONFIG = { "host": os.getenv("DB_HOST", "127.0.0.1"), "port": int(os.getenv("DB_PORT", "5432")), "dbname": os.getenv("DB_NAME", "baza"), "user": os.getenv("DB_USER", "super"), "password": os.getenv("DB_PASS", ""), } LLM_URL = os.getenv("LLM_API_URL", "https://api.aillm.ru/v1/chat/completions") LLM_KEY = os.getenv("LLM_API_KEY", "") LLM_MODEL = os.getenv("LLM_MODEL", "gpt-oss-120b") MAX_CONTENT_LENGTH = 200 * 1024 * 1024 # 200 MB VERSION = "2.0.0" ``` --- ### Шаг 4: Blueprint `upload_bp.py` — загрузка файлов Самый критичный blueprint. Замена `deploy/compare/upload.py` + `deploy/compare/unzip.py`. ```python # site/routes/upload_bp.py from flask import Blueprint, request, jsonify from site.services.parse import parse_file from site.db import documents, contracts from site.config import MAX_CONTENT_LENGTH import hashlib, base64, zipfile, io, os upload_bp = Blueprint("upload", __name__) ALLOWED = {"pdf", "docx", "doc", "zip"} @upload_bp.route("/upload", methods=["POST"]) def upload(): """Загрузка одного файла → парсинг → БД.""" f = request.files.get("files") if not f: return jsonify(ok=False, error="no file"), 400 ext = f.filename.rsplit(".", 1)[-1].lower() if "." in f.filename else "" if ext not in ALLOWED: return jsonify(ok=False, error=f"unsupported: .{ext}"), 400 data = f.read() content_hash = hashlib.sha256(data).hexdigest()[:16] batch_id = request.form.get("batch_id") zip_source = request.form.get("zip_source") # Проверка дубликата по хешу if batch_id: existing = documents.get_by_hash(batch_id, content_hash) if existing: return jsonify(ok=False, error="duplicate", doc_id=existing["id"]) doc = documents.insert( filename=f.filename, mime_type=f.content_type or "application/octet-stream", original_bytes=base64.b64encode(data).decode(), batch_id=batch_id, zip_source=zip_source, content_hash=content_hash, ) # Парсинг try: result = parse_file(f.filename, data) if result["status"] == "parsed": documents.set_parsed(doc["id"], result["elements"]) else: documents.set_error(doc["id"], result.get("error", "parse failed")) except Exception as e: documents.set_error(doc["id"], str(e)) result = {"status": "error", "error": str(e)} contract_id = request.form.get("contract_id") return jsonify(ok=True, doc_id=doc["id"], contract_id=contract_id, parsed={"status": result["status"], "element_count": result.get("element_count", 0)}) @upload_bp.route("/convert-doc", methods=["POST"]) def convert_doc(): """.doc → .docx через libreoffice.""" import subprocess, tempfile f = request.files.get("files") if not f: return jsonify(ok=False, error="no file"), 400 data = f.read() with tempfile.NamedTemporaryFile(suffix=".doc", delete=False) as tmp: tmp.write(data) doc_path = tmp.name tmpdir = tempfile.mkdtemp() try: subprocess.run(["libreoffice", "--headless", "--convert-to", "docx", "--outdir", tmpdir, doc_path], timeout=30, capture_output=True) docx_files = [x for x in os.listdir(tmpdir) if x.endswith(".docx")] if docx_files: with open(os.path.join(tmpdir, docx_files[0]), "rb") as out: from flask import send_file return send_file(io.BytesIO(out.read()), mimetype="application/vnd.openxmlformats-officedocument.wordprocessingml.document") return jsonify(ok=False, error="conversion produced no output"), 500 finally: os.unlink(doc_path) for x in os.listdir(tmpdir): os.unlink(os.path.join(tmpdir, x)) os.rmdir(tmpdir) @upload_bp.route("/unzip-upload", methods=["POST"]) def unzip_upload(): """Распаковать ZIP → список файлов (base64).""" f = request.files.get("files") if not f: return jsonify(ok=False, error="no file"), 400 data = f.read() MAX_FILES = 500 MAX_UNCOMPRESSED = 500 * 1024 * 1024 files = [] total = 0 with zipfile.ZipFile(io.BytesIO(data)) as zf: if len(zf.namelist()) > MAX_FILES: return jsonify(ok=False, error=f"too many files (max {MAX_FILES})"), 400 for info in zf.infolist(): if info.is_dir(): continue name = os.path.basename(info.filename) if not name or ".." in name: continue raw = zf.read(info) total += len(raw) if total > MAX_UNCOMPRESSED: return jsonify(ok=False, error="total uncompressed > 500MB"), 400 ext = name.rsplit(".", 1)[-1].lower() if "." in name else "" files.append({ "filename": name, "ext": ext, "size": len(raw), "data_b64": base64.b64encode(raw).decode(), }) return jsonify(ok=True, files=files) ``` **Критично:** - `request.files` вместо `cgi.FieldStorage` — файл уже в памяти - `base64` всё ещё нужен (JS на фронте делает `atob()`) - Таймауты: Flask не режет сам — поставить `timeout` на libreoffice --- ### Шаг 5: Blueprint `pipeline_bp.py` — SSE + classify ```python # site/routes/pipeline_bp.py from flask import Blueprint, request, jsonify, Response, stream_with_context from site.services.process import run_pipeline # адаптированный process.py from site.services.classify import classify_batch from site.llm_prompt import build_prompt from site.db import documents import json, re, os, threading, sys pipeline_bp = Blueprint("pipeline", __name__) @pipeline_bp.route("/process-v2", methods=["GET"]) def process_v2(): """SSE-стриминг сравнения.""" cid = request.args.get("contract_id") if not cid or not re.fullmatch(r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}', cid, re.I): return jsonify(ok=False, error="invalid contract_id"), 400 order_ids = request.args.get("order", "") def generate(): yield ": ok\n\n" try: for event in run_pipeline(cid, order_ids, build_prompt): yield f"data: {json.dumps(event, ensure_ascii=False)}\n\n" except GeneratorExit: return except Exception as e: yield f"data: {json.dumps({'type': 'error', 'message': str(e)}, ensure_ascii=False)}\n\n" return Response( stream_with_context(generate()), content_type="text/event-stream; charset=utf-8", headers={ "Cache-Control": "no-cache", "X-Accel-Buffering": "no", # ← nginx не буферизует } ) @pipeline_bp.route("/api/classify-batch", methods=["POST"]) def classify_batch_route(): """Запустить классификацию. Синхронно для малых батчей, 202 для больших.""" body = request.get_json() batch_id = body.get("batch_id") if not batch_id: return jsonify(ok=False, error="batch_id required"), 400 pending = documents.list_pending(batch_id) total = len(pending) if total == 0: return jsonify(ok=False, error="no pending documents"), 400 # Для ≤10 файлов — синхронно (быстрее, проще) if total <= 10: result = classify_batch(batch_id) return jsonify(result) # Для >10 файлов — в отдельном потоке, сразу 202 lock_path = f"/tmp/classify_{batch_id}.lock" if os.path.exists(lock_path): return jsonify(ok=False, error="classify already running"), 409 with open(lock_path, "w") as lf: lf.write(str(os.getpid())) def _run(): try: classify_batch(batch_id) finally: if os.path.exists(lock_path): os.remove(lock_path) threading.Thread(target=_run, daemon=True).start() return jsonify(ok=True, total=total), 202 ``` **Критично:** - `compare/process.py` нужно адаптировать: `run_pipeline` должен стать **генератором** (yield события), а не принимать `sse_send` callback - `GeneratorExit` в генераторе — обязательно - `X-Accel-Buffering: no` — иначе nginx буферизует SSE - classify: синхронно для ≤10 файлов, Thread для >10 (вместо subprocess) --- ### Шаг 6: Blueprint `api_bp.py` — всё остальное API ```python # site/routes/api_bp.py from flask import Blueprint, request, jsonify from site.db import documents, supplements, contracts, spec_current from site.services.grouping import group_documents, apply_groups from site.db.connection import execute, query api_bp = Blueprint("api", __name__) @api_bp.route("/api/supplements") def api_supplements(): cid = request.args.get("contract_id") if not cid: return jsonify(ok=False, error="contract_id required"), 400 rows = supplements.list_by_contract(cid) return jsonify(ok=True, supplements=rows) @api_bp.route("/api/documents/") def api_document(doc_id): doc = documents.get(doc_id) if not doc: return jsonify(ok=False, error="not found"), 404 return jsonify(ok=True, **{k: doc.get(k) for k in [ "id", "filename", "status", "elements_json", "doc_type", "own_number", "parent_number", "doc_date", "counterparty", "classify_status", "classify_raw", "classify_input" ]}) @api_bp.route("/api/documents/", methods=["DELETE"]) def api_document_delete(doc_id): supps = query("SELECT id, contract_id FROM supplements WHERE document_id = %s", (doc_id,)) for s in (supps or []): execute("DELETE FROM spec_current WHERE contract_id = %s AND last_event_id IN (SELECT id FROM spec_events WHERE supplement_id = %s)", (s["contract_id"], s["id"])) execute("DELETE FROM spec_events WHERE supplement_id = %s", (s["id"],)) execute("DELETE FROM supplements WHERE id = %s", (s["id"],)) execute("DELETE FROM documents WHERE id = %s", (doc_id,)) return jsonify(ok=True) @api_bp.route("/api/sync", methods=["POST"]) def api_sync(): body = request.get_json() keep_ids = set(body.get("keep_ids", [])) if len(keep_ids) > 1000: return jsonify(ok=False, error="too many keep_ids"), 400 docs = query("SELECT id FROM documents", ()) deleted = 0 for d in (docs or []): if d["id"] in keep_ids: continue supps = query("SELECT id, contract_id FROM supplements WHERE document_id = %s", (d["id"],)) for s in (supps or []): execute("DELETE FROM spec_current WHERE contract_id = %s AND last_event_id IN (SELECT id FROM spec_events WHERE supplement_id = %s)", (s["contract_id"], s["id"])) execute("DELETE FROM spec_events WHERE supplement_id = %s", (s["id"],)) execute("DELETE FROM supplements WHERE id = %s", (s["id"],)) execute("DELETE FROM documents WHERE id = %s", (d["id"],)) deleted += 1 execute("DELETE FROM contracts WHERE id NOT IN (SELECT DISTINCT contract_id FROM supplements)") return jsonify(ok=True, deleted=deleted) @api_bp.route("/api/groups") def api_groups(): batch_id = request.args.get("batch") if not batch_id: return jsonify(ok=False, error="batch required"), 400 result = group_documents(batch_id) return jsonify(result) @api_bp.route("/api/batch-progress") def api_batch_progress(): batch_id = request.args.get("batch") if not batch_id: return jsonify(ok=False, error="batch required"), 400 counts = documents.count_by_status(batch_id) docs = documents.list_by_batch(batch_id) return jsonify(ok=True, counts=counts, total=len(docs)) @api_bp.route("/api/apply-groups", methods=["POST"]) def api_apply_groups(): body = request.get_json() batch_id = body.get("batch_id") groups = body.get("groups", []) if not batch_id: return jsonify(ok=False, error="batch_id required"), 400 result = apply_groups(batch_id, groups) return jsonify(result) @api_bp.route("/api/spec-current") def api_spec_current(): cid = request.args.get("contract_id") if not cid: return jsonify(ok=False, error="contract_id required"), 400 rows = spec_current.list_by_contract(cid) return jsonify(ok=True, rows=rows) @api_bp.route("/api/cleanup", methods=["POST"]) def api_cleanup(): execute("DELETE FROM spec_current") execute("DELETE FROM spec_events") execute("DELETE FROM supplements") execute("DELETE FROM upload_chunks") execute("DELETE FROM documents") execute("DELETE FROM contracts") return jsonify(ok=True, message="all data cleaned") ``` --- ### Шаг 7: `app.py` — точка входа ```python # site/app.py from flask import Flask, render_template from site.config import VERSION, MAX_CONTENT_LENGTH def create_app(): app = Flask(__name__) app.config["VERSION"] = VERSION app.config["MAX_CONTENT_LENGTH"] = MAX_CONTENT_LENGTH # Blueprints from site.routes.upload_bp import upload_bp from site.routes.pipeline_bp import pipeline_bp from site.routes.api_bp import api_bp from site.routes.prompts_bp import prompts_bp from site.routes.health_bp import health_bp from site.routes.pages_bp import pages_bp app.register_blueprint(upload_bp) app.register_blueprint(pipeline_bp) app.register_blueprint(api_bp) app.register_blueprint(prompts_bp) app.register_blueprint(health_bp) app.register_blueprint(pages_bp) @app.after_request def no_cache(response): response.headers["Cache-Control"] = "no-cache, no-store, must-revalidate" response.headers["Pragma"] = "no-cache" response.headers["Expires"] = "0" return response return app app = create_app() ``` --- ### Шаг 8: Адаптировать `compare/process.py` в генератор Текущий `run_pipeline(cid, order_ids, sse_send, build_prompt_fn)` принимает callback `sse_send`. Нужно сделать генератором: ```python # site/services/process.py def run_pipeline(contract_id, order_ids, build_prompt_fn): """Generator: yield SSE events.""" # ... та же логика, но вместо sse_send(event) → yield event # ... GeneratorExit уже обрабатывается во view ``` --- ## 3. Что НЕ трогаем (остаётся как есть) | Что | Почему | |-----|--------| | `templates/index.html` | Уже работает, меняются только URL (с ВМ на свои) | | `static/*.js` (6 файлов) | Меняется только `VM_API` → `""` (свои же endpoints) | | `convert_doc.py` | Остаётся как утилита (libreoffice), вызывается из upload_bp | | CSS, иконки, модалки | Без изменений | --- ## 4. Изменения во фронтенде (минимальные) В `deploy/app.js` (статика) поменять ОДНУ строку: ```javascript // Было: var VM_API = 'https://contracts.kube5s.ru'; // Стало: var VM_API = ''; // все API на том же домене ``` ВСЁ. Больше ничего не меняется — XHR, SSE, fetch работают с теми же путями. --- ## 5. Последовательность выполнения | # | Шаг | Сложность | Риск | |---|-----|-----------|------| | 1 | `db/` → `site/db/` | Низкая | Низкий — просто копирование | | 2 | `compare/` → `site/services/` | Низкая | Низкий — правим импорты | | 3 | `config.py` | Низкая | Низкий | | 4 | `upload_bp.py` | Средняя | **Высокий** — ключевой функционал | | 5 | `pipeline_bp.py` + адаптация `process.py` | Высокая | **Высокий** — SSE критичен | | 6 | `api_bp.py` | Средняя | Средний — много ручек | | 7 | `prompts_bp.py`, `health_bp.py`, `pages_bp.py` | Низкая | Низкий | | 8 | `app.py` + тесты | Средняя | Средний | --- ## 6. Чек-лист перед деплоем - [ ] `python3 -c "from site.app import app"` — приложение создаётся без ошибок - [ ] `/health` → `{"ok": true}` - [ ] `POST /upload` с реальным PDF → `{"ok": true, "doc_id": "..."}` - [ ] `POST /unzip-upload` с ZIP → список файлов - [ ] `GET /process-v2?contract_id=...` → SSE-поток (curl test) - [ ] `POST /api/classify-batch` → классификация работает - [ ] `GET /api/groups?batch=...` → группы - [ ] `POST /api/apply-groups` → создаются supplements - [ ] `GET /api/spec-current?contract_id=...` → спецификация - [ ] `GET /` → HTML с таблицей файлов - [ ] `no_cache` after_request — есть - [ ] `X-Accel-Buffering: no` на SSE - [ ] `GeneratorExit` в SSE-генераторе - [ ] `stream_with_context` на SSE - [ ] `MAX_CONTENT_LENGTH = 200 MB` - [ ] `VM_API = ''` в app.js (фронтенд) - [ ] Все старые тесты проходят --- ## 7. Риски и mitigation | Риск | Mitigation | |------|-----------| | SSE зависает под нагрузкой | `stream_with_context` + `X-Accel-Buffering: no` | | classify блокирует HTTP | Thread для >10 файлов, sync для ≤10 | | libreoffice падает | timeout=30, отдельный процесс | | Коннекты к БД исчерпываются | ThreadedConnectionPool уже есть, minconn=1, maxconn=10 | | ZIP-бомба | Проверка MAX_UNCOMPRESSED = 500MB, MAX_RATIO | | Загрузка больших PDF (>100MB) | MAX_CONTENT_LENGTH = 200MB, XHR timeout = 180s |