app/utils/file_utils.py

"""
文件操作工具
处理ZIP文件、文件验证、目录操作等
"""

import os
import json
import zipfile
import shutil
import hashlib
from pathlib import Path
from typing import Optional, List, Tuple
import logging
from datetime import datetime
from app.config import constants

logger = logging.getLogger(__name__)


class FileUtils:
    """文件工具类"""

    @staticmethod
    def extract_zip(zip_path: Path, extract_to: Path) -> Tuple[bool, str]:
        """解压ZIP文件"""
        try:
            if not zip_path.exists():
                return False, "ZIP文件不存在"

            # 创建目标目录
            extract_to.mkdir(parents=True, exist_ok=True)

            with zipfile.ZipFile(zip_path, "r") as zip_ref:
                # 验证ZIP文件
                if zip_ref.testzip() is not None:
                    return False, "ZIP文件损坏"

                # 检查是否有路径遍历攻击
                for file_info in zip_ref.infolist():
                    if file_info.filename.startswith("/") or ".." in file_info.filename:
                        return False, "ZIP文件包含非法路径"

                # 解压文件
                zip_ref.extractall(extract_to)

            logger.info(f"成功解压ZIP文件: {zip_path} -> {extract_to}")
            return True, "解压成功"

        except zipfile.BadZipFile:
            return False, "ZIP文件格式错误"
        except Exception as e:
            logger.error(f"解压ZIP文件失败: {e}")
            return False, f"解压失败: {str(e)}"

    @staticmethod
    def validate_plugin_structure(plugin_dir: Path) -> Tuple[bool, str, Optional[dict]]:
        """验证插件目录结构

        兼容两种情况:
        1. 直接结构: plugin_dir/plugin.json, plugin_dir/main.py
        2. 嵌套结构: plugin_dir/SomeFolder/plugin.json, plugin_dir/SomeFolder/main.py
        """
        try:
            # 确保是 Path 对象
            if not isinstance(plugin_dir, Path):
                plugin_dir = Path(plugin_dir)

            # 检查必需文件
            required_files = constants.PLUGIN_REQUIRED_FILES
            actual_plugin_dir = plugin_dir

            # 首先检查直接结构
            if all(
                (plugin_dir / file).exists() for file in constants.PLUGIN_REQUIRED_FILES
            ):
                actual_plugin_dir = plugin_dir
            else:
                # 检查嵌套结构 - 在第一层子目录中寻找
                found = False
                for item in plugin_dir.iterdir():
                    if item.is_dir():
                        if all(
                            (item / file).exists()
                            for file in constants.PLUGIN_REQUIRED_FILES
                        ):
                            actual_plugin_dir = item
                            found = True
                            break

                if not found:
                    # 没有找到任何结构，返回错误
                    return (
                        False,
                        f"缺少必需文件: {' 或 '.join(constants.PLUGIN_REQUIRED_FILES)}",
                        None,
                    )

            # 读取和验证 plugin.json
            plugin_json_path = actual_plugin_dir / "plugin.json"
            try:
                with open(plugin_json_path, "r", encoding="utf-8") as f:
                    plugin_metadata = json.load(f)

                # 基本字段验证
                required_fields = ["name", "version", "description", "author"]
                for field in required_fields:
                    if field not in plugin_metadata:
                        return False, f"plugin.json 缺少必需字段: {field}", None

                # 名称验证
                name = plugin_metadata["name"]
                if not name.replace("_", "").isalnum():
                    return False, "插件名称只能包含字母、数字和下划线", None

                # 返回元数据和实际路径（用于后续安装）
                plugin_metadata["_actual_path"] = str(actual_plugin_dir)

                return True, "验证成功", plugin_metadata

            except json.JSONDecodeError:
                return False, "plugin.json 格式错误", None

        except Exception as e:
            logger.error(f"验证插件结构失败: {e}")
            return False, f"验证失败: {str(e)}", None

    @staticmethod
    def safe_copy(src: Path, dst: Path) -> bool:
        """安全复制文件/目录"""
        try:
            if src.is_file():
                shutil.copy2(src, dst)
            elif src.is_dir():
                shutil.copytree(src, dst, dirs_exist_ok=True)
            return True
        except Exception as e:
            logger.error(f"复制文件失败 {src} -> {dst}: {e}")
            return False

    @staticmethod
    def safe_remove(path: Path) -> bool:
        """安全删除文件/目录"""
        try:
            if path.is_file():
                path.unlink()
            elif path.is_dir():
                shutil.rmtree(path)
            return True
        except Exception as e:
            logger.error(f"删除文件失败 {path}: {e}")
            return False

    @staticmethod
    def create_backup(src: Path, backup_dir: Path) -> Tuple[bool, str]:
        """创建备份"""
        try:
            if not src.exists():
                return False, "源文件不存在"

            backup_dir.mkdir(parents=True, exist_ok=True)

            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            backup_name = f"{src.name}_{timestamp}"
            backup_path = backup_dir / backup_name

            if src.is_file():
                shutil.copy2(src, backup_path)
            else:
                shutil.copytree(src, backup_path)

            return True, str(backup_path)

        except Exception as e:
            logger.error(f"创建备份失败: {e}")
            return False, f"备份失败: {str(e)}"

    @staticmethod
    def restore_backup(backup_path: Path, target_path: Path) -> bool:
        """恢复备份"""
        try:
            if not backup_path.exists():
                return False

            # 如果目标路径存在，先删除
            if target_path.exists():
                FileUtils.safe_remove(target_path)

            if backup_path.is_file():
                shutil.copy2(backup_path, target_path)
            else:
                shutil.copytree(backup_path, target_path)

            return True

        except Exception as e:
            logger.error(f"恢复备份失败: {e}")
            return False

    @staticmethod
    def calculate_file_hash(file_path: Path) -> Optional[str]:
        """计算文件哈希值"""
        try:
            hasher = hashlib.sha256()
            with open(file_path, "rb") as f:
                for chunk in iter(lambda: f.read(4096), b""):
                    hasher.update(chunk)
            return hasher.hexdigest()
        except Exception as e:
            logger.error(f"计算文件哈希失败 {file_path}: {e}")
            return None

    @staticmethod
    def cleanup_old_files(directory: Path, keep_count: int = 5):
        """清理旧文件，只保留指定数量的最新文件"""
        try:
            if not directory.exists():
                return

            # 获取所有文件并按修改时间排序
            files = list(directory.iterdir())
            files.sort(key=lambda x: x.stat().st_mtime, reverse=True)

            # 删除旧文件
            for file in files[keep_count:]:
                FileUtils.safe_remove(file)

        except Exception as e:
            logger.error(f"清理旧文件失败: {e}")


# 创建全局实例
file_utils = FileUtils()