contracts/History/migration-vm-to-flask-plan-2026-07-14.md

План миграции: ВМ → 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

# 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.

# 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

# 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

# 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/<doc_id>")
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/<doc_id>", 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 — точка входа

# 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. Нужно сделать генератором:

# 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 (статика) поменять ОДНУ строку:

// Было:
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