From a7b53fde1915bc4762fc6ae5e8207f92d1a44ae8 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E2=80=9CNaeel=E2=80=9D?= <tazet@narod.ru>
Date: Wed, 15 Jul 2026 10:56:42 +0400
Subject: [PATCH] =?UTF-8?q?feat:=20PostgreSQL=20=E2=86=92=20SQLite=20?=
 =?UTF-8?q?=E2=80=94=20thread-local,=20WAL,=20session-key,=20os.remove=20c?=
 =?UTF-8?q?leanup?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 requirements.txt      |   1 -
 site/app.py           |  30 +----
 site/config.py        |   8 --
 site/db/connection.py | 281 ++++++++++++++++++++++++++++++------------
 site/routes/api_bp.py |  26 +---
 5 files changed, 214 insertions(+), 132 deletions(-)

diff --git a/requirements.txt b/requirements.txt
index 7513a80..2fa29fb 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -3,4 +3,3 @@ gunicorn
 httpx
 python-docx
 pdfplumber
-psycopg2-binary
diff --git a/site/app.py b/site/app.py
index db96b02..38d98a7 100644
--- a/site/app.py
+++ b/site/app.py
@@ -29,36 +29,14 @@ def create_app():
 
 
 def _init_db():
-    """Schema migration + recovery: сброс застрявших classify."""
+    """Создать БД + схему + seed prompts. SQLite — всё в одном файле /tmp."""
+    from site.db.connection import init_db
+    init_db()
     try:
-        from site.db.connection import execute
-
-        for sql in [
-            "ALTER TABLE documents ADD COLUMN IF NOT EXISTS classify_raw text",
-            "ALTER TABLE documents ADD COLUMN IF NOT EXISTS classify_input text",
-            "ALTER TABLE documents ADD COLUMN IF NOT EXISTS zip_source text",
-            "ALTER TABLE documents ADD COLUMN IF NOT EXISTS content_hash text",
-            "ALTER TABLE documents ADD COLUMN IF NOT EXISTS batch_id uuid",
-            "ALTER TABLE documents ADD COLUMN IF NOT EXISTS classify_status text DEFAULT 'pending'",
-            "ALTER TABLE documents ALTER COLUMN classify_status SET DEFAULT 'pending'",
-            "ALTER TABLE documents ADD COLUMN IF NOT EXISTS doc_type text",
-            "ALTER TABLE documents ADD COLUMN IF NOT EXISTS own_number text",
-            "ALTER TABLE documents ADD COLUMN IF NOT EXISTS parent_number text",
-            "ALTER TABLE documents ADD COLUMN IF NOT EXISTS doc_date date",
-            "ALTER TABLE documents ADD COLUMN IF NOT EXISTS counterparty text",
-        ]:
-            execute(sql)
-
         from site.db import prompts as db_prompts
         db_prompts.seed_defaults()
-
-        # Crash recovery: сбросить застрявшие classify
-        execute(
-            "UPDATE documents SET classify_status='pending', error_message=NULL "
-            "WHERE classify_status='processing'"
-        )
     except Exception:
-        pass  # БД недоступна при старте — не фатально
+        pass
 
 
 app = create_app()
diff --git a/site/config.py b/site/config.py
index 3217fb6..1dec7f4 100644
--- a/site/config.py
+++ b/site/config.py
@@ -3,14 +3,6 @@ import os
 
 VERSION = "2.0.0"
 
-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")
diff --git a/site/db/connection.py b/site/db/connection.py
index 6afefb4..33d6faa 100644
--- a/site/db/connection.py
+++ b/site/db/connection.py
@@ -1,91 +1,220 @@
-"""
-Database connection — psycopg2 connection pool.
+"""Database connection — SQLite with thread-local connections.
 
