"""SQLite — сохранение сессий диагностики. Таблица sessions (35+ колонок): client_ip, real_ip, user_agent, content_length, created_at 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 """ import json import sqlite3 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._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): 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, -- 'bt' | 'tcp' 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); CREATE INDEX IF NOT EXISTS idx_sessions_aid ON sessions(android_id); CREATE INDEX IF NOT EXISTS idx_sessions_request_id ON sessions(request_id); CREATE INDEX IF NOT EXISTS idx_sessions_uuid ON sessions(device_uuid); """) 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 (идемпотентность). """ 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]): """Сохраняет быстрый скан кодов ошибок.""" 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()