movie-create/taskflow/run_manager.py

"""
运行管理模块
管理每次运行的输出目录，确保每次运行的结果独立存储
"""

from datetime import datetime
from pathlib import Path
from typing import Optional

from .logger import get_logger


class RunManager:
    """
    运行管理类：为每次运行创建独立的输出目录

    每次运行都会在基础输出目录下创建一个独立的子目录
    用于存储该次运行的所有结果文件，便于区分和管理不同运行的结果

    使用示例：
        >>> run_manager = RunManager(base_output_dir="output")
        >>> run_dir = run_manager.create_run_directory()
        >>> print(run_dir) # 输出: output/run_20250101_120000
    """

    def __init__(self, base_output_dir: str = "output"):
        """
        初始化运行管理器

        Args:
            base_output_dir: 基础输出目录
        """
        self.base_output_dir = Path(base_output_dir)
        self.base_output_dir.mkdir(parents=True, exist_ok=True)
        self.current_run_dir: Optional[Path] = None
        self.run_id: Optional[str] = None
        self.logger = get_logger("taskflow.run_manager")

    def create_run_directory(self, run_id: Optional[str] = None) -> str:
        """
        创建本次运行的输出目录

        Args:
            run_id: 运行ID（如果为None，使用当前时间戳）

        Returns:
            str: 本次运行的输出目录路径
        """
        if run_id is None:
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            run_id = f"run_{timestamp}"

        self.run_id = run_id
        self.current_run_dir = self.base_output_dir / run_id
        self.current_run_dir.mkdir(parents=True, exist_ok=True)

        self.logger.info(f"本次运行的输出目录: {self.current_run_dir}")
        return str(self.current_run_dir)

    def get_run_directory(self) -> str:
        """
        获取本次运行的输出目录

        Returns:
            str: 本次运行的输出目录路径

        Raises:
            RuntimeError: 未创建运行目录
        """
        if self.current_run_dir is None:
            raise RuntimeError("未创建运行目录, 请先调用 create_run_directory 方法")
        return str(self.current_run_dir)

    def get_run_id(self) -> str:
        """
        获取本次运行的运行ID

        Returns:
            str: 本次运行的运行ID
        """
        if self.run_id is None:
            raise RuntimeError("未创建运行目录, 请先调用 create_run_directory 方法")
        return self.run_id
    
    def list_runs(self) -> list:
        """
        列出所有以存在的运行目录

        Returns:
            运行ID列表，按时间排序，最新的运行排在前面
        """
        runs = []
        for item in self.base_output_dir.iterdir():
            if item.is_dir() and item.name.startswith("run_"):
                runs.append({
                    "run_id": item.name,
                    "path": str(item),
                    "created_at": datetime.fromtimestamp(item.stat().st_ctime).isoformat()
                })
        
        runs.sort(key=lambda x: x["created_at"], reverse=True)
        return runs