"""SQLite 存储:表结构与插入字段由协议 schema 自动生成。 消息类 -> 表名 映射见 TABLE_NAMES;插入时从 msg.to_dict() 取字段, 与 protocol.py 保持单一数据源,避免手写列定义漂移。 """ import sqlite3 import threading from . import protocol def _table_name(msg_cls): name = msg_cls.__name__ if name.startswith("MsgDis"): name = name[len("MsgDis"):] # MsgDisHighVolBusFb -> msg_disHighVolBusFbMsg return "msg_dis" + name + "Msg" TABLE_NAMES = {cls: _table_name(cls) for cls in protocol.ALL_MESSAGES.values()} class DB: def __init__(self, db_path): self.db_path = db_path self.conn = None self.lock = threading.Lock() def connect(self): self.conn = sqlite3.connect(self.db_path, check_same_thread=False) self.conn.row_factory = sqlite3.Row self.create_tables() def disconnect(self): """与 close() 等价,兼容历史调用。""" self.close() def close(self): if self.conn is not None: self.conn.close() self.conn = None def create_tables(self): with self.lock: cursor = self.conn.cursor() for msg_cls, table in TABLE_NAMES.items(): cols = ", ".join(f'"{name}" INTEGER' for name in msg_cls.field_names()) cursor.execute( f'CREATE TABLE IF NOT EXISTS "{table}" (' f'id INTEGER PRIMARY KEY AUTOINCREMENT, ' f'timestamp DATETIME DEFAULT (datetime(\'now\', \'localtime\')), ' f'{cols})' ) self.conn.commit() def insert_message(self, msg): """按消息类型路由到对应数据表。返回 True/False。""" table = TABLE_NAMES.get(msg.__class__) if table is None: return False return self.insert_dict(table, msg.to_dict()) def insert_dict(self, table, data_dict): with self.lock: try: if self.conn is None: self.connect() if self.conn is None: return False columns = ", ".join(f'"{c}"' for c in data_dict.keys()) placeholders = ", ".join(["?"] * len(data_dict)) query = f'INSERT INTO "{table}" ({columns}) VALUES ({placeholders})' self.conn.cursor().execute(query, list(data_dict.values())) self.conn.commit() return True except sqlite3.Error: return False