feat:[FL-212][任务监控页面支持查看执行日志]

- 后端添加任务日志存储和查询API
  - 新增 /api/v1/admin/task-logs 端点支持上传、获取和列出任务日志
  - 日志存储到MinIO，路径格式: logs/YYYY/MM/DD/{task_id}.log
  - 新增 task_log_service 处理MinIO存储交互
  - 新增 task_log schema 定义API请求响应格式

- 前端任务监控页面添加查看日志功能
  - 在任务表格和卡片视图中添加"查看日志"按钮
  - 点击按钮打开模态框显示任务执行日志
  - 支持桌面端表格和移动端卡片两种视图

- 新增单元测试验证日志路径生成和错误处理

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Co-authored-by: multica-agent <github@multica.ai>

api/app/api/router.py

@@ -16,6 +16,7 @@ from .v1.lines import router as lines_router
 from .v1.scheduled_tasks import router as scheduled_tasks_router
 from .v1.system_messages import router as system_messages_router
 from .v1.system_params import router as system_params_router
+from .v1.task_logs import router as task_logs_router
 from .v1.task_monitor import router as task_monitor_router
 from .v1.tower_models import router as tower_models_router
 from .v1.tower_profiles import router as tower_profiles_router
@@ -33,6 +34,7 @@ v1_router.include_router(atp_assets_router)
 v1_router.include_router(dimensions_router)
 v1_router.include_router(documents_router)
 v1_router.include_router(task_monitor_router)
+v1_router.include_router(task_logs_router)
 v1_router.include_router(scheduled_tasks_router)
 v1_router.include_router(system_messages_router)
 v1_router.include_router(system_params_router)

api/app/api/v1/task_logs.py

@@ -0,0 +1,76 @@
+from __future__ import annotations
+
+from datetime import datetime
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+
+from ...core.dependencies import CurrentUser, require_any_permission, require_enabled_menu_route
+from ...schemas.task_log import TaskLogResponse, TaskLogUploadRequest, TaskLogUploadResponse
+from ...services.task_log_service import (
+    TaskLogNotFoundError,
+    TaskLogServiceError,
+    get_task_log,
+    list_task_logs,
+    upload_task_log,
+)
+
+router = APIRouter(prefix="/admin/task-logs", tags=["admin-task-logs"], dependencies=[Depends(require_enabled_menu_route)])
+
+
+@router.post("/upload", response_model=TaskLogUploadResponse)
+def upload_log(
+    request: TaskLogUploadRequest,
+    _: CurrentUser = Depends(require_any_permission("celery.manage")),
+) -> TaskLogUploadResponse:
+    """
+    Upload task execution log to MinIO storage.
+    Logs are stored in the path: logs/YYYY/MM/DD/{task_id}.log
+    """
+    try:
+        log_path, uploaded_at = upload_task_log(request.task_id, request.log_content)
+        return TaskLogUploadResponse(
+            task_id=request.task_id,
+            log_path=log_path,
+            uploaded_at=uploaded_at,
+        )
+    except TaskLogServiceError as exc:
+        raise HTTPException(status_code=500, detail=str(exc)) from exc
+
+
+@router.get("/{task_id}", response_model=TaskLogResponse)
+def get_log(
+    task_id: str,
+    log_date: datetime | None = Query(default=None, description="Date of the log (defaults to today)"),
+    _: CurrentUser = Depends(require_any_permission("celery.read", "celery.manage")),
+) -> TaskLogResponse:
+    """
+    Retrieve task execution log from MinIO storage.
+    If log_date is not provided, tries to find the log for today.
+    """
+    try:
+        log_content, log_path = get_task_log(task_id, log_date)
+        return TaskLogResponse(
+            task_id=task_id,
+            log_content=log_content,
+            log_path=log_path,
+            exists=True,
+        )
+    except TaskLogNotFoundError as exc:
+        raise HTTPException(status_code=404, detail=str(exc)) from exc
+    except TaskLogServiceError as exc:
+        raise HTTPException(status_code=500, detail=str(exc)) from exc
+
+
+@router.get("/{task_id}/list", response_model=list[str])
+def list_logs(
+    task_id: str,
+    _: CurrentUser = Depends(require_any_permission("celery.read", "celery.manage")),
+) -> list[str]:
+    """
+    List all available logs for a task across all dates.
+    Returns list of log paths, sorted by most recent first.
+    """
+    try:
+        return list_task_logs(task_id)
+    except TaskLogServiceError as exc:
+        raise HTTPException(status_code=500, detail=str(exc)) from exc

