237 lines
9.6 KiB
Python
237 lines
9.6 KiB
Python
from .file_handler import FileHandler
|
|
import json
|
|
from pathlib import Path
|
|
from .logger import is_backend_logging_enabled, log, set_backend_logging_enabled
|
|
|
|
|
|
class WorkflowManager:
|
|
"""工作流管理器 - 单例模式"""
|
|
_instance = None
|
|
_initialized = False
|
|
|
|
def __new__(cls, base_path=None):
|
|
if cls._instance is None:
|
|
cls._instance = super().__new__(cls)
|
|
return cls._instance
|
|
|
|
def __init__(self, base_path=None):
|
|
# 防止重复初始化
|
|
if self._initialized:
|
|
return
|
|
|
|
if base_path is None:
|
|
raise ValueError("base_path required on first initialization")
|
|
|
|
self._load_config()
|
|
self.file_handler = FileHandler(
|
|
base_path,
|
|
auto_backup_enabled=self.auto_backup_enabled,
|
|
max_auto_backups=self.max_auto_backups,
|
|
latest_backup_enabled=self.latest_backup_enabled,
|
|
)
|
|
WorkflowManager._initialized = True
|
|
|
|
@staticmethod
|
|
def _read_config_file():
|
|
config_file = Path(__file__).parent / "config.json"
|
|
if not config_file.exists():
|
|
return {}
|
|
return json.loads(config_file.read_text(encoding="utf-8"))
|
|
|
|
@classmethod
|
|
def resolve_data_path(cls, comfyui_path):
|
|
config = cls._read_config_file()
|
|
data_dir = config.get("data_dir", "yeban-workflows") or "yeban-workflows"
|
|
data_path = Path(data_dir).expanduser()
|
|
if not data_path.is_absolute():
|
|
data_path = Path(comfyui_path) / data_path
|
|
return data_path
|
|
|
|
def _load_config(self):
|
|
"""从 config.json 加载配置"""
|
|
try:
|
|
config = self._read_config_file()
|
|
self.data_dir = config.get("data_dir", "yeban-workflows")
|
|
self.auto_save_interval = config.get("auto_save_interval", 60)
|
|
self.max_versions = config.get("max_versions", 20)
|
|
self.auto_backup_enabled = config.get("auto_backup_enabled", True)
|
|
self.max_auto_backups = config.get("max_auto_backups", 50)
|
|
self.latest_backup_enabled = config.get("latest_backup_enabled", True)
|
|
log(f"[yeban-WM] \u914d\u7f6e\u5df2\u52a0\u8f7d: \u81ea\u52a8\u4fdd\u5b58\u95f4\u9694={self.auto_save_interval}\u79d2, \u6700\u5927\u7248\u672c\u6570={self.max_versions}, \u81ea\u52a8\u5907\u4efd\u6570={self.max_auto_backups}, \u6700\u65b0\u5907\u4efd={self.latest_backup_enabled}")
|
|
except Exception as e:
|
|
log(f"[yeban-WM] \u52a0\u8f7d\u914d\u7f6e\u5931\u8d25: {e}")
|
|
self.data_dir = "yeban-workflows"
|
|
self.auto_save_interval = 60
|
|
self.max_versions = 20
|
|
self.auto_backup_enabled = True
|
|
self.max_auto_backups = 50
|
|
self.latest_backup_enabled = True
|
|
|
|
@classmethod
|
|
def reset(cls):
|
|
"""重置单例（用于测试或重载）"""
|
|
cls._instance = None
|
|
cls._initialized = False
|
|
|
|
# ── groups ────────────────────────────────────────────────────────────────
|
|
|
|
def create_group(self, name, parent_id=None):
|
|
"""创建分组"""
|
|
if not name or not isinstance(name, str):
|
|
raise ValueError("Group name must be non-empty string")
|
|
name = name.strip()
|
|
return self.file_handler.create_group(name, parent_id)
|
|
|
|
def delete_group(self, group_id):
|
|
"""删除分组"""
|
|
return self.file_handler.delete_group(group_id)
|
|
|
|
def rename_group(self, group_id, new_name):
|
|
"""重命名分组"""
|
|
if not new_name or not isinstance(new_name, str):
|
|
raise ValueError("Group name must be non-empty string")
|
|
new_name = new_name.strip()
|
|
return self.file_handler.rename_group(group_id, new_name)
|
|
|
|
# ── workflows ─────────────────────────────────────────────────────────────
|
|
|
|
def save_workflow(self, data, name, group_id=None):
|
|
"""保存工作流"""
|
|
if not name or not isinstance(name, str):
|
|
raise ValueError("Workflow name must be non-empty string")
|
|
name = name.strip()
|
|
if not isinstance(data, dict):
|
|
raise ValueError("Workflow data must be a dict")
|
|
return self.file_handler.save_workflow(data, name, group_id)
|
|
|
|
def load_workflow(self, wf_id):
|
|
"""加载工作流"""
|
|
if not wf_id:
|
|
raise ValueError("Workflow ID required")
|
|
return self.file_handler.load_workflow(wf_id)
|
|
|
|
def delete_workflow(self, wf_id):
|
|
"""删除工作流"""
|
|
if not wf_id:
|
|
raise ValueError("Workflow ID required")
|
|
return self.file_handler.delete_workflow(wf_id)
|
|
|
|
def rename_workflow(self, wf_id, new_name):
|
|
"""重命名工作流"""
|
|
if not wf_id:
|
|
raise ValueError("Workflow ID required")
|
|
if not new_name or not isinstance(new_name, str):
|
|
raise ValueError("New name must be non-empty string")
|
|
new_name = new_name.strip()
|
|
return self.file_handler.rename_workflow(wf_id, new_name)
|
|
|
|
def move_workflow(self, wf_id, target_group_id):
|
|
"""移动工作流到其他分组"""
|
|
if not wf_id:
|
|
raise ValueError("Workflow ID required")
|
|
return self.file_handler.move_workflow(wf_id, target_group_id)
|
|
|
|
def copy_workflow(self, wf_id, target_group_id, new_name=None):
|
|
"""复制工作流"""
|
|
if not wf_id:
|
|
raise ValueError("Workflow ID required")
|
|
if new_name:
|
|
new_name = new_name.strip()
|
|
return self.file_handler.copy_workflow(wf_id, target_group_id, new_name)
|
|
|
|
# ── thumbnail ─────────────────────────────────────────────────────────────
|
|
|
|
def save_thumbnail(self, wf_id, data_url):
|
|
"""保存缩略图"""
|
|
if not wf_id:
|
|
raise ValueError("Workflow ID required")
|
|
if not data_url:
|
|
raise ValueError("Data URL required")
|
|
return self.file_handler.save_thumbnail(wf_id, data_url)
|
|
|
|
def get_thumbnail_path(self, wf_id):
|
|
"""获取缩略图路径"""
|
|
if not wf_id:
|
|
raise ValueError("Workflow ID required")
|
|
return self.file_handler.get_thumbnail_path(wf_id)
|
|
|
|
# ── version history ───────────────────────────────────────────────────────
|
|
|
|
def save_version(self, wf_id):
|
|
"""创建版本快照"""
|
|
if not wf_id:
|
|
raise ValueError("Workflow ID required")
|
|
return self.file_handler.save_version(wf_id, self.max_versions)
|
|
|
|
def list_versions(self, wf_id):
|
|
"""列出所有版本"""
|
|
if not wf_id:
|
|
raise ValueError("Workflow ID required")
|
|
return self.file_handler.list_versions(wf_id)
|
|
|
|
def restore_version(self, wf_id, version_filename):
|
|
"""恢复到指定版本"""
|
|
if not wf_id:
|
|
raise ValueError("Workflow ID required")
|
|
if not version_filename:
|
|
raise ValueError("Version filename required")
|
|
return self.file_handler.restore_version(wf_id, version_filename)
|
|
|
|
# ── misc ──────────────────────────────────────────────────────────────────
|
|
|
|
def search_workflows(self, query):
|
|
"""搜索工作流"""
|
|
if not isinstance(query, str):
|
|
query = ""
|
|
return self.file_handler.search_workflows(query)
|
|
|
|
def get_all_workflows(self):
|
|
"""获取所有工作流元数据"""
|
|
return self.file_handler.get_all_workflows()
|
|
|
|
def get_all_groups(self):
|
|
"""获取所有分组"""
|
|
return self.file_handler.get_all_groups()
|
|
|
|
def backup_workflow(self, wf_id):
|
|
"""备份工作流"""
|
|
if not wf_id:
|
|
raise ValueError("Workflow ID required")
|
|
return self.file_handler.backup_workflow(wf_id)
|
|
|
|
def import_workflow(self, import_path, workflow_name, group_id=None):
|
|
"""导入工作流"""
|
|
if not import_path:
|
|
raise ValueError("Import path required")
|
|
if not workflow_name or not isinstance(workflow_name, str):
|
|
raise ValueError("Workflow name must be non-empty string")
|
|
workflow_name = workflow_name.strip()
|
|
return self.file_handler.import_workflow(import_path, workflow_name, group_id)
|
|
|
|
def toggle_star(self, wf_id):
|
|
"""收藏/取消收藏工作流"""
|
|
if not wf_id:
|
|
raise ValueError("Workflow ID required")
|
|
return self.file_handler.toggle_star(wf_id)
|
|
|
|
def update_tags(self, wf_id, tags):
|
|
"""更新工作流标签"""
|
|
if not wf_id:
|
|
raise ValueError("Workflow ID required")
|
|
if not isinstance(tags, list):
|
|
tags = []
|
|
tags = [str(t).strip() for t in tags if str(t).strip()]
|
|
return self.file_handler.update_tags(wf_id, tags)
|
|
|
|
def get_settings(self):
|
|
return {
|
|
"backend_logging_enabled": is_backend_logging_enabled(),
|
|
}
|
|
|
|
def update_settings(self, backend_logging_enabled=None):
|
|
if backend_logging_enabled is not None:
|
|
if isinstance(backend_logging_enabled, str):
|
|
backend_logging_enabled = backend_logging_enabled.strip().lower() not in {"0", "false", "no", "off"}
|
|
set_backend_logging_enabled(bool(backend_logging_enabled))
|
|
return self.get_settings()
|