mirror of
https://github.com/HuangYuChuh/ComfyUI_Skills_OpenClaw.git
synced 2026-07-30 03:23:40 +00:00
e948ab84c2
* feat: add workflow execution history backend * build: refresh bundled workflow history UI assets * fix: simplify common workflow parameter aliases * fix: resolve output_dir symlinks and allow workflows with no image output --------- Co-authored-by: Kelin <kelin@KelindeMacBook-Air.local>
76 lines
2.3 KiB
Python
76 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
BASE_DIR = Path(__file__).resolve().parent.parent.parent
|
|
CONFIG_PATH = BASE_DIR / "config.json"
|
|
DATA_DIR = BASE_DIR / "data"
|
|
OUTPUTS_DIR = BASE_DIR / "outputs"
|
|
WORKFLOW_FILENAME = "workflow.json"
|
|
SCHEMA_FILENAME = "schema.json"
|
|
HISTORY_DIRNAME = "history"
|
|
DEFAULT_COMFYUI_SERVER_URL = "http://127.0.0.1:8188"
|
|
DEFAULT_SERVER_ID = "local"
|
|
DEFAULT_OUTPUT_DIR = "./outputs"
|
|
|
|
|
|
def default_server() -> dict[str, object]:
|
|
return {
|
|
"id": DEFAULT_SERVER_ID,
|
|
"name": "Local",
|
|
"url": DEFAULT_COMFYUI_SERVER_URL,
|
|
"enabled": True,
|
|
"output_dir": DEFAULT_OUTPUT_DIR,
|
|
}
|
|
|
|
|
|
def default_config() -> dict[str, object]:
|
|
return {
|
|
"servers": [default_server()],
|
|
"default_server": DEFAULT_SERVER_ID,
|
|
}
|
|
|
|
|
|
def get_server_data_dir(server_id: str) -> Path:
|
|
"""Return the data directory for a specific server."""
|
|
return DATA_DIR / server_id
|
|
|
|
|
|
def get_server_workflow_dir(server_id: str, workflow_id: str) -> Path:
|
|
"""Return the directory that stores one workflow and its schema."""
|
|
return get_server_data_dir(server_id) / workflow_id
|
|
|
|
|
|
def get_server_workflow_path(server_id: str, workflow_id: str) -> Path:
|
|
return get_server_workflow_dir(server_id, workflow_id) / WORKFLOW_FILENAME
|
|
|
|
|
|
def get_server_schema_path(server_id: str, workflow_id: str) -> Path:
|
|
return get_server_workflow_dir(server_id, workflow_id) / SCHEMA_FILENAME
|
|
|
|
|
|
def get_server_history_dir(server_id: str, workflow_id: str) -> Path:
|
|
return get_server_workflow_dir(server_id, workflow_id) / HISTORY_DIRNAME
|
|
|
|
|
|
def get_server_history_entry_path(server_id: str, workflow_id: str, run_id: str) -> Path:
|
|
return get_server_history_dir(server_id, workflow_id) / f"{run_id}.json"
|
|
|
|
|
|
def list_server_workflow_dirs(server_id: str) -> list[Path]:
|
|
server_dir = get_server_data_dir(server_id)
|
|
if not server_dir.exists():
|
|
return []
|
|
return sorted(
|
|
[path for path in server_dir.iterdir() if path.is_dir() and not path.name.startswith(".")],
|
|
key=lambda path: path.name.lower(),
|
|
)
|
|
|
|
|
|
def get_legacy_server_workflows_dir(server_id: str) -> Path:
|
|
return get_server_data_dir(server_id) / "workflows"
|
|
|
|
|
|
def get_legacy_server_schemas_dir(server_id: str) -> Path:
|
|
return get_server_data_dir(server_id) / "schemas"
|