api/app/schemas/task_log.py

@@ -0,0 +1,23 @@
+from __future__ import annotations
+
+from datetime import datetime
+
+from pydantic import BaseModel, Field
+
+
+class TaskLogUploadRequest(BaseModel):
+    task_id: str = Field(min_length=1, max_length=255)
+    log_content: str
+
+
+class TaskLogUploadResponse(BaseModel):
+    task_id: str
+    log_path: str
+    uploaded_at: datetime
+
+
+class TaskLogResponse(BaseModel):
+    task_id: str
+    log_content: str
+    log_path: str
+    exists: bool = True

api/app/services/task_log_service.py

@@ -0,0 +1,205 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+from ..core.config import get_settings
+
+
+class TaskLogServiceError(RuntimeError):
+    pass
+
+
+class TaskLogNotFoundError(TaskLogServiceError):
+    pass
+
+
+def _get_log_path(task_id: str, timestamp: datetime | None = None) -> str:
+    """
+    Generate log path in MinIO following the pattern: logs/YYYY/MM/DD/{task_id}.log
+    """
+    if timestamp is None:
+        timestamp = datetime.now(timezone.utc)
+
+    year = timestamp.strftime("%Y")
+    month = timestamp.strftime("%m")
+    day = timestamp.strftime("%d")
+
+    return f"logs/{year}/{month}/{day}/{task_id}.log"
+
+
+def upload_task_log(task_id: str, log_content: str) -> tuple[str, datetime]:
+    """
+    Upload task execution log to MinIO.
+    Returns: (log_path, uploaded_at)
+    """
+    settings = get_settings()
+
+    if not settings.minio_enabled:
+        raise TaskLogServiceError("MinIO storage is not enabled")
+
+    try:
+        import boto3
+        from botocore.config import Config
+        from botocore.exceptions import BotoCoreError, ClientError
+    except ImportError as exc:
+        raise TaskLogServiceError("boto3 library is required for MinIO storage") from exc
+
+    uploaded_at = datetime.now(timezone.utc)
+    log_path = _get_log_path(task_id, uploaded_at)
+
+    try:
+        client_config = Config(
+            connect_timeout=3.0,
+            read_timeout=10.0,
+            retries={"max_attempts": 2},
+        )
+
+        session = boto3.session.Session(
+            aws_access_key_id=settings.minio_access_key,
+            aws_secret_access_key=settings.minio_secret_key,
+        )
+
+        client = session.client(
+            "s3",
+            endpoint_url=settings.minio_endpoint,
+            region_name=settings.minio_region,
+            config=client_config,
+        )
+
+        # Upload log content as text file
+        client.put_object(
+            Bucket=settings.minio_bucket,
+            Key=log_path,
+            Body=log_content.encode("utf-8"),
+            ContentType="text/plain; charset=utf-8",
+        )
+
+        return log_path, uploaded_at
+
+    except (BotoCoreError, ClientError) as exc:
+        raise TaskLogServiceError(f"Failed to upload log to MinIO: {exc}") from exc
+    except Exception as exc:
+        raise TaskLogServiceError(f"Unexpected error during log upload: {exc}") from exc
+
+
+def get_task_log(task_id: str, log_date: datetime | None = None) -> tuple[str, str]:
+    """
+    Retrieve task execution log from MinIO.
+    If log_date is not provided, tries to find the log for today.
+    Returns: (log_content, log_path)
+    """
+    settings = get_settings()
+
+    if not settings.minio_enabled:
+        raise TaskLogServiceError("MinIO storage is not enabled")
+
+    try:
+        import boto3
+        from botocore.config import Config
+        from botocore.exceptions import BotoCoreError, ClientError
+    except ImportError as exc:
+        raise TaskLogServiceError("boto3 library is required for MinIO storage") from exc
+
+    if log_date is None:
+        log_date = datetime.now(timezone.utc)
+
+    log_path = _get_log_path(task_id, log_date)
+
+    try:
+        client_config = Config(
+            connect_timeout=3.0,
+            read_timeout=10.0,
+            retries={"max_attempts": 2},
+        )
+
+        session = boto3.session.Session(
+            aws_access_key_id=settings.minio_access_key,
+            aws_secret_access_key=settings.minio_secret_key,
+        )
+
+        client = session.client(
+            "s3",
+            endpoint_url=settings.minio_endpoint,
+            region_name=settings.minio_region,
+            config=client_config,
+        )
+
+        response = client.get_object(
+            Bucket=settings.minio_bucket,
+            Key=log_path,
+        )
+
+        body = response.get("Body")
+        if body is None:
+            raise TaskLogServiceError("Log file body is empty")
+
+        log_content = body.read().decode("utf-8")
+        return log_content, log_path
+
+    except ClientError as exc:
+        error_code = exc.response.get("Error", {}).get("Code", "")
+        if error_code in {"404", "NoSuchKey", "NotFound"}:
+            raise TaskLogNotFoundError(f"Log not found for task {task_id} at path {log_path}") from exc
+        raise TaskLogServiceError(f"Failed to retrieve log from MinIO: {exc}") from exc
+    except (BotoCoreError, UnicodeDecodeError) as exc:
+        raise TaskLogServiceError(f"Failed to read log content: {exc}") from exc
+    except Exception as exc:
+        raise TaskLogServiceError(f"Unexpected error during log retrieval: {exc}") from exc
+
+
+def list_task_logs(task_id: str) -> list[str]:
+    """
+    List all available logs for a task across all dates.
+    Returns: list of log paths
+    """
+    settings = get_settings()
+
+    if not settings.minio_enabled:
+        raise TaskLogServiceError("MinIO storage is not enabled")
+
+    try:
+        import boto3
+        from botocore.config import Config
+        from botocore.exceptions import BotoCoreError, ClientError
+    except ImportError as exc:
+        raise TaskLogServiceError("boto3 library is required for MinIO storage") from exc
+
+    try:
+        client_config = Config(
+            connect_timeout=3.0,
+            read_timeout=10.0,
+            retries={"max_attempts": 2},
+        )
+
+        session = boto3.session.Session(
+            aws_access_key_id=settings.minio_access_key,
+            aws_secret_access_key=settings.minio_secret_key,
+        )
+
+        client = session.client(
+            "s3",
+            endpoint_url=settings.minio_endpoint,
+            region_name=settings.minio_region,
+            config=client_config,
+        )
+
+        # Search for all logs matching the task_id pattern
+        prefix = "logs/"
+        suffix = f"/{task_id}.log"
+
+        log_paths: list[str] = []
+        paginator = client.get_paginator("list_objects_v2")
+        pages = paginator.paginate(Bucket=settings.minio_bucket, Prefix=prefix)
+
+        for page in pages:
+            for item in page.get("Contents", []):
+                key = item.get("Key", "")
+                if key.endswith(suffix):
+                    log_paths.append(key)
+
+        return sorted(log_paths, reverse=True)  # Most recent first
+
+    except (BotoCoreError, ClientError) as exc:
+        raise TaskLogServiceError(f"Failed to list logs from MinIO: {exc}") from exc
+    except Exception as exc:
+        raise TaskLogServiceError(f"Unexpected error during log listing: {exc}") from exc

api/tests/test_task_log_service.py

@@ -0,0 +1,64 @@
+"""
+Basic unit tests for task log service.
+"""
+from __future__ import annotations
+
+import pytest
+from datetime import datetime, timezone
+from unittest.mock import MagicMock, patch
+
+from app.services.task_log_service import (
+    TaskLogServiceError,
+    TaskLogNotFoundError,
+    _get_log_path,
+    upload_task_log,
+    get_task_log,
+)
+
+
+def test_get_log_path_with_timestamp():
+    """Test log path generation with specific timestamp"""
+    timestamp = datetime(2026, 6, 28, 12, 30, 45, tzinfo=timezone.utc)
+    task_id = "test-task-123"
+
+    result = _get_log_path(task_id, timestamp)
+
+    assert result == "logs/2026/06/28/test-task-123.log"
+
+
+def test_get_log_path_without_timestamp():
+    """Test log path generation with current timestamp"""
+    task_id = "test-task-456"
+
+    result = _get_log_path(task_id)
+
+    # Should contain the task_id and follow the pattern
+    assert task_id in result
+    assert result.startswith("logs/")
+    assert result.endswith(f"/{task_id}.log")
+
+
+@patch("app.services.task_log_service.get_settings")
+def test_upload_task_log_minio_disabled(mock_get_settings):
+    """Test upload fails when MinIO is disabled"""
+    mock_settings = MagicMock()
+    mock_settings.minio_enabled = False
+    mock_get_settings.return_value = mock_settings
+
+    with pytest.raises(TaskLogServiceError) as exc_info:
+        upload_task_log("task-123", "log content")
+
+    assert "MinIO storage is not enabled" in str(exc_info.value)
+
+
+@patch("app.services.task_log_service.get_settings")
+def test_get_task_log_minio_disabled(mock_get_settings):
+    """Test retrieval fails when MinIO is disabled"""
+    mock_settings = MagicMock()
+    mock_settings.minio_enabled = False
+    mock_get_settings.return_value = mock_settings
+
+    with pytest.raises(TaskLogServiceError) as exc_info:
+        get_task_log("task-123")
+
+    assert "MinIO storage is not enabled" in str(exc_info.value)

web/src/app/admin/task-monitor/page.tsx

@@ -12,6 +12,7 @@ import {
   Empty,
   Form,
   Input,
+  Modal,
   Row,
   Select,
   Space,
@@ -169,6 +170,10 @@ export default function AdminTaskMonitorPage() {
   const [isLoadingMore, setIsLoadingMore] = useState(false);
   const pageCardRef = useRef<HTMLDivElement | null>(null);
   const { current: paginationCurrent, pageSize: paginationPageSize } = pagination;
+  const [logModalVisible, setLogModalVisible] = useState(false);
+  const [logModalContent, setLogModalContent] = useState("");
+  const [logModalTaskId, setLogModalTaskId] = useState("");
+  const [logModalLoading, setLogModalLoading] = useState(false);
 
   const resetTaskListPagination = useCallback(() => {
     setPagination((prev) => ({ ...prev, current: 1 }));
@@ -177,6 +182,34 @@ export default function AdminTaskMonitorPage() {
     setIsLoadingMore(false);
   }, []);
 
+  const handleViewLog = async (taskId: string) => {
+    setLogModalTaskId(taskId);
+    setLogModalVisible(true);
+    setLogModalLoading(true);
+    setLogModalContent("");
+
+    try {
+      const response = await fetchWithAuth(`/api/v1/admin/task-logs/${encodeURIComponent(taskId)}`);
+      if (!response.ok) {
+        const errorText = await readApiError(response);
+        throw new Error(errorText);
+      }
+      const data = await response.json();
+      setLogModalContent(data.log_content || "日志内容为空");
+    } catch (error) {
+      const errorMessage = error instanceof Error ? error.message : "获取日志失败";
+      setLogModalContent(`错误：${errorMessage}`);
+    } finally {
+      setLogModalLoading(false);
+    }
+  };
+
+  const handleCloseLogModal = () => {
+    setLogModalVisible(false);
+    setLogModalTaskId("");
+    setLogModalContent("");
+  };
+
   const workersOverviewQuery = useQuery({
     queryKey: ["flower-workers-overview"],
     enabled: Boolean(user) && canRead,
@@ -318,8 +351,19 @@ export default function AdminTaskMonitorPage() {
             "-"
           ),
       },
+      {
+        title: "操作",
+        key: "actions",
+        width: 100,
+        fixed: "right",
+        render: (_: unknown, record: TaskTableRow) => (
+          <Button size="small" onClick={() => handleViewLog(record.task_id)}>
+            查看日志
+          </Button>
+        ),
+      },
     ],
-    [],
+    [handleViewLog],
   );
 
   const filteredWorkers = useMemo(() => {
@@ -615,6 +659,11 @@ export default function AdminTaskMonitorPage() {
             </Typography.Text>
           </div>
         )}
+        <div style={{ marginTop: 8 }}>
+          <Button size="small" block onClick={() => handleViewLog(task.task_id)}>
+            查看日志
+          </Button>
+        </div>
       </Space>
     </AntCard>
   );
@@ -897,6 +946,41 @@ export default function AdminTaskMonitorPage() {
           </div>
         ) : null}
       </AntCard>
+
+      <Modal
+        title={`任务执行日志 - ${logModalTaskId}`}
+        open={logModalVisible}
+        onCancel={handleCloseLogModal}
+        footer={[
+          <Button key="close" onClick={handleCloseLogModal}>
+            关闭
+          </Button>,
+        ]}
+        width={800}
+        style={{ top: 20 }}
+      >
+        {logModalLoading ? (
+          <div style={{ textAlign: "center", padding: "40px 0" }}>
+            <Spin tip="加载日志中..." />
+          </div>
+        ) : (
+          <pre
+            style={{
+              maxHeight: "60vh",
+              overflow: "auto",
+              padding: "16px",
+              backgroundColor: "#f5f5f5",
+              borderRadius: "4px",
+              fontSize: "12px",
+              lineHeight: "1.5",
+              whiteSpace: "pre-wrap",
+              wordBreak: "break-word",
+            }}
+          >
+            {logModalContent}
+          </pre>
+        )}
+      </Modal>
     </div>
   );
 }