-Архитектурное решение (Opus):
-- ThreadedConnectionPool(minconn=1, maxconn=10) — оптимально для ThreadingMixIn HTTP-сервера.
-  Каждый HTTP-запрос в отдельном потоке получает своё соединение из пула.
-- RealDictCursor для query() — возвращает dict с lowercase ключами (консистентно с Python API).
-- execute()/execute_returning() — для INSERT/UPDATE/DELETE с автокоммитом и откатом при ошибке.
-- Пул создаётся лениво (при первом запросе) через get_pool().
+Архитектура (Sonnet review 2026-07-15):
+- threading.local() — каждому потоку своё соединение
+- WAL-режим — readers не блокируют writer
+- _db_session_key — защита от inode split-brain при cleanup+reuse потоков
+- busy_timeout=5000 — ждать при конкурентной записи
 """
+import sqlite3
+import threading
+import time
 import os
-import psycopg2
-import psycopg2.pool
-import psycopg2.extras
 
-# Глобальный пул соединений (singleton)
-_pool = None
-
-# Параметры подключения из переменных окружения (systemd Environment)
-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", ""),
-}
+DB_PATH = "/tmp/contracts.db"
+_db_session_key = None
+_local = threading.local()
 
 
-def get_pool():
-    """Возвращает глобальный пул соединений. Создаёт при первом вызове."""
-    global _pool
-    if _pool is None:
-        _pool = psycopg2.pool.ThreadedConnectionPool(
-            minconn=1, maxconn=10, **DB_CONFIG
-        )
-    return _pool
+def init_db():
+    """Создать/пересоздать БД + схему. Вызывается при старте и после cleanup."""
+    global _db_session_key
+    _db_session_key = time.time()
+
+    # Удалить старый файл + WAL-сателлиты
+    for f in (DB_PATH, DB_PATH + "-wal", DB_PATH + "-shm"):
+        try:
+            os.remove(f)
+        except FileNotFoundError:
+            pass
+
+    conn = get_conn()
+    _create_schema(conn)
+    return conn
+
+
+def _create_schema(conn):
+    """Создать все таблицы (idempotent)."""
+    conn.executescript("""
+        CREATE TABLE IF NOT EXISTS documents (
+            id TEXT PRIMARY KEY,
+            filename TEXT NOT NULL,
+            mime_type TEXT DEFAULT 'application/octet-stream',
+            original_bytes TEXT,
+            status TEXT DEFAULT 'uploaded',
+            error_message TEXT,
+            elements_json TEXT,
+            batch_id TEXT,
+            zip_source TEXT,
+            content_hash TEXT,
+            classify_status TEXT DEFAULT 'pending',
+            doc_type TEXT,
+            own_number TEXT,
+            parent_number TEXT,
+            doc_date TEXT,
+            counterparty TEXT,
+            classify_raw TEXT,
+            classify_input TEXT,
+            created_at TEXT DEFAULT (datetime('now'))
+        );
+
+        CREATE TABLE IF NOT EXISTS contracts (
+            id TEXT PRIMARY KEY,
+            number TEXT NOT NULL,
+            client TEXT DEFAULT '',
+            created_at TEXT DEFAULT (datetime('now'))
+        );
+
+        CREATE TABLE IF NOT EXISTS supplements (
+            id TEXT PRIMARY KEY,
+            contract_id TEXT NOT NULL REFERENCES contracts(id),
+            document_id TEXT NOT NULL REFERENCES documents(id),
+            type TEXT DEFAULT 'additional',
+            created_at TEXT DEFAULT (datetime('now'))
+        );
+
+        CREATE TABLE IF NOT EXISTS spec_events (
+            id TEXT PRIMARY KEY,
+            contract_id TEXT NOT NULL,
+            supplement_id TEXT NOT NULL,
+            event_type TEXT NOT NULL,
+            payload TEXT,
+            created_at TEXT DEFAULT (datetime('now'))
+        );
+
+        CREATE TABLE IF NOT EXISTS spec_current (
+            id TEXT PRIMARY KEY,
+            contract_id TEXT NOT NULL,
+            name_hash TEXT NOT NULL,
+            name TEXT,
+            price REAL,
+            qty REAL,
+            sum REAL,
+            date_start TEXT,
+            last_event_id TEXT,
+            created_at TEXT DEFAULT (datetime('now'))
+        );
+
+        CREATE TABLE IF NOT EXISTS prompts (
+            id TEXT PRIMARY KEY,
+            role TEXT NOT NULL,
+            name TEXT DEFAULT '',
+            body TEXT NOT NULL,
+            is_active INTEGER DEFAULT 0,
+            notes TEXT,
+            created_at TEXT DEFAULT (datetime('now'))
+        );
+
+        CREATE TABLE IF NOT EXISTS upload_chunks (
+            id TEXT PRIMARY KEY,
+            upload_id TEXT NOT NULL,
+            chunk_index INTEGER NOT NULL,
+            data TEXT,
+            created_at TEXT DEFAULT (datetime('now'))
+        );
+
+        CREATE INDEX IF NOT EXISTS idx_documents_batch ON documents(batch_id);
+        CREATE INDEX IF NOT EXISTS idx_documents_hash ON documents(batch_id, content_hash);
+        CREATE INDEX IF NOT EXISTS idx_supplements_contract ON supplements(contract_id);
+        CREATE INDEX IF NOT EXISTS idx_spec_current_contract ON spec_current(contract_id);
+        CREATE INDEX IF NOT EXISTS idx_spec_events_supplement ON spec_events(supplement_id);
+    """)
+
+
+def cleanup_db():
+    """Удалить БД полностью. Вызывает init_db() для создания новой."""
+    init_db()
+
+
+def get_conn():
+    """Thread-local соединение. Пересоздаётся при смене сессии."""
+    global _db_session_key
+
+    conn = getattr(_local, "conn", None)
+    local_key = getattr(_local, "session_key", None)
+
+    if conn is None or local_key != _db_session_key:
+        if conn is not None:
+            try:
+                conn.close()
+            except Exception:
+                pass
+        conn = sqlite3.connect(DB_PATH, timeout=5, check_same_thread=False)
+        conn.row_factory = sqlite3.Row
+        conn.execute("PRAGMA journal_mode=WAL")
+        conn.execute("PRAGMA busy_timeout=5000")
+        conn.execute("PRAGMA foreign_keys=ON")
+        _local.conn = conn
+        _local.session_key = _db_session_key
+
+    return conn
 
 
 def query(sql, params=None):
-    """
-    SELECT → list[dict] с lowercase ключами.
-    Используется всеми db/*.py модулями для чтения данных.
-    """
-    pool = get_pool()
-    conn = pool.getconn()
-    try:
-        with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
-            cur.execute(sql, params)
-            rows = cur.fetchall()
-            return [{k.lower(): v for k, v in r.items()} for r in rows]
-    finally:
-        pool.putconn(conn)
+    """SELECT → list[dict]."""
+    conn = get_conn()
+    sql = _pg_to_sqlite(sql)
+    cur = conn.execute(sql, params or [])
+    return [dict(r) for r in cur.fetchall()]
 
 
 def execute(sql, params=None):
-    """
-    INSERT/UPDATE/DELETE → количество затронутых строк.
-    Автокоммит. При ошибке — rollback и проброс исключения.
-    """
-    pool = get_pool()
-    conn = pool.getconn()
-    try:
-        with conn.cursor() as cur:
-            cur.execute(sql, params)
-            conn.commit()
-            return cur.rowcount
-    except Exception:
-        conn.rollback()
-        raise
-    finally:
-        pool.putconn(conn)
+    """INSERT/UPDATE/DELETE → rowcount."""
+    conn = get_conn()
+    sql = _pg_to_sqlite(sql)
+    cur = conn.execute(sql, params or [])
+    conn.commit()
+    return cur.rowcount
 
 
 def execute_returning(sql, params=None):
-    """
-    INSERT/UPDATE/DELETE с RETURNING → dict первой строки.
-    Используется для insert с автогенерацией UUID (gen_random_uuid()).
-    """
-    pool = get_pool()
-    conn = pool.getconn()
-    try:
-        with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
-            cur.execute(sql, params)
-            conn.commit()
-            row = cur.fetchone()
-            return {k.lower(): v for k, v in row.items()} if row else None
-    except Exception:
-        conn.rollback()
-        raise
-    finally:
-        pool.putconn(conn)
+    """INSERT с RETURNING → dict (эмулируется через lastrowid)."""
+    conn = get_conn()
+    table = _extract_table(sql)
+    sql = _pg_to_sqlite(sql)
+
+    if " RETURNING " in sql.upper():
+        sql = sql[:sql.upper().rfind(" RETURNING ")]
+
+    cur = conn.execute(sql, params or [])
+    conn.commit()
+    rowid = cur.lastrowid
+
+    if table and rowid:
+        row = conn.execute(f"SELECT * FROM {table} WHERE rowid = ?", (rowid,)).fetchone()
+        if row:
+            return dict(row)
+
+    # Fallback: если нет таблицы или rowid — просто вернуть последнюю строку
+    if table:
+        row = conn.execute(f"SELECT * FROM {table} ORDER BY rowid DESC LIMIT 1").fetchone()
+        return dict(row) if row else None
+    return None
+
+
+def _pg_to_sqlite(sql):
+    """Конвертировать PostgreSQL-специфичный SQL в SQLite."""
+    # %s → ?
+    sql = sql.replace("%s", "?")
+    # ::jsonb → убрать (SQLite не типизирует)
+    sql = sql.replace("::jsonb", "")
+    # gen_random_uuid() → хекс-UUID через randomblob
+    if "gen_random_uuid()" in sql:
+        import uuid
+        sql = sql.replace("gen_random_uuid()", "?")
+    # BOOLEAN → INTEGER
+    sql = sql.replace(" BOOLEAN ", " INTEGER ")
+    sql = sql.replace(" bool ", " INTEGER ")
+    # ILIKE → LIKE (SQLite LIKE case-insensitive для ASCII)
+    sql = sql.replace(" ILIKE ", " LIKE ")
+    # FALSE/TRUE → 0/1
+    sql = sql.replace(" FALSE", " 0").replace(" TRUE", " 1")
+    sql = sql.replace(" false", " 0").replace(" true", " 1")
+    return sql
+
+
+def _extract_table(sql):
+    """Извлечь имя таблицы из INSERT INTO <table>."""
+    import re
+    m = re.search(r"INSERT\s+INTO\s+(\w+)", sql, re.IGNORECASE)
+    return m.group(1) if m else None
diff --git a/site/routes/api_bp.py b/site/routes/api_bp.py
index ece7a8f..0e1e597 100644
--- a/site/routes/api_bp.py
+++ b/site/routes/api_bp.py
@@ -3,21 +3,12 @@ from flask import Blueprint, request, jsonify
 from site.db import documents, supplements, spec_current
 from site.db.connection import execute, query
 from site.services.grouping import group_documents, apply_groups
-from site.config import API_KEY, LLM_URL, LLM_KEY, LLM_MODEL
+from site.config import LLM_URL, LLM_KEY, LLM_MODEL
 import httpx
 
 api_bp = Blueprint("api", __name__)
 
 
-def _require_auth():
-    """Проверка API-ключа для опасных операций."""
-    if API_KEY:
-        key = request.headers.get("X-Api-Key", "")
-        if key != API_KEY:
-            return False
-    return True
-
-
 # ── Supplements ──────────────────────────────────────────────────
 
 @api_bp.route("/api/supplements")
@@ -181,18 +172,11 @@ def chat():
         return jsonify(OK=False, ERROR=f"LLM error: {e}")
 
 
-# ── Cleanup (требует API-ключ) ────────────────────────────────────
+# ── Cleanup (атомарное удаление БД) ──────────────────────────────
 
 @api_bp.route("/api/cleanup", methods=["POST"])
 def api_cleanup():
-    """Полная очистка всех данных (кроме prompts)."""
-    if not _require_auth():
-        return jsonify(ok=False, error="unauthorized"), 401
-
-    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")
+    """Полная очистка: os.remove(DB) + init новой. Данные гарантированно стёрты."""
+    from site.db.connection import cleanup_db
+    cleanup_db()
     return jsonify(ok=True, message="all data cleaned")