831 lines
32 KiB
Python
831 lines
32 KiB
Python
import json
|
|
import uuid
|
|
import shutil
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
from threading import RLock
|
|
import re
|
|
from .logger import log
|
|
|
|
|
|
class FileHandler:
|
|
def __init__(self, base_path, auto_backup_enabled=True, max_auto_backups=50, latest_backup_enabled=True):
|
|
self.base_path = Path(base_path)
|
|
self.workflows_dir = self.base_path / "workflows"
|
|
self.thumbnails_dir = self.base_path / "thumbnails"
|
|
self.backups_dir = self.base_path / "backups"
|
|
self.auto_backups_dir = self.backups_dir / "auto"
|
|
self.latest_backups_dir = self.backups_dir / "latest"
|
|
self.legacy_name_files_dir = self.backups_dir / "legacy-name-files"
|
|
self.versions_dir = self.base_path / "versions"
|
|
self.metadata_file = self.base_path / "metadata.json"
|
|
self.metadata_backup_file = self.base_path / "metadata.json.bak"
|
|
self.auto_backup_enabled = bool(auto_backup_enabled)
|
|
self.max_auto_backups = max(1, int(max_auto_backups or 50))
|
|
self.latest_backup_enabled = bool(latest_backup_enabled)
|
|
|
|
# 线程锁，保证并发安全
|
|
self._metadata_lock = RLock()
|
|
|
|
self._init_directories()
|
|
self._init_metadata()
|
|
|
|
def _init_directories(self):
|
|
"""初始化所需目录"""
|
|
for d in [self.workflows_dir, self.thumbnails_dir,
|
|
self.backups_dir, self.auto_backups_dir,
|
|
self.latest_backups_dir, self.legacy_name_files_dir,
|
|
self.versions_dir]:
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
|
|
def _init_metadata(self):
|
|
"""初始化元数据文件"""
|
|
if not self.metadata_file.exists():
|
|
self._write_metadata({"groups": {}, "workflows": {}})
|
|
self._recover_metadata_from_files()
|
|
self._migrate_named_workflow_files()
|
|
|
|
def _read_metadata(self):
|
|
"""线程安全地读取元数据"""
|
|
try:
|
|
with self._metadata_lock:
|
|
if not self.metadata_file.exists():
|
|
return {"groups": {}, "workflows": {}}
|
|
return json.loads(self.metadata_file.read_text(encoding="utf-8-sig"))
|
|
except Exception as e:
|
|
log(f"[yeban-WM] \u8bfb\u53d6\u5143\u6570\u636e\u5931\u8d25: {e}")
|
|
return {"groups": {}, "workflows": {}}
|
|
|
|
def _metadata_is_valid(self, data):
|
|
return (
|
|
isinstance(data, dict)
|
|
and isinstance(data.get("groups"), dict)
|
|
and isinstance(data.get("workflows"), dict)
|
|
)
|
|
|
|
def _read_workflow_file(self, path):
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
if isinstance(data, dict) and isinstance(data.get("nodes"), list):
|
|
return data
|
|
except Exception as e:
|
|
log(f"[yeban-WM] \u5df2\u5ffd\u7565\u65e0\u6548\u5de5\u4f5c\u6d41\u6587\u4ef6 {path.name}: {e}")
|
|
return None
|
|
|
|
def _workflow_file_for_name(self, workflow_name):
|
|
safe_filename = self._sanitize_filename(workflow_name or "")
|
|
return self.workflows_dir / f"{safe_filename}.json"
|
|
|
|
def _workflow_file_for_id(self, workflow_id):
|
|
return self.workflows_dir / f"{workflow_id}.json"
|
|
|
|
def _find_existing_workflow_id(self, meta, workflow_name, group_id):
|
|
matches = []
|
|
for workflow_id, workflow in meta.get("workflows", {}).items():
|
|
if not isinstance(workflow, dict):
|
|
continue
|
|
if workflow.get("name") == workflow_name and workflow.get("group") == group_id:
|
|
matches.append((workflow_id, workflow))
|
|
|
|
if not matches:
|
|
return None
|
|
|
|
matches.sort(
|
|
key=lambda match: (
|
|
match[1].get("modified_at") or match[1].get("created_at") or "",
|
|
match[0],
|
|
),
|
|
reverse=True,
|
|
)
|
|
if len(matches) > 1:
|
|
ids = ", ".join(workflow_id[:8] for workflow_id, _ in matches)
|
|
log(f"[yeban-WM] Detected duplicate workflow name/group for {workflow_name}: {ids}")
|
|
return matches[0][0]
|
|
|
|
def _archive_legacy_workflow_file(self, workflow_id, legacy_file):
|
|
if not legacy_file.exists():
|
|
return
|
|
try:
|
|
stamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
|
|
archived = self.legacy_name_files_dir / f"{workflow_id}_{stamp}_{legacy_file.name}"
|
|
shutil.move(str(legacy_file), str(archived))
|
|
except Exception as e:
|
|
log(f"[yeban-WM] \u5f52\u6863\u65e7\u5de5\u4f5c\u6d41\u6587\u4ef6\u5931\u8d25 {legacy_file.name}: {e}")
|
|
|
|
def _migrate_named_workflow_files(self):
|
|
meta = self._read_metadata()
|
|
if not self._metadata_is_valid(meta):
|
|
return
|
|
|
|
changed = False
|
|
for workflow_id, wf in list(meta["workflows"].items()):
|
|
if not isinstance(wf, dict):
|
|
continue
|
|
|
|
workflow_name = wf.get("name") or workflow_id
|
|
id_file = self._workflow_file_for_id(workflow_id)
|
|
legacy_file = self._workflow_file_for_name(workflow_name)
|
|
|
|
if not id_file.exists() and legacy_file.exists():
|
|
try:
|
|
shutil.copy2(legacy_file, id_file)
|
|
wf["size"] = id_file.stat().st_size
|
|
changed = True
|
|
log(f"[yeban-WM] \u5df2\u8fc1\u79fb\u5de5\u4f5c\u6d41\u6587\u4ef6\u4e3a ID \u547d\u540d: {workflow_id}.json")
|
|
except Exception as e:
|
|
log(f"[yeban-WM] \u8fc1\u79fb\u5de5\u4f5c\u6d41\u6587\u4ef6\u5931\u8d25 {legacy_file.name}: {e}")
|
|
continue
|
|
|
|
if legacy_file.exists() and legacy_file != id_file:
|
|
self._archive_legacy_workflow_file(workflow_id, legacy_file)
|
|
|
|
if changed:
|
|
self._save_metadata_snapshot("id_filename_migration")
|
|
self._write_metadata(meta)
|
|
|
|
def _version_workflow_id(self, version_file):
|
|
match = re.match(r"^(.+)_\d{8}_\d{6}\.json$", version_file.name)
|
|
return match.group(1) if match else None
|
|
|
|
def _save_metadata_snapshot(self, reason):
|
|
if not self.metadata_file.exists():
|
|
return
|
|
try:
|
|
snapshot_dir = self.base_path / "metadata_backups"
|
|
snapshot_dir.mkdir(parents=True, exist_ok=True)
|
|
stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
shutil.copy2(self.metadata_file, snapshot_dir / f"{reason}_{stamp}.json")
|
|
except Exception as e:
|
|
log(f"[yeban-WM] \u4fdd\u5b58\u5143\u6570\u636e\u5feb\u7167\u5931\u8d25: {e}")
|
|
|
|
def _is_generated_workflow_id(self, value):
|
|
try:
|
|
uuid.UUID(str(value))
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
def _load_metadata_backup_workflows(self, limit=12):
|
|
backup_workflows = {}
|
|
backup_dir = self.base_path / "metadata_backups"
|
|
candidates = []
|
|
|
|
if self.metadata_backup_file.exists():
|
|
candidates.append(self.metadata_backup_file)
|
|
if backup_dir.exists():
|
|
candidates.extend(
|
|
sorted(
|
|
backup_dir.glob("*.json"),
|
|
key=lambda path: path.stat().st_mtime,
|
|
reverse=True,
|
|
)[:limit]
|
|
)
|
|
|
|
for path in candidates:
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8-sig"))
|
|
except Exception as e:
|
|
log(f"[yeban-WM] Ignored invalid metadata backup {path.name}: {e}")
|
|
continue
|
|
|
|
if not self._metadata_is_valid(data):
|
|
continue
|
|
|
|
for workflow_id, item in data["workflows"].items():
|
|
if not isinstance(item, dict):
|
|
continue
|
|
existing = backup_workflows.get(workflow_id)
|
|
existing_name = existing.get("name") if isinstance(existing, dict) else None
|
|
item_name = item.get("name")
|
|
if (
|
|
existing is None
|
|
or (
|
|
existing_name == workflow_id
|
|
and item_name
|
|
and item_name != workflow_id
|
|
)
|
|
):
|
|
backup_workflows[workflow_id] = item
|
|
|
|
return backup_workflows
|
|
|
|
def _recover_metadata_from_files(self):
|
|
"""
|
|
Rebuild missing metadata entries from workflow files and version snapshots.
|
|
|
|
Some ComfyUI update/reinstall flows can leave metadata.json empty while
|
|
workflows and versions are still on disk. This keeps the panel from
|
|
showing an empty list after that kind of index loss.
|
|
"""
|
|
meta = self._read_metadata()
|
|
if not self._metadata_is_valid(meta):
|
|
meta = {"groups": {}, "workflows": {}}
|
|
|
|
workflows = meta["workflows"]
|
|
changed = False
|
|
backup_workflows = self._load_metadata_backup_workflows()
|
|
|
|
workflow_files = {}
|
|
for path in self.workflows_dir.glob("*.json"):
|
|
if self._read_workflow_file(path) is not None:
|
|
workflow_files[path.stem] = path
|
|
|
|
latest_versions = {}
|
|
for path in self.versions_dir.glob("*.json"):
|
|
workflow_id = self._version_workflow_id(path)
|
|
if not workflow_id or self._read_workflow_file(path) is None:
|
|
continue
|
|
current = latest_versions.get(workflow_id)
|
|
if current is None or path.stat().st_mtime > current.stat().st_mtime:
|
|
latest_versions[workflow_id] = path
|
|
|
|
def fallback_metadata(workflow_id, source_file):
|
|
backup_item = backup_workflows.get(workflow_id)
|
|
source_modified_at = datetime.fromtimestamp(source_file.stat().st_mtime).isoformat()
|
|
if isinstance(backup_item, dict):
|
|
name = backup_item.get("name") or workflow_id
|
|
if name == workflow_id and self._is_generated_workflow_id(name):
|
|
name = workflow_id
|
|
return {
|
|
"name": name,
|
|
"group": backup_item.get("group"),
|
|
"created_at": backup_item.get("created_at") or source_modified_at,
|
|
"modified_at": backup_item.get("modified_at") or source_modified_at,
|
|
"tags": backup_item.get("tags") if isinstance(backup_item.get("tags"), list) else [],
|
|
"starred": bool(backup_item.get("starred", False)),
|
|
}
|
|
return {
|
|
"name": workflow_id,
|
|
"group": None,
|
|
"created_at": source_modified_at,
|
|
"modified_at": source_modified_at,
|
|
"tags": [],
|
|
"starred": False,
|
|
}
|
|
|
|
def add_entry(workflow_id, name, source_file):
|
|
nonlocal changed
|
|
item = workflows.get(workflow_id)
|
|
modified_at = datetime.fromtimestamp(source_file.stat().st_mtime).isoformat()
|
|
has_thumbnail = (self.thumbnails_dir / f"{workflow_id}.png").exists()
|
|
fallback = fallback_metadata(workflow_id, source_file)
|
|
restored_name = fallback.get("name") or name
|
|
if item is None:
|
|
workflows[workflow_id] = {
|
|
"id": workflow_id,
|
|
"name": restored_name,
|
|
"group": fallback.get("group"),
|
|
"created_at": fallback.get("created_at") or modified_at,
|
|
"modified_at": fallback.get("modified_at") or modified_at,
|
|
"size": source_file.stat().st_size,
|
|
"tags": fallback.get("tags", []),
|
|
"starred": bool(fallback.get("starred", False)),
|
|
"has_thumbnail": has_thumbnail,
|
|
}
|
|
changed = True
|
|
return
|
|
|
|
if (
|
|
item.get("name") == workflow_id
|
|
and restored_name
|
|
and restored_name != workflow_id
|
|
):
|
|
item["name"] = restored_name
|
|
changed = True
|
|
|
|
for key, default in (
|
|
("id", workflow_id),
|
|
("group", fallback.get("group")),
|
|
("created_at", fallback.get("created_at") or modified_at),
|
|
("name", restored_name),
|
|
("tags", fallback.get("tags", [])),
|
|
("starred", bool(fallback.get("starred", False))),
|
|
):
|
|
if key not in item:
|
|
item[key] = default
|
|
changed = True
|
|
if item.get("size") != source_file.stat().st_size:
|
|
item["size"] = source_file.stat().st_size
|
|
changed = True
|
|
if item.get("has_thumbnail") != has_thumbnail:
|
|
item["has_thumbnail"] = has_thumbnail
|
|
changed = True
|
|
|
|
for name, path in workflow_files.items():
|
|
existing_name = workflows.get(name, {}).get("name")
|
|
add_entry(name, existing_name or name, path)
|
|
|
|
if not workflows:
|
|
for workflow_id, version_file in latest_versions.items():
|
|
matching_file = self.workflows_dir / f"{workflow_id}.json"
|
|
try:
|
|
shutil.copy2(version_file, matching_file)
|
|
workflow_files[workflow_id] = matching_file
|
|
add_entry(workflow_id, workflow_id, matching_file)
|
|
log(f"[yeban-WM] \u5df2\u4ece\u7248\u672c\u6062\u590d\u5de5\u4f5c\u6d41\u6587\u4ef6: {matching_file.name}")
|
|
except Exception as e:
|
|
log(f"[yeban-WM] \u6062\u590d\u5de5\u4f5c\u6d41\u6587\u4ef6\u5931\u8d25 {workflow_id}: {e}")
|
|
|
|
if changed:
|
|
self._save_metadata_snapshot("recovery")
|
|
self._write_metadata(meta)
|
|
log(f"[yeban-WM] \u5df2\u4ece\u78c1\u76d8\u6062\u590d\u5143\u6570\u636e: {len(workflows)} \u4e2a\u5de5\u4f5c\u6d41")
|
|
|
|
def _write_metadata(self, data):
|
|
"""线程安全地写入元数据"""
|
|
try:
|
|
with self._metadata_lock:
|
|
# 先写到临时文件，再原子性重命名
|
|
if self.metadata_file.exists():
|
|
try:
|
|
shutil.copy2(self.metadata_file, self.metadata_backup_file)
|
|
except Exception as e:
|
|
log(f"[yeban-WM] 更新元数据备份失败: {e}")
|
|
temp_file = self.metadata_file.with_suffix('.tmp')
|
|
temp_file.write_text(
|
|
json.dumps(data, indent=2, ensure_ascii=False),
|
|
encoding="utf-8"
|
|
)
|
|
temp_file.replace(self.metadata_file)
|
|
except Exception as e:
|
|
log(f"[yeban-WM] 写入元数据失败: {e}")
|
|
raise
|
|
|
|
def _sanitize_filename(self, filename):
|
|
"""
|
|
将文件名中的特殊字符替换为下划线
|
|
保留中文字符和常见符号
|
|
"""
|
|
# 替换不安全的字符
|
|
filename = re.sub(r'[<>:"/\\|?*]', '_', filename)
|
|
# 移除前后空格
|
|
filename = filename.strip()
|
|
# 如果为空，使用默认名称
|
|
if not filename:
|
|
filename = "untitled"
|
|
return filename
|
|
|
|
def _auto_backup_existing_workflow(self, workflow_id, workflow_name, wf_file):
|
|
if not self.auto_backup_enabled or not wf_file.exists():
|
|
return None
|
|
|
|
try:
|
|
safe_name = self._sanitize_filename(workflow_name)
|
|
ts = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
|
|
backup_file = self.auto_backups_dir / f"{workflow_id}_{safe_name}_{ts}.json"
|
|
shutil.copy2(wf_file, backup_file)
|
|
self._cleanup_old_auto_backups(workflow_id)
|
|
return backup_file
|
|
except Exception as e:
|
|
log(f"[yeban-WM] 自动备份工作流失败 {workflow_id}: {e}")
|
|
return None
|
|
|
|
def _cleanup_old_auto_backups(self, workflow_id):
|
|
try:
|
|
backups = sorted(
|
|
self.auto_backups_dir.glob(f"{workflow_id}_*.json"),
|
|
key=lambda f: f.stat().st_mtime
|
|
)
|
|
while len(backups) > self.max_auto_backups:
|
|
old_file = backups.pop(0)
|
|
try:
|
|
old_file.unlink()
|
|
except Exception as e:
|
|
log(f"[yeban-WM] 删除旧自动备份失败: {e}")
|
|
except Exception as e:
|
|
log(f"[yeban-WM] 清理自动备份失败: {e}")
|
|
|
|
# ── groups ────────────────────────────────────────────────────────────────
|
|
|
|
def _sync_latest_backup(self, workflow_id, workflow_name, workflow_data):
|
|
if not self.latest_backup_enabled:
|
|
return None
|
|
|
|
try:
|
|
safe_name = self._sanitize_filename(workflow_name)
|
|
backup_file = self.latest_backups_dir / f"{workflow_id}_{safe_name}.json"
|
|
temp_file = backup_file.with_suffix(".tmp")
|
|
temp_file.write_text(
|
|
json.dumps(workflow_data, indent=2, ensure_ascii=False),
|
|
encoding="utf-8"
|
|
)
|
|
temp_file.replace(backup_file)
|
|
return backup_file
|
|
except Exception as e:
|
|
log(f"[yeban-WM] 同步最新备份失败 {workflow_id}: {e}")
|
|
return None
|
|
|
|
def create_group(self, group_name, parent_id=None):
|
|
"""创建分组"""
|
|
meta = self._read_metadata()
|
|
group_id = str(uuid.uuid4())
|
|
meta["groups"][group_id] = {
|
|
"id": group_id,
|
|
"name": group_name,
|
|
"parent": parent_id or None,
|
|
"created_at": datetime.now().isoformat(),
|
|
}
|
|
self._write_metadata(meta)
|
|
return group_id
|
|
|
|
def delete_group(self, group_id):
|
|
"""删除分组，子分组提升到根"""
|
|
meta = self._read_metadata()
|
|
if group_id not in meta["groups"]:
|
|
return False
|
|
|
|
# 子分组提升到根
|
|
for gid in list(meta["groups"].keys()):
|
|
if meta["groups"][gid].get("parent") == group_id:
|
|
meta["groups"][gid]["parent"] = None
|
|
|
|
del meta["groups"][group_id]
|
|
self._write_metadata(meta)
|
|
return True
|
|
|
|
def rename_group(self, group_id, new_name):
|
|
"""重命名分组"""
|
|
meta = self._read_metadata()
|
|
if group_id not in meta["groups"]:
|
|
return None
|
|
meta["groups"][group_id]["name"] = new_name
|
|
self._write_metadata(meta)
|
|
return group_id
|
|
|
|
# ── workflows ─────────────────────────────────────────────────────────────
|
|
|
|
def save_workflow(self, workflow_data, workflow_name, group_id=None):
|
|
"""
|
|
保存工作流
|
|
如果同名同分组的工作流已存在，则覆盖；否则创建新的
|
|
文件名使用工作流名称而不是UUID
|
|
"""
|
|
meta = self._read_metadata()
|
|
|
|
existing_id = self._find_existing_workflow_id(meta, workflow_name, group_id)
|
|
|
|
# 如果是新工作流，生成 UUID 用作 ID
|
|
# 但文件名改为使用工作流名称
|
|
wf_id = existing_id or str(uuid.uuid4())
|
|
|
|
# Store the primary workflow by id so duplicate display names cannot collide.
|
|
wf_file = self._workflow_file_for_id(wf_id)
|
|
legacy_file = self._workflow_file_for_name(workflow_name)
|
|
|
|
try:
|
|
# 先写到临时文件，再原子性重命名
|
|
backup_source = wf_file if wf_file.exists() else legacy_file
|
|
if existing_id and backup_source.exists():
|
|
self._auto_backup_existing_workflow(existing_id, workflow_name, backup_source)
|
|
temp_file = wf_file.with_suffix('.tmp')
|
|
temp_file.write_text(
|
|
json.dumps(workflow_data, indent=2, ensure_ascii=False),
|
|
encoding="utf-8"
|
|
)
|
|
temp_file.replace(wf_file)
|
|
if legacy_file.exists() and legacy_file != wf_file:
|
|
self._archive_legacy_workflow_file(wf_id, legacy_file)
|
|
self._sync_latest_backup(wf_id, workflow_name, workflow_data)
|
|
except Exception as e:
|
|
log(f"[yeban-WM] 写入工作流文件失败: {e}")
|
|
raise
|
|
|
|
now = datetime.now().isoformat()
|
|
|
|
if existing_id:
|
|
# 覆盖现有工作流
|
|
meta["workflows"][wf_id].update({
|
|
"name": workflow_name,
|
|
"group": group_id,
|
|
"modified_at": now,
|
|
"size": len(json.dumps(workflow_data)),
|
|
})
|
|
else:
|
|
# 创建新工作流
|
|
meta["workflows"][wf_id] = {
|
|
"id": wf_id,
|
|
"name": workflow_name,
|
|
"group": group_id,
|
|
"created_at": now,
|
|
"modified_at": now,
|
|
"size": len(json.dumps(workflow_data)),
|
|
"tags": [],
|
|
"starred": False,
|
|
"has_thumbnail": False,
|
|
}
|
|
|
|
self._write_metadata(meta)
|
|
return wf_id
|
|
|
|
def load_workflow(self, workflow_id):
|
|
"""加载工作流数据"""
|
|
meta = self._read_metadata()
|
|
|
|
# 从元数据中获取工作流名称
|
|
if workflow_id not in meta["workflows"]:
|
|
return None
|
|
|
|
workflow_name = meta["workflows"][workflow_id]["name"]
|
|
wf_file = self._workflow_file_for_id(workflow_id)
|
|
if not wf_file.exists():
|
|
wf_file = self._workflow_file_for_name(workflow_name)
|
|
|
|
if not wf_file.exists():
|
|
return None
|
|
try:
|
|
return json.loads(wf_file.read_text(encoding="utf-8"))
|
|
except Exception as e:
|
|
log(f"[yeban-WM] 加载工作流失败: {e}")
|
|
return None
|
|
|
|
def delete_workflow(self, workflow_id):
|
|
"""删除工作流及其相关文件"""
|
|
meta = self._read_metadata()
|
|
if workflow_id not in meta["workflows"]:
|
|
return False
|
|
|
|
# 根据工作流名称找到文件
|
|
workflow_name = meta["workflows"][workflow_id]["name"]
|
|
candidate_files = [
|
|
self._workflow_file_for_id(workflow_id),
|
|
self._workflow_file_for_name(workflow_name),
|
|
]
|
|
|
|
for wf_file in candidate_files:
|
|
if not wf_file.exists():
|
|
continue
|
|
try:
|
|
wf_file.unlink()
|
|
except Exception as e:
|
|
log(f"[yeban-WM] 删除工作流文件失败: {e}")
|
|
|
|
# 删除缩略图
|
|
thumb = self.thumbnails_dir / f"{workflow_id}.png"
|
|
if thumb.exists():
|
|
try:
|
|
thumb.unlink()
|
|
except Exception as e:
|
|
log(f"[yeban-WM] 删除缩略图失败: {e}")
|
|
|
|
# 删除所有版本
|
|
for ver_file in self.versions_dir.glob(f"{workflow_id}_*.json"):
|
|
try:
|
|
ver_file.unlink()
|
|
except Exception as e:
|
|
log(f"[yeban-WM] 删除版本文件失败: {e}")
|
|
|
|
del meta["workflows"][workflow_id]
|
|
self._write_metadata(meta)
|
|
return True
|
|
|
|
def rename_workflow(self, workflow_id, new_name):
|
|
"""重命名工作流"""
|
|
meta = self._read_metadata()
|
|
if workflow_id not in meta["workflows"]:
|
|
return None
|
|
|
|
old_name = meta["workflows"][workflow_id]["name"]
|
|
id_file = self._workflow_file_for_id(workflow_id)
|
|
old_legacy_file = self._workflow_file_for_name(old_name)
|
|
|
|
# Older releases stored the primary file by display name. Migrate it if needed.
|
|
if not id_file.exists() and old_legacy_file.exists():
|
|
try:
|
|
shutil.copy2(old_legacy_file, id_file)
|
|
self._archive_legacy_workflow_file(workflow_id, old_legacy_file)
|
|
except Exception as e:
|
|
log(f"[yeban-WM] 重命名时迁移工作流文件失败: {e}")
|
|
|
|
# 更新元数据
|
|
meta["workflows"][workflow_id]["name"] = new_name
|
|
meta["workflows"][workflow_id]["modified_at"] = datetime.now().isoformat()
|
|
self._write_metadata(meta)
|
|
return workflow_id
|
|
|
|
def move_workflow(self, workflow_id, target_group_id):
|
|
"""移动工作流到其他分组"""
|
|
meta = self._read_metadata()
|
|
if workflow_id not in meta["workflows"]:
|
|
return None
|
|
meta["workflows"][workflow_id]["group"] = target_group_id or None
|
|
meta["workflows"][workflow_id]["modified_at"] = datetime.now().isoformat()
|
|
self._write_metadata(meta)
|
|
return workflow_id
|
|
|
|
def copy_workflow(self, workflow_id, target_group_id, new_name=None):
|
|
"""复制工作流到目标分组"""
|
|
meta = self._read_metadata()
|
|
if workflow_id not in meta["workflows"]:
|
|
return None
|
|
|
|
workflow_data = self.load_workflow(workflow_id)
|
|
if workflow_data is None:
|
|
return None
|
|
|
|
original_name = meta["workflows"][workflow_id]["name"]
|
|
name = new_name or f"{original_name}_copy"
|
|
return self.save_workflow(workflow_data, name, target_group_id)
|
|
|
|
# ── thumbnail ─────────────────────────────────────────────────────────────
|
|
|
|
def save_thumbnail(self, workflow_id, data_url):
|
|
"""保存缩略图"""
|
|
import base64
|
|
|
|
meta = self._read_metadata()
|
|
if workflow_id not in meta["workflows"]:
|
|
return False
|
|
|
|
try:
|
|
# 去掉 data:image/png;base64, 前缀
|
|
b64 = re.sub(r"^data:image/\w+;base64,", "", data_url)
|
|
thumb_file = self.thumbnails_dir / f"{workflow_id}.png"
|
|
thumb_file.write_bytes(base64.b64decode(b64))
|
|
|
|
meta["workflows"][workflow_id]["has_thumbnail"] = True
|
|
self._write_metadata(meta)
|
|
return True
|
|
except Exception as e:
|
|
log(f"[yeban-WM] 保存缩略图失败: {e}")
|
|
return False
|
|
|
|
def get_thumbnail_path(self, workflow_id):
|
|
"""获取缩略图路径"""
|
|
thumb = self.thumbnails_dir / f"{workflow_id}.png"
|
|
return thumb if thumb.exists() else None
|
|
|
|
# ── version history ───────────────────────────────────────────────────────
|
|
|
|
def save_version(self, workflow_id, max_versions=20):
|
|
"""创建工作流版本快照"""
|
|
meta = self._read_metadata()
|
|
if workflow_id not in meta["workflows"]:
|
|
return None
|
|
|
|
workflow_data = self.load_workflow(workflow_id)
|
|
if workflow_data is None:
|
|
return None
|
|
|
|
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
ver_file = self.versions_dir / f"{workflow_id}_{ts}.json"
|
|
|
|
try:
|
|
ver_file.write_text(
|
|
json.dumps(workflow_data, indent=2, ensure_ascii=False),
|
|
encoding="utf-8"
|
|
)
|
|
except Exception as e:
|
|
log(f"[yeban-WM] 保存版本快照失败: {e}")
|
|
return None
|
|
|
|
# 清理超出 max_versions 的旧版本
|
|
self._cleanup_old_versions(workflow_id, max_versions)
|
|
return ver_file.name
|
|
|
|
def _cleanup_old_versions(self, workflow_id, max_versions):
|
|
"""清理超出限制的旧版本"""
|
|
try:
|
|
all_versions = sorted(
|
|
self.versions_dir.glob(f"{workflow_id}_*.json"),
|
|
key=lambda f: f.stat().st_mtime
|
|
)
|
|
|
|
while len(all_versions) > max_versions:
|
|
old_file = all_versions.pop(0)
|
|
try:
|
|
old_file.unlink()
|
|
except Exception as e:
|
|
log(f"[yeban-WM] 删除旧版本失败: {e}")
|
|
except Exception as e:
|
|
log(f"[yeban-WM] 清理版本历史失败: {e}")
|
|
|
|
def list_versions(self, workflow_id):
|
|
"""列出工作流的所有版本"""
|
|
try:
|
|
files = sorted(
|
|
self.versions_dir.glob(f"{workflow_id}_*.json"),
|
|
key=lambda f: f.stat().st_mtime,
|
|
reverse=True
|
|
)
|
|
return [f.name for f in files]
|
|
except Exception as e:
|
|
log(f"[yeban-WM] 读取版本列表失败: {e}")
|
|
return []
|
|
|
|
def restore_version(self, workflow_id, version_filename):
|
|
"""恢复到指定版本"""
|
|
if Path(version_filename).name != version_filename:
|
|
return False
|
|
if not version_filename.startswith(f"{workflow_id}_"):
|
|
return False
|
|
ver_file = self.versions_dir / version_filename
|
|
if not ver_file.exists():
|
|
return False
|
|
|
|
try:
|
|
# 先把当前版本存一个快照
|
|
self.save_version(workflow_id)
|
|
|
|
workflow_data = json.loads(ver_file.read_text(encoding="utf-8"))
|
|
|
|
# 根据工作流名称生成文件路径
|
|
meta = self._read_metadata()
|
|
if workflow_id not in meta["workflows"]:
|
|
return False
|
|
|
|
wf_file = self._workflow_file_for_id(workflow_id)
|
|
|
|
wf_file.write_text(
|
|
json.dumps(workflow_data, indent=2, ensure_ascii=False),
|
|
encoding="utf-8"
|
|
)
|
|
|
|
meta["workflows"][workflow_id]["modified_at"] = datetime.now().isoformat()
|
|
self._write_metadata(meta)
|
|
return True
|
|
except Exception as e:
|
|
log(f"[yeban-WM] 恢复版本失败: {e}")
|
|
return False
|
|
|
|
# ── search ────────────────────────────────────────────────────────────────
|
|
|
|
def search_workflows(self, query):
|
|
"""按名称或标签搜索工作流"""
|
|
meta = self._read_metadata()
|
|
q = query.lower()
|
|
results = []
|
|
|
|
for wf_id, wf in meta["workflows"].items():
|
|
name_match = q in wf["name"].lower()
|
|
tag_match = any(q in t.lower() for t in wf.get("tags", []))
|
|
|
|
if name_match or tag_match:
|
|
results.append(wf_id)
|
|
|
|
return results
|
|
|
|
# ── misc ──────────────────────────────────────────────────────────────────
|
|
|
|
def get_all_workflows(self):
|
|
"""获取所有工作流元数据"""
|
|
return self._read_metadata()["workflows"]
|
|
|
|
def get_all_groups(self):
|
|
"""获取所有分组"""
|
|
return self._read_metadata()["groups"]
|
|
|
|
def backup_workflow(self, workflow_id):
|
|
"""备份工作流到 backups 目录"""
|
|
meta = self._read_metadata()
|
|
if workflow_id not in meta["workflows"]:
|
|
return None
|
|
|
|
workflow_data = self.load_workflow(workflow_id)
|
|
if workflow_data is None:
|
|
return None
|
|
|
|
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
name = meta["workflows"][workflow_id]["name"]
|
|
safe_name = self._sanitize_filename(name)
|
|
backup_file = self.backups_dir / f"{workflow_id}_{safe_name}_{ts}.json"
|
|
|
|
try:
|
|
backup_file.write_text(
|
|
json.dumps(workflow_data, indent=2, ensure_ascii=False),
|
|
encoding="utf-8"
|
|
)
|
|
return str(backup_file)
|
|
except Exception as e:
|
|
log(f"[yeban-WM] 备份工作流失败: {e}")
|
|
return None
|
|
|
|
def import_workflow(self, import_path, workflow_name, group_id=None):
|
|
"""从文件导入工作流"""
|
|
try:
|
|
workflow_data = json.loads(Path(import_path).read_text(encoding="utf-8"))
|
|
return self.save_workflow(workflow_data, workflow_name, group_id)
|
|
except Exception as e:
|
|
log(f"[yeban-WM] 导入工作流失败: {e}")
|
|
return None
|
|
|
|
def toggle_star(self, workflow_id):
|
|
"""切换工作流收藏状态"""
|
|
meta = self._read_metadata()
|
|
if workflow_id not in meta["workflows"]:
|
|
return None
|
|
|
|
meta["workflows"][workflow_id]["starred"] = not meta["workflows"][workflow_id].get("starred", False)
|
|
self._write_metadata(meta)
|
|
return meta["workflows"][workflow_id]["starred"]
|
|
|
|
def update_tags(self, workflow_id, tags):
|
|
"""更新工作流标签"""
|
|
meta = self._read_metadata()
|
|
if workflow_id not in meta["workflows"]:
|
|
return False
|
|
|
|
# 去重并排序
|
|
meta["workflows"][workflow_id]["tags"] = sorted(list(set(tags)))
|
|
self._write_metadata(meta)
|
|
return True
|