# 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) -> bool: """ 保存图谱样式配置（增强版：自动处理分组逻辑） """ # 1. 处理分组逻辑：查不到就建，查到了就用 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: # 如果不存在，新建一个组（is_active 和 is_default 默认为 false） 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)) get_id_sql = "SELECT LAST_INSERT_ID() as last_id" id_res = mysql_client.execute_query(get_id_sql) target_group_id = id_res[0]['last_id'] if id_res else 1 # 2. 转换样式 JSON config_json = json.dumps(styles_dict, ensure_ascii=False) # 3. 插入配置表 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: """ 核心优化：获取嵌套结构的方案列表 增加 is_active 和 is_default 字段支持，并按默认/激活状态排序 """ # 1. 查询所有方案组：让默认方案排在最上面 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 [] # 2. 查询所有配置项 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 [] # 3. 内存聚合 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: # 兼容 MySQL 布尔值转换 (某些驱动返回 0/1) 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: """ 核心新增：应用全案 逻辑：将该组设为 is_active=true，其余所有组设为 false """ try: # 1. 全部重置为非激活 reset_sql = "UPDATE graph_style_groups SET is_active = %s" mysql_client.execute_update(reset_sql, (False,)) # 2. 激活指定组 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: # 1. 全部取消默认 reset_sql = "UPDATE graph_style_groups SET is_default = %s" mysql_client.execute_update(reset_sql, (False,)) # 2. 设置新的默认项 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, create_time DESC" return mysql_client.execute_query(sql) or []