1790 lines
70 KiB
Python
1790 lines
70 KiB
Python
import os
|
||
import json
|
||
import csv
|
||
import tempfile
|
||
import shutil
|
||
import folder_paths
|
||
|
||
class ConfigManager:
|
||
def __init__(self):
|
||
# 插件目录
|
||
self.dir_path = os.path.dirname(os.path.abspath(__file__))
|
||
|
||
# 获取 ComfyUI 用户目录
|
||
try:
|
||
user_dir = folder_paths.get_user_directory()
|
||
if user_dir and os.path.isdir(user_dir):
|
||
# 使用 user/default/prompt-assistant 作为基础目录
|
||
self.base_dir = os.path.join(user_dir, "default", "prompt-assistant")
|
||
# self._log(f"使用用户配置目录: {self.base_dir}")
|
||
else:
|
||
# 回退到插件目录
|
||
self.base_dir = self.dir_path
|
||
self._log(f"回退到插件配置目录: {self.base_dir}")
|
||
except Exception as e:
|
||
# 异常处理，回退到插件目录
|
||
self.base_dir = self.dir_path
|
||
self._log(f"无法获取用户目录({str(e)})，使用插件配置目录")
|
||
|
||
# 定义各个子目录
|
||
self.config_dir = os.path.join(self.base_dir, "config")
|
||
self.rules_dir = os.path.join(self.base_dir, "rules")
|
||
self.tags_dir = os.path.join(self.base_dir, "tags")
|
||
|
||
# 确保目录存在
|
||
os.makedirs(self.config_dir, exist_ok=True)
|
||
os.makedirs(self.rules_dir, exist_ok=True)
|
||
os.makedirs(self.tags_dir, exist_ok=True)
|
||
|
||
# 配置文件路径（用户配置和选择）
|
||
self.config_path = os.path.join(self.config_dir, "config.json")
|
||
self.active_prompts_path = os.path.join(self.config_dir, "active_prompts.json")
|
||
self.tags_user_path = os.path.join(self.config_dir, "tags_user.json")
|
||
self.tags_selection_path = os.path.join(self.config_dir, "tags_selection.json")
|
||
|
||
# 规则文件路径（规则定义和模板）
|
||
self.system_prompts_path = os.path.join(self.rules_dir, "system_prompts.json")
|
||
self.kontext_presets_path = os.path.join(self.rules_dir, "kontext_presets.json")
|
||
|
||
# ---模板目录（插件内置）---
|
||
self.templates_dir = os.path.join(self.dir_path, "config")
|
||
|
||
# 存储模板版本号（用于版本比对）
|
||
self._template_versions = {}
|
||
|
||
# ---加载默认配置（从模板文件）---
|
||
self.default_config = self._load_template("config", {"version": "2.0", "model_services": []})
|
||
self.default_system_prompts = self._load_template("system_prompts", {})
|
||
self.default_kontext_presets = self._load_template("kontext_presets", {})
|
||
|
||
# ---简单默认配置（无需模板，直接定义）---
|
||
self.default_active_prompts = {
|
||
"expand": "expand_扩写-通用",
|
||
"vision_zh": "vision_zh_图像描述-Tag风格",
|
||
"vision_en": "vision_en_Detail_Caption"
|
||
}
|
||
self.default_user_tags = {"favorites": []}
|
||
|
||
# 默认标签选择
|
||
self.default_tags_selection = {"selected_file": "用户标签.csv"}
|
||
|
||
|
||
|
||
# 执行数据迁移和配置文件初始化
|
||
# migration_tool 统一处理：确保文件存在 -> CSV标签迁移 -> 旧版迁移 -> 增量更新
|
||
self._run_migrations()
|
||
|
||
# 验证并修复激活提示词（静默模式，仅异常时修复）
|
||
self.validate_and_fix_active_prompts()
|
||
|
||
# 验证并修复模型参数配置
|
||
self.validate_and_fix_model_params()
|
||
|
||
# --- 数据迁移 ---
|
||
def _run_migrations(self):
|
||
"""
|
||
执行数据迁移（按需调用，不影响性能）
|
||
仅在需要时才导入和运行迁移工具
|
||
"""
|
||
try:
|
||
from .utils.migration_tool import run_migrations
|
||
|
||
# 准备默认配置数据用于增量更新
|
||
default_configs = {
|
||
'config': self.default_config,
|
||
'system_prompts': self.default_system_prompts,
|
||
'active_prompts': self.default_active_prompts,
|
||
'tags_user': self.default_user_tags,
|
||
'kontext_presets': self.default_kontext_presets
|
||
}
|
||
|
||
# 运行迁移
|
||
results = run_migrations(
|
||
plugin_dir=self.dir_path,
|
||
user_base_dir=self.base_dir,
|
||
logger=self._log,
|
||
default_configs=default_configs
|
||
)
|
||
|
||
# 记录迁移结果
|
||
if results.get('tags_migration'):
|
||
self._log("[用户标签.csv] 数据迁移完成")
|
||
|
||
except Exception as e:
|
||
self._log(f"数据迁移失败: {str(e)}")
|
||
# 迁移失败不影响正常运行，仅记录日志
|
||
|
||
# --- 统一日志输出 ---
|
||
def _log(self, msg: str):
|
||
"""统一控制台日志前缀"""
|
||
from .utils.common import _ANSI_CLEAR_EOL
|
||
print(f"\r{_ANSI_CLEAR_EOL}✨ {msg}", flush=True)
|
||
|
||
# ---模板加载---
|
||
def _load_template(self, template_name: str, fallback: dict = None) -> dict:
|
||
"""
|
||
从模板文件加载默认配置
|
||
|
||
参数:
|
||
template_name: 模板名称（不含扩展名和_template后缀）
|
||
fallback: 加载失败时的回退默认值
|
||
|
||
返回:
|
||
配置字典（包含 __config_version 用于版本管理）
|
||
"""
|
||
template_path = os.path.join(self.templates_dir, f"{template_name}_template.json")
|
||
try:
|
||
with open(template_path, "r", encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
# 获取版本号并保存，用于后续比对
|
||
template_version = data.get("__config_version", "2.0")
|
||
self._template_versions[template_name] = template_version
|
||
return data
|
||
except Exception as e:
|
||
self._log(f"加载模板 {template_name} 失败: {str(e)}，使用回退值")
|
||
# 确保 fallback 也包含版本号
|
||
if fallback is None:
|
||
fallback = {}
|
||
# 如果 fallback 没有版本号，添加默认版本号
|
||
if "__config_version" not in fallback:
|
||
fallback = {"__config_version": "2.0", **fallback}
|
||
self._template_versions[template_name] = "2.0"
|
||
return fallback
|
||
|
||
def _get_config_version(self, config: dict) -> str:
|
||
"""
|
||
获取配置版本号（兼容新旧两种版本字段）
|
||
|
||
版本字段优先级:
|
||
1. __config_version (新版本字段，如 "2.0.0")
|
||
2. version (旧版本字段，如 "2.0" 或 "1.0")
|
||
3. 默认返回 "1.0"（无版本字段视为最旧版本）
|
||
|
||
返回:
|
||
版本字符串，如 "2.0.0"、"2.0" 或 "1.0"
|
||
"""
|
||
# 优先使用新版本字段
|
||
if "__config_version" in config:
|
||
return config["__config_version"]
|
||
# 兼容旧版本字段
|
||
return config.get("version", "1.0")
|
||
|
||
def _is_v2_config(self, config: dict) -> bool:
|
||
"""
|
||
检查配置是否为 v2.0 或更高版本
|
||
|
||
返回:
|
||
True 表示 v2.0 或更高版本 (1.9 也视为 v2 格式，用于增量测试)
|
||
"""
|
||
version = self._get_config_version(config)
|
||
try:
|
||
v_float = float(version)
|
||
return v_float >= 1.9
|
||
except ValueError:
|
||
# 如果不是数字（如 "2.0.0"），取主版本号比较
|
||
major_version = version.split(".")[0]
|
||
try:
|
||
return int(major_version) >= 2
|
||
except ValueError:
|
||
return False
|
||
|
||
# --- 注意：以下方法已迁移到 migration_tool.py ---
|
||
# - _apply_migrated_api_keys
|
||
# - _migrate_provider_to_service
|
||
# - _create_or_update_custom_service
|
||
# - _match_service_by_provider
|
||
# - _check_and_add_missing_services
|
||
# 配置文件的创建、迁移和增量更新统一由 migration_tool 处理
|
||
|
||
|
||
def _atomic_write_json(self, file_path: str, data: dict) -> bool:
|
||
"""
|
||
原子性写入 JSON 文件
|
||
|
||
采用"写临时文件 + 原子性重命名"的策略，确保文件写入的原子性：
|
||
- 如果写入成功，新文件会替换旧文件
|
||
- 如果写入失败或被中断，旧文件保持不变
|
||
|
||
参数:
|
||
file_path: 目标文件路径
|
||
data: 要保存的数据字典
|
||
|
||
返回:
|
||
bool: 保存成功返回 True，失败返回 False
|
||
"""
|
||
temp_fd = None
|
||
temp_path = None
|
||
|
||
try:
|
||
# ---步骤1：写入临时文件---
|
||
# 在同一目录下创建临时文件（确保在同一文件系统，rename 才是原子的）
|
||
temp_fd, temp_path = tempfile.mkstemp(
|
||
dir=os.path.dirname(file_path),
|
||
suffix='.tmp',
|
||
prefix='.tmp_'
|
||
)
|
||
|
||
# 完整写入新配置到临时文件
|
||
with os.fdopen(temp_fd, 'w', encoding='utf-8') as f:
|
||
json.dump(data, f, ensure_ascii=False, indent=2)
|
||
temp_fd = None # 文件已关闭，避免重复关闭
|
||
|
||
# ---步骤2：原子性替换---
|
||
# rename 操作是原子的，要么成功替换，要么失败不变
|
||
shutil.move(temp_path, file_path)
|
||
temp_path = None # 已移动，避免清理时删除
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
self._log(f"原子性写入 JSON 文件失败 [{os.path.basename(file_path)}]: {str(e)}")
|
||
return False
|
||
|
||
finally:
|
||
# 清理临时文件（如果写入失败）
|
||
if temp_fd is not None:
|
||
try:
|
||
os.close(temp_fd)
|
||
except:
|
||
pass
|
||
|
||
if temp_path is not None and os.path.exists(temp_path):
|
||
try:
|
||
os.unlink(temp_path)
|
||
except:
|
||
pass
|
||
|
||
def load_config(self):
|
||
"""加载配置文件"""
|
||
try:
|
||
with open(self.config_path, "r", encoding="utf-8") as f:
|
||
return json.load(f)
|
||
except Exception as e:
|
||
self._log(f"加载配置文件失败: {str(e)}")
|
||
return self.default_config
|
||
|
||
def save_config(self, config):
|
||
"""保存配置文件"""
|
||
return self._atomic_write_json(self.config_path, config)
|
||
|
||
def load_system_prompts(self):
|
||
"""加载系统提示词配置"""
|
||
try:
|
||
with open(self.system_prompts_path, "r", encoding="utf-8") as f:
|
||
return json.load(f)
|
||
except Exception as e:
|
||
self._log(f"加载系统提示词配置失败: {str(e)}")
|
||
return self.default_system_prompts
|
||
|
||
def save_system_prompts(self, system_prompts):
|
||
"""保存系统提示词配置"""
|
||
return self._atomic_write_json(self.system_prompts_path, system_prompts)
|
||
|
||
def load_active_prompts(self):
|
||
"""加载激活的提示词配置"""
|
||
try:
|
||
with open(self.active_prompts_path, "r", encoding="utf-8") as f:
|
||
return json.load(f)
|
||
except Exception as e:
|
||
self._log(f"加载激活的提示词配置失败: {str(e)}")
|
||
return self.default_active_prompts
|
||
|
||
def save_active_prompts(self, active_prompts):
|
||
"""保存激活的提示词配置"""
|
||
return self._atomic_write_json(self.active_prompts_path, active_prompts)
|
||
|
||
def load_user_tags(self):
|
||
"""加载用户标签配置"""
|
||
try:
|
||
with open(self.tags_user_path, "r", encoding="utf-8") as f:
|
||
return json.load(f)
|
||
except Exception as e:
|
||
self._log(f"加载用户标签配置失败: {str(e)}")
|
||
return self.default_user_tags
|
||
|
||
def save_user_tags(self, user_tags):
|
||
"""保存用户标签配置"""
|
||
return self._atomic_write_json(self.tags_user_path, user_tags)
|
||
|
||
def load_kontext_presets(self):
|
||
"""加载Kontext预设配置"""
|
||
try:
|
||
with open(self.kontext_presets_path, "r", encoding="utf-8") as f:
|
||
return json.load(f)
|
||
except Exception as e:
|
||
self._log(f"加载Kontext预设配置失败: {str(e)}")
|
||
return {}
|
||
|
||
def save_kontext_presets(self, kontext_presets):
|
||
"""保存Kontext预设配置"""
|
||
return self._atomic_write_json(self.kontext_presets_path, kontext_presets)
|
||
|
||
|
||
|
||
# --- 注意：ensure_tags_csv_exists 和 CSV 标签迁移已迁移到 migration_tool.py ---
|
||
|
||
|
||
|
||
def list_tags_files(self) -> list:
|
||
"""列出tags目录下所有CSV文件"""
|
||
try:
|
||
files = []
|
||
for filename in os.listdir(self.tags_dir):
|
||
if filename.endswith(".csv"):
|
||
files.append(filename)
|
||
return sorted(files)
|
||
except Exception as e:
|
||
self._log(f"列出标签文件失败: {str(e)}")
|
||
return []
|
||
|
||
def load_tags_csv(self, filename: str) -> dict:
|
||
"""加载CSV标签文件，返回嵌套字典结构"""
|
||
csv_path = os.path.join(self.tags_dir, filename)
|
||
if not os.path.exists(csv_path):
|
||
self._log(f"CSV文件不存在: {filename}")
|
||
return {}
|
||
|
||
# 尝试多种编码，优先尝试 utf-8-sig (Excel默认UTF-8)，然后是 gbk (Excel默认ANSI)，最后是 utf-8
|
||
encodings = ['utf-8-sig', 'gbk', 'gb18030', 'utf-8']
|
||
|
||
for encoding in encodings:
|
||
try:
|
||
result = {}
|
||
with open(csv_path, "r", encoding=encoding, newline="") as f:
|
||
reader = csv.reader(f)
|
||
try:
|
||
header = next(reader, None) # 跳过表头
|
||
except StopIteration:
|
||
return {} # 空文件
|
||
|
||
for row in reader:
|
||
# 过滤无效行
|
||
if not row or not any(cell.strip() for cell in row):
|
||
continue
|
||
|
||
# 至少需要两列：标签名, 标签值
|
||
if len(row) < 2:
|
||
continue
|
||
|
||
tag_name = row[0].strip()
|
||
tag_value = row[1].strip()
|
||
|
||
if not tag_name:
|
||
continue
|
||
|
||
# 分类路径：从第3列开始，过滤空值
|
||
categories = [c.strip() for c in row[2:] if c.strip()]
|
||
|
||
# 构建嵌套结构
|
||
current = result
|
||
for cat in categories:
|
||
if cat not in current or not isinstance(current[cat], dict):
|
||
current[cat] = {}
|
||
current = current[cat]
|
||
|
||
# 处理空分类占位符：只创建分类结构，不添加标签
|
||
if tag_name == "__empty__" or tag_name == "__placeholder__":
|
||
continue
|
||
|
||
# 添加标签
|
||
current[tag_name] = tag_value
|
||
|
||
return result
|
||
except UnicodeDecodeError:
|
||
continue
|
||
except Exception as e:
|
||
self._log(f"加载CSV标签失败 ({encoding}): {str(e)}")
|
||
continue
|
||
|
||
self._log(f"无法加载CSV文件: {filename}，尝试了所有编码均失败")
|
||
return {}
|
||
|
||
def save_tags_csv(self, filename: str, tags: dict) -> bool:
|
||
"""保存标签数据到CSV文件"""
|
||
csv_path = os.path.join(self.tags_dir, filename)
|
||
|
||
try:
|
||
rows = []
|
||
max_depth = 0
|
||
|
||
def extract_tags(obj, path: list):
|
||
nonlocal max_depth
|
||
# 确保 obj 是字典类型
|
||
if not isinstance(obj, dict):
|
||
return
|
||
|
||
# 如果是空分类（空字典），添加占位行
|
||
if len(obj) == 0 and path:
|
||
# 使用 __empty__ 作为占位符标记空分类
|
||
rows.append(["__empty__", ""] + path)
|
||
max_depth = max(max_depth, len(path))
|
||
return
|
||
|
||
for key, value in obj.items():
|
||
if isinstance(value, str):
|
||
rows.append([key, value] + path)
|
||
max_depth = max(max_depth, len(path))
|
||
elif isinstance(value, dict):
|
||
extract_tags(value, path + [key])
|
||
|
||
# 提取所有标签
|
||
extract_tags(tags, [])
|
||
|
||
if not rows:
|
||
self._log(f"保存CSV标签: 数据为空")
|
||
# 如果数据为空，写入只含表头的文件或保持现状？
|
||
# 通常为了防止误删，如果 tags 为空暂不操作或清空文件。
|
||
# 这里选择写入表头：
|
||
with open(csv_path, "w", encoding="utf-8-sig", newline="") as f:
|
||
writer = csv.writer(f)
|
||
writer.writerow(["标签名", "标签值"])
|
||
return True
|
||
|
||
# 动态构建表头
|
||
header = ["标签名", "标签值"]
|
||
for i in range(max_depth):
|
||
num_zh = ["一", "二", "三", "四", "五", "六", "七", "八", "九", "十"]
|
||
suffix = num_zh[i] if i < len(num_zh) else str(i + 1)
|
||
header.append(f"{suffix}级分类")
|
||
|
||
with open(csv_path, "w", encoding="utf-8-sig", newline="") as f:
|
||
writer = csv.writer(f)
|
||
writer.writerow(header)
|
||
for row in rows:
|
||
# 补齐长度以匹配表头
|
||
while len(row) < len(header):
|
||
row.append("")
|
||
# 确保 row 长度不超过表头（防御性）
|
||
writer.writerow(row[:len(header)])
|
||
|
||
return True
|
||
except Exception as e:
|
||
self._log(f"保存CSV标签失败: {str(e)}")
|
||
return False
|
||
|
||
def get_tags_selection(self) -> dict:
|
||
"""获取用户选择的标签文件"""
|
||
try:
|
||
if os.path.exists(self.tags_selection_path):
|
||
with open(self.tags_selection_path, "r", encoding="utf-8") as f:
|
||
return json.load(f)
|
||
return self.default_tags_selection
|
||
except Exception as e:
|
||
self._log(f"读取标签选择失败: {str(e)}")
|
||
return self.default_tags_selection
|
||
|
||
def save_tags_selection(self, selection: dict) -> bool:
|
||
"""保存用户选择的标签文件"""
|
||
try:
|
||
with open(self.tags_selection_path, "w", encoding="utf-8") as f:
|
||
json.dump(selection, f, ensure_ascii=False, indent=2)
|
||
return True
|
||
except Exception as e:
|
||
self._log(f"保存标签选择失败: {str(e)}")
|
||
return False
|
||
|
||
def get_favorites(self) -> dict:
|
||
"""获取收藏列表"""
|
||
user_tags = self.load_user_tags()
|
||
favorites = user_tags.get("favorites", {})
|
||
|
||
# 兼容性处理：如果是列表，转换为字典
|
||
if isinstance(favorites, list):
|
||
new_favorites = {}
|
||
for item in favorites:
|
||
if isinstance(item, str):
|
||
new_favorites[item] = item
|
||
elif isinstance(item, dict):
|
||
name = item.get("name", item.get("value"))
|
||
value = item.get("value")
|
||
if name and value:
|
||
new_favorites[name] = value
|
||
return new_favorites
|
||
|
||
return favorites
|
||
|
||
def add_favorite(self, tag_value: str, tag_name: str = None, category: str = "默认") -> bool:
|
||
"""添加收藏"""
|
||
try:
|
||
user_tags = self.load_user_tags()
|
||
favorites = user_tags.get("favorites", {})
|
||
|
||
# 兼容性迁移：如果是一维字典 {name: value}，无需强制迁移，但新添加的会放入 category
|
||
# 如果是列表，先迁移为字典
|
||
if isinstance(favorites, list):
|
||
favorites = self.get_favorites()
|
||
|
||
name = tag_name if tag_name else tag_value
|
||
|
||
# 使用嵌套结构 {分类: {名称: 值}}
|
||
if category not in favorites:
|
||
# 检查是否存在旧的平铺结构，如果有，且category是默认，可能混杂
|
||
# 这里简单处理：如果 favorites 只有键值对且都不是字典，说明是旧版平铺
|
||
# 但为了不破坏旧数据，我们在顶层只存储分类字典
|
||
# 如果 favorites 中已有非字典的值，说明是旧版平铺结构 {name: value}
|
||
# 我们将它们移动到 "默认" 分类
|
||
has_legacy = any(not isinstance(v, dict) for v in favorites.values())
|
||
if has_legacy:
|
||
legacy_items = {k: v for k, v in favorites.items() if not isinstance(v, dict)}
|
||
# 清除旧项
|
||
for k in legacy_items:
|
||
del favorites[k]
|
||
# 初始化默认分类
|
||
if "默认" not in favorites:
|
||
favorites["默认"] = {}
|
||
favorites["默认"].update(legacy_items)
|
||
|
||
if category not in favorites:
|
||
favorites[category] = {}
|
||
|
||
# 如果 favorites[category] 不是字典（防御性编程），初始化为字典
|
||
if not isinstance(favorites.get(category), dict):
|
||
favorites[category] = {}
|
||
|
||
favorites[category][name] = tag_value
|
||
|
||
user_tags["favorites"] = favorites
|
||
return self.save_user_tags(user_tags)
|
||
except Exception as e:
|
||
self._log(f"添加收藏失败: {str(e)}")
|
||
return False
|
||
|
||
def remove_favorite(self, tag_value: str, category: str = None) -> bool:
|
||
"""移除收藏"""
|
||
try:
|
||
user_tags = self.load_user_tags()
|
||
favorites = user_tags.get("favorites", {})
|
||
|
||
# 兼容性迁移
|
||
if isinstance(favorites, list):
|
||
favorites = self.get_favorites()
|
||
|
||
removed = False
|
||
|
||
# 如果指定了分类，只在指定分类中删除
|
||
if category:
|
||
# 尝试直接匹配分类（完全匹配）
|
||
target_categories = [category]
|
||
|
||
# 如果没找到，尝试模糊匹配（处理文件名后缀差异）
|
||
if category not in favorites:
|
||
# 比如 category 是 "foo"，favorites里有 "foo.csv" 或相反
|
||
# 但通常 favorites 里的 key 已经是去后缀的
|
||
pass
|
||
|
||
for cat in target_categories:
|
||
if cat in favorites and isinstance(favorites[cat], dict):
|
||
# 根据值删除
|
||
keys_to_remove = [k for k, v in favorites[cat].items() if v == tag_value]
|
||
for k in keys_to_remove:
|
||
del favorites[cat][k]
|
||
removed = True
|
||
|
||
# 如果该分类空了，是否删除分类键？暂时保留
|
||
else:
|
||
# 未指定分类，递归全部删除（旧逻辑）
|
||
# 如果是旧版平铺结构
|
||
if any(not isinstance(v, dict) for v in favorites.values()):
|
||
keys_to_remove = [k for k, v in favorites.items() if not isinstance(v, dict) and v == tag_value]
|
||
for k in keys_to_remove:
|
||
del favorites[k]
|
||
removed = True
|
||
|
||
# 如果是新版嵌套结构
|
||
for cat, items in favorites.items():
|
||
if isinstance(items, dict):
|
||
keys_to_remove = [k for k, v in items.items() if v == tag_value]
|
||
for k in keys_to_remove:
|
||
del items[k]
|
||
removed = True
|
||
|
||
if removed:
|
||
user_tags["favorites"] = favorites
|
||
return self.save_user_tags(user_tags)
|
||
|
||
return True
|
||
except Exception as e:
|
||
self._log(f"移除收藏失败: {str(e)}")
|
||
return False
|
||
|
||
def get_system_prompts(self):
|
||
"""获取系统提示词配置 (合并提示词定义和激活状态)"""
|
||
system_prompts = self.load_system_prompts()
|
||
active_prompts = self.load_active_prompts()
|
||
system_prompts['active_prompts'] = active_prompts
|
||
return system_prompts
|
||
|
||
def update_system_prompts(self, system_prompts):
|
||
"""更新系统提示词配置 (仅更新提示词定义)"""
|
||
prompts_to_save = system_prompts.copy()
|
||
if 'active_prompts' in prompts_to_save:
|
||
del prompts_to_save['active_prompts']
|
||
return self.save_system_prompts(prompts_to_save)
|
||
|
||
def update_active_prompts(self, active_prompts):
|
||
"""更新所有激活的提示词"""
|
||
return self.save_active_prompts(active_prompts)
|
||
|
||
def update_active_prompt(self, prompt_type, prompt_id):
|
||
"""更新单个激活的提示词"""
|
||
active_prompts = self.load_active_prompts()
|
||
active_prompts[prompt_type] = prompt_id
|
||
return self.save_active_prompts(active_prompts)
|
||
|
||
def get_baidu_translate_config(self):
|
||
"""获取百度翻译配置"""
|
||
config = self.load_config()
|
||
return config.get("baidu_translate", self.default_config["baidu_translate"])
|
||
|
||
def get_llm_config(self):
|
||
"""获取LLM配置"""
|
||
config = self.load_config()
|
||
current_service_info = config.get('current_services', {}).get('llm')
|
||
|
||
# 适配新旧格式:支持字符串(旧)和字典(新)
|
||
if isinstance(current_service_info, str):
|
||
# 旧格式: "service_id"
|
||
current_service_id = current_service_info
|
||
current_model_name = None
|
||
elif isinstance(current_service_info, dict):
|
||
# 新格式: {"service": "service_id", "model": "model_name"}
|
||
current_service_id = current_service_info.get('service')
|
||
current_model_name = current_service_info.get('model')
|
||
else:
|
||
# 未设置
|
||
current_service_id = None
|
||
current_model_name = None
|
||
|
||
if not current_service_id:
|
||
# 没有选中的服务，返回默认结构
|
||
return self._get_empty_llm_config()
|
||
|
||
# 查找对应的服务
|
||
service = self._get_service_by_id(current_service_id)
|
||
if not service:
|
||
return self._get_empty_llm_config()
|
||
|
||
# 获取LLM模型列表
|
||
llm_models = service.get('llm_models', [])
|
||
|
||
# 如果指定了模型名称,尝试查找
|
||
target_model = None
|
||
if current_model_name:
|
||
target_model = next((m for m in llm_models if m.get('name') == current_model_name), None)
|
||
|
||
# 如果未找到指定模型,使用默认模型或第一个模型
|
||
if not target_model:
|
||
target_model = next((m for m in llm_models if m.get('is_default')),
|
||
llm_models[0] if llm_models else None)
|
||
|
||
if not target_model:
|
||
return self._get_empty_llm_config()
|
||
|
||
# 直接获取API Key（明文存储）
|
||
api_key = service.get('api_key', '')
|
||
|
||
# 返回配置
|
||
return {
|
||
"provider": service.get('id', ''), # 使用service_id作为provider
|
||
"model": target_model.get('name', ''),
|
||
"base_url": service.get('base_url', ''),
|
||
"api_key": api_key,
|
||
"temperature": target_model.get('temperature', 0.7),
|
||
"max_tokens": target_model.get('max_tokens', 1000),
|
||
"top_p": target_model.get('top_p', 0.9),
|
||
"auto_unload": service.get('auto_unload', True) if service.get('type') == 'ollama' else None,
|
||
"providers": {} # v2.0中不再使用此字段
|
||
}
|
||
|
||
|
||
def _get_empty_llm_config(self):
|
||
"""返回空的LLM配置"""
|
||
return {
|
||
"provider": "",
|
||
"model": "",
|
||
"base_url": "",
|
||
"api_key": "",
|
||
"temperature": 0.7,
|
||
"max_tokens": 1000,
|
||
"top_p": 0.9,
|
||
"providers": {}
|
||
}
|
||
|
||
def _get_service_by_id(self, service_id: str) -> dict:
|
||
"""根据ID获取服务配置"""
|
||
config = self.load_config()
|
||
services = config.get('model_services', [])
|
||
for service in services:
|
||
if service.get('id') == service_id:
|
||
return service
|
||
return None
|
||
|
||
def get_vision_config(self):
|
||
"""获取视觉模型配置"""
|
||
config = self.load_config()
|
||
current_service_info = config.get('current_services', {}).get('vlm')
|
||
|
||
# 适配新旧格式:支持字符串(旧)和字典(新)
|
||
if isinstance(current_service_info, str):
|
||
# 旧格式: "service_id"
|
||
current_service_id = current_service_info
|
||
current_model_name = None
|
||
elif isinstance(current_service_info, dict):
|
||
# 新格式: {"service": "service_id", "model": "model_name"}
|
||
current_service_id = current_service_info.get('service')
|
||
current_model_name = current_service_info.get('model')
|
||
else:
|
||
# 未设置
|
||
current_service_id = None
|
||
current_model_name = None
|
||
|
||
if not current_service_id:
|
||
# 没有选中的服务，返回默认结构
|
||
return self._get_empty_vision_config()
|
||
|
||
# 查找对应的服务
|
||
service = self._get_service_by_id(current_service_id)
|
||
if not service:
|
||
return self._get_empty_vision_config()
|
||
|
||
# 获取VLM模型列表
|
||
vlm_models = service.get('vlm_models', [])
|
||
|
||
# 如果指定了模型名称,尝试查找
|
||
target_model = None
|
||
if current_model_name:
|
||
target_model = next((m for m in vlm_models if m.get('name') == current_model_name), None)
|
||
|
||
# 如果未找到指定模型,使用默认模型或第一个模型
|
||
if not target_model:
|
||
target_model = next((m for m in vlm_models if m.get('is_default')),
|
||
vlm_models[0] if vlm_models else None)
|
||
|
||
if not target_model:
|
||
return self._get_empty_vision_config()
|
||
|
||
# 直接获取API Key（明文存储）
|
||
api_key = service.get('api_key', '')
|
||
|
||
# 返回配置
|
||
return {
|
||
"provider": service.get('id', ''), # 使用service_id作为provider
|
||
"model": target_model.get('name', ''),
|
||
"base_url": service.get('base_url', ''),
|
||
"api_key": api_key,
|
||
"temperature": target_model.get('temperature', 0.7),
|
||
"max_tokens": target_model.get('max_tokens', 4096),
|
||
"top_p": target_model.get('top_p', 0.9),
|
||
"auto_unload": service.get('auto_unload', True) if service.get('type') == 'ollama' else None,
|
||
"providers": {} # v2.0中不再使用此字段
|
||
}
|
||
|
||
def _get_empty_vision_config(self):
|
||
"""返回空的视觉模型配置"""
|
||
return {
|
||
"provider": "",
|
||
"model": "",
|
||
"base_url": "",
|
||
"api_key": "",
|
||
"temperature": 0.7,
|
||
"max_tokens": 4096,
|
||
"top_p": 0.9,
|
||
"providers": {}
|
||
}
|
||
|
||
def get_translate_config(self):
|
||
"""获取翻译服务配置（支持百度翻译和LLM翻译）"""
|
||
config = self.load_config()
|
||
current_service_info = config.get('current_services', {}).get('translate')
|
||
|
||
# 适配新旧格式:支持字符串(旧)和字典(新)
|
||
if isinstance(current_service_info, str):
|
||
# 旧格式: "service_id"
|
||
current_service_id = current_service_info
|
||
current_model_name = None
|
||
elif isinstance(current_service_info, dict):
|
||
# 新格式: {"service": "service_id", "model": "model_name"}
|
||
current_service_id = current_service_info.get('service')
|
||
current_model_name = current_service_info.get('model')
|
||
else:
|
||
# 未设置，默认使用百度翻译
|
||
current_service_id = 'baidu'
|
||
current_model_name = None
|
||
|
||
# 百度翻译特殊处理（使用独立的baidu_translate配置）
|
||
if current_service_id == 'baidu':
|
||
baidu_config = self.get_baidu_translate_config()
|
||
return {
|
||
"provider": "baidu",
|
||
"model": "",
|
||
"base_url": "",
|
||
"api_key": baidu_config.get('app_id', ''),
|
||
"secret_key": baidu_config.get('secret_key', ''),
|
||
"temperature": 0.7,
|
||
"max_tokens": 1000,
|
||
"top_p": 0.9,
|
||
"providers": {}
|
||
}
|
||
|
||
# 查找对应的LLM服务
|
||
service = self._get_service_by_id(current_service_id)
|
||
if not service:
|
||
# 服务不存在，回退到百度翻译
|
||
baidu_config = self.get_baidu_translate_config()
|
||
return {
|
||
"provider": "baidu",
|
||
"model": "",
|
||
"base_url": "",
|
||
"api_key": baidu_config.get('app_id', ''),
|
||
"secret_key": baidu_config.get('secret_key', ''),
|
||
"temperature": 0.7,
|
||
"max_tokens": 1000,
|
||
"top_p": 0.9,
|
||
"providers": {}
|
||
}
|
||
|
||
# 获取LLM模型列表
|
||
llm_models = service.get('llm_models', [])
|
||
|
||
# 如果指定了模型名称,尝试查找
|
||
target_model = None
|
||
if current_model_name:
|
||
target_model = next((m for m in llm_models if m.get('name') == current_model_name), None)
|
||
|
||
# 如果未找到指定模型,使用默认模型或第一个模型
|
||
if not target_model:
|
||
target_model = next((m for m in llm_models if m.get('is_default')),
|
||
llm_models[0] if llm_models else None)
|
||
|
||
if not target_model:
|
||
# 没有可用模型，回退到百度翻译
|
||
baidu_config = self.get_baidu_translate_config()
|
||
return {
|
||
"provider": "baidu",
|
||
"model": "",
|
||
"base_url": "",
|
||
"api_key": baidu_config.get('app_id', ''),
|
||
"secret_key": baidu_config.get('secret_key', ''),
|
||
"temperature": 0.7,
|
||
"max_tokens": 1000,
|
||
"top_p": 0.9,
|
||
"providers": {}
|
||
}
|
||
|
||
# 返回LLM翻译配置
|
||
api_key = service.get('api_key', '')
|
||
return {
|
||
"provider": service.get('id', ''),
|
||
"model": target_model.get('name', ''),
|
||
"base_url": service.get('base_url', ''),
|
||
"api_key": api_key,
|
||
"temperature": target_model.get('temperature', 0.7),
|
||
"max_tokens": target_model.get('max_tokens', 1000),
|
||
"top_p": target_model.get('top_p', 0.9),
|
||
"auto_unload": service.get('auto_unload', True) if service.get('type') == 'ollama' else None,
|
||
"providers": {}
|
||
}
|
||
|
||
def get_settings(self):
|
||
"""获取ComfyUI用户设置（从设置文件读取）"""
|
||
try:
|
||
# ComfyUI的设置文件通常位于 user/default/comfy.settings.json
|
||
# 需要找到ComfyUI的根目录
|
||
import sys
|
||
|
||
# 尝试从多个可能的路径查找设置文件
|
||
possible_paths = []
|
||
|
||
# 方法1: 通过当前文件路径向上查找
|
||
current_dir = os.path.dirname(os.path.abspath(__file__))
|
||
# custom_nodes/comfyui_prompt_assistant -> custom_nodes -> ComfyUI
|
||
comfyui_root = os.path.dirname(os.path.dirname(current_dir))
|
||
possible_paths.append(os.path.join(comfyui_root, "user", "default", "comfy.settings.json"))
|
||
|
||
# 方法2: 通过sys.path查找
|
||
for path in sys.path:
|
||
if 'ComfyUI' in path:
|
||
possible_paths.append(os.path.join(path, "user", "default", "comfy.settings.json"))
|
||
|
||
# 尝试读取设置文件
|
||
for settings_path in possible_paths:
|
||
if os.path.exists(settings_path):
|
||
try:
|
||
with open(settings_path, 'r', encoding='utf-8') as f:
|
||
settings_data = json.load(f)
|
||
# 返回设置数据
|
||
return settings_data
|
||
except Exception as e:
|
||
self._log(f"读取设置文件失败: {settings_path}, 错误: {str(e)}")
|
||
continue
|
||
|
||
# 如果都找不到，返回空字典
|
||
return {}
|
||
|
||
except Exception as e:
|
||
# 如果无法获取，返回空字典
|
||
self._log(f"获取用户设置失败: {str(e)}")
|
||
return {}
|
||
|
||
def update_baidu_translate_config(self, app_id=None, secret_key=None):
|
||
"""更新百度翻译配置"""
|
||
config = self.load_config()
|
||
if "baidu_translate" not in config:
|
||
config["baidu_translate"] = {}
|
||
|
||
# 仅更新提供的参数
|
||
if app_id is not None:
|
||
config["baidu_translate"]["app_id"] = app_id
|
||
if secret_key is not None:
|
||
config["baidu_translate"]["secret_key"] = secret_key
|
||
|
||
return self.save_config(config)
|
||
|
||
|
||
|
||
|
||
# --- 注意：validate_and_fix_system_prompts 已迁移到 migration_tool.py ---
|
||
# 系统提示词的验证和补全由 migration_tool 的增量更新逻辑统一处理
|
||
|
||
|
||
def validate_and_fix_active_prompts(self):
|
||
"""
|
||
验证激活提示词是否存在，如果不存在则修复
|
||
|
||
注意：此方法只修复 active_prompts.json（切换到存在的提示词）
|
||
不会恢复 system_prompts.json 中被删除的内容（尊重用户的删除操作）
|
||
"""
|
||
try:
|
||
system_prompts = self.load_system_prompts()
|
||
active_prompts = self.load_active_prompts()
|
||
|
||
# 标记是否需要更新激活提示词
|
||
modified = False
|
||
|
||
# 检查并修复扩写提示词
|
||
if "expand" in active_prompts:
|
||
expand_id = active_prompts["expand"]
|
||
expand_prompts = system_prompts.get("expand_prompts", {})
|
||
|
||
if expand_id not in expand_prompts:
|
||
# 激活的提示词不存在，切换到第一个可用的
|
||
if expand_prompts:
|
||
first_expand_id = next(iter(expand_prompts))
|
||
active_prompts["expand"] = first_expand_id
|
||
self._log(f"激活的扩写提示词 '{expand_id}' 不存在，已切换到 '{first_expand_id}'")
|
||
modified = True
|
||
else:
|
||
# 没有可用的扩写提示词，清空激活
|
||
active_prompts["expand"] = ""
|
||
self._log(f"警告：没有可用的扩写提示词")
|
||
modified = True
|
||
|
||
# 检查并修复中文反推提示词
|
||
if "vision_zh" in active_prompts:
|
||
vision_zh_id = active_prompts["vision_zh"]
|
||
vision_prompts = system_prompts.get("vision_prompts", {})
|
||
zh_prompts = {k: v for k, v in vision_prompts.items() if k.startswith("vision_zh_")}
|
||
|
||
if vision_zh_id not in vision_prompts:
|
||
if zh_prompts:
|
||
first_id = next(iter(zh_prompts))
|
||
active_prompts["vision_zh"] = first_id
|
||
self._log(f"激活的中文反推提示词 '{vision_zh_id}' 不存在，已切换到 '{first_id}'")
|
||
modified = True
|
||
else:
|
||
active_prompts["vision_zh"] = ""
|
||
self._log(f"警告：没有可用的中文反推提示词")
|
||
modified = True
|
||
|
||
# 检查并修复英文反推提示词
|
||
if "vision_en" in active_prompts:
|
||
vision_en_id = active_prompts["vision_en"]
|
||
vision_prompts = system_prompts.get("vision_prompts", {})
|
||
en_prompts = {k: v for k, v in vision_prompts.items() if k.startswith("vision_en_")}
|
||
|
||
if vision_en_id not in vision_prompts:
|
||
if en_prompts:
|
||
first_id = next(iter(en_prompts))
|
||
active_prompts["vision_en"] = first_id
|
||
self._log(f"激活的英文反推提示词 '{vision_en_id}' 不存在，已切换到 '{first_id}'")
|
||
modified = True
|
||
else:
|
||
active_prompts["vision_en"] = ""
|
||
self._log(f"警告：没有可用的英文反推提示词")
|
||
modified = True
|
||
|
||
# 如果需要更新，保存修复后的激活提示词
|
||
if modified:
|
||
self.save_active_prompts(active_prompts)
|
||
self._log("已完成激活提示词的验证和修复")
|
||
|
||
except Exception as e:
|
||
self._log(f"验证激活提示词异常: {str(e)}")
|
||
|
||
|
||
|
||
def validate_and_fix_model_params(self):
|
||
"""
|
||
验证并修复模型参数配置
|
||
注意: v2.0版本中，模型参数直接存储在 model_services 数组的模型对象中，
|
||
这个方法主要用于确保配置文件存在和格式正确
|
||
"""
|
||
try:
|
||
config = self.load_config()
|
||
|
||
# 确保是 v2.0 格式
|
||
if not self._is_v2_config(config):
|
||
self._log("[config.json] 警告: 检测到旧版本配置，请手动创建新的配置文件或使用默认配置")
|
||
return
|
||
|
||
# v2.0 格式中，参数已经在各个服务的模型列表中，无需额外验证
|
||
# 如果需要补全缺失的服务或模型参数，应该在服务商管理API中处理
|
||
|
||
except Exception as e:
|
||
self._log(f"[config.json] 验证模型参数配置时出错: {str(e)}")
|
||
|
||
|
||
# --- API Key 安全相关方法（方案A）---
|
||
|
||
@staticmethod
|
||
def mask_api_key(api_key: str) -> str:
|
||
"""
|
||
掩码API Key，只显示首尾部分
|
||
用于前端安全显示，防止API Key在Network中明文可见
|
||
|
||
参数:
|
||
api_key: 明文API Key
|
||
|
||
返回:
|
||
str: 掩码后的API Key
|
||
|
||
示例:
|
||
- sk-abc123xyz789 -> sk-abc***xyz789
|
||
- 短Key (< 8字符) -> ***
|
||
- 空字符串 -> ""
|
||
"""
|
||
if not api_key:
|
||
return ""
|
||
if len(api_key) < 8:
|
||
return "***"
|
||
# 显示前6个字符和后4个字符
|
||
return f"{api_key[:6]}***{api_key[-4:]}"
|
||
|
||
def get_llm_config_masked(self):
|
||
"""
|
||
获取LLM配置（API Key掩码版本）
|
||
用于前端显示，不暴露完整API Key
|
||
|
||
返回:
|
||
Dict: LLM配置，api_key字段被掩码
|
||
"""
|
||
config = self.get_llm_config()
|
||
|
||
if 'api_key' in config:
|
||
# 掩码API Key
|
||
config['api_key_masked'] = self.mask_api_key(config['api_key'])
|
||
config['api_key_exists'] = bool(config['api_key'])
|
||
# 移除明文API Key
|
||
del config['api_key']
|
||
|
||
# 处理所有providers的API Key
|
||
if 'providers' in config:
|
||
for provider_name, provider_config in config['providers'].items():
|
||
if 'api_key' in provider_config:
|
||
provider_config['api_key_masked'] = self.mask_api_key(provider_config['api_key'])
|
||
provider_config['api_key_exists'] = bool(provider_config['api_key'])
|
||
del provider_config['api_key']
|
||
|
||
return config
|
||
|
||
def get_vision_config_masked(self):
|
||
"""
|
||
获取视觉模型配置（API Key掩码版本）
|
||
用于前端显示，不暴露完整API Key
|
||
|
||
返回:
|
||
Dict: 视觉模型配置，api_key字段被掩码
|
||
"""
|
||
config = self.get_vision_config()
|
||
|
||
if 'api_key' in config:
|
||
# 掩码API Key
|
||
config['api_key_masked'] = self.mask_api_key(config['api_key'])
|
||
config['api_key_exists'] = bool(config['api_key'])
|
||
# 移除明文API Key
|
||
del config['api_key']
|
||
|
||
# 处理所有providers的API Key
|
||
if 'providers' in config:
|
||
for provider_name, provider_config in config['providers'].items():
|
||
if 'api_key' in provider_config:
|
||
provider_config['api_key_masked'] = self.mask_api_key(provider_config['api_key'])
|
||
provider_config['api_key_exists'] = bool(provider_config['api_key'])
|
||
del provider_config['api_key']
|
||
|
||
return config
|
||
|
||
# --- 服务商管理方法（CRUD）---
|
||
|
||
def get_all_services(self):
|
||
"""
|
||
获取所有服务商列表
|
||
|
||
返回:
|
||
List[Dict]: 服务商列表
|
||
"""
|
||
config = self.load_config()
|
||
|
||
if self._is_v2_config(config):
|
||
return config.get('model_services', [])
|
||
else:
|
||
# v1.0不支持此功能
|
||
return []
|
||
|
||
def get_service(self, service_id: str):
|
||
"""
|
||
获取指定服务商的完整配置
|
||
|
||
参数:
|
||
service_id: 服务商ID
|
||
|
||
返回:
|
||
Dict: 服务商配置，不存在返回None
|
||
"""
|
||
return self._get_service_by_id(service_id)
|
||
|
||
def create_service(self, service_type: str, name: str = "", base_url: str = "",
|
||
api_key: str = "", description: str = ""):
|
||
"""
|
||
创建新的服务商
|
||
|
||
参数:
|
||
service_type: 服务类型 ('openai_compatible' 或 'ollama')
|
||
name: 服务商名称（如果为空，自动生成）
|
||
base_url: Base URL
|
||
api_key: API Key（明文存储）
|
||
description: 描述
|
||
|
||
返回:
|
||
str: 新创建的service_id，失败返回None
|
||
"""
|
||
try:
|
||
config = self.load_config()
|
||
|
||
if not self._is_v2_config(config):
|
||
self._log("创建服务商失败: 配置版本过低，请先迁移到v2.0")
|
||
return None
|
||
|
||
# 获取现有服务商列表
|
||
current_services = config.get('model_services', [])
|
||
|
||
# 生成服务商ID和名称
|
||
service_id, auto_name = self._generate_service_id_and_name(service_type, current_services)
|
||
|
||
# 如果用户没有提供名称，使用自动生成的名称
|
||
if not name:
|
||
name = auto_name
|
||
|
||
# 创建服务配置
|
||
new_service = {
|
||
"id": service_id,
|
||
"type": service_type,
|
||
"name": name,
|
||
"description": description,
|
||
"base_url": base_url,
|
||
"api_key": api_key or "",
|
||
"disable_thinking": True,
|
||
"enable_advanced_params": True,
|
||
"filter_thinking_output": True,
|
||
"llm_models": [],
|
||
"vlm_models": []
|
||
}
|
||
|
||
# Ollama特有配置
|
||
if service_type == "ollama":
|
||
new_service["auto_unload"] = True
|
||
|
||
# 添加到配置
|
||
if 'model_services' not in config:
|
||
config['model_services'] = []
|
||
|
||
config['model_services'].append(new_service)
|
||
|
||
# 保存配置
|
||
if self.save_config(config):
|
||
self._log(f"成功创建服务商: {name} (ID: {service_id})")
|
||
return service_id
|
||
else:
|
||
self._log(f"保存服务商配置失败: {name}")
|
||
return None
|
||
|
||
except Exception as e:
|
||
self._log(f"创建服务商异常: {str(e)}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return None
|
||
|
||
def _generate_service_id_and_name(self, service_type: str, current_services: list) -> tuple:
|
||
"""
|
||
生成服务商ID和默认名称
|
||
|
||
参数:
|
||
service_type: 服务类型
|
||
current_services: 现有服务商列表
|
||
|
||
返回:
|
||
tuple: (service_id, default_name)
|
||
"""
|
||
import random
|
||
|
||
# 类型映射
|
||
type_map = {
|
||
"ollama": {
|
||
"name_prefix": "Ollama服务",
|
||
"id_prefix": "ollama"
|
||
},
|
||
"openai_compatible": {
|
||
"name_prefix": "通用服务",
|
||
"id_prefix": "service"
|
||
}
|
||
}
|
||
|
||
# 获取类型配置
|
||
type_config = type_map.get(service_type, {
|
||
"name_prefix": "新服务",
|
||
"id_prefix": service_type
|
||
})
|
||
|
||
name_prefix = type_config["name_prefix"]
|
||
id_prefix = type_config["id_prefix"]
|
||
|
||
# 收集已使用的编号
|
||
existing_numbers = set()
|
||
for service in current_services:
|
||
sid = service.get('id', '')
|
||
# 匹配格式：{id_prefix}_{数字}
|
||
if sid.startswith(f"{id_prefix}_"):
|
||
try:
|
||
num_str = sid.split('_')[-1]
|
||
if num_str.isdigit():
|
||
existing_numbers.add(int(num_str))
|
||
except:
|
||
pass
|
||
|
||
# 生成随机三位数（100-999），最多尝试100次
|
||
max_attempts = 100
|
||
for _ in range(max_attempts):
|
||
random_number = random.randint(100, 999)
|
||
if random_number not in existing_numbers:
|
||
break
|
||
else:
|
||
# 如果100次都重复，使用更大的随机数（4位数）
|
||
random_number = random.randint(1000, 9999)
|
||
while random_number in existing_numbers:
|
||
random_number = random.randint(1000, 9999)
|
||
|
||
# 生成ID和名称
|
||
service_id = f"{id_prefix}_{random_number}"
|
||
default_name = f"{name_prefix}-{random_number}"
|
||
|
||
return service_id, default_name
|
||
|
||
def delete_service(self, service_id: str):
|
||
"""
|
||
删除服务商
|
||
|
||
参数:
|
||
service_id: 服务商ID
|
||
|
||
返回:
|
||
bool: 成功返回True
|
||
"""
|
||
try:
|
||
config = self.load_config()
|
||
|
||
if not self._is_v2_config(config):
|
||
self._log("删除服务商失败: 配置版本过低")
|
||
return False
|
||
|
||
services = config.get('model_services', [])
|
||
|
||
# 查找并删除服务
|
||
original_length = len(services)
|
||
config['model_services'] = [s for s in services if s.get('id') != service_id]
|
||
|
||
if len(config['model_services']) == original_length:
|
||
self._log(f"删除服务商失败: 服务商不存在 (ID: {service_id})")
|
||
return False
|
||
|
||
# 如果删除的是当前服务，清除current_services引用
|
||
current_services = config.get('current_services', {})
|
||
if current_services.get('llm') == service_id:
|
||
current_services['llm'] = None
|
||
if current_services.get('vlm') == service_id:
|
||
current_services['vlm'] = None
|
||
if current_services.get('translate') == service_id:
|
||
current_services['translate'] = None
|
||
|
||
# 保存配置
|
||
if self.save_config(config):
|
||
self._log(f"成功删除服务商: {service_id}")
|
||
return True
|
||
else:
|
||
self._log(f"保存配置失败")
|
||
return False
|
||
|
||
except Exception as e:
|
||
self._log(f"删除服务商异常: {str(e)}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
def update_services_order(self, service_ids: list) -> bool:
|
||
"""
|
||
更新服务商顺序
|
||
|
||
参数:
|
||
service_ids: 服务商ID列表,按新顺序排列
|
||
|
||
返回:
|
||
bool: 成功返回True
|
||
"""
|
||
try:
|
||
config = self.load_config()
|
||
|
||
if not self._is_v2_config(config):
|
||
self._log("更新服务商顺序失败: 配置版本过低")
|
||
return False
|
||
|
||
services = config.get('model_services', [])
|
||
|
||
# 创建ID到服务的映射
|
||
service_map = {s.get('id'): s for s in services}
|
||
|
||
# 验证所有service_id都存在
|
||
for service_id in service_ids:
|
||
if service_id not in service_map:
|
||
self._log(f"更新服务商顺序失败: 服务商不存在 (ID: {service_id})")
|
||
return False
|
||
|
||
# 按新顺序重建services数组
|
||
new_services = []
|
||
for service_id in service_ids:
|
||
new_services.append(service_map[service_id])
|
||
|
||
# 添加未在service_ids中的服务(防止遗漏)
|
||
for service_id, service in service_map.items():
|
||
if service_id not in service_ids:
|
||
new_services.append(service)
|
||
self._log(f"警告: 服务商 {service_id} 不在新顺序中,已追加到末尾")
|
||
|
||
config['model_services'] = new_services
|
||
|
||
# 保存配置
|
||
if self.save_config(config):
|
||
self._log(f"成功更新服务商顺序: {', '.join(service_ids)}")
|
||
return True
|
||
else:
|
||
self._log("保存配置失败")
|
||
return False
|
||
|
||
except Exception as e:
|
||
self._log(f"更新服务商顺序异常: {str(e)}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
|
||
def update_service(self, service_id: str, **kwargs):
|
||
"""
|
||
更新服务商配置
|
||
|
||
参数:
|
||
service_id: 服务商ID
|
||
**kwargs: 要更新的字段（name, description, base_url, api_key, auto_unload等）
|
||
|
||
返回:
|
||
bool: 成功返回True
|
||
"""
|
||
try:
|
||
config = self.load_config()
|
||
|
||
if not self._is_v2_config(config):
|
||
self._log("更新服务商失败: 配置版本过低")
|
||
return False
|
||
|
||
# 查找服务
|
||
services = config.get('model_services', [])
|
||
service = None
|
||
service_index = -1
|
||
|
||
for i, s in enumerate(services):
|
||
if s.get('id') == service_id:
|
||
service = s
|
||
service_index = i
|
||
break
|
||
|
||
if not service:
|
||
self._log(f"更新服务商失败: 服务商不存在 (ID: {service_id})")
|
||
return False
|
||
|
||
# 更新字段
|
||
if 'name' in kwargs:
|
||
service['name'] = kwargs['name']
|
||
|
||
if 'description' in kwargs:
|
||
service['description'] = kwargs['description']
|
||
|
||
if 'base_url' in kwargs:
|
||
service['base_url'] = kwargs['base_url']
|
||
|
||
if 'api_key' in kwargs:
|
||
# 直接使用明文API Key
|
||
service['api_key'] = kwargs['api_key'] or ""
|
||
|
||
if 'auto_unload' in kwargs and service.get('type') == 'ollama':
|
||
service['auto_unload'] = kwargs['auto_unload']
|
||
|
||
if 'disable_thinking' in kwargs:
|
||
service['disable_thinking'] = kwargs['disable_thinking']
|
||
|
||
if 'enable_advanced_params' in kwargs:
|
||
service['enable_advanced_params'] = kwargs['enable_advanced_params']
|
||
|
||
if 'filter_thinking_output' in kwargs:
|
||
service['filter_thinking_output'] = kwargs['filter_thinking_output']
|
||
|
||
# 更新services数组
|
||
config['model_services'][service_index] = service
|
||
|
||
# 保存配置
|
||
if self.save_config(config):
|
||
self._log(f"成功更新服务商: {service_id}")
|
||
return True
|
||
else:
|
||
self._log(f"保存配置失败")
|
||
return False
|
||
|
||
except Exception as e:
|
||
self._log(f"更新服务商异常: {str(e)}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
def set_current_service(self, service_type: str, service_id: str, model_name: str = None):
|
||
"""
|
||
设置当前使用的服务商和模型
|
||
|
||
参数:
|
||
service_type: 服务类型 ('llm', 'vlm', 或 'translate')
|
||
service_id: 服务商ID
|
||
model_name: 模型名称(可选,如果不提供则使用该服务的默认模型或第一个模型)
|
||
|
||
返回:
|
||
bool: 成功返回True
|
||
"""
|
||
try:
|
||
config = self.load_config()
|
||
|
||
if not self._is_v2_config(config):
|
||
self._log("设置当前服务商失败: 配置版本过低")
|
||
return False
|
||
|
||
# ---百度翻译特殊处理---
|
||
# 百度翻译使用独立的baidu_translate配置,不在model_services中
|
||
if service_id == 'baidu':
|
||
# 百度翻译支持LLM服务类型(旧兼容)和translate服务类型
|
||
if service_type not in ['llm', 'translate']:
|
||
self._log(f"设置当前服务商失败: 百度翻译不支持{service_type}服务类型")
|
||
return False
|
||
|
||
# 确保baidu_translate配置存在
|
||
if 'baidu_translate' not in config:
|
||
config['baidu_translate'] = {"app_id": "", "secret_key": ""}
|
||
|
||
# 确保current_services结构存在
|
||
if 'current_services' not in config:
|
||
config['current_services'] = {}
|
||
|
||
# 设置百度为当前服务(无模型概念)
|
||
config['current_services'][service_type] = {
|
||
"service": "baidu",
|
||
"model": ""
|
||
}
|
||
|
||
# 保存配置
|
||
if self.save_config(config):
|
||
self._log(f"当前服务商已切换: 百度翻译 ({service_type})")
|
||
return True
|
||
else:
|
||
self._log("设置当前服务商失败: 保存配置失败")
|
||
return False
|
||
|
||
# ---其他服务:验证服务存在---
|
||
service = self._get_service_by_id(service_id)
|
||
if not service:
|
||
self._log(f"设置当前服务商失败: 服务商不存在 (ID: {service_id})")
|
||
return False
|
||
|
||
# 根据service_type确定模型列表字段
|
||
model_list_key = f'{service_type}_models'
|
||
if service_type == 'translate':
|
||
model_list_key = 'llm_models'
|
||
|
||
# 如果提供了model_name,验证模型是否存在
|
||
if model_name:
|
||
model_list = service.get(model_list_key, [])
|
||
model_exists = any(m.get('name') == model_name for m in model_list)
|
||
|
||
if not model_exists:
|
||
self._log(f"设置当前服务商失败: 模型不存在 (模型: {model_name}, 服务: {service_id})")
|
||
return False
|
||
|
||
# 确保current_services结构存在
|
||
if 'current_services' not in config:
|
||
config['current_services'] = {}
|
||
|
||
# 获取当前服务信息(兼容旧格式)
|
||
current_info = config['current_services'].get(service_type)
|
||
|
||
# 设置新格式的current_services
|
||
if model_name:
|
||
# 明确指定了模型
|
||
config['current_services'][service_type] = {
|
||
"service": service_id,
|
||
"model": model_name
|
||
}
|
||
else:
|
||
# 未指定模型,使用默认模型或第一个模型
|
||
model_list = service.get(model_list_key, [])
|
||
|
||
# 如果是百度服务,没有模型
|
||
if service.get('id') == 'baidu' or service.get('type') == 'baidu':
|
||
config['current_services'][service_type] = {
|
||
"service": service_id,
|
||
"model": ""
|
||
}
|
||
else:
|
||
# 查找默认模型或第一个模型
|
||
default_model = next((m for m in model_list if m.get('is_default')),
|
||
model_list[0] if model_list else None)
|
||
|
||
if default_model:
|
||
config['current_services'][service_type] = {
|
||
"service": service_id,
|
||
"model": default_model.get('name', '')
|
||
}
|
||
else:
|
||
# 没有模型,只设置服务
|
||
config['current_services'][service_type] = {
|
||
"service": service_id,
|
||
"model": ""
|
||
}
|
||
|
||
# 保存配置
|
||
if self.save_config(config):
|
||
service_name = service.get('name', service_id)
|
||
log_model = f" | 模型:{model_name}" if model_name else ""
|
||
self._log(f"成功设置当前{service_type}服务: {service_name}{log_model}")
|
||
return True
|
||
else:
|
||
self._log(f"保存配置失败")
|
||
return False
|
||
|
||
except Exception as e:
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
# --- 模型管理方法 ---
|
||
|
||
def add_model_to_service(self, service_id: str, model_type: str, model_name: str,
|
||
temperature: float = 0.7, top_p: float = 0.9, max_tokens: int = 4096):
|
||
"""添加模型到服务商"""
|
||
try:
|
||
config = self.load_config()
|
||
services = config.get('model_services', [])
|
||
|
||
for i, service in enumerate(services):
|
||
if service.get('id') == service_id:
|
||
model_list_key = 'llm_models' if model_type == 'llm' else 'vlm_models'
|
||
|
||
if model_list_key not in service:
|
||
service[model_list_key] = []
|
||
|
||
# 检查是否已存在
|
||
if any(m.get('name') == model_name for m in service[model_list_key]):
|
||
self._log(f"模型已存在: {model_name}")
|
||
return False
|
||
|
||
# 添加新模型
|
||
new_model = {
|
||
"name": model_name,
|
||
"is_default": len(service[model_list_key]) == 0,
|
||
"temperature": temperature,
|
||
"top_p": top_p,
|
||
"max_tokens": max_tokens
|
||
}
|
||
service[model_list_key].append(new_model)
|
||
config['model_services'][i] = service
|
||
|
||
if self.save_config(config):
|
||
self._log(f"成功添加模型: {model_name}")
|
||
return True
|
||
return False
|
||
|
||
self._log(f"服务商不存在: {service_id}")
|
||
return False
|
||
except Exception as e:
|
||
self._log(f"添加模型异常: {str(e)}")
|
||
return False
|
||
|
||
def delete_model_from_service(self, service_id: str, model_type: str, model_name: str):
|
||
"""从服务商删除模型"""
|
||
try:
|
||
config = self.load_config()
|
||
services = config.get('model_services', [])
|
||
|
||
for i, service in enumerate(services):
|
||
if service.get('id') == service_id:
|
||
model_list_key = 'llm_models' if model_type == 'llm' else 'vlm_models'
|
||
|
||
if model_list_key not in service:
|
||
return False
|
||
|
||
original_length = len(service[model_list_key])
|
||
service[model_list_key] = [m for m in service[model_list_key] if m.get('name') != model_name]
|
||
|
||
if len(service[model_list_key]) == original_length:
|
||
self._log(f"模型不存在: {model_name}")
|
||
return False
|
||
|
||
# 如果删除的是默认模型，设置第一个为默认
|
||
if len(service[model_list_key]) > 0:
|
||
if not any(m.get('is_default') for m in service[model_list_key]):
|
||
service[model_list_key][0]['is_default'] = True
|
||
|
||
config['model_services'][i] = service
|
||
|
||
if self.save_config(config):
|
||
self._log(f"成功删除模型: {model_name}")
|
||
return True
|
||
return False
|
||
|
||
self._log(f"服务商不存在: {service_id}")
|
||
return False
|
||
except Exception as e:
|
||
self._log(f"删除模型异常: {str(e)}")
|
||
return False
|
||
|
||
def set_default_model(self, service_id: str, model_type: str, model_name: str):
|
||
"""设置默认模型"""
|
||
try:
|
||
config = self.load_config()
|
||
services = config.get('model_services', [])
|
||
|
||
for i, service in enumerate(services):
|
||
if service.get('id') == service_id:
|
||
model_list_key = 'llm_models' if model_type == 'llm' else 'vlm_models'
|
||
|
||
if model_list_key not in service:
|
||
return False
|
||
|
||
found = False
|
||
for model in service[model_list_key]:
|
||
if model.get('name') == model_name:
|
||
model['is_default'] = True
|
||
found = True
|
||
else:
|
||
model['is_default'] = False
|
||
|
||
if not found:
|
||
self._log(f"模型不存在: {model_name}")
|
||
return False
|
||
|
||
config['model_services'][i] = service
|
||
|
||
if self.save_config(config):
|
||
self._log(f"成功设置默认模型: {model_name}")
|
||
return True
|
||
return False
|
||
|
||
self._log(f"服务商不存在: {service_id}")
|
||
return False
|
||
except Exception as e:
|
||
self._log(f"设置默认模型异常: {str(e)}")
|
||
return False
|
||
|
||
def update_model_order(self, service_id: str, model_type: str, model_names: list):
|
||
"""更新模型顺序"""
|
||
try:
|
||
config = self.load_config()
|
||
services = config.get('model_services', [])
|
||
|
||
for i, service in enumerate(services):
|
||
if service.get('id') == service_id:
|
||
model_list_key = 'llm_models' if model_type == 'llm' else 'vlm_models'
|
||
|
||
if model_list_key not in service:
|
||
return False
|
||
|
||
# 创建模型字典
|
||
model_dict = {m.get('name'): m for m in service[model_list_key]}
|
||
|
||
# 按新顺序重新排列
|
||
new_model_list = []
|
||
for name in model_names:
|
||
if name in model_dict:
|
||
new_model_list.append(model_dict[name])
|
||
|
||
service[model_list_key] = new_model_list
|
||
config['model_services'][i] = service
|
||
|
||
if self.save_config(config):
|
||
self._log(f"成功更新模型顺序")
|
||
return True
|
||
return False
|
||
|
||
self._log(f"服务商不存在: {service_id}")
|
||
return False
|
||
except Exception as e:
|
||
self._log(f"更新模型顺序异常: {str(e)}")
|
||
return False
|
||
|
||
def update_model_parameter(self, service_id: str, model_type: str, model_name: str,
|
||
parameter_name: str, parameter_value):
|
||
"""更新模型参数"""
|
||
try:
|
||
config = self.load_config()
|
||
services = config.get('model_services', [])
|
||
|
||
for i, service in enumerate(services):
|
||
if service.get('id') == service_id:
|
||
model_list_key = 'llm_models' if model_type == 'llm' else 'vlm_models'
|
||
|
||
if model_list_key not in service:
|
||
return False
|
||
|
||
# 查找模型并更新参数
|
||
for model in service[model_list_key]:
|
||
if model.get('name') == model_name:
|
||
model[parameter_name] = parameter_value
|
||
config['model_services'][i] = service
|
||
|
||
if self.save_config(config):
|
||
self._log(f"成功更新模型参数: {model_name}.{parameter_name} = {parameter_value}")
|
||
return True
|
||
return False
|
||
|
||
self._log(f"模型不存在: {model_name}")
|
||
return False
|
||
|
||
self._log(f"服务商不存在: {service_id}")
|
||
return False
|
||
except Exception as e:
|
||
self._log(f"更新模型参数异常: {str(e)}")
|
||
return False
|
||
|
||
# 创建全局配置管理器实例
|
||
config_manager = ConfigManager()
|