v1.1.50: scenario_definitions DB+CRUD, seed, runner from DB, 409 lock, UI definitions, drop config.yaml scenarios

2026-07-30 15:06:05 +04:00
parent 67f6390509
commit 983b387708
8 changed files with 418 additions and 161 deletions
@@ -17,6 +17,7 @@
 """

 import os
+import json
 from db.pool import get_conn, put_conn

 SCHEMA_SQL = """
@@ -80,9 +81,65 @@ CREATE TABLE IF NOT EXISTS scenario_runs (

 CREATE INDEX IF NOT EXISTS idx_scenario_runs_client_stand
    ON scenario_runs (client_id, stand, created_at DESC);
+
+CREATE INDEX IF NOT EXISTS idx_scenario_runs_status
+    ON scenario_runs (client_id, stand, status);
+
+ALTER TABLE scenario_runs ADD COLUMN IF NOT EXISTS definition_id INTEGER;
+ALTER TABLE scenario_runs ADD COLUMN IF NOT EXISTS definition_version INTEGER;
+
+-- Таблица определений сценариев — редактируются без редеплоя
+CREATE TABLE IF NOT EXISTS scenario_definitions (
+    id            SERIAL PRIMARY KEY,
+    client_id     VARCHAR(64) NOT NULL,
+    stand         VARCHAR(16) NOT NULL,
+    name          VARCHAR(200) NOT NULL,
+    steps         JSONB NOT NULL DEFAULT '[]',
+    version       INTEGER NOT NULL DEFAULT 1,
+    is_active     BOOLEAN NOT NULL DEFAULT TRUE,
+    created_at    TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    updated_at    TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    updated_by    VARCHAR(128)
+);
+
+CREATE UNIQUE INDEX IF NOT EXISTS idx_scenario_defs_unique
+    ON scenario_definitions (client_id, stand, LOWER(name));
+
+CREATE INDEX IF NOT EXISTS idx_scenario_defs_active
+    ON scenario_definitions (client_id, stand, is_active);
 """


+def _seed_scenarios():
+    """Импорт дефолтных сценариев из scenario_seed.yaml при первом старте."""
+    try:
+        import yaml, os
+        path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "scenario_seed.yaml")
+        if not os.path.exists(path):
+            return
+        with open(path) as f:
+            data = yaml.safe_load(f) or {}
+        seed_list = data.get("scenarios", [])
+        if not seed_list:
+            return
+        conn = get_conn()
+        if not conn:
+            return
+        cur = conn.cursor()
+        for s in seed_list:
+            cur.execute("""
+                INSERT INTO scenario_definitions (client_id, stand, name, steps)
+                VALUES (%s, %s, %s, %s)
+                ON CONFLICT (client_id, stand, LOWER(name)) DO NOTHING
+            """, (s.get("client_id", ""), s.get("stand", ""), s["name"], json.dumps(s.get("steps", []))))
+        conn.commit()
+        cur.close()
+        put_conn(conn)
+        print("[DB] Seed scenarios OK", flush=True)
+    except Exception as e:
+        print(f"[DB] Seed scenarios error: {e}", flush=True)
+
+
 def init_db():
    """Применить схему + миграции — идемпотентно, безопасно для конкурентного вызова."""
    if not os.getenv("DB_USER"):
@@ -107,3 +164,6 @@ def init_db():
        conn.rollback()
    finally:
        put_conn(conn)
+
+    # Seed сценариев — только при первом старте (INSERT ON CONFLICT DO NOTHING)
+    _seed_scenarios()
@@ -0,0 +1,168 @@
+"""
+CRUD для таблицы scenario_definitions.
+
+Функции:
+  list_definitions(client_id, stand)   — список активных
+  get_definition(def_id, cid, stand)   — один
+  create_definition(cid, stand, name, steps, updated_by) → id
+  update_definition(def_id, cid, stand, name, steps, version, updated_by) → ok|conflict
+  delete_definition(def_id, cid, stand) — мягкое (is_active=false)
+  lock_check(cid, stand)               — есть ли RUNNING запуск → 409 если да
+"""
+
+import json
+from db.pool import get_conn, put_conn
+
+
+def list_definitions(client_id, stand):
+    conn = get_conn()
+    if not conn:
+        return []
+    try:
+        cur = conn.cursor()
+        cur.execute("""
+            SELECT id, name, steps, version, is_active, created_at, updated_at
+            FROM scenario_definitions
+            WHERE client_id = %s AND stand = %s AND is_active = TRUE
+            ORDER BY name
+        """, (client_id, stand))
+        rows = cur.fetchall()
+        cols = [d[0] for d in cur.description]
+        result = []
+        for r in rows:
+            d = dict(zip(cols, r))
+            d["created_at"] = d["created_at"].isoformat() if d["created_at"] else None
+            d["updated_at"] = d["updated_at"].isoformat() if d["updated_at"] else None
+            result.append(d)
+        return result
+    except Exception as e:
+        print(f"[DEFS] list error: {e}", flush=True)
+        return []
+    finally:
+        put_conn(conn)
+
+
+def get_definition(def_id, client_id, stand):
+    conn = get_conn()
+    if not conn:
+        return None
+    try:
+        cur = conn.cursor()
+        cur.execute("""
+            SELECT id, name, steps, version, is_active, created_at, updated_at, updated_by
+            FROM scenario_definitions
+            WHERE id = %s AND client_id = %s AND stand = %s
+        """, (def_id, client_id, stand))
+        row = cur.fetchone()
+        if not row:
+            return None
+        cols = [d[0] for d in cur.description]
+        d = dict(zip(cols, row))
+        d["created_at"] = d["created_at"].isoformat() if d["created_at"] else None
+        d["updated_at"] = d["updated_at"].isoformat() if d["updated_at"] else None
+        return d
+    except Exception as e:
+        print(f"[DEFS] get error: {e}", flush=True)
+        return None
+    finally:
+        put_conn(conn)
+
+
+def create_definition(client_id, stand, name, steps, updated_by=""):
+    conn = get_conn()
+    if not conn:
+        return None
+    try:
+        cur = conn.cursor()
+        cur.execute("""
+            INSERT INTO scenario_definitions (client_id, stand, name, steps, updated_by)
+            VALUES (%s, %s, %s, %s, %s)
+            ON CONFLICT (client_id, stand, LOWER(name))
+            DO UPDATE SET steps = EXCLUDED.steps, version = scenario_definitions.version + 1,
+                          is_active = TRUE, updated_at = NOW(), updated_by = EXCLUDED.updated_by
+            RETURNING id, version
+        """, (client_id, stand, name, json.dumps(steps), updated_by))
+        row = cur.fetchone()
+        conn.commit()
+        cur.close()
+        return {"id": row[0], "version": row[1]}
+    except Exception as e:
+        print(f"[DEFS] create error: {e}", flush=True)
+        conn.rollback()
+        return None
+    finally:
+        put_conn(conn)
+
+
+def update_definition(def_id, client_id, stand, name, steps, version, updated_by=""):
+    """Возвращает dict {ok, version} или {conflict: True}."""
+    conn = get_conn()
+    if not conn:
+        return {"error": "no db"}
+    try:
+        cur = conn.cursor()
+        cur.execute("""
+            UPDATE scenario_definitions
+            SET name = %s, steps = %s, version = version + 1,
+                updated_at = NOW(), updated_by = %s
+            WHERE id = %s AND client_id = %s AND stand = %s AND version = %s
+            RETURNING version
+        """, (name, json.dumps(steps), updated_by, def_id, client_id, stand, version))
+        row = cur.fetchone()
+        conn.commit()
+        cur.close()
+        if row:
+            return {"ok": True, "version": row[0]}
+        return {"conflict": True}
+    except Exception as e:
+        print(f"[DEFS] update error: {e}", flush=True)
+        conn.rollback()
+        return {"error": str(e)}
+    finally:
+        put_conn(conn)
+
+
+def delete_definition(def_id, client_id, stand):
+    """Мягкое удаление."""
+    conn = get_conn()
+    if not conn:
+        return False
+    try:
+        cur = conn.cursor()
+        cur.execute("""
+            UPDATE scenario_definitions
+            SET is_active = FALSE, updated_at = NOW()
+            WHERE id = %s AND client_id = %s AND stand = %s
+        """, (def_id, client_id, stand))
+        ok = cur.rowcount > 0
+        conn.commit()
+        cur.close()
+        return ok
+    except Exception as e:
+        print(f"[DEFS] delete error: {e}", flush=True)
+        conn.rollback()
+        return False
+    finally:
+        put_conn(conn)
+
+
+def lock_check(client_id, stand):
+    """Проверить нет ли активного RUNNING запуска. True = можно запускать."""
+    conn = get_conn()
+    if not conn:
+        return True  # без БД — разрешаем
+    try:
+        cur = conn.cursor()
+        cur.execute("""
+            SELECT id FROM scenario_runs
+            WHERE client_id = %s AND stand = %s AND status = 'RUNNING'
+            LIMIT 1
+        """, (client_id, stand))
+        row = cur.fetchone()
+        cur.close()
+        return row is None
+    except Exception as e:
+        print(f"[DEFS] lock_check error: {e}", flush=True)
+        return True
+    finally:
+        put_conn(conn)