movie-create/taskflow/io_handler.py

"""
文件I/O处理器模块
提供统一的文件读写接口，支持多种文件格式
"""

import asyncio
import aiofiles
import json
import os
import re
from pathlib import Path
from typing import Any


class FileIOHandler:
    """
    文件I/O处理器：统一管理文件的读写

    提供JSON和纯文本的读写功能，可扩展支持其他格式
    
    使用示例：
        >>> handler = FileIOHandler()
        >>> data = handler.read_json("input.json")
        >>> handler.write_json(data, "output.json")
    """

    @staticmethod
    def string_to_json(data: str) -> Any:
        """
        将字符串转换为JSON
        """
        try:
            json_data = re.sub(r'^```json|\n```$', '', data, flags=re.MULTILINE).strip()
            if not json_data:
                json_data = data
            return json.loads(json_data)
        except Exception as e:
            raise e

    @staticmethod
    def read_json(file_path: str) -> Any:
        """
        读取JSON文件
        
        Args:
            file_path: 文件路径
        
        Returns:
            解析后的JSON数据

        Raises:
            FileNotFoundError: 文件不存在
            json.JSONDecodeError: JSON解析错误
        """
        if not os.path.exists(file_path):
            raise FileNotFoundError(f"文件不存在: {file_path}")

        with open(file_path, 'r', encoding='utf-8') as f:
            return json.load(f)

    @staticmethod
    def write_json(data: Any, file_path: str, indent: int = 2):
        """
        保存JSON文件

        Args:
            data: 要保存的数据
            file_path: 文件路径
            indent: JSON缩进空格数
        """
        output_path = Path(file_path)
        output_path.parent.mkdir(parents=True, exist_ok=True)

        # 如果数据是字符串，则先转换为JSON
        if isinstance(data, str):
            data = FileIOHandler.string_to_json(data)

        with open(output_path, 'w', encoding='utf-8') as f:
            json.dump(data, f, ensure_ascii=False, indent=indent)

    @staticmethod
    async def write_json_async(data: Any, file_path: str, indent: int = 2):
        """
        异步保存JSON文件
        
        Args:
            data: 要保存的数据
            file_path: 文件路径
            indent: JSON缩进空格数
        """
        output_path = Path(file_path)
        output_path.parent.mkdir(parents=True, exist_ok=True)

        async with aiofiles.open(output_path, 'w', encoding='utf-8') as f:
            await f.write(json.dumps(data, ensure_ascii=False, indent=indent))


    @staticmethod
    def read_text(file_path: str) -> str:
        """
        读取文本文件
        
        Args:
            file_path: 文件路径
        
        Returns:
            文件内容字符串
        
        Raises:
            FileNotFoundError: 文件不存在
        """
        if not os.path.exists(file_path):
            raise FileNotFoundError(f"文件不存在: {file_path}")
        
        with open(file_path, 'r', encoding='utf-8') as f:
            return f.read()
    
    @staticmethod
    def write_text(content: str, file_path: str):
        """
        保存文本文件
        
        Args:
            content: 要保存的内容
            file_path: 文件路径
        """
        output_path = Path(file_path)
        output_path.parent.mkdir(parents=True, exist_ok=True)
        
        with open(file_path, 'w', encoding='utf-8') as f:
            f.write(content)

    @staticmethod
    async def write_text_async(content: str, file_path: str):
        """
        异步保存文本文件
        
        Args:
            content: 要保存的内容
            file_path: 文件路径
        """
        output_path = Path(file_path)
        output_path.parent.mkdir(parents=True, exist_ok=True)

        async with aiofiles.open(output_path, 'w', encoding='utf-8') as f:
            await f.write(content)

    @staticmethod
    def file_exists(file_path: str) -> bool:
        """
        检查文件是否存在

        Args:
            file_path: 文件路径

        Returns:
            bool: 文件是否存在
        """
        return os.path.exists(file_path)