apuc
04a7b31a75
- Загрузка/сканирование видео, извлечение кадров, анализ через SDK ФПГ - Понятный диагноз с сохранением в SQLite и share-ссылками - Экспорт PDF/DOCX, скачивание записанного видео - Камерное сканирование на мобильных (canvas + MediaRecorder) - Секреты вынесены в переменные окружения
124 lines
3.8 KiB
Python
124 lines
3.8 KiB
Python
#!/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] |