987 lines
40 KiB
Python
987 lines
40 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
from typing import Any
|
||
|
||
from fastapi import APIRouter, HTTPException
|
||
|
||
from calliope.comfyui.endpoints import client_for, endpoint_snapshot
|
||
from calliope.comfyui.lint import blocking_problem, workflow_warnings
|
||
from calliope.comfyui.metadata import media_io_from_schema, suggested_metadata
|
||
from calliope.comfyui.output_map import (
|
||
auto_output_map,
|
||
load_output_map,
|
||
mapping_problem,
|
||
needs_user_output_map,
|
||
validate_output_map,
|
||
)
|
||
from calliope.comfyui.parser import (
|
||
classify_usage,
|
||
effective_input_schema,
|
||
parse_dynamic_inputs,
|
||
parse_dynamic_outputs,
|
||
)
|
||
from calliope.comfyui.preflight import check_workflow_against
|
||
from calliope.comfyui.profiles import detect_prompt_profile
|
||
from calliope.comfyui.roles import usage_to_kind
|
||
from calliope.comfyui.ui_convert import convert_ui_workflow
|
||
from calliope.comfyui.validate import validate_input_values
|
||
from calliope.config import settings
|
||
from calliope.db import get_db, row_to_dict
|
||
from calliope.events.bus import event_bus
|
||
from calliope.models.schemas import (
|
||
WorkflowAnalyze,
|
||
WorkflowCheckRequest,
|
||
WorkflowConvertRequest,
|
||
WorkflowCreate,
|
||
WorkflowImportBatchRequest,
|
||
WorkflowImportScanRequest,
|
||
WorkflowTestRequest,
|
||
WorkflowUIFileItem,
|
||
WorkflowUpdate,
|
||
)
|
||
from calliope.queue.manager import queue_manager
|
||
from calliope.queue.workflow_test import (
|
||
TEST_SOURCE,
|
||
find_active_test_job,
|
||
list_test_jobs,
|
||
purge_workflow_tests,
|
||
)
|
||
|
||
router = APIRouter()
|
||
|
||
# HEL-681: metadata columns the lightweight list view exposes (everything
|
||
# except the heavy workflow_json / input_schema / output_map payloads).
|
||
_META_COLUMNS = (
|
||
"id, name, kind, usage, is_enabled, category, model_family, media_io, tags,"
|
||
" favorite, last_used_at, description, source, source_path, prompt_profile,"
|
||
" created_at, endpoint_id, usage_source, output_map_source"
|
||
)
|
||
|
||
|
||
def _serialize_workflow(row: Any) -> dict[str, Any]:
|
||
data = row_to_dict(row)
|
||
data["workflow_json"] = json.loads(data["workflow_json"])
|
||
stored = json.loads(data["input_schema"]) if data.get("input_schema") else []
|
||
data["input_schema"] = effective_input_schema(stored, data["workflow_json"])
|
||
# Outputs are pure graph derivation (no manual overrides) — always fresh,
|
||
# so parser upgrades reach old rows without a re-import.
|
||
data["output_schema"] = parse_dynamic_outputs(data["workflow_json"])
|
||
data["is_enabled"] = bool(data.get("is_enabled"))
|
||
data["warnings"] = workflow_warnings(data["workflow_json"])
|
||
raw_map = data.get("output_map")
|
||
data["output_map"] = load_output_map(raw_map)
|
||
# Pending flag: unknown output candidates the user has not mapped yet.
|
||
data["needs_mapping"] = (
|
||
mapping_problem(data["workflow_json"], raw_map, data.get("output_map_source")) is not None
|
||
)
|
||
# HEL-681: structured library metadata (raw JSON columns → values).
|
||
data["media_io"] = _resolved_media_io(
|
||
data.get("media_io"), data["input_schema"], data["output_schema"]
|
||
)
|
||
data["tags"] = _load_json_col(data.get("tags"), [])
|
||
data["favorite"] = bool(data.get("favorite"))
|
||
return data
|
||
|
||
|
||
def _load_json_col(raw: Any, default: Any) -> Any:
|
||
try:
|
||
value = json.loads(raw) if raw else None
|
||
except (json.JSONDecodeError, TypeError):
|
||
value = None
|
||
return value if value is not None else default
|
||
|
||
|
||
def _resolved_media_io(
|
||
raw: Any, inputs: list[dict[str, Any]], outputs: list[dict[str, Any]]
|
||
) -> dict[str, list[str]]:
|
||
"""Fill missing declarations only; an explicit empty list is meaningful."""
|
||
stored = _load_json_col(raw, {})
|
||
derived = media_io_from_schema(inputs, outputs)
|
||
if not isinstance(stored, dict):
|
||
return derived
|
||
return {
|
||
key: stored[key] if isinstance(stored.get(key), list) else derived[key]
|
||
for key in ("in", "out")
|
||
}
|
||
|
||
|
||
async def _connection_check(
|
||
endpoint_id: str | None, workflow_json: dict[str, Any], refresh: bool = False
|
||
) -> dict[str, Any]:
|
||
"""L2 preflight (HEL-643): workflow vs the target ComfyUI's /object_info.
|
||
|
||
An unreachable connection degrades to ``reachable: False`` with an honest
|
||
「无法核查」 message — it never blocks and never pretends to pass.
|
||
"""
|
||
endpoint = settings.resolve_comfyui_endpoint(endpoint_id)
|
||
if endpoint is None:
|
||
return {
|
||
"ok": False,
|
||
"reachable": False,
|
||
"endpoint_id": endpoint_id,
|
||
"endpoint_name": None,
|
||
"findings": [],
|
||
"message": "没有可用的 ComfyUI 连接，无法核查节点与模型",
|
||
}
|
||
base = {
|
||
"endpoint_id": endpoint.get("id"),
|
||
"endpoint_name": endpoint.get("name"),
|
||
"base_url": endpoint.get("base_url"),
|
||
}
|
||
client = client_for(endpoint)
|
||
try:
|
||
object_info = await client.object_info(force_refresh=refresh)
|
||
except Exception as exc:
|
||
return {
|
||
**base,
|
||
"ok": False,
|
||
"reachable": False,
|
||
"findings": [],
|
||
"message": f"无法核查（连不上该 ComfyUI）：{type(exc).__name__}",
|
||
}
|
||
finally:
|
||
await client.close()
|
||
findings = check_workflow_against(object_info, workflow_json)
|
||
ok = not any(f.get("level") == "error" for f in findings)
|
||
return {**base, "ok": ok, "reachable": True, "findings": findings}
|
||
|
||
|
||
@router.post("/analyze")
|
||
async def analyze_workflow(payload: WorkflowAnalyze) -> dict[str, Any]:
|
||
inputs = parse_dynamic_inputs(payload.workflow_json)
|
||
outputs = parse_dynamic_outputs(payload.workflow_json)
|
||
judgment = classify_usage(payload.workflow_json)
|
||
result: dict[str, Any] = {
|
||
"inputs": inputs,
|
||
"outputs": outputs,
|
||
"suggested_profile": detect_prompt_profile(payload.workflow_json),
|
||
"warnings": workflow_warnings(payload.workflow_json),
|
||
"usage": judgment["usage"],
|
||
"usage_ambiguous": judgment["ambiguous"],
|
||
"needs_mapping": judgment["needs_mapping"],
|
||
"usage_problem": judgment["problem"],
|
||
"usage_evidence": judgment["evidence"],
|
||
"usable_usage": judgment["usable_usage"],
|
||
}
|
||
# L2 (optional): with a target connection the analyze response also
|
||
# reports which nodes/models that ComfyUI actually has (HEL-643).
|
||
if payload.endpoint_id:
|
||
result["connection_check"] = await _connection_check(
|
||
payload.endpoint_id, payload.workflow_json, payload.refresh
|
||
)
|
||
return result
|
||
|
||
|
||
def _stored_schema(payload_schema: list[dict[str, Any]] | None, workflow: dict[str, Any]) -> str:
|
||
"""Schema to persist: the confirm page's corrected copy, else the auto one."""
|
||
if payload_schema and all(isinstance(e, dict) for e in payload_schema):
|
||
return json.dumps(payload_schema)
|
||
return json.dumps(parse_dynamic_inputs(workflow))
|
||
|
||
|
||
@router.post("/{workflow_id}/reanalyze")
|
||
async def reanalyze_workflow(workflow_id: int) -> dict[str, Any]:
|
||
"""Recompute usage and input/output schemas from the stored workflow JSON.
|
||
|
||
Workflows imported before a parser upgrade keep stale (often empty)
|
||
schemas — this refreshes them without re-importing. A usage or output
|
||
mapping the user set by hand is never overwritten; only ``auto`` (and
|
||
pending NULL) rows get the new judgment.
|
||
"""
|
||
conn = get_db(settings.db_path)
|
||
try:
|
||
row = conn.execute("SELECT * FROM workflows WHERE id = ?", (workflow_id,)).fetchone()
|
||
if not row:
|
||
raise HTTPException(status_code=404, detail="Workflow not found")
|
||
data = row_to_dict(row)
|
||
workflow = json.loads(data["workflow_json"])
|
||
judgment = classify_usage(workflow)
|
||
usage = data["usage"]
|
||
usage_source = data["usage_source"]
|
||
if data["usage_source"] != "user" and judgment["usage"]:
|
||
usage = judgment["usage"]
|
||
usage_source = "auto"
|
||
stored = json.loads(data["input_schema"]) if data["input_schema"] else []
|
||
schema = effective_input_schema(stored, workflow)
|
||
outputs = parse_dynamic_outputs(workflow)
|
||
if data["output_map_source"] == "user":
|
||
output_map, output_map_source = data["output_map"], "user"
|
||
elif needs_user_output_map(workflow):
|
||
# Unknown output candidates: no auto map, and the row must not
|
||
# generate until the user designates the output nodes.
|
||
output_map, output_map_source = None, None
|
||
else:
|
||
output_map, output_map_source = json.dumps(auto_output_map(workflow)), "auto"
|
||
media_io = _resolved_media_io(data["media_io"], schema, outputs)
|
||
media_io["in"] = media_io_from_schema(schema, outputs)["in"]
|
||
is_enabled = data["is_enabled"]
|
||
if output_map_source is None or usage is None:
|
||
is_enabled = 0
|
||
conn.execute(
|
||
"UPDATE workflows SET input_schema = ?, output_schema = ?,"
|
||
" usage = ?, usage_source = ?, output_map = ?, output_map_source = ?,"
|
||
" is_enabled = ?, media_io = ? WHERE id = ?",
|
||
(
|
||
json.dumps(schema),
|
||
json.dumps(outputs),
|
||
usage,
|
||
usage_source,
|
||
output_map,
|
||
output_map_source,
|
||
is_enabled,
|
||
json.dumps(media_io),
|
||
workflow_id,
|
||
),
|
||
)
|
||
conn.commit()
|
||
fresh = conn.execute(
|
||
"SELECT * FROM workflows WHERE id = ?", (workflow_id,)
|
||
).fetchone()
|
||
return _serialize_workflow(fresh)
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def _last_test_summary(job: dict[str, Any]) -> dict[str, Any]:
|
||
"""Slim "上次测试" record for a workflow row (queried live, never stored)."""
|
||
try:
|
||
outputs = json.loads(job.get("output_paths_json") or "[]")
|
||
except (json.JSONDecodeError, TypeError):
|
||
outputs = []
|
||
return {
|
||
"id": job["id"],
|
||
"status": job.get("status"),
|
||
"phase": job.get("phase"),
|
||
"error": job.get("error"),
|
||
"created_at": job.get("created_at"),
|
||
"completed_at": job.get("completed_at"),
|
||
"output_count": len(outputs) if isinstance(outputs, list) else 0,
|
||
}
|
||
|
||
|
||
@router.get("")
|
||
async def list_workflows(view: str | None = None) -> list[dict[str, Any]]:
|
||
conn = get_db(settings.db_path)
|
||
try:
|
||
if view == "meta":
|
||
# HEL-681: metadata-only list for 150+ scale browsing — no
|
||
# workflow_json in the response. Only legacy/missing media
|
||
# declarations need graph parsing. `needs_mapping` here is the
|
||
# cheap pending check (usage/output_map not yet settled); rows
|
||
# that never generated keep showing as pending exactly like the
|
||
# full view reports them.
|
||
rows = conn.execute(
|
||
f"SELECT {_META_COLUMNS} FROM workflows ORDER BY favorite DESC,"
|
||
" last_used_at DESC, id DESC"
|
||
).fetchall()
|
||
out = []
|
||
for row in rows:
|
||
data = row_to_dict(row)
|
||
data["is_enabled"] = bool(data.get("is_enabled"))
|
||
data["favorite"] = bool(data.get("favorite"))
|
||
media = _load_json_col(data.get("media_io"), {})
|
||
if isinstance(media, dict) and all(
|
||
isinstance(media.get(key), list) for key in ("in", "out")
|
||
):
|
||
data["media_io"] = {key: media[key] for key in ("in", "out")}
|
||
else:
|
||
legacy = conn.execute(
|
||
"SELECT workflow_json, input_schema FROM workflows WHERE id = ?",
|
||
(data["id"],),
|
||
).fetchone()
|
||
graph = json.loads(legacy["workflow_json"])
|
||
schema = effective_input_schema(
|
||
_load_json_col(legacy["input_schema"], []), graph
|
||
)
|
||
data["media_io"] = _resolved_media_io(
|
||
data.get("media_io"), schema, parse_dynamic_outputs(graph)
|
||
)
|
||
data["tags"] = _load_json_col(data.get("tags"), [])
|
||
data["needs_mapping"] = (
|
||
data.get("usage") is None or data.get("output_map_source") is None
|
||
)
|
||
out.append(data)
|
||
return out
|
||
rows = conn.execute("SELECT * FROM workflows ORDER BY id DESC").fetchall()
|
||
out = [_serialize_workflow(r) for r in rows]
|
||
# 上次测试 state is read live from the newest test job per workflow —
|
||
# no cached column that could go stale (HEL-643).
|
||
latest: dict[int, dict[str, Any]] = {}
|
||
for jr in conn.execute(
|
||
"SELECT * FROM jobs WHERE origin = 'workflow_test' AND id IN"
|
||
" (SELECT MAX(id) FROM jobs WHERE origin = 'workflow_test' GROUP BY workflow_id)"
|
||
).fetchall():
|
||
latest[int(jr["workflow_id"])] = _last_test_summary(dict(jr))
|
||
for data in out:
|
||
data["last_test"] = latest.get(int(data["id"]))
|
||
return out
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
@router.post("/{workflow_id}/check")
|
||
async def check_workflow(workflow_id: int, payload: WorkflowCheckRequest) -> dict[str, Any]:
|
||
"""L2 re-check of a saved workflow against its (or a chosen) connection.
|
||
|
||
The user installs missing nodes/models in ComfyUI, comes back here and
|
||
re-checks — no re-import needed.
|
||
"""
|
||
conn = get_db(settings.db_path)
|
||
try:
|
||
row = conn.execute(
|
||
"SELECT workflow_json, endpoint_id FROM workflows WHERE id = ?", (workflow_id,)
|
||
).fetchone()
|
||
finally:
|
||
conn.close()
|
||
if not row:
|
||
raise HTTPException(status_code=404, detail="Workflow not found")
|
||
endpoint_id = payload.endpoint_id or row["endpoint_id"]
|
||
return await _connection_check(endpoint_id, json.loads(row["workflow_json"]), payload.refresh)
|
||
|
||
|
||
@router.post("/dedupe")
|
||
async def dedupe_workflows() -> dict[str, Any]:
|
||
"""Delete byte-identical duplicate workflows, keeping the lowest id.
|
||
|
||
Two rows are duplicates only when name, kind AND workflow_json all match —
|
||
same name with different JSON is a user's variant, never touched.
|
||
"""
|
||
conn = get_db(settings.db_path)
|
||
try:
|
||
rows = conn.execute(
|
||
"SELECT id, name, kind, workflow_json FROM workflows ORDER BY id"
|
||
).fetchall()
|
||
seen: dict[tuple[str, str, str], int] = {}
|
||
drop_ids: list[int] = []
|
||
for r in rows:
|
||
key = (r["name"], r["kind"], r["workflow_json"])
|
||
if key in seen:
|
||
drop_ids.append(r["id"])
|
||
else:
|
||
seen[key] = r["id"]
|
||
removed = 0
|
||
for wid in drop_ids:
|
||
cur = conn.execute("DELETE FROM workflows WHERE id = ?", (wid,))
|
||
removed += cur.rowcount
|
||
conn.commit()
|
||
return {
|
||
"ok": True,
|
||
"removed": removed,
|
||
"kept_ids": sorted(seen.values()),
|
||
"removed_ids": drop_ids,
|
||
}
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
@router.post("")
|
||
async def create_workflow(payload: WorkflowCreate) -> dict[str, Any]:
|
||
judgment = classify_usage(payload.workflow_json)
|
||
usage = payload.usage or judgment["usage"]
|
||
usage_source = "user" if payload.usage else "auto"
|
||
kind = payload.kind or usage_to_kind(usage)
|
||
if kind is None:
|
||
# Indeterminate and nothing chosen: classify_output hint keeps old
|
||
# callers working while the workflow stays pending confirmation.
|
||
out_kinds = {
|
||
o.get("kind")
|
||
for o in parse_dynamic_outputs(payload.workflow_json)
|
||
if o.get("kind") in ("image", "video")
|
||
}
|
||
kind = "video" if out_kinds == {"video"} else "image"
|
||
profile = payload.prompt_profile or detect_prompt_profile(payload.workflow_json)
|
||
# HEL-638: an explicit user output map wins; otherwise the trusted auto
|
||
# map. Either way a graph with unknown output candidates is unusable
|
||
# until the user designates the output nodes.
|
||
if payload.output_map:
|
||
problems = validate_output_map(payload.workflow_json, payload.output_map)
|
||
if problems:
|
||
raise HTTPException(status_code=422, detail="；".join(problems))
|
||
output_map = json.dumps(payload.output_map)
|
||
output_map_source = payload.output_map_source or "user"
|
||
else:
|
||
auto_map = auto_output_map(payload.workflow_json)
|
||
output_map = json.dumps(auto_map) if auto_map else None
|
||
output_map_source = "auto" if auto_map else None
|
||
mapping_ok = not needs_user_output_map(payload.workflow_json) or bool(payload.output_map)
|
||
schema = _stored_schema(payload.input_schema, payload.workflow_json)
|
||
outputs = parse_dynamic_outputs(payload.workflow_json)
|
||
media_io = media_io_from_schema(json.loads(schema), outputs)
|
||
conn = get_db(settings.db_path)
|
||
try:
|
||
# Byte-identical re-imports are the main way the library fills up with
|
||
# same-named clones (the "@" typeahead then shows 20 identical rows).
|
||
existing = conn.execute(
|
||
"SELECT id, name FROM workflows WHERE name = ? AND kind = ? AND workflow_json = ?",
|
||
(payload.name, kind, json.dumps(payload.workflow_json)),
|
||
).fetchone()
|
||
if existing:
|
||
raise HTTPException(
|
||
status_code=409,
|
||
detail=(
|
||
f"An identical workflow already exists (id {existing['id']},"
|
||
f" “{existing['name']}”)."
|
||
),
|
||
)
|
||
# Unconfirmed workflows (ambiguous usage / unknown outputs without a
|
||
# user mapping) are stored but disabled: they must not reach the
|
||
# generate pages until the user confirms usage AND, for unknown
|
||
# output nodes, designates the outputs on the connections page.
|
||
is_enabled = 1 if (usage and mapping_ok) else 0
|
||
cur = conn.execute(
|
||
"""
|
||
INSERT INTO workflows (name, kind, workflow_json, input_schema,
|
||
output_schema, description, prompt_profile,
|
||
endpoint_id, is_enabled, usage, usage_source,
|
||
output_map, output_map_source, media_io)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
""",
|
||
(
|
||
payload.name,
|
||
kind,
|
||
json.dumps(payload.workflow_json),
|
||
schema,
|
||
json.dumps(outputs),
|
||
payload.description,
|
||
profile,
|
||
payload.endpoint_id,
|
||
is_enabled,
|
||
usage,
|
||
usage_source,
|
||
output_map,
|
||
output_map_source,
|
||
json.dumps(media_io),
|
||
),
|
||
)
|
||
conn.commit()
|
||
row = conn.execute("SELECT * FROM workflows WHERE id = ?", (cur.lastrowid,)).fetchone()
|
||
return _serialize_workflow(row)
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
@router.get("/{workflow_id}")
|
||
async def get_workflow(workflow_id: int) -> dict[str, Any]:
|
||
conn = get_db(settings.db_path)
|
||
try:
|
||
row = conn.execute("SELECT * FROM workflows WHERE id = ?", (workflow_id,)).fetchone()
|
||
if not row:
|
||
raise HTTPException(status_code=404, detail="Workflow not found")
|
||
return _serialize_workflow(row)
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
@router.patch("/{workflow_id}")
|
||
async def update_workflow(workflow_id: int, payload: WorkflowUpdate) -> dict[str, Any]:
|
||
conn = get_db(settings.db_path)
|
||
try:
|
||
existing = conn.execute("SELECT * FROM workflows WHERE id = ?", (workflow_id,)).fetchone()
|
||
if not existing:
|
||
raise HTTPException(status_code=404, detail="Workflow not found")
|
||
data = payload.model_dump(exclude_unset=True)
|
||
if "is_enabled" in data and data["is_enabled"] is not None:
|
||
data["is_enabled"] = 1 if data["is_enabled"] else 0
|
||
# HEL-681: library metadata patch fields (favorite/tags/category/
|
||
# model_family) — booleans to ints, lists to JSON text.
|
||
if "favorite" in data and data["favorite"] is not None:
|
||
data["favorite"] = 1 if data["favorite"] else 0
|
||
if data.get("tags") is not None:
|
||
data["tags"] = json.dumps(data["tags"])
|
||
workflow_json = json.loads(existing["workflow_json"])
|
||
# HEL-638: output_map payload. A non-empty list validates against the
|
||
# stored JSON and becomes the user decision; an explicit empty list
|
||
# resets the row to pending (disabled).
|
||
if "output_map" in data:
|
||
if data["output_map"]:
|
||
problems = validate_output_map(workflow_json, data["output_map"])
|
||
if problems:
|
||
raise HTTPException(status_code=422, detail="；".join(problems))
|
||
data["output_map"] = json.dumps(data["output_map"])
|
||
data.setdefault("output_map_source", "user")
|
||
else:
|
||
data["output_map"] = None
|
||
data["output_map_source"] = None
|
||
data["is_enabled"] = 0
|
||
if data.get("usage"):
|
||
# Confirming / correcting the usage is an explicit user decision:
|
||
# kind (output media) follows the usage. Whether it re-enables is
|
||
# decided by the gate below — with unknown output nodes the user
|
||
# must also designate the output nodes (HEL-638).
|
||
data.setdefault("usage_source", "user")
|
||
data["kind"] = usage_to_kind(data["usage"]) or existing["kind"]
|
||
if "input_schema" in data and data["input_schema"] is not None:
|
||
if not all(isinstance(e, dict) for e in data["input_schema"]):
|
||
raise HTTPException(status_code=422, detail="input_schema 必须是参数对象数组")
|
||
media = _resolved_media_io(
|
||
existing["media_io"], [], parse_dynamic_outputs(workflow_json)
|
||
)
|
||
schema = effective_input_schema(data["input_schema"], workflow_json)
|
||
media["in"] = media_io_from_schema(schema, [])["in"]
|
||
data["media_io"] = json.dumps(media)
|
||
data["input_schema"] = json.dumps(data["input_schema"])
|
||
elif existing["media_io"] is None:
|
||
schema = effective_input_schema(
|
||
_load_json_col(existing["input_schema"], []), workflow_json
|
||
)
|
||
data["media_io"] = json.dumps(
|
||
media_io_from_schema(schema, parse_dynamic_outputs(workflow_json))
|
||
)
|
||
# endpoint_id: explicit null clears the binding; other Nones are skips
|
||
# (an output_map reset keeps its explicit NULLs).
|
||
explicit_null = {"endpoint_id", "output_map", "output_map_source"} & {
|
||
k for k, v in data.items() if v is None
|
||
}
|
||
data = {k: v for k, v in data.items() if v is not None or k in explicit_null}
|
||
|
||
# Enable gate: usage must be settled and, when the graph carries
|
||
# unknown custom output nodes, a user output mapping must exist.
|
||
usage_ok = bool(data.get("usage") or existing["usage"])
|
||
map_raw = data.get("output_map", existing["output_map"])
|
||
map_source = data.get("output_map_source", existing["output_map_source"])
|
||
map_problem = mapping_problem(workflow_json, map_raw, map_source)
|
||
if data.get("is_enabled"):
|
||
if not usage_ok:
|
||
raise HTTPException(status_code=400, detail="请先选择工作流用途，再启用该工作流")
|
||
if map_problem:
|
||
raise HTTPException(status_code=400, detail=map_problem)
|
||
elif (data.get("usage") or data.get("output_map")) and "is_enabled" not in data:
|
||
# A usage-confirm or output-map-confirm path: enabled only when
|
||
# both decisions are settled.
|
||
data["is_enabled"] = 1 if (usage_ok and not map_problem) else 0
|
||
if data:
|
||
fields = ", ".join(f"{k} = :{k}" for k in data)
|
||
data["id"] = workflow_id
|
||
conn.execute(f"UPDATE workflows SET {fields} WHERE id = :id", data)
|
||
conn.commit()
|
||
row = conn.execute("SELECT * FROM workflows WHERE id = ?", (workflow_id,)).fetchone()
|
||
return _serialize_workflow(row)
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def _job_public(job: dict[str, Any]) -> dict[str, Any]:
|
||
out = dict(job)
|
||
out["payload"] = json.loads(job.get("payload_json") or "{}")
|
||
out["output_paths"] = json.loads(job.get("output_paths_json") or "[]")
|
||
out.pop("payload_json", None)
|
||
out.pop("output_paths_json", None)
|
||
return out
|
||
|
||
|
||
@router.post("/{workflow_id}/test")
|
||
async def test_workflow(workflow_id: int, payload: WorkflowTestRequest) -> dict[str, Any]:
|
||
"""Launch (or re-attach to) one REAL test run of a workflow (HEL-643 L3).
|
||
|
||
The job goes through the ordinary queue → ComfyUI → outputs pipeline;
|
||
only its artifacts and bookkeeping are isolated (origin='workflow_test').
|
||
A dry-run setting is refused outright — placeholder output must never
|
||
count as a test pass. Repeat clicks reuse the in-flight job.
|
||
"""
|
||
from calliope import config
|
||
|
||
if config.settings.dry_run:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail=(
|
||
"测试工作流必须真实运行：当前设置开启了 Dry-run（占位模式），"
|
||
"请到「设置」关闭 Dry-run 后再测试"
|
||
),
|
||
)
|
||
conn = get_db(settings.db_path)
|
||
try:
|
||
wf = conn.execute(
|
||
"SELECT id, kind, workflow_json, input_schema, output_map, output_map_source,"
|
||
" endpoint_id FROM workflows WHERE id = ?",
|
||
(workflow_id,),
|
||
).fetchone()
|
||
if not wf:
|
||
raise HTTPException(status_code=404, detail="Workflow not found")
|
||
workflow_json = json.loads(wf["workflow_json"])
|
||
broken = blocking_problem(workflow_json)
|
||
if broken:
|
||
raise HTTPException(status_code=400, detail=broken)
|
||
map_problem = mapping_problem(workflow_json, wf["output_map"], wf["output_map_source"])
|
||
if map_problem:
|
||
raise HTTPException(status_code=400, detail=map_problem)
|
||
schema = effective_input_schema(
|
||
json.loads(wf["input_schema"]) if wf["input_schema"] else [], workflow_json
|
||
)
|
||
input_values, problems = validate_input_values(schema, payload.input_values)
|
||
if problems:
|
||
raise HTTPException(status_code=400, detail="参数有误：" + "；".join(problems))
|
||
|
||
endpoint_id = payload.endpoint_id or wf["endpoint_id"] or None
|
||
snapshot = endpoint_snapshot(endpoint_id)
|
||
active = find_active_test_job(workflow_id, (snapshot or {}).get("id"))
|
||
if active is not None:
|
||
return {"ok": True, "job": _job_public(active), "reused": True}
|
||
|
||
from calliope.routers.playground import ensure_playground_project
|
||
|
||
pid = ensure_playground_project(conn)
|
||
finally:
|
||
conn.close()
|
||
|
||
job = queue_manager.enqueue(
|
||
project_id=pid,
|
||
kind=wf["kind"],
|
||
workflow_id=workflow_id,
|
||
scene_id=None,
|
||
payload={
|
||
"input_values": input_values,
|
||
"source": TEST_SOURCE,
|
||
**({"endpoint_id": payload.endpoint_id} if payload.endpoint_id else {}),
|
||
},
|
||
origin="workflow_test",
|
||
)
|
||
await event_bus.publish(
|
||
"job.created",
|
||
{"job_id": job["id"], "kind": job["kind"], "source": TEST_SOURCE,
|
||
"project_id": job["project_id"], "workflow_test": True},
|
||
)
|
||
return {"ok": True, "job": _job_public(job), "reused": False}
|
||
|
||
|
||
@router.get("/{workflow_id}/tests")
|
||
async def list_workflow_tests(workflow_id: int) -> list[dict[str, Any]]:
|
||
conn = get_db(settings.db_path)
|
||
try:
|
||
row = conn.execute("SELECT id FROM workflows WHERE id = ?", (workflow_id,)).fetchone()
|
||
finally:
|
||
conn.close()
|
||
if not row:
|
||
raise HTTPException(status_code=404, detail="Workflow not found")
|
||
return list_test_jobs(workflow_id)
|
||
|
||
|
||
@router.post("/{workflow_id}/clear-tests")
|
||
async def clear_workflow_tests(workflow_id: int) -> dict[str, Any]:
|
||
"""Delete every test record and artifact for this workflow."""
|
||
conn = get_db(settings.db_path)
|
||
try:
|
||
row = conn.execute("SELECT id FROM workflows WHERE id = ?", (workflow_id,)).fetchone()
|
||
finally:
|
||
conn.close()
|
||
if not row:
|
||
raise HTTPException(status_code=404, detail="Workflow not found")
|
||
try:
|
||
removed = purge_workflow_tests(workflow_id)
|
||
except RuntimeError as exc:
|
||
raise HTTPException(status_code=409, detail=str(exc)) from exc
|
||
await event_bus.publish("job.deleted", {"workflow_id": workflow_id, "source": TEST_SOURCE})
|
||
return {"ok": True, "removed": removed}
|
||
|
||
|
||
@router.delete("/{workflow_id}")
|
||
async def delete_workflow(workflow_id: int) -> dict[str, bool]:
|
||
conn = get_db(settings.db_path)
|
||
try:
|
||
# HEL-619: refuse to delete while queued/running jobs still use the
|
||
# workflow — deleting sets their workflow_id NULL and the job dies
|
||
# with "工作流已被删除" only after the user waits for it.
|
||
row = conn.execute(
|
||
"SELECT id FROM workflows WHERE id = ?", (workflow_id,)
|
||
).fetchone()
|
||
if not row:
|
||
raise HTTPException(status_code=404, detail="Workflow not found")
|
||
pending = conn.execute(
|
||
"SELECT COUNT(*) AS n FROM jobs"
|
||
" WHERE workflow_id = ? AND status IN ('pending', 'running')",
|
||
(workflow_id,),
|
||
).fetchone()["n"]
|
||
if pending:
|
||
raise HTTPException(
|
||
status_code=409,
|
||
detail=(
|
||
f"该工作流还有 {pending} 个排队或执行中的任务，"
|
||
"请先等任务完成或取消任务，再删除该工作流"
|
||
),
|
||
)
|
||
# Test records/artifacts belong to the workflow — cascade (HEL-643).
|
||
# The 409 above guarantees none are in flight here.
|
||
purge_workflow_tests(workflow_id, allow_active=True)
|
||
conn.execute("DELETE FROM workflows WHERE id = ?", (workflow_id,))
|
||
conn.commit()
|
||
return {"ok": True}
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# HEL-681 (S2): UI→API conversion dry-run + batch import + repo scan.
|
||
# The convert endpoint never writes; import-batch converts again server-side
|
||
# and skips anything uncertain (needs_confirmation / lint-blocked / conflict)
|
||
# — repeated runs import nothing new, so the batch is idempotent by design.
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _convert_one(item: WorkflowUIFileItem) -> tuple[dict[str, Any], dict[str, Any] | None]:
|
||
"""(report, converted workflow) for one UI file — shared by every
|
||
convert/import path so the analysis can never drift between them."""
|
||
result = convert_ui_workflow(item.workflow_json)
|
||
stem = (
|
||
item.source_path.replace("\\", "/").rsplit("/", 1)[-1].removesuffix(".json")
|
||
if item.source_path
|
||
else None
|
||
)
|
||
name = item.name or stem or "未命名工作流"
|
||
report: dict[str, Any] = {
|
||
"name": name,
|
||
"source_path": item.source_path,
|
||
"status": result.status,
|
||
"needs_confirmation": result.needs_confirmation,
|
||
"issues": [
|
||
{"node_id": i.node_id, "code": i.code, "level": i.level, "message": i.message}
|
||
for i in result.issues
|
||
],
|
||
}
|
||
if result.workflow is None:
|
||
report["importable"] = False
|
||
report["skip_reason"] = "转换失败：" + (
|
||
result.issues[0].message if result.issues else "未知错误"
|
||
)
|
||
return report, None
|
||
suggested = suggested_metadata(result.workflow, item.source_path, item.workflow_json)
|
||
report["suggested"] = suggested
|
||
# The import gate re-runs the product's own checks — a converted graph
|
||
# that the lint would block (missing model link etc.) is reported, not
|
||
# silently imported.
|
||
broken = blocking_problem(result.workflow)
|
||
if broken:
|
||
report["importable"] = False
|
||
report["skip_reason"] = broken
|
||
report["status"] = "warn"
|
||
return report, result.workflow
|
||
if result.needs_confirmation:
|
||
report["importable"] = False
|
||
report["skip_reason"] = "存在无法可靠映射的控件，需人工确认后重导"
|
||
return report, result.workflow
|
||
report["importable"] = True
|
||
return report, result.workflow
|
||
|
||
|
||
@router.post("/convert")
|
||
async def convert_workflows(payload: WorkflowConvertRequest) -> dict[str, Any]:
|
||
"""Dry-run UI→API conversion for a batch of files: per-file ok/warn/fail,
|
||
issues, suggested metadata and importability. Nothing is stored."""
|
||
files = []
|
||
object_info: dict[str, Any] | None = None
|
||
client = None
|
||
if payload.endpoint_id:
|
||
endpoint = settings.resolve_comfyui_endpoint(payload.endpoint_id)
|
||
if endpoint is not None:
|
||
client = client_for(endpoint)
|
||
try:
|
||
object_info = await client.object_info(force_refresh=payload.refresh)
|
||
except Exception:
|
||
object_info = None
|
||
try:
|
||
for item in payload.files:
|
||
report, converted = _convert_one(item)
|
||
if object_info is not None and report.get("importable") and converted:
|
||
findings = check_workflow_against(object_info, converted)
|
||
errors = [f for f in findings if f.get("level") == "error"]
|
||
report["preflight"] = {
|
||
"endpoint_id": payload.endpoint_id,
|
||
"ok": not errors,
|
||
"findings": findings,
|
||
}
|
||
if errors:
|
||
report["importable"] = False
|
||
report["skip_reason"] = errors[0].get("message") or "preflight 未通过"
|
||
if payload.include_workflows:
|
||
report["workflow_json"] = converted
|
||
files.append(report)
|
||
finally:
|
||
if client is not None:
|
||
await client.close()
|
||
counts = {key: sum(1 for f in files if f["status"] == key) for key in ("ok", "warn", "fail")}
|
||
return {
|
||
"total": len(files),
|
||
**counts,
|
||
"needs_confirmation": sum(1 for f in files if f["needs_confirmation"]),
|
||
"importable": sum(1 for f in files if f.get("importable")),
|
||
"files": files,
|
||
}
|
||
|
||
|
||
def _insert_converted(
|
||
conn: Any, item: WorkflowUIFileItem, report: dict[str, Any], converted: dict[str, Any]
|
||
) -> dict[str, Any]:
|
||
"""Insert one converted workflow row (import-batch's write path).
|
||
|
||
Mirrors create_workflow's enable gate: usage must be settled (auto or the
|
||
wizard's confirmation) and unknown outputs need a mapping; anything less
|
||
lands disabled exactly like a manual import would.
|
||
"""
|
||
suggested = report["suggested"]
|
||
judgment = classify_usage(converted)
|
||
usage = item.usage or judgment["usage"]
|
||
usage_source = "user" if item.usage else "auto"
|
||
kind = usage_to_kind(usage) or suggested["kind"]
|
||
profile = detect_prompt_profile(converted)
|
||
if item.output_map:
|
||
problems = validate_output_map(converted, item.output_map)
|
||
if problems:
|
||
raise HTTPException(status_code=422, detail="；".join(problems))
|
||
output_map = json.dumps(item.output_map)
|
||
output_map_source = "user"
|
||
else:
|
||
auto_map = auto_output_map(converted)
|
||
output_map = json.dumps(auto_map) if auto_map else None
|
||
output_map_source = "auto" if auto_map else None
|
||
mapping_ok = not needs_user_output_map(converted) or bool(item.output_map)
|
||
workflow_text = json.dumps(converted)
|
||
existing = conn.execute(
|
||
"SELECT id FROM workflows WHERE name = ? AND kind = ? AND workflow_json = ?",
|
||
(report["name"], kind, workflow_text),
|
||
).fetchone()
|
||
if existing:
|
||
return {
|
||
"result": "skipped",
|
||
"skip_reason": "conflict",
|
||
"existing_id": existing["id"],
|
||
}
|
||
is_enabled = 1 if (usage and mapping_ok) else 0
|
||
schema = _stored_schema(item.input_schema, converted)
|
||
outputs = parse_dynamic_outputs(converted)
|
||
cur = conn.execute(
|
||
"""
|
||
INSERT INTO workflows (name, kind, workflow_json, input_schema, output_schema,
|
||
description, prompt_profile, endpoint_id, is_enabled,
|
||
usage, usage_source, output_map, output_map_source,
|
||
category, model_family, media_io, tags, favorite,
|
||
source, source_path)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, 'ui_convert', ?)
|
||
""",
|
||
(
|
||
report["name"],
|
||
kind,
|
||
workflow_text,
|
||
schema,
|
||
json.dumps(outputs),
|
||
item.description if item.description is not None else suggested["description"],
|
||
profile,
|
||
is_enabled,
|
||
usage,
|
||
usage_source,
|
||
output_map,
|
||
output_map_source,
|
||
suggested["category"],
|
||
suggested["model_family"],
|
||
json.dumps(media_io_from_schema(json.loads(schema), outputs)),
|
||
json.dumps(item.tags or []),
|
||
item.source_path,
|
||
),
|
||
)
|
||
return {"result": "imported", "id": cur.lastrowid}
|
||
|
||
|
||
async def _run_import(items: list[WorkflowUIFileItem]) -> dict[str, Any]:
|
||
reports: list[dict[str, Any]] = []
|
||
conn = get_db(settings.db_path)
|
||
try:
|
||
for item in items:
|
||
report, converted = _convert_one(item)
|
||
if not report.get("importable"):
|
||
report["result"] = "skipped"
|
||
reports.append(report)
|
||
continue
|
||
try:
|
||
outcome = _insert_converted(conn, item, report, converted)
|
||
except HTTPException as exc:
|
||
report["result"] = "failed"
|
||
report["skip_reason"] = str(exc.detail)
|
||
reports.append(report)
|
||
continue
|
||
conn.commit()
|
||
report.update(outcome)
|
||
reports.append(report)
|
||
finally:
|
||
conn.close()
|
||
imported = [r for r in reports if r.get("result") == "imported"]
|
||
skipped = [
|
||
{"name": r["name"], "source_path": r["source_path"], "reason": r.get("skip_reason")}
|
||
for r in reports
|
||
if r.get("result") != "imported"
|
||
]
|
||
return {
|
||
"total": len(reports),
|
||
"imported": len(imported),
|
||
"skipped": len(skipped),
|
||
"imported_ids": [r["id"] for r in imported],
|
||
"skipped_files": skipped,
|
||
"files": reports,
|
||
}
|
||
|
||
|
||
@router.post("/import-batch")
|
||
async def import_workflows_batch(payload: WorkflowImportBatchRequest) -> dict[str, Any]:
|
||
"""Batch-import UI-format workflows: convert server-side, insert what is
|
||
cleanly convertible, skip conflicts and anything uncertain. Re-running
|
||
with the same files creates zero new rows."""
|
||
return await _run_import(payload.files)
|
||
|
||
|
||
@router.post("/import-scan")
|
||
async def import_workflows_scan(payload: WorkflowImportScanRequest) -> dict[str, Any]:
|
||
"""Scan a server-side directory for UI-format workflow JSONs and run the
|
||
convert (dry_run, default) or import pipeline over every file found."""
|
||
from pathlib import Path
|
||
|
||
root = Path(payload.root)
|
||
if not root.is_absolute():
|
||
candidates = [
|
||
Path.cwd() / root,
|
||
Path(__file__).resolve().parents[4] / root,
|
||
]
|
||
root = next((c for c in candidates if c.is_dir()), root)
|
||
if not root.is_dir():
|
||
raise HTTPException(status_code=404, detail=f"目录不存在：{payload.root}")
|
||
items: list[WorkflowUIFileItem] = []
|
||
for path in sorted(root.rglob("*.json")):
|
||
try:
|
||
data = json.loads(path.read_text(encoding="utf-8"))
|
||
except (json.JSONDecodeError, OSError):
|
||
continue
|
||
if not (isinstance(data, dict) and isinstance(data.get("nodes"), list)):
|
||
continue # API-format or non-workflow files: not this pipeline's job
|
||
items.append(
|
||
WorkflowUIFileItem(
|
||
source_path=str(path.relative_to(root)), workflow_json=data
|
||
)
|
||
)
|
||
if payload.dry_run:
|
||
files = [_convert_one(item)[0] for item in items]
|
||
counts = {
|
||
key: sum(1 for f in files if f["status"] == key) for key in ("ok", "warn", "fail")
|
||
}
|
||
return {
|
||
"total": len(files),
|
||
**counts,
|
||
"needs_confirmation": sum(1 for f in files if f["needs_confirmation"]),
|
||
"importable": sum(1 for f in files if f.get("importable")),
|
||
"files": files,
|
||
}
|
||
return await _run_import(items)
|
||
|
||
|
||
@router.post("/{workflow_id}/touch")
|
||
async def touch_workflow(workflow_id: int) -> dict[str, bool]:
|
||
"""Record a use (最近使用) — called by the generate pipeline on enqueue."""
|
||
conn = get_db(settings.db_path)
|
||
try:
|
||
row = conn.execute("SELECT id FROM workflows WHERE id = ?", (workflow_id,)).fetchone()
|
||
if not row:
|
||
raise HTTPException(status_code=404, detail="Workflow not found")
|
||
conn.execute(
|
||
"UPDATE workflows SET last_used_at = CURRENT_TIMESTAMP WHERE id = ?",
|
||
(workflow_id,),
|
||
)
|
||
conn.commit()
|
||
return {"ok": True}
|
||
finally:
|
||
conn.close()
|