app/plugins/run_log.py

"""
插件运行日志服务

为所有插件提供统一的 run/log/artifact 事实源。
所有插件运行页的日志必须来自此服务，禁止前端假步骤。
"""

import json
import logging
import uuid
import shutil
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional, Dict, Any, List
from enum import Enum

from pydantic import BaseModel, Field

from app.config.settings import settings

logger = logging.getLogger(__name__)


class RunStatus(str, Enum):
    """Run 状态枚举"""
    QUEUED = "queued"
    RUNNING = "running"
    SUCCEEDED = "succeeded"
    FAILED = "failed"


class EventLevel(str, Enum):
    """事件级别"""
    DEBUG = "debug"
    INFO = "info"
    WARNING = "warning"
    ERROR = "error"


class PluginRun(BaseModel):
    """插件运行记录"""
    run_id: str = Field(..., description="运行唯一标识")
    plugin_name: str = Field(..., description="插件名称")
    status: RunStatus = Field(default=RunStatus.QUEUED, description="运行状态")
    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
    updated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
    result: Optional[Dict[str, Any]] = Field(None, description="运行结果")
    error: Optional[str] = Field(None, description="错误信息")


class PluginRunEvent(BaseModel):
    """插件运行事件（日志条目）"""
    seq: int = Field(..., description="事件序号，单调递增")
    ts: datetime = Field(default_factory=lambda: datetime.now(timezone.utc), description="事件时间戳")
    stage: str = Field(..., description="处理阶段，如 upload/model_load/transcribe")
    level: EventLevel = Field(default=EventLevel.INFO, description="事件级别")
    message: str = Field(..., description="事件消息")
    detail: Optional[str] = Field(None, description="详细描述")
    artifact_id: Optional[str] = Field(None, description="关联的 artifact ID")


class PluginRunArtifact(BaseModel):
    """插件运行产物元数据"""
    artifact_id: str = Field(..., description="产物唯一标识")
    run_id: str = Field(..., description="所属 run ID")
    filename: str = Field(..., description="原始文件名")
    media_type: str = Field(default="application/octet-stream", description="MIME 类型")
    size: int = Field(default=0, description="文件大小（字节）")
    path: str = Field(..., description="artifact 相对路径（相对于 run 目录）")
    stage: str = Field(..., description="生成阶段")
    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))


class PluginRunService:
    """插件运行日志服务

    基于文件系统的持久化实现：
    - data/plugin_runs/{run_id}/run.json — 运行元数据
    - data/plugin_runs/{run_id}/events.jsonl — 事件日志
    - data/plugin_runs/{run_id}/artifacts.json — artifact 元数据
    - data/plugin_runs/{run_id}/artifacts/ — artifact 文件
    """

    def __init__(self):
        self._runs_dir = settings.PLUGIN_RUNS_DIR
        self._runs_dir.mkdir(parents=True, exist_ok=True)

    # ---- 内部辅助 ----

    def _run_dir(self, run_id: str) -> Path:
        """获取 run 目录路径，并做安全校验"""
        run_dir = self._runs_dir / run_id
        # 防止路径穿越
        resolved = run_dir.resolve()
        if not str(resolved).startswith(str(self._runs_dir.resolve())):
            raise ValueError(f"无效的 run_id（路径穿越检测）: {run_id}")
        return run_dir

    def _ensure_run_dir(self, run_id: str) -> Path:
        """确保 run 目录存在"""
        run_dir = self._run_dir(run_id)
        run_dir.mkdir(parents=True, exist_ok=True)
        return run_dir

    def _save_run(self, run: PluginRun) -> None:
        """持久化 run 元数据到 JSON"""
        run_dir = self._ensure_run_dir(run.run_id)
        run_file = run_dir / "run.json"
        run_file.write_text(
            run.model_dump_json(indent=2), encoding="utf-8"
        )

    def _load_run(self, run_id: str) -> Optional[PluginRun]:
        """从 JSON 加载 run 元数据"""
        run_dir = self._run_dir(run_id)
        run_file = run_dir / "run.json"
        if not run_file.exists():
            return None
        data = json.loads(run_file.read_text(encoding="utf-8"))
        return PluginRun(**data)

    def _append_event(self, run_id: str, event: PluginRunEvent) -> None:
        """追加事件到 JSONL 文件"""
        run_dir = self._ensure_run_dir(run_id)
        events_file = run_dir / "events.jsonl"
        with open(events_file, "a", encoding="utf-8") as f:
            f.write(event.model_dump_json() + "\n")

    def _load_events(self, run_id: str) -> List[PluginRunEvent]:
        """从 JSONL 加载事件列表"""
        run_dir = self._run_dir(run_id)
        events_file = run_dir / "events.jsonl"
        if not events_file.exists():
            return []
        events = []
        with open(events_file, "r", encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if line:
                    events.append(PluginRunEvent(**json.loads(line)))
        return events

    def _save_artifacts_meta(self, run_id: str, artifacts: List[PluginRunArtifact]) -> None:
        """持久化 artifact 元数据列表"""
        run_dir = self._ensure_run_dir(run_id)
        meta_file = run_dir / "artifacts.json"
        meta_file.write_text(
            json.dumps(
                [a.model_dump(mode="json") for a in artifacts],
                indent=2,
                ensure_ascii=False,
            ),
            encoding="utf-8",
        )

    def _load_artifacts_meta(self, run_id: str) -> List[PluginRunArtifact]:
        """加载 artifact 元数据列表"""
        run_dir = self._run_dir(run_id)
        meta_file = run_dir / "artifacts.json"
        if not meta_file.exists():
            return []
        data = json.loads(meta_file.read_text(encoding="utf-8"))
        return [PluginRunArtifact(**item) for item in data]

    # ---- 公开 API ----

    def create_run(self, plugin_name: str) -> PluginRun:
        """创建新的运行记录

        Args:
            plugin_name: 插件名称

        Returns:
            新创建的 PluginRun 实例
        """
        run_id = uuid.uuid4().hex[:12]
        run = PluginRun(
            run_id=run_id,
            plugin_name=plugin_name,
            status=RunStatus.RUNNING,
        )
        self._save_run(run)
        # 同时写入创建事件
        self.add_event(
            run_id=run_id,
            stage="init",
            level=EventLevel.INFO,
            message=f"运行已创建: {plugin_name}",
        )
        logger.info(f"创建 run: run_id={run_id}, plugin={plugin_name}")
        return run

    def add_event(
        self,
        run_id: str,
        stage: str,
        message: str,
        level: EventLevel = EventLevel.INFO,
        detail: Optional[str] = None,
        artifact_id: Optional[str] = None,
    ) -> PluginRunEvent:
        """添加运行事件

        Args:
            run_id: 运行 ID
            stage: 处理阶段
            message: 事件消息
            level: 事件级别
            detail: 详细描述
            artifact_id: 关联 artifact ID

        Returns:
            新创建的 PluginRunEvent
        """
        # 获取当前 seq
        existing = self._load_events(run_id)
        seq = len(existing) + 1

        event = PluginRunEvent(
            seq=seq,
            stage=stage,
            level=level,
            message=message,
            detail=detail,
            artifact_id=artifact_id,
        )
        self._append_event(run_id, event)
        return event

    def add_artifact(
        self,
        run_id: str,
        filename: str,
        source_path: Path,
        stage: str,
        media_type: str = "application/octet-stream",
    ) -> PluginRunArtifact:
        """添加运行产物

        将文件复制到 run 的 artifacts 目录并记录元数据。

        Args:
            run_id: 运行 ID
            filename: 原始文件名
            source_path: 源文件路径
            stage: 生成阶段
            media_type: MIME 类型

        Returns:
            新创建的 PluginRunArtifact
        """
        run_dir = self._ensure_run_dir(run_id)
        artifacts_dir = run_dir / "artifacts"
        artifacts_dir.mkdir(exist_ok=True)

        artifact_id = uuid.uuid4().hex[:8]
        # 保持原始扩展名
        dest_name = f"{artifact_id}_{filename}"
        dest_path = artifacts_dir / dest_name

        # 复制文件
        shutil.copy2(str(source_path), str(dest_path))
        file_size = dest_path.stat().st_size

        artifact = PluginRunArtifact(
            artifact_id=artifact_id,
            run_id=run_id,
            filename=filename,
            media_type=media_type,
            size=file_size,
            path=f"artifacts/{dest_name}",
            stage=stage,
        )

        # 更新元数据
        artifacts = self._load_artifacts_meta(run_id)
        artifacts.append(artifact)
        self._save_artifacts_meta(run_id, artifacts)

        # 写入事件
        self.add_event(
            run_id=run_id,
            stage=stage,
            message=f"产物已保存: {filename} ({file_size} bytes)",
            level=EventLevel.INFO,
            artifact_id=artifact_id,
        )

        logger.info(f"添加 artifact: run_id={run_id}, artifact_id={artifact_id}, filename={filename}")
        return artifact

    def finish_run(
        self,
        run_id: str,
        status: RunStatus = RunStatus.SUCCEEDED,
        result: Optional[Dict[str, Any]] = None,
        error: Optional[str] = None,
    ) -> Optional[PluginRun]:
        """完成运行

        Args:
            run_id: 运行 ID
            status: 最终状态
            result: 运行结果
            error: 错误信息（status=FAILED 时）

        Returns:
            更新后的 PluginRun，不存在则返回 None
        """
        run = self._load_run(run_id)
        if run is None:
            return None

        run.status = status
        run.updated_at = datetime.now(timezone.utc)
        if result is not None:
            run.result = result
        if error is not None:
            run.error = error

        self._save_run(run)

        # 写入完成事件
        event_level = EventLevel.ERROR if status == RunStatus.FAILED else EventLevel.INFO
        finish_msg = f"运行{'失败' if status == RunStatus.FAILED else '完成'}: {run_id}"
        if error:
            finish_msg += f" — {error}"
        self.add_event(
            run_id=run_id,
            stage="finish",
            level=event_level,
            message=finish_msg,
            detail=error,
        )

        logger.info(f"完成 run: run_id={run_id}, status={status.value}")
        return run

    def get_run(self, run_id: str) -> Optional[PluginRun]:
        """获取运行记录"""
        return self._load_run(run_id)

    def list_events(self, run_id: str) -> List[PluginRunEvent]:
        """获取运行事件列表"""
        return self._load_events(run_id)

    def get_artifact(self, run_id: str, artifact_id: str) -> Optional[PluginRunArtifact]:
        """获取产物元数据"""
        artifacts = self._load_artifacts_meta(run_id)
        for a in artifacts:
            if a.artifact_id == artifact_id:
                return a
        return None

    def list_artifacts(self, run_id: str) -> List[PluginRunArtifact]:
        """获取运行的所有产物"""
        return self._load_artifacts_meta(run_id)

    def get_artifact_path(self, run_id: str, artifact_id: str) -> Optional[Path]:
        """获取产物文件的真实路径（用于下载）"""
        artifact = self.get_artifact(run_id, artifact_id)
        if artifact is None:
            return None
        run_dir = self._run_dir(run_id)
        artifact_path = run_dir / artifact.path
        # 安全校验
        resolved = artifact_path.resolve()
        if not str(resolved).startswith(str(run_dir.resolve())):
            raise ValueError(f"artifact 路径穿越检测: {artifact.path}")
        if not resolved.exists():
            return None
        return resolved


# 全局单例
_run_log_service: Optional[PluginRunService] = None


def get_run_log_service() -> PluginRunService:
    """获取 PluginRunService 全局单例"""
    global _run_log_service
    if _run_log_service is None:
        _run_log_service = PluginRunService()
    return _run_log_service