"""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() # ── 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()