351 lines
15 KiB
Python
351 lines
15 KiB
Python
"""SQLite — сохранение сессий диагностики + профили ELM-устройств.
|
||
|
||
Таблица sessions (35+ колонок):
|
||
... (см. ниже)
|
||
|
||
Таблица device_profiles:
|
||
mac (TEXT PK), level (INT), elm_version, elm_desc, protocol,
|
||
supported (JSON), unsupported (JSON), first_seen, last_seen
|
||
"""
|
||
|
||
import json
|
||
import sqlite3
|
||
import threading
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
|
||
|
||
class Database:
|
||
def __init__(self, path: str | Path = "elmer.db"):
|
||
self.path = Path(path)
|
||
self.conn = sqlite3.connect(str(self.path), timeout=30, check_same_thread=False)
|
||
self.conn.row_factory = sqlite3.Row
|
||
self.conn.execute("PRAGMA journal_mode=WAL")
|
||
self.conn.execute("PRAGMA busy_timeout=30000")
|
||
self._lock = threading.Lock()
|
||
self._init_schema()
|
||
|
||
def __enter__(self):
|
||
"""Контекстный менеджер: with Database() as db."""
|
||
return self
|
||
|
||
def __exit__(self, exc_type, exc_val, exc_tb):
|
||
"""Закрытие соединения при выходе из with-блока."""
|
||
self.close()
|
||
return False
|
||
|
||
def close(self):
|
||
"""Закрыть соединение с SQLite."""
|
||
if self.conn:
|
||
self.conn.close()
|
||
self.conn = None
|
||
|
||
def _init_schema(self):
|
||
# Основная схема (может упасть на индексах старых БД — ловим)
|
||
try:
|
||
self.conn.executescript("""
|
||
CREATE TABLE IF NOT EXISTS sessions (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
|
||
-- Сервер
|
||
client_ip TEXT,
|
||
real_ip TEXT,
|
||
user_agent TEXT,
|
||
content_length INTEGER,
|
||
created_at TEXT NOT NULL DEFAULT (datetime('now')),
|
||
|
||
-- Телефон
|
||
phone_model TEXT,
|
||
phone_maker TEXT,
|
||
android_version TEXT,
|
||
android_sdk INTEGER,
|
||
app_version TEXT,
|
||
android_id TEXT,
|
||
device_uuid TEXT,
|
||
phone_lang TEXT,
|
||
phone_tz TEXT,
|
||
phone_display TEXT,
|
||
|
||
-- ELM327
|
||
elm_mac TEXT,
|
||
elm_bt_name TEXT,
|
||
obd_protocol TEXT,
|
||
|
||
-- Авто
|
||
vin TEXT,
|
||
dtc_count INTEGER DEFAULT 0,
|
||
pid_count INTEGER DEFAULT 0,
|
||
|
||
-- Сессия
|
||
duration_ms INTEGER,
|
||
response_count INTEGER DEFAULT 0,
|
||
error_count INTEGER DEFAULT 0,
|
||
retry_count INTEGER DEFAULT 0,
|
||
timeout_count INTEGER DEFAULT 0,
|
||
script_mode TEXT,
|
||
transport TEXT,
|
||
mock_mode INTEGER DEFAULT 0,
|
||
car_info TEXT,
|
||
|
||
-- LLM
|
||
diagnosis_text TEXT,
|
||
diagnosis_len INTEGER,
|
||
llm_model TEXT,
|
||
llm_duration_ms INTEGER,
|
||
llm_success INTEGER DEFAULT 0,
|
||
|
||
-- Сырые данные (JSON)
|
||
raw_responses TEXT,
|
||
|
||
-- Идемпотентность
|
||
request_id TEXT UNIQUE,
|
||
response_json TEXT
|
||
);
|
||
|
||
CREATE INDEX IF NOT EXISTS idx_sessions_created ON sessions(created_at);
|
||
CREATE INDEX IF NOT EXISTS idx_sessions_vin ON sessions(vin);
|
||
CREATE INDEX IF NOT EXISTS idx_sessions_mac ON sessions(elm_mac);
|
||
|
||
-- Профили ELM-устройств
|
||
CREATE TABLE IF NOT EXISTS device_profiles (
|
||
mac TEXT PRIMARY KEY,
|
||
level INTEGER NOT NULL,
|
||
elm_version TEXT,
|
||
elm_desc TEXT,
|
||
protocol TEXT,
|
||
voltage TEXT,
|
||
response_time_ms INTEGER DEFAULT 250,
|
||
supported TEXT,
|
||
unsupported TEXT,
|
||
errors TEXT,
|
||
first_seen TEXT NOT NULL DEFAULT (datetime('now')),
|
||
last_seen TEXT NOT NULL DEFAULT (datetime('now'))
|
||
);
|
||
CREATE INDEX IF NOT EXISTS idx_profiles_level ON device_profiles(level);
|
||
""")
|
||
except sqlite3.OperationalError:
|
||
pass # старая БД без новых колонок — применим миграции ниже
|
||
|
||
self.conn.commit()
|
||
|
||
# Миграции: добавляем колонки, которых нет в старых БД
|
||
migrations = [
|
||
"ALTER TABLE device_profiles ADD COLUMN response_time_ms INTEGER DEFAULT 250",
|
||
"ALTER TABLE sessions ADD COLUMN device_uuid TEXT",
|
||
"ALTER TABLE sessions ADD COLUMN phone_lang TEXT",
|
||
"ALTER TABLE sessions ADD COLUMN phone_tz TEXT",
|
||
"ALTER TABLE sessions ADD COLUMN phone_display TEXT",
|
||
"ALTER TABLE sessions ADD COLUMN android_id TEXT",
|
||
"ALTER TABLE sessions ADD COLUMN car_info TEXT",
|
||
"ALTER TABLE sessions ADD COLUMN request_id TEXT",
|
||
"ALTER TABLE sessions ADD COLUMN response_json TEXT",
|
||
]
|
||
for sql in migrations:
|
||
try:
|
||
self.conn.execute(sql)
|
||
except sqlite3.OperationalError:
|
||
pass
|
||
|
||
# Индексы для новых колонок (могут отсутствовать в старых БД)
|
||
index_migrations = [
|
||
"CREATE INDEX IF NOT EXISTS idx_sessions_aid ON sessions(android_id)",
|
||
"CREATE INDEX IF NOT EXISTS idx_sessions_uuid ON sessions(device_uuid)",
|
||
"CREATE INDEX IF NOT EXISTS idx_sessions_request_id ON sessions(request_id)",
|
||
]
|
||
for sql in index_migrations:
|
||
try:
|
||
self.conn.execute(sql)
|
||
except sqlite3.OperationalError:
|
||
pass
|
||
|
||
self.conn.commit()
|
||
|
||
# command_queue — для raw-ретранслятора (gunicorn-safe)
|
||
self.conn.execute("""
|
||
CREATE TABLE IF NOT EXISTS command_queue (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
device_id TEXT NOT NULL,
|
||
seq INTEGER NOT NULL,
|
||
cmd TEXT NOT NULL,
|
||
timeout_ms INTEGER DEFAULT 500,
|
||
drain_first INTEGER DEFAULT 0,
|
||
status TEXT NOT NULL DEFAULT 'pending',
|
||
raw_response TEXT,
|
||
elapsed_ms INTEGER,
|
||
prompt INTEGER,
|
||
error TEXT,
|
||
created_at TEXT NOT NULL DEFAULT (datetime('now')),
|
||
sent_at TEXT,
|
||
responded_at TEXT
|
||
)
|
||
""")
|
||
self.conn.execute(
|
||
"CREATE INDEX IF NOT EXISTS idx_cq_device_status_seq ON command_queue(device_id, status, seq)"
|
||
)
|
||
self.conn.commit()
|
||
|
||
# ── sessions ──────────────────────────────────────────
|
||
|
||
def get_cached_response(self, request_id: str) -> dict | None:
|
||
"""Возвращает сохранённый ответ сессии по request_id, или None."""
|
||
row = self.conn.execute(
|
||
"SELECT response_json FROM sessions WHERE request_id = ?", (request_id,)
|
||
).fetchone()
|
||
if row and row["response_json"]:
|
||
return json.loads(row["response_json"])
|
||
return None
|
||
|
||
def save_session(self, client_info: dict, responses: list[dict],
|
||
diagnosis: str = "", llm_model: str = "",
|
||
llm_duration_ms: int = 0, llm_success: bool = False,
|
||
request_id: str = "", response_json: dict | None = None):
|
||
"""Сохраняет сводную запись о сессии.
|
||
|
||
Если request_id передан и уже существует — silently return (идемпотентность).
|
||
"""
|
||
with self._lock:
|
||
ci = client_info
|
||
|
||
# Подсчёт DTC/PID из ответов
|
||
dtc_count = 0
|
||
pid_count = 0
|
||
for r in responses:
|
||
dec = (r.get("decoded") or "").lower()
|
||
if dec.startswith("dtc"):
|
||
dtc_count += 1
|
||
elif ":" in dec and not dec.startswith(("vin", "dtc", "elm", "protocol")):
|
||
pid_count += 1
|
||
|
||
# VIN из ответов
|
||
vin = None
|
||
for r in responses:
|
||
dec = (r.get("decoded") or "")
|
||
if dec.startswith("VIN:"):
|
||
vin = dec[4:].strip()
|
||
if len(vin) != 17:
|
||
vin = None
|
||
break
|
||
|
||
resp_json_str = json.dumps(response_json, ensure_ascii=False) if response_json else None
|
||
|
||
self.conn.execute("""
|
||
INSERT OR IGNORE INTO sessions (
|
||
client_ip, real_ip, user_agent, content_length,
|
||
phone_model, phone_maker, android_version, android_sdk,
|
||
app_version, android_id, device_uuid, phone_lang, phone_tz, phone_display,
|
||
elm_mac, elm_bt_name, obd_protocol,
|
||
vin, dtc_count, pid_count,
|
||
duration_ms, response_count, error_count,
|
||
retry_count, timeout_count, script_mode,
|
||
transport, mock_mode, car_info,
|
||
diagnosis_text, diagnosis_len, llm_model,
|
||
llm_duration_ms, llm_success,
|
||
raw_responses, request_id, response_json
|
||
) VALUES (?,?,?,?, ?,?,?,?,?, ?,?,?,?,?, ?,?,?, ?,?,?, ?,?,?, ?,?,?, ?,?, ?,?,?,?, ?,?,?,?,?)
|
||
""", (
|
||
ci.get("client_ip"), ci.get("real_ip"), ci.get("user_agent"),
|
||
ci.get("content_length"),
|
||
ci.get("phone_model"), ci.get("phone_maker"), ci.get("android_version"),
|
||
ci.get("android_sdk"), ci.get("app_version"), ci.get("android_id"),
|
||
ci.get("device_uuid"),
|
||
ci.get("phone_lang"), ci.get("phone_tz"), ci.get("phone_display"),
|
||
ci.get("elm_mac"), ci.get("elm_bt_name"), ci.get("obd_protocol"),
|
||
vin, dtc_count, pid_count,
|
||
ci.get("duration_ms"), len(responses), ci.get("error_count", 0),
|
||
ci.get("retry_count", 0), ci.get("timeout_count", 0),
|
||
ci.get("script_mode"), ci.get("transport"), ci.get("mock_mode", 0),
|
||
ci.get("car_info", ""),
|
||
diagnosis, len(diagnosis), llm_model,
|
||
llm_duration_ms, 1 if llm_success else 0,
|
||
json.dumps(responses, ensure_ascii=False) if responses else None,
|
||
request_id if request_id else None,
|
||
resp_json_str,
|
||
))
|
||
self.conn.commit()
|
||
|
||
def get_recent_sessions(self, limit: int = 50) -> list[dict]:
|
||
"""Последние N сессий."""
|
||
rows = self.conn.execute(
|
||
"SELECT * FROM sessions ORDER BY created_at DESC LIMIT ?", (limit,)
|
||
).fetchall()
|
||
return [dict(r) for r in rows]
|
||
|
||
def save_dtc_scan(self, client_info: dict, dtc_codes: list[str]):
|
||
"""Сохраняет быстрый скан кодов ошибок."""
|
||
with self._lock:
|
||
self.conn.execute("""
|
||
INSERT INTO sessions (
|
||
client_ip, real_ip, user_agent,
|
||
phone_model, phone_maker, android_version, android_sdk,
|
||
app_version, android_id, device_uuid,
|
||
elm_mac, elm_bt_name,
|
||
dtc_count, response_count,
|
||
script_mode, transport,
|
||
raw_responses
|
||
) VALUES (?,?,?, ?,?,?,?, ?,?,?, ?,?, ?,?, ?,?,?)
|
||
""", (
|
||
client_info.get("client_ip"), client_info.get("real_ip"), client_info.get("user_agent"),
|
||
client_info.get("phone_model"), client_info.get("phone_maker"), client_info.get("android_version"),
|
||
client_info.get("android_sdk"), client_info.get("app_version"), client_info.get("android_id"),
|
||
client_info.get("device_uuid"),
|
||
client_info.get("elm_mac"), client_info.get("elm_bt_name"),
|
||
len(dtc_codes), 0,
|
||
"dtc_scan", client_info.get("transport", "bt"),
|
||
json.dumps([{"decoded": f"DTC stored: {c}"} for c in dtc_codes], ensure_ascii=False)
|
||
))
|
||
self.conn.commit()
|
||
|
||
# ── device_profiles ──────────────────────────────────
|
||
|
||
def get_device_profile(self, mac: str) -> dict | None:
|
||
"""Возвращает сохранённый профиль устройства по MAC, или None."""
|
||
row = self.conn.execute(
|
||
"SELECT * FROM device_profiles WHERE mac = ?", (mac,)
|
||
).fetchone()
|
||
if not row:
|
||
return None
|
||
p = dict(row)
|
||
for f in ("supported", "unsupported", "errors"):
|
||
p[f] = json.loads(p[f]) if p.get(f) else []
|
||
return p
|
||
|
||
def save_device_profile(self, mac: str, profile: dict):
|
||
"""Сохраняет или обновляет профиль устройства.
|
||
|
||
profile — результат obd.probe.probe() + response_time_ms.
|
||
"""
|
||
with self._lock:
|
||
now = datetime.now(timezone.utc).isoformat()
|
||
self.conn.execute("""
|
||
INSERT INTO device_profiles
|
||
(mac, level, elm_version, elm_desc, protocol, voltage,
|
||
response_time_ms, supported, unsupported, errors,
|
||
first_seen, last_seen)
|
||
VALUES (?,?,?,?,?,?, ?,?,?,?, ?,?)
|
||
ON CONFLICT(mac) DO UPDATE SET
|
||
level = excluded.level,
|
||
elm_version = excluded.elm_version,
|
||
elm_desc = excluded.elm_desc,
|
||
protocol = excluded.protocol,
|
||
voltage = excluded.voltage,
|
||
response_time_ms = excluded.response_time_ms,
|
||
supported = excluded.supported,
|
||
unsupported = excluded.unsupported,
|
||
errors = excluded.errors,
|
||
last_seen = excluded.last_seen
|
||
""", (
|
||
mac,
|
||
profile.get("level", -1),
|
||
profile.get("elm_version"),
|
||
profile.get("elm_desc"),
|
||
profile.get("protocol"),
|
||
profile.get("voltage"),
|
||
profile.get("response_time_ms", 250),
|
||
json.dumps(profile.get("supported", []), ensure_ascii=False),
|
||
json.dumps(profile.get("unsupported", []), ensure_ascii=False),
|
||
json.dumps(profile.get("errors", []), ensure_ascii=False),
|
||
now, now,
|
||
))
|
||
self.conn.commit()
|
||
|