# service/GraphStyleService.py import json from util.mysql_utils import mysql_client class GraphStyleService: @staticmethod def save_config(canvas_name: str, current_label: str, styles_dict: dict, group_name: str = None, config_id: int = None) -> bool: """ 保存图谱样式配置（修复版：防止自动保存导致的分组乱跑） """ # 2. 转换样式 JSON config_json = json.dumps(styles_dict, ensure_ascii=False) # 3. 【核心修改点】：区分 更新 还是 新建 if config_id: # --- 更新逻辑 --- # 如果带了 ID，我们要极其谨慎地处理 group_id，防止在自动保存时被误改 # A. 如果调用者明确传了 group_name，说明是“移动”或“初次保存到某组” if group_name and group_name.strip() != "": # 检查/创建 目标方案组 check_group_sql = "SELECT id FROM graph_style_groups WHERE group_name = %s LIMIT 1" existing_group = mysql_client.execute_query(check_group_sql, (group_name,)) if existing_group: target_group_id = existing_group[0]['id'] else: create_group_sql = "INSERT INTO graph_style_groups (group_name, is_active, is_default) VALUES (%s, %s, %s)" mysql_client.execute_update(create_group_sql, (group_name, False, False)) target_group_id = mysql_client.execute_query("SELECT LAST_INSERT_ID() as last_id")[0]['last_id'] # 执行带分组更新的 SQL sql = """ UPDATE graph_configs SET canvas_name = %s, current_label = %s, config_json = %s, group_id = %s WHERE id = %s """ affected_rows = mysql_client.execute_update(sql, (canvas_name, current_label, config_json, target_group_id, config_id)) else: # B. 如果没有传 group_name，说明是“实时自动保存”，严禁修改 group_id # 这样即使前端变量乱了，数据库的分组也不会变 sql = """ UPDATE graph_configs SET canvas_name = %s, current_label = %s, config_json = %s WHERE id = %s """ affected_rows = mysql_client.execute_update(sql, (canvas_name, current_label, config_json, config_id)) else: # --- 新建逻辑 --- # 新建时必须有组名，默认“默认方案” if not group_name or group_name.strip() == "": group_name = "默认方案" check_group_sql = "SELECT id FROM graph_style_groups WHERE group_name = %s LIMIT 1" existing_group = mysql_client.execute_query(check_group_sql, (group_name,)) if existing_group: target_group_id = existing_group[0]['id'] else: create_group_sql = "INSERT INTO graph_style_groups (group_name, is_active, is_default) VALUES (%s, %s, %s)" mysql_client.execute_update(create_group_sql, (group_name, False, False)) target_group_id = mysql_client.execute_query("SELECT LAST_INSERT_ID() as last_id")[0]['last_id'] sql = """ INSERT INTO graph_configs (canvas_name, current_label, config_json, group_id) VALUES (%s, %s, %s, %s) """ affected_rows = mysql_client.execute_update(sql, (canvas_name, current_label, config_json, target_group_id)) return affected_rows > 0 @staticmethod def get_grouped_configs() -> list: """ 获取嵌套结构的方案列表，按默认/激活状态排序 """ groups_sql = """ SELECT id, group_name, is_active, is_default FROM graph_style_groups ORDER BY is_default DESC, id ASC """ groups = mysql_client.execute_query(groups_sql) or [] configs_sql = "SELECT id, group_id, canvas_name, current_label, config_json, create_time FROM graph_configs" configs = mysql_client.execute_query(configs_sql) or [] for conf in configs: if conf.get('config_json'): try: conf['styles'] = json.loads(conf['config_json']) except: conf['styles'] = {} del conf['config_json'] if conf.get('create_time') and not isinstance(conf['create_time'], str): conf['create_time'] = conf['create_time'].strftime('%Y-%m-%d %H:%M:%S') result = [] for g in groups: g['is_active'] = bool(g['is_active']) g['is_default'] = bool(g['is_default']) g_children = [c for c in configs if c['group_id'] == g['id']] g['configs'] = g_children g['expanded'] = g['is_active'] result.append(g) return result @staticmethod def apply_group_all(group_id: int) -> bool: """应用全案：设置激活状态""" try: reset_sql = "UPDATE graph_style_groups SET is_active = %s" mysql_client.execute_update(reset_sql, (False,)) apply_sql = "UPDATE graph_style_groups SET is_active = %s WHERE id = %s" affected_rows = mysql_client.execute_update(apply_sql, (True, group_id)) return affected_rows > 0 except Exception as e: print(f"Apply group error: {e}") return False @staticmethod def set_default_group(group_id: int) -> bool: """设为系统初始默认方案""" try: reset_sql = "UPDATE graph_style_groups SET is_default = %s" mysql_client.execute_update(reset_sql, (False,)) set_sql = "UPDATE graph_style_groups SET is_default = %s WHERE id = %s" affected_rows = mysql_client.execute_update(set_sql, (True, group_id)) return affected_rows > 0 except Exception as e: print(f"Set default error: {e}") return False @staticmethod def get_all_configs() -> list: """获取扁平查询""" sql = """ SELECT c.id, c.group_id, c.canvas_name, c.current_label, c.config_json, c.create_time, g.is_active FROM graph_configs c LEFT JOIN graph_style_groups g ON c.group_id = g.id ORDER BY c.create_time DESC """ rows = mysql_client.execute_query(sql) if not rows: return [] for row in rows: if row.get('config_json'): try: row['styles'] = json.loads(row['config_json']) except: row['styles'] = {} del row['config_json'] if row.get('create_time') and not isinstance(row['create_time'], str): row['create_time'] = row['create_time'].strftime('%Y-%m-%d %H:%M:%S') row['is_active'] = bool(row.get('is_active', False)) return rows @staticmethod def delete_group(group_id: int) -> bool: """级联删除""" del_configs_sql = "DELETE FROM graph_configs WHERE group_id = %s" mysql_client.execute_update(del_configs_sql, (group_id,)) del_group_sql = "DELETE FROM graph_style_groups WHERE id = %s" affected_rows = mysql_client.execute_update(del_group_sql, (group_id,)) return affected_rows > 0 @staticmethod def delete_config(config_id: int) -> bool: """删除配置""" sql = "DELETE FROM graph_configs WHERE id = %s" affected_rows = mysql_client.execute_update(sql, (config_id,)) return affected_rows > 0 @staticmethod def batch_delete_configs(config_ids: list) -> int: """批量删除""" if not config_ids: return 0 try: clean_ids = [int(cid) for cid in config_ids if str(cid).isdigit()] except: return 0 if not clean_ids: return 0 placeholders = ', '.join(['%s'] * len(clean_ids)) sql = f"DELETE FROM graph_configs WHERE id IN ({placeholders})" return mysql_client.execute_update(sql, tuple(clean_ids)) @staticmethod def get_group_list() -> list: """获取方案列表""" sql = "SELECT id, group_name, is_active, is_default FROM graph_style_groups ORDER BY is_default DESC, id DESC" return mysql_client.execute_query(sql) or []