| 1 | import logging |
| 2 | from typing import Any, Optional |
| 3 | |
| 4 | from models.image_client import ImageClient |
| 5 | from models.config_model import ( |
| 6 | image_capabilities, |
| 7 | list_api_models, |
| 8 | media_capabilities, |
| 9 | model_ability_tags, |
| 10 | parse_api_model, |
| 11 | video_capabilities, |
| 12 | ) |
| 13 | from models.video_client import VideoClient |
| 14 | |
| 15 | logger = logging.getLogger(__name__) |
| 16 | |
| 17 | |
| 18 | def list_api_workflows( |
| 19 | media_type: Optional[str] = None, |
| 20 | required_adapter_abilities: Optional[list[str]] = None, |
| 21 | verified_only: bool = False, |
| 22 | ) -> list[dict[str, Any]]: |
| 23 | return list_api_models( |
| 24 | media_type=media_type, |
| 25 | required_adapter_abilities=required_adapter_abilities, |
| 26 | verified_only=verified_only, |
| 27 | ) |
| 28 | |
| 29 | |
| 30 | def parse_api_workflow(workflow: str, media_type: str) -> tuple[str, str]: |
| 31 | return parse_api_model(workflow, media_type) |
| 32 | |
| 33 | |
| 34 | def normalize_video_duration(provider: str, model: str, duration: int) -> int: |
| 35 | contract = video_capabilities(provider, model).get("duration") or {} |
| 36 | if contract.get("verified"): |
| 37 | return min(max(int(duration), int(contract.get("min", duration))), int(contract.get("max", duration))) |
| 38 | if provider == "dashscope": |
| 39 | return 10 if duration >= 8 else 5 |
| 40 | if provider == "seedance" or "seedance" in (model or "").lower(): |
| 41 | return min(max(duration, 5), 10) |
| 42 | return max(duration, 1) |
| 43 | |
| 44 | |
| 45 | def generate_image_api( |
| 46 | *, |
| 47 | prompt: str, |
| 48 | model: str, |
| 49 | output_dir: str, |
| 50 | task_id: str, |
| 51 | image_paths: Optional[list[str]] = None, |
| 52 | video_ratio: str = "9:16", |
| 53 | resolution: str = "1080P", |
| 54 | ) -> str: |
| 55 | _, resolved_model = parse_api_workflow(model, "image") |
| 56 | logger.info( |
| 57 | "Generating API image: model=%s refs=%d ratio=%s resolution=%s", |
| 58 | resolved_model, |
| 59 | len(image_paths or []), |
| 60 | video_ratio, |
| 61 | resolution, |
| 62 | ) |
| 63 | paths = ImageClient().generate_image( |
| 64 | prompt=prompt, |
| 65 | image_paths=image_paths, |
| 66 | model=resolved_model, |
| 67 | save_dir=output_dir, |
| 68 | session_id=task_id, |
| 69 | video_ratio=video_ratio, |
| 70 | resolution=resolution, |
| 71 | ) |
| 72 | if not paths: |
| 73 | raise RuntimeError(f"Image API returned no result for model={resolved_model}") |
| 74 | return paths[0] |
| 75 | |
| 76 | |
| 77 | def generate_video_api( |
| 78 | *, |
| 79 | prompt: str, |
| 80 | model: str, |
| 81 | output_path: str, |
| 82 | image_path: Optional[str] = None, |
| 83 | duration: int = 5, |
| 84 | video_ratio: str = "9:16", |
| 85 | video_resolution: Optional[str] = None, |
| 86 | **params, |
| 87 | ) -> str: |
| 88 | provider, resolved_model = parse_api_workflow(model, "video") |
| 89 | safe_duration = normalize_video_duration(provider, resolved_model, int(duration)) |
| 90 | logger.info( |
| 91 | "Generating API video: provider=%s model=%s duration=%ss ratio=%s output=%s", |
| 92 | provider or "unknown", |
| 93 | resolved_model, |
| 94 | safe_duration, |
| 95 | video_ratio, |
| 96 | output_path, |
| 97 | ) |
| 98 | VideoClient().generate_video( |
| 99 | prompt=prompt, |
| 100 | image_path=image_path, |
| 101 | save_path=output_path, |
| 102 | model=resolved_model, |
| 103 | duration=safe_duration, |
| 104 | video_ratio=video_ratio, |
| 105 | resolution=video_resolution or params.pop("resolution", None), |
| 106 | **params, |
| 107 | ) |
| 108 | return output_path |
| 109 |