Веб-приложение диагностики по видео (ФПГ GigaDoc)

- Загрузка/сканирование видео, извлечение кадров, анализ через SDK ФПГ
- Понятный диагноз с сохранением в SQLite и share-ссылками
- Экспорт PDF/DOCX, скачивание записанного видео
- Камерное сканирование на мобильных (canvas + MediaRecorder)
- Секреты вынесены в переменные окружения

storage.py

@@ -0,0 +1,124 @@
+#!/usr/bin/env python3
+"""
+Хранение результатов анализов в SQLite.
+
+Таблица results:
+  - id          INTEGER PRIMARY KEY AUTOINCREMENT
+  - share_id    TEXT UNIQUE  — короткий ID для публичной ссылки
+  - created_at  TEXT         — дата создания (ISO)
+  - raw_json    TEXT         — сырой JSON от SDK
+  - summary     TEXT         — сводка (для OG-тегов)
+  - warnings    TEXT         — JSON-список предупреждений
+  - metrics     TEXT         — JSON-список показателей
+  - advice      TEXT         — JSON-список рекомендаций
+"""
+
+import json
+import os
+import sqlite3
+from datetime import datetime
+
+DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "results.db")
+
+# Первый ID результата
+FIRST_ID = 10005
+
+
+def _connect():
+    conn = sqlite3.connect(DB_PATH)
+    conn.row_factory = sqlite3.Row
+    return conn
+
+
+def init_db():
+    """Создать таблицу, если её нет."""
+    conn = _connect()
+    conn.execute("""
+        CREATE TABLE IF NOT EXISTS results (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            share_id TEXT UNIQUE NOT NULL,
+            created_at TEXT NOT NULL,
+            raw_json TEXT NOT NULL,
+            summary TEXT NOT NULL,
+            warnings TEXT NOT NULL DEFAULT '[]',
+            metrics TEXT NOT NULL DEFAULT '[]',
+            advice TEXT NOT NULL DEFAULT '[]'
+        )
+    """)
+    conn.commit()
+    conn.close()
+
+
+def save_result(interpretation: dict, raw_result: dict) -> str:
+    """
+    Сохранить результат анализа. Возвращает share_id (числовой, начиная с 10005).
+    """
+    init_db()
+
+    conn = _connect()
+
+    # Следующий ID: MAX(id) + 1, но не меньше FIRST_ID
+    row = conn.execute("SELECT MAX(id) AS max_id FROM results").fetchone()
+    next_id = (row["max_id"] + 1) if row["max_id"] is not None else FIRST_ID
+    if next_id < FIRST_ID:
+        next_id = FIRST_ID
+
+    share_id = str(next_id)
+
+    conn.execute(
+        """
+        INSERT INTO results (id, share_id, created_at, raw_json, summary, warnings, metrics, advice)
+        VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+        """,
+        (
+            next_id,
+            share_id,
+            datetime.now().isoformat(timespec="seconds"),
+            json.dumps(raw_result, ensure_ascii=False),
+            interpretation.get("summary", ""),
+            json.dumps(interpretation.get("warnings", []), ensure_ascii=False),
+            json.dumps(interpretation.get("metrics", []), ensure_ascii=False),
+            json.dumps(interpretation.get("advice", []), ensure_ascii=False),
+        ),
+    )
+    conn.commit()
+    conn.close()
+    return share_id
+
+
+def get_result(share_id: str):
+    """Получить результат по share_id. Возвращает dict или None."""
+    init_db()
+
+    conn = _connect()
+    row = conn.execute(
+        "SELECT * FROM results WHERE share_id = ?", (share_id,)
+    ).fetchone()
+    conn.close()
+
+    if row is None:
+        return None
+
+    return {
+        "share_id": row["share_id"],
+        "created_at": row["created_at"],
+        "raw_json": json.loads(row["raw_json"]),
+        "summary": row["summary"],
+        "warnings": json.loads(row["warnings"]),
+        "metrics": json.loads(row["metrics"]),
+        "advice": json.loads(row["advice"]),
+    }
+
+
+def list_results(limit: int = 50):
+    """Список последних результатов (для админки/истории)."""
+    init_db()
+
+    conn = _connect()
+    rows = conn.execute(
+        "SELECT share_id, created_at, summary FROM results ORDER BY id DESC LIMIT ?",
+        (limit,),
+    ).fetchall()
+    conn.close()
+
+    return [dict(r) for r in rows]