cbpark84
1c919f1106
- run_batch(): sequential|concurrent|auto - chain(): 파이프라인 체이닝 - schema(): 스키마 자동 추출 - ResultCache: LRU+TTL 캐시 - preview_callback: 실시간 프리뷰 - upload_file/audio/video() - BridgeInputSEED: random/increment/fixed - MCPServer: MCP Tool 자동 생성
105 lines
3.1 KiB
Python
105 lines
3.1 KiB
Python
"""BridgeInput/Output 스키마 자동 추출 및 검증."""
|
|
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
BRIDGE_INPUT_PREFIX = "BridgeInput"
|
|
BRIDGE_OUTPUT = "BridgeOutput"
|
|
|
|
|
|
def extract_schema(workflow: dict) -> Dict[str, Any]:
|
|
"""
|
|
워크플로우에서 BridgeInput/BridgeOutput 스키마를 추출한다.
|
|
|
|
Returns:
|
|
{
|
|
"inputs": {
|
|
"input_name": {
|
|
"type": "STRING" | "INT" | "FLOAT" | "BOOLEAN" | "IMAGE" | ...
|
|
"node_id": "1",
|
|
"required": True,
|
|
}
|
|
},
|
|
"outputs": {
|
|
"output_name": {
|
|
"type": "IMAGE" | "STRING" | ... (from BridgeOutput)
|
|
"node_id": "9",
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
inputs: Dict[str, Dict[str, Any]] = {}
|
|
outputs: Dict[str, Dict[str, Any]] = {}
|
|
|
|
for node_id, node in workflow.items():
|
|
if not isinstance(node, dict):
|
|
continue
|
|
|
|
class_type = node.get("class_type", "")
|
|
node_inputs = node.get("inputs", {})
|
|
|
|
# BridgeInput 노드
|
|
if class_type.startswith(BRIDGE_INPUT_PREFIX):
|
|
input_name = node_inputs.get("name")
|
|
if input_name:
|
|
type_name = class_type[len(BRIDGE_INPUT_PREFIX):]
|
|
inputs[input_name] = {
|
|
"type": type_name,
|
|
"node_id": node_id,
|
|
"required": True,
|
|
}
|
|
|
|
# BridgeOutput 노드
|
|
elif class_type == BRIDGE_OUTPUT:
|
|
output_name = node_inputs.get("name")
|
|
if output_name:
|
|
outputs[output_name] = {
|
|
"node_id": node_id,
|
|
}
|
|
|
|
return {"inputs": inputs, "outputs": outputs}
|
|
|
|
|
|
def validate_inputs(schema: Dict[str, Any], inputs: Dict[str, Any]) -> Tuple[bool, List[str]]:
|
|
"""
|
|
입력이 스키마와 일치하는지 검증한다.
|
|
|
|
Returns:
|
|
(is_valid, error_messages)
|
|
"""
|
|
errors = []
|
|
required_inputs = schema.get("inputs", {})
|
|
|
|
for name, spec in required_inputs.items():
|
|
if spec.get("required") and name not in inputs:
|
|
errors.append(f"Required input '{name}' ({spec['type']}) not provided")
|
|
|
|
for name in inputs:
|
|
if name not in required_inputs:
|
|
errors.append(f"Unknown input '{name}' (not found in workflow)")
|
|
|
|
return len(errors) == 0, errors
|
|
|
|
|
|
def describe_schema(schema: Dict[str, Any]) -> str:
|
|
"""스키마를 인간이 읽을 수 있는 형태로 반환한다."""
|
|
lines = ["Schema:"]
|
|
|
|
inputs = schema.get("inputs", {})
|
|
if inputs:
|
|
lines.append(" Inputs:")
|
|
for name, spec in inputs.items():
|
|
required = " (required)" if spec.get("required") else " (optional)"
|
|
lines.append(f" - {name}: {spec.get('type', 'UNKNOWN')}{required}")
|
|
else:
|
|
lines.append(" Inputs: (none)")
|
|
|
|
outputs = schema.get("outputs", {})
|
|
if outputs:
|
|
lines.append(" Outputs:")
|
|
for name, spec in outputs.items():
|
|
lines.append(f" - {name}")
|
|
else:
|
|
lines.append(" Outputs: (none)")
|
|
|
|
return "\n".join(lines)
|