import sqlite3 from typing import List, Optional, Tuple from pathlib import Path import os from .models import FilterData, ParamData, ConfigData, ProjectData class DatabaseManager: _instance = None def __new__(cls, db_path: str = None): if cls._instance is None: cls._instance = super(DatabaseManager, cls).__new__(cls) if db_path is None: app_dir = Path(os.path.dirname(os.path.abspath(__file__))).parent data_dir = app_dir / 'data' db_path = str(data_dir / 'database.db') print(f"db_path: {db_path}") cls._instance.db_path = db_path cls._instance._init_database() return cls._instance def __init__(self): # self.init_database() pass def _init_database(self): Path(self.db_path).parent.mkdir(parents=True, exist_ok=True) with sqlite3.connect(self.db_path) as conn: # 首先创建 projects 表（如果不存在） conn.execute(""" CREATE TABLE IF NOT EXISTS projects ( id INTEGER PRIMARY KEY, name TEXT NOT NULL UNIQUE, description TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); """) # 检查 configs 表是否存在 cursor = conn.execute(""" SELECT name FROM sqlite_master WHERE type='table' AND name='configs'; """) if cursor.fetchone() is None: # 如果表不存在，创建新表 conn.execute(""" CREATE TABLE configs ( id INTEGER PRIMARY KEY, name TEXT NOT NULL, channel_id INTEGER NOT NULL, project_id INTEGER, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, UNIQUE(name, channel_id, project_id), FOREIGN KEY (project_id) REFERENCES projects(id) ); """) else: # 如果表已存在，检查是否需要添加 project_id 列 cursor = conn.execute("PRAGMA table_info(configs)") columns = [row[1] for row in cursor.fetchall()] if 'project_id' not in columns: # 添加 project_id 列 conn.execute(""" ALTER TABLE configs ADD COLUMN project_id INTEGER REFERENCES projects(id); """) # 其他表的创建保持不变 conn.executescript(""" CREATE TABLE IF NOT EXISTS params ( id INTEGER PRIMARY KEY, config_id INTEGER, channel_id INTEGER NOT NULL, delay_data1 FLOAT, ENC_volume_data1 FLOAT, ENT_mx_right_data FLOAT, ENT_mix_left_data FLOAT, FOREIGN KEY (config_id) REFERENCES configs(id), FOREIGN KEY (channel_id) REFERENCES configs(channel_id) ); CREATE TABLE IF NOT EXISTS filters ( id INTEGER PRIMARY KEY, config_id INTEGER, channel_id INTEGER NOT NULL, filter_type TEXT NOT NULL, freq FLOAT, q FLOAT, gain FLOAT, slope FLOAT, enabled BOOLEAN DEFAULT TRUE, position INTEGER, FOREIGN KEY (config_id) REFERENCES configs(id), FOREIGN KEY (channel_id) REFERENCES configs(channel_id) ); """) def get_connection(self): return sqlite3.connect(self.db_path) def get_all_configs(self) -> List[ConfigData]: with self.get_connection() as conn: cursor = conn.execute("SELECT id, name, channel_id, created_at FROM configs") return [ConfigData(*row) for row in cursor.fetchall()] def load_config(self, config_id: int) -> Tuple[Optional[ParamData], List[FilterData]]: with self.get_connection() as conn: # 1. 首先加载参数数据 cursor = conn.execute( """SELECT config_id, channel_id, delay_data1, ENC_volume_data1, ENT_mx_right_data, ENT_mix_left_data FROM params WHERE config_id = ?""", (config_id,) ) param_row = cursor.fetchone() if param_row is None: return None, [] # 2. 创建 ParamData 对象 params = ParamData( config_id=param_row[0], channel_id=param_row[1], delay_data1=param_row[2], ENC_volume_data1=param_row[3], ENT_mx_right_data=param_row[4], ENT_mix_left_data=param_row[5] ) # 3. 加载滤波器数据 cursor = conn.execute( """SELECT config_id, channel_id, filter_type, freq, q, gain, slope, enabled, position, id FROM filters WHERE config_id = ? ORDER BY position""", (config_id,) ) filters = [] for row in cursor.fetchall(): filters.append(FilterData( channel_id=row[1], filter_type=row[2], freq=row[3], q=row[4], gain=row[5], slope=row[6], enabled=bool(row[7]), position=row[8], config_id=row[0], id=row[9] )) print(f"load_config filters: {filters}") return params, filters def save_config(self, name: str, channel_id: int, params: ParamData, filters: List[FilterData], project_id: Optional[int] = None) -> int: with self.get_connection() as conn: # 检查是否存在相同的配置 cursor = conn.execute( """SELECT id FROM configs WHERE name = ? AND channel_id = ? AND (project_id = ? OR (project_id IS NULL AND ? IS NULL))""", (name, channel_id, project_id, project_id) ) existing_config = cursor.fetchone() if existing_config: # 更新现有配置 config_id = existing_config[0] # 更新参数 params.config_id = config_id self.update_params(config_id, params) # 删除旧的滤波器 conn.execute("DELETE FROM filters WHERE config_id = ?", (config_id,)) # 插入新的滤波器 for filter_data in filters: filter_data.config_id = config_id cursor = conn.execute( """INSERT INTO filters (config_id, channel_id, filter_type, freq, q, gain, slope, enabled, position) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""", (filter_data.config_id, filter_data.channel_id, filter_data.filter_type, filter_data.freq, filter_data.q, filter_data.gain, filter_data.slope, filter_data.enabled, filter_data.position) ) filter_data.id = cursor.lastrowid return config_id else: # 创建新配置 cursor = conn.execute( "INSERT INTO configs (name, channel_id, project_id) VALUES (?, ?, ?)", (name, channel_id, project_id) ) config_id = cursor.lastrowid params.config_id = config_id conn.execute( """INSERT INTO params (config_id, channel_id, delay_data1, ENC_volume_data1, ENT_mx_right_data, ENT_mix_left_data) VALUES (?, ?, ?, ?, ?, ?)""", (params.config_id, params.channel_id, params.delay_data1, params.ENC_volume_data1, params.ENT_mx_right_data, params.ENT_mix_left_data) ) for filter_data in filters: filter_data.config_id = config_id cursor = conn.execute( """INSERT INTO filters (config_id, channel_id, filter_type, freq, q, gain, slope, enabled, position) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""", (filter_data.config_id, filter_data.channel_id, filter_data.filter_type, filter_data.freq, filter_data.q, filter_data.gain, filter_data.slope, filter_data.enabled, filter_data.position) ) filter_data.id = cursor.lastrowid conn.commit() return config_id def update_filter(self, filter_id: int, **kwargs): with self.get_connection() as conn: set_clause = ", ".join(f"{k} = ?" for k in kwargs.keys()) query = f"UPDATE filters SET {set_clause} WHERE id = ?" conn.execute(query, (*kwargs.values(), filter_id)) conn.commit() def update_params(self, config_id: int, params: ParamData): with self.get_connection() as conn: conn.execute( """UPDATE params SET delay_data1 = ?, ENC_volume_data1 = ?, ENT_mx_right_data = ?, ENT_mix_left_data = ? WHERE config_id = ?""", (params.delay_data1, params.ENC_volume_data1, params.ENT_mx_right_data, params.ENT_mix_left_data, config_id) ) conn.commit() def init_database(self): """初始化数据库，如果没有数据则创建默认配置""" configs = self.get_all_configs() if not configs: # 创建默认配置 default_params = ParamData( channel_id=1, delay_data1=0.0, ENC_volume_data1=0.0, ENT_mx_right_data=0.0, ENT_mix_left_data=0.0, config_id=None # 将在 save_config 中设置 ) # 创建一些默认滤波器 default_filters = [ FilterData( channel_id=1, filter_type="PEQ", freq=1000.0, q=1.0, gain=0.0, slope=12.0, position=0, config_id=None # 将在 save_config 中设置 ), FilterData( channel_id=1, filter_type="HPF", freq=20.0, q=0.707, gain=0.0, slope=12.0, position=1, config_id=None # 将在 save_config 中设置 ) ] # 保存默认配置 self.save_config("Default Config", 1, default_params, default_filters) def add_project(self, name: str, description: str) -> int: with self.get_connection() as conn: cursor = conn.execute( "INSERT INTO projects (name, description) VALUES (?, ?)", (name, description) ) conn.commit() return cursor.lastrowid def get_all_projects(self) -> List[ProjectData]: with self.get_connection() as conn: cursor = conn.execute( "SELECT id, name, description, created_at FROM projects" ) return [ProjectData(*row) for row in cursor.fetchall()] def update_project_name(self, project_id: int, new_name: str) -> bool: """更新项目名称""" try: with self.get_connection() as conn: cursor = conn.execute( "UPDATE projects SET name = ? WHERE id = ?", (new_name, project_id) ) conn.commit() return cursor.rowcount > 0 except sqlite3.IntegrityError: # 项目名已存在 return False def get_project_configs(self, project_id: int) -> List[ConfigData]: """获取项目关联的所有配置""" with self.get_connection() as conn: cursor = conn.execute( """SELECT id, name, channel_id, created_at, project_id FROM configs WHERE project_id = ?""", (project_id,) ) return [ConfigData(*row) for row in cursor.fetchall()]