| 1 | # Copyright (C) 2025 AIDC-AI |
| 2 | # |
| 3 | # Licensed under the Apache License, Version 2.0 (the "License"); |
| 4 | # you may not use this file except in compliance with the License. |
| 5 | # You may obtain a copy of the License at |
| 6 | # http://www.apache.org/licenses/LICENSE-2.0 |
| 7 | # Unless required by applicable law or agreed to in writing, software |
| 8 | # distributed under the License is distributed on an "AS IS" BASIS, |
| 9 | # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 10 | # See the License for the specific language governing permissions and |
| 11 | # limitations under the License. |
| 12 | |
| 13 | """ |
| 14 | OS utilities for file and path management |
| 15 | |
| 16 | Provides utilities for managing paths and files in Pixelle-Video. |
| 17 | Inspired by Pixelle-MCP's os_util.py. |
| 18 | """ |
| 19 | |
| 20 | import os |
| 21 | import random |
| 22 | from datetime import datetime |
| 23 | from pathlib import Path |
| 24 | from typing import Optional, Tuple, Literal |
| 25 | |
| 26 | |
| 27 | def get_pixelle_video_root_path() -> str: |
| 28 | """ |
| 29 | Get Pixelle-Video root path |
| 30 | |
| 31 | Uses PIXELLE_VIDEO_ROOT environment variable to determine project root. |
| 32 | This ensures reliable path resolution in both development and packaged environments. |
| 33 | |
| 34 | Returns: |
| 35 | Project root path as string |
| 36 | """ |
| 37 | # Check environment variable (required for reliable operation) |
| 38 | env_root = os.environ.get("PIXELLE_VIDEO_ROOT") |
| 39 | if env_root and Path(env_root).exists(): |
| 40 | return str(Path(env_root).resolve()) |
| 41 | |
| 42 | # Fallback to current working directory if environment variable not set |
| 43 | # (for development environments where env var might not be set) |
| 44 | return str(Path.cwd()) |
| 45 | |
| 46 | |
| 47 | def ensure_pixelle_video_root_path() -> str: |
| 48 | """ |
| 49 | Ensure Pixelle-Video root path exists and return the path |
| 50 | |
| 51 | Returns: |
| 52 | Root path as string |
| 53 | """ |
| 54 | root_path = get_pixelle_video_root_path() |
| 55 | root_path_obj = Path(root_path) |
| 56 | output_dir = root_path_obj / 'output' |
| 57 | output_dir.mkdir(parents=True, exist_ok=True) |
| 58 | |
| 59 | return root_path |
| 60 | |
| 61 | |
| 62 | def get_root_path(*paths: str) -> str: |
| 63 | """ |
| 64 | Get path relative to Pixelle-Video root path |
| 65 | |
| 66 | Args: |
| 67 | *paths: Path components to join |
| 68 | |
| 69 | Returns: |
| 70 | Absolute path as string |
| 71 | |
| 72 | Example: |
| 73 | get_root_path("temp", "audio.mp3") |
| 74 | # Returns: "/path/to/project/temp/audio.mp3" |
| 75 | """ |
| 76 | root_path = ensure_pixelle_video_root_path() |
| 77 | if paths: |
| 78 | return os.path.join(root_path, *paths) |
| 79 | return root_path |
| 80 | |
| 81 | |
| 82 | def get_temp_path(*paths: str) -> str: |
| 83 | """ |
| 84 | Get path relative to Pixelle-Video temp folder |
| 85 | |
| 86 | Ensures temp directory exists before returning path. |
| 87 | |
| 88 | Args: |
| 89 | *paths: Path components to join |
| 90 | |
| 91 | Returns: |
| 92 | Absolute path to temp directory or file |
| 93 | |
| 94 | Example: |
| 95 | get_temp_path("audio.mp3") |
| 96 | # Returns: "/path/to/project/temp/audio.mp3" |
| 97 | """ |
| 98 | temp_path = get_root_path("temp") |
| 99 | |
| 100 | # Ensure temp directory exists |
| 101 | os.makedirs(temp_path, exist_ok=True) |
| 102 | |
| 103 | if paths: |
| 104 | return os.path.join(temp_path, *paths) |
| 105 | return temp_path |
| 106 | |
| 107 | |
| 108 | def get_data_path(*paths: str) -> str: |
| 109 | """ |
| 110 | Get path relative to Pixelle-Video data folder |
| 111 | |
| 112 | Ensures data directory exists before returning path. |
| 113 | |
| 114 | Args: |
| 115 | *paths: Path components to join |
| 116 | |
| 117 | Returns: |
| 118 | Absolute path to data directory or file |
| 119 | |
| 120 | Example: |
| 121 | get_data_path("videos", "output.mp4") |
| 122 | # Returns: "/path/to/project/data/videos/output.mp4" |
| 123 | """ |
| 124 | data_path = get_root_path("data") |
| 125 | |
| 126 | # Ensure data directory exists |
| 127 | os.makedirs(data_path, exist_ok=True) |
| 128 | |
| 129 | if paths: |
| 130 | return os.path.join(data_path, *paths) |
| 131 | return data_path |
| 132 | |
| 133 | |
| 134 | def get_output_path(*paths: str) -> str: |
| 135 | """ |
| 136 | Get path relative to Pixelle-Video output folder |
| 137 | |
| 138 | Ensures output directory exists before returning path. |
| 139 | |
| 140 | Args: |
| 141 | *paths: Path components to join |
| 142 | |
| 143 | Returns: |
| 144 | Absolute path to output directory or file |
| 145 | |
| 146 | Example: |
| 147 | get_output_path("video.mp4") |
| 148 | # Returns: "/path/to/project/output/video.mp4" |
| 149 | """ |
| 150 | output_path = get_root_path("output") |
| 151 | |
| 152 | # Ensure output directory exists |
| 153 | os.makedirs(output_path, exist_ok=True) |
| 154 | |
| 155 | if paths: |
| 156 | return os.path.join(output_path, *paths) |
| 157 | return output_path |
| 158 | |
| 159 | |
| 160 | def save_bytes_to_file(data: bytes, file_path: str) -> str: |
| 161 | """ |
| 162 | Save bytes data to file |
| 163 | |
| 164 | Creates parent directories if they don't exist. |
| 165 | |
| 166 | Args: |
| 167 | data: Binary data to save |
| 168 | file_path: Target file path |
| 169 | |
| 170 | Returns: |
| 171 | Absolute path of saved file |
| 172 | |
| 173 | Example: |
| 174 | save_bytes_to_file(audio_data, get_temp_path("audio.mp3")) |
| 175 | """ |
| 176 | # Ensure parent directory exists |
| 177 | os.makedirs(os.path.dirname(file_path), exist_ok=True) |
| 178 | |
| 179 | # Write binary data |
| 180 | with open(file_path, "wb") as f: |
| 181 | f.write(data) |
| 182 | |
| 183 | return os.path.abspath(file_path) |
| 184 | |
| 185 | |
| 186 | def ensure_dir(path: str) -> str: |
| 187 | """ |
| 188 | Ensure directory exists, create if not |
| 189 | |
| 190 | Args: |
| 191 | path: Directory path |
| 192 | |
| 193 | Returns: |
| 194 | Absolute path of directory |
| 195 | """ |
| 196 | os.makedirs(path, exist_ok=True) |
| 197 | return os.path.abspath(path) |
| 198 | |
| 199 | |
| 200 | # ========== Task Directory Management ========== |
| 201 | |
| 202 | def create_task_id() -> str: |
| 203 | """ |
| 204 | Create unique task ID with timestamp + random suffix |
| 205 | |
| 206 | Format: {timestamp}_{random_hex} |
| 207 | Example: "20251028_143052_ab3d" |
| 208 | |
| 209 | Collision probability: < 0.0001% (65536 combinations per second) |
| 210 | |
| 211 | Returns: |
| 212 | Task ID string |
| 213 | """ |
| 214 | timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') |
| 215 | random_suffix = f"{random.randint(0, 0xFFFF):04x}" # 4-digit hex (0000-ffff) |
| 216 | return f"{timestamp}_{random_suffix}" |
| 217 | |
| 218 | |
| 219 | def create_task_output_dir(task_id: Optional[str] = None) -> Tuple[str, str]: |
| 220 | """ |
| 221 | Create isolated output directory for single video generation task |
| 222 | |
| 223 | Directory structure: |
| 224 | output/{task_id}/ |
| 225 | ├── final.mp4 # Final video output |
| 226 | ├── frames/ # All frame-related files |
| 227 | │ ├── 01_audio.mp3 |
| 228 | │ ├── 01_image.png |
| 229 | │ ├── 01_composed.png |
| 230 | │ ├── 01_segment.mp4 |
| 231 | │ └── ... |
| 232 | └── metadata.json # Optional: task metadata |
| 233 | |
| 234 | Args: |
| 235 | task_id: Optional task ID (auto-generated if None) |
| 236 | |
| 237 | Returns: |
| 238 | (task_dir, task_id) tuple |
| 239 | |
| 240 | Example: |
| 241 | >>> task_dir, task_id = create_task_output_dir() |
| 242 | >>> # task_dir = "/path/to/project/output/20251028_143052_ab3d" |
| 243 | >>> # task_id = "20251028_143052_ab3d" |
| 244 | """ |
| 245 | if task_id is None: |
| 246 | task_id = create_task_id() |
| 247 | |
| 248 | task_dir = get_output_path(task_id) |
| 249 | frames_dir = os.path.join(task_dir, "frames") |
| 250 | |
| 251 | # Create directories |
| 252 | os.makedirs(frames_dir, exist_ok=True) |
| 253 | |
| 254 | return task_dir, task_id |
| 255 | |
| 256 | |
| 257 | def get_task_path(task_id: str, *paths: str) -> str: |
| 258 | """ |
| 259 | Get path within task directory |
| 260 | |
| 261 | Args: |
| 262 | task_id: Task ID |
| 263 | *paths: Path components to join |
| 264 | |
| 265 | Returns: |
| 266 | Absolute path within task directory |
| 267 | |
| 268 | Example: |
| 269 | >>> get_task_path("20251028_143052_ab3d", "final.mp4") |
| 270 | >>> # Returns: "/path/to/project/output/20251028_143052_ab3d/final.mp4" |
| 271 | """ |
| 272 | task_dir = get_output_path(task_id) |
| 273 | if paths: |
| 274 | return os.path.join(task_dir, *paths) |
| 275 | return task_dir |
| 276 | |
| 277 | |
| 278 | def get_task_frame_path( |
| 279 | task_id: str, |
| 280 | frame_index: int, |
| 281 | file_type: Literal["audio", "image", "video", "composed", "segment"] |
| 282 | ) -> str: |
| 283 | """ |
| 284 | Get frame file path within task directory |
| 285 | |
| 286 | Args: |
| 287 | task_id: Task ID |
| 288 | frame_index: Frame index (0-based internally, but filename starts from 01) |
| 289 | file_type: File type (audio/image/video/composed/segment) |
| 290 | |
| 291 | Returns: |
| 292 | Absolute path to frame file |
| 293 | |
| 294 | Example: |
| 295 | >>> get_task_frame_path("20251028_143052_ab3d", 0, "audio") |
| 296 | >>> # Returns: ".../output/20251028_143052_ab3d/frames/01_audio.mp3" |
| 297 | """ |
| 298 | ext_map = { |
| 299 | "audio": "mp3", |
| 300 | "image": "png", |
| 301 | "video": "mp4", |
| 302 | "composed": "png", |
| 303 | "segment": "mp4" |
| 304 | } |
| 305 | |
| 306 | # Frame number starts from 01 for better human readability |
| 307 | filename = f"{frame_index + 1:02d}_{file_type}.{ext_map[file_type]}" |
| 308 | return get_task_path(task_id, "frames", filename) |
| 309 | |
| 310 | |
| 311 | def get_task_final_video_path(task_id: str) -> str: |
| 312 | """ |
| 313 | Get final video path within task directory |
| 314 | |
| 315 | Args: |
| 316 | task_id: Task ID |
| 317 | |
| 318 | Returns: |
| 319 | Absolute path to final video |
| 320 | |
| 321 | Example: |
| 322 | >>> get_task_final_video_path("20251028_143052_ab3d") |
| 323 | >>> # Returns: ".../output/20251028_143052_ab3d/final.mp4" |
| 324 | """ |
| 325 | return get_task_path(task_id, "final.mp4") |
| 326 | |
| 327 | |
| 328 | # ========== Resource Management (Templates/BGM/Workflows) ========== |
| 329 | |
| 330 | def get_resource_path(resource_type: Literal["bgm", "templates", "workflows"], *paths: str) -> str: |
| 331 | """ |
| 332 | Get resource file path with custom override support |
| 333 | |
| 334 | Search priority: |
| 335 | 1. data/{resource_type}/*paths (custom, higher priority) |
| 336 | 2. {resource_type}/*paths (default, fallback) |
| 337 | |
| 338 | Args: |
| 339 | resource_type: Resource type ("bgm", "templates", "workflows") |
| 340 | *paths: Path components relative to resource directory |
| 341 | |
| 342 | Returns: |
| 343 | Absolute path to resource file (custom if exists, otherwise default) |
| 344 | |
| 345 | Raises: |
| 346 | FileNotFoundError: If file not found in either location |
| 347 | |
| 348 | Examples: |
| 349 | >>> get_resource_path("bgm", "happy.mp3") |
| 350 | # Returns: "data/bgm/happy.mp3" (if exists) or "bgm/happy.mp3" |
| 351 | |
| 352 | >>> get_resource_path("templates", "1080x1920", "default.html") |
| 353 | # Returns: "data/templates/1080x1920/default.html" or "templates/1080x1920/default.html" |
| 354 | |
| 355 | >>> get_resource_path("workflows", "selfhost", "image_flux.json") |
| 356 | # Returns: "data/workflows/selfhost/image_flux.json" or "workflows/selfhost/image_flux.json" |
| 357 | """ |
| 358 | # Build custom path (data/*) |
| 359 | custom_path = get_data_path(resource_type, *paths) |
| 360 | |
| 361 | # Build default path (root/*) |
| 362 | default_path = get_root_path(resource_type, *paths) |
| 363 | |
| 364 | # Priority: custom > default |
| 365 | if os.path.exists(custom_path): |
| 366 | return custom_path |
| 367 | |
| 368 | if os.path.exists(default_path): |
| 369 | return default_path |
| 370 | |
| 371 | # Not found in either location |
| 372 | raise FileNotFoundError( |
| 373 | f"Resource not found: {os.path.join(resource_type, *paths)}\n" |
| 374 | f" Searched locations:\n" |
| 375 | f" 1. {custom_path} (custom)\n" |
| 376 | f" 2. {default_path} (default)" |
| 377 | ) |
| 378 | |
| 379 | |
| 380 | def list_resource_files( |
| 381 | resource_type: Literal["bgm", "templates", "workflows"], |
| 382 | subdir: str = "" |
| 383 | ) -> list[str]: |
| 384 | """ |
| 385 | List resource files with custom override support |
| 386 | |
| 387 | Merges files from both default and custom locations: |
| 388 | - Files from data/{resource_type}/* (custom, higher priority) |
| 389 | - Files from {resource_type}/* (default) |
| 390 | - Duplicate names are deduplicated (custom takes precedence) |
| 391 | |
| 392 | Args: |
| 393 | resource_type: Resource type ("bgm", "templates", "workflows") |
| 394 | subdir: Optional subdirectory (e.g., "1080x1920" for templates) |
| 395 | |
| 396 | Returns: |
| 397 | Sorted list of filenames (deduplicated, custom overrides default) |
| 398 | |
| 399 | Examples: |
| 400 | >>> list_resource_files("bgm") |
| 401 | # Returns: ["custom.mp3", "default.mp3", "happy.mp3"] |
| 402 | # (merged from bgm/ and data/bgm/) |
| 403 | |
| 404 | >>> list_resource_files("templates", "1080x1920") |
| 405 | # Returns: ["custom.html", "default.html", "modern.html"] |
| 406 | # (merged from templates/1080x1920/ and data/templates/1080x1920/) |
| 407 | """ |
| 408 | files = {} # Use dict to track source priority: {filename: path} |
| 409 | |
| 410 | # Build directory paths |
| 411 | default_dir = Path(get_root_path(resource_type, subdir)) if subdir else Path(get_root_path(resource_type)) |
| 412 | custom_dir = Path(get_data_path(resource_type, subdir)) if subdir else Path(get_data_path(resource_type)) |
| 413 | |
| 414 | # Scan default directory first (lower priority) |
| 415 | if default_dir.exists() and default_dir.is_dir(): |
| 416 | for item in default_dir.iterdir(): |
| 417 | if item.is_file(): |
| 418 | files[item.name] = str(item) |
| 419 | |
| 420 | # Scan custom directory (higher priority, overwrites) |
| 421 | if custom_dir.exists() and custom_dir.is_dir(): |
| 422 | for item in custom_dir.iterdir(): |
| 423 | if item.is_file(): |
| 424 | files[item.name] = str(item) # Overwrite if exists |
| 425 | |
| 426 | return sorted(files.keys()) |
| 427 | |
| 428 | |
| 429 | def list_resource_dirs( |
| 430 | resource_type: Literal["bgm", "templates", "workflows"] |
| 431 | ) -> list[str]: |
| 432 | """ |
| 433 | List subdirectories in resource directory |
| 434 | |
| 435 | Merges directories from both default and custom locations. |
| 436 | |
| 437 | Args: |
| 438 | resource_type: Resource type ("bgm", "templates", "workflows") |
| 439 | |
| 440 | Returns: |
| 441 | Sorted list of directory names (deduplicated) |
| 442 | |
| 443 | Examples: |
| 444 | >>> list_resource_dirs("templates") |
| 445 | # Returns: ["1080x1080", "1080x1920", "1920x1080"] |
| 446 | |
| 447 | >>> list_resource_dirs("workflows") |
| 448 | # Returns: ["runninghub", "selfhost"] |
| 449 | """ |
| 450 | dirs = set() |
| 451 | |
| 452 | # Build directory paths |
| 453 | default_dir = Path(get_root_path(resource_type)) |
| 454 | custom_dir = Path(get_data_path(resource_type)) |
| 455 | |
| 456 | # Scan default directory |
| 457 | if default_dir.exists() and default_dir.is_dir(): |
| 458 | for item in default_dir.iterdir(): |
| 459 | if item.is_dir(): |
| 460 | dirs.add(item.name) |
| 461 | |
| 462 | # Scan custom directory |
| 463 | if custom_dir.exists() and custom_dir.is_dir(): |
| 464 | for item in custom_dir.iterdir(): |
| 465 | if item.is_dir(): |
| 466 | dirs.add(item.name) |
| 467 | |
| 468 | return sorted(dirs) |
| 469 | |
| 470 | |
| 471 | def resource_exists(resource_type: Literal["bgm", "templates", "workflows"], *paths: str) -> bool: |
| 472 | """ |
| 473 | Check if resource file exists (in custom or default location) |
| 474 | |
| 475 | Args: |
| 476 | resource_type: Resource type ("bgm", "templates", "workflows") |
| 477 | *paths: Path components relative to resource directory |
| 478 | |
| 479 | Returns: |
| 480 | True if exists in either location, False otherwise |
| 481 | |
| 482 | Examples: |
| 483 | >>> resource_exists("bgm", "happy.mp3") |
| 484 | True |
| 485 | |
| 486 | >>> resource_exists("templates", "1080x1920", "default.html") |
| 487 | True |
| 488 | """ |
| 489 | custom_path = get_data_path(resource_type, *paths) |
| 490 | default_path = get_root_path(resource_type, *paths) |
| 491 | |
| 492 | return os.path.exists(custom_path) or os.path.exists(default_path) |
| 493 | |
| 494 |