From 849a068a46aac0054fcab1bd265ab9d86afdd70a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=94=A8=E6=88=B7CI=20Test?= Date: Fri, 10 Jul 2026 10:21:53 +0800 Subject: [PATCH 1/4] =?UTF-8?q?feat:=20=E4=B8=80=E9=94=AE=E7=94=9F?= =?UTF-8?q?=E6=88=90=E9=93=BE=E8=B7=AF=E6=97=A5=E5=BF=97=E6=9C=80=E5=B0=8F?= =?UTF-8?q?=E9=9B=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Alembic 037: generation_tasks 表添加 logs TEXT 字段 - Domain 层: GenerationTask 添加 append_log/get_logs 方法（上限200条） - Repository 层: _to_domain/create/update 映射 logs 字段 - API Schema: GenerationTaskResponse 添加 logs 字段 + JSON validator - API Route: create_generation_task 添加请求/校验/入队/错误日志 - Worker: generate_video 全流程结构化日志 - [task_id=xxx] [阶段] 消息 统一格式 - 接收任务/下载素材/剪辑计划/渲染/OSS上传/完成/失败 各阶段埋点 - _flush_logs 实时持久化到 DB（独立 session，异常安全） - 18 个单元测试全部通过 Co-Authored-By: Claude Fable 5 --- .../037_add_logs_to_generation_tasks.py | 26 ++ apps/api/app/api/routes/generation_tasks.py | 74 ++++-- apps/api/app/schemas/generation_task.py | 19 +- apps/worker/worker_app/tasks/generation.py | 167 +++++++++++- .../generation_task_repository.py | 3 + packages/adapters/sqlalchemy_impl/models.py | 1 + packages/domain/generation_task.py | 39 +++ tests/unit/test_generation_logging.py | 244 ++++++++++++++++++ 8 files changed, 545 insertions(+), 28 deletions(-) create mode 100644 alembic/versions/037_add_logs_to_generation_tasks.py create mode 100644 tests/unit/test_generation_logging.py diff --git a/alembic/versions/037_add_logs_to_generation_tasks.py b/alembic/versions/037_add_logs_to_generation_tasks.py new file mode 100644 index 000000000..d5af0f74a --- /dev/null +++ b/alembic/versions/037_add_logs_to_generation_tasks.py @@ -0,0 +1,26 @@ +"""Add logs field to generation_tasks + +Revision ID: 037_generation_logs +Revises: 036_expand_uuid_36 +Create Date: 2026-07-10 +""" + +import sqlalchemy as sa + +from alembic import op + +revision = "037_generation_logs" +down_revision = "036_expand_uuid_36" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column( + "generation_tasks", + sa.Column("logs", sa.Text(), nullable=False, server_default="[]"), + ) + + +def downgrade() -> None: + op.drop_column("generation_tasks", "logs") diff --git a/apps/api/app/api/routes/generation_tasks.py b/apps/api/app/api/routes/generation_tasks.py index 4f95856a4..36083b5ce 100644 --- a/apps/api/app/api/routes/generation_tasks.py +++ b/apps/api/app/api/routes/generation_tasks.py @@ -1,3 +1,4 @@ +import logging import random import uuid from typing import Any @@ -31,6 +32,8 @@ from packages.application import ( ListGeneratedVideosByTaskUseCase, ) +logger = logging.getLogger(__name__) + router = APIRouter() @@ -57,6 +60,7 @@ def _to_generation_task_response(task) -> GenerationTaskResponse: source_edit_plan_id=task.source_edit_plan_id or "", asset_select_mode=getattr(task, "asset_select_mode", ""), batch_id=getattr(task, "batch_id", ""), + logs=getattr(task, "logs", "[]"), status=task.status, progress=task.progress, result_count=task.result_count, @@ -181,19 +185,37 @@ def create_generation_task( asset_library_repository: Any = Depends(get_asset_library_repository), asset_repository: Any = Depends(get_asset_repository), ) -> BatchGenerationTaskResponse: - project_id, asset_library_id = _resolve_project_and_library( - request, project_repository, asset_library_repository, asset_repository, authenticated_user + logger.info( + "[生成任务] 接收请求: user_id=%s, template_id=%s, asset_count=%d, mode=%s, count=%d", + authenticated_user.user.id, + request.template_id, + len(request.asset_ids), + request.asset_select_mode, + request.count, ) + try: + project_id, asset_library_id = _resolve_project_and_library( + request, project_repository, asset_library_repository, asset_repository, authenticated_user + ) + except HTTPException as e: + logger.warning("[生成任务] 校验失败: %s", e.detail) + raise + # asset_library 存在性校验（仅在提供了 asset_library_id 时） resolved_asset_ids: list[str] = list(request.asset_ids) if asset_library_id: library = asset_library_repository.get(asset_library_id) if library is None or (project_id and library.project_id != project_id): + logger.warning("[生成任务] 素材库不存在: library_id=%s", asset_library_id) raise HTTPException(status_code=404, detail=f"AssetLibrary {asset_library_id} not found") assets = asset_repository.find_by_library(asset_library_id) - _ensure_library_has_ready_video_assets(assets) + try: + _ensure_library_has_ready_video_assets(assets) + except HTTPException as e: + logger.warning("[生成任务] 素材校验失败: %s", e.detail) + raise # 素材库自动匹配：当未显式指定 asset_ids 时，按模式自动选取 if not resolved_asset_ids: @@ -209,25 +231,35 @@ def create_generation_task( # 同批次任务共享 batch_id，用于视频查重时批次内比对 batch_id = uuid.uuid4().hex if count > 1 else "" - for _ in range(count): - task = use_case.execute( - CreateGenerationTaskCommand( - project_id=project_id, - asset_library_id=asset_library_id, - strategy_id=request.strategy_id, - voice_library_id=request.voice_library_id, - template_id=request.template_id, - asset_ids=resolved_asset_ids, - title_ids=request.title_ids, - voice_ids=request.voice_ids, - created_by_user_id=authenticated_user.user.id, - source_edit_plan_id=request.source_edit_plan_id, - asset_select_mode=request.asset_select_mode, - batch_id=batch_id, + try: + for _ in range(count): + task = use_case.execute( + CreateGenerationTaskCommand( + project_id=project_id, + asset_library_id=asset_library_id, + strategy_id=request.strategy_id, + voice_library_id=request.voice_library_id, + template_id=request.template_id, + asset_ids=resolved_asset_ids, + title_ids=request.title_ids, + voice_ids=request.voice_ids, + created_by_user_id=authenticated_user.user.id, + source_edit_plan_id=request.source_edit_plan_id, + asset_select_mode=request.asset_select_mode, + batch_id=batch_id, + ) ) - ) - celery_app.send_task("worker.generate_video", args=[task.id]) - created_tasks.append(task) + celery_app.send_task("worker.generate_video", args=[task.id]) + created_tasks.append(task) + logger.info( + "[生成任务] 入队成功: task_id=%s, status=%s, batch_id=%s", + task.id, + task.status, + batch_id, + ) + except Exception as e: + logger.error("[生成任务] 创建失败: %s", e, exc_info=True) + raise HTTPException(status_code=500, detail=f"创建生成任务失败: {e}") items = [_to_generation_task_response(t) for t in created_tasks] return BatchGenerationTaskResponse(items=items, total=len(items)) diff --git a/apps/api/app/schemas/generation_task.py b/apps/api/app/schemas/generation_task.py index e75479954..1724d3f74 100644 --- a/apps/api/app/schemas/generation_task.py +++ b/apps/api/app/schemas/generation_task.py @@ -1,4 +1,6 @@ -from pydantic import BaseModel, Field, model_validator +import json + +from pydantic import BaseModel, Field, field_validator, model_validator class CreateGenerationTaskRequest(BaseModel): @@ -62,6 +64,21 @@ class GenerationTaskResponse(BaseModel): progress: float result_count: int error_message: str + logs: list[dict] = Field(default_factory=list) + + @field_validator("logs", mode="before") + @classmethod + def _parse_logs(cls, v: object) -> list[dict]: + """将 JSON 字符串解析为 list[dict]。""" + if isinstance(v, str): + try: + parsed = json.loads(v) + return parsed if isinstance(parsed, list) else [] + except (json.JSONDecodeError, TypeError): + return [] + if isinstance(v, list): + return v + return [] class BatchGenerationTaskResponse(BaseModel): diff --git a/apps/worker/worker_app/tasks/generation.py b/apps/worker/worker_app/tasks/generation.py index 3c84f45c8..58322695f 100755 --- a/apps/worker/worker_app/tasks/generation.py +++ b/apps/worker/worker_app/tasks/generation.py @@ -13,10 +13,13 @@ from __future__ import annotations +import json import logging import os import tempfile +import time from dataclasses import dataclass, field +from datetime import datetime, timezone from pathlib import Path from typing import Any, Optional @@ -81,6 +84,26 @@ def _update_task_status(task_id: str, status_action: str, **kwargs) -> bool: return False +# ── 日志持久化辅助 ──────────────────────────────────────────────────────────── + + +def _flush_logs(task_id: str, gen_task) -> None: + """将 gen_task.logs 持久化到 DB（独立 session，失败不抛异常）。""" + try: + session = SessionLocal() + try: + from packages.adapters.sqlalchemy_impl.models import GenerationTaskModel + + model = session.query(GenerationTaskModel).filter(GenerationTaskModel.id == task_id).first() + if model: + model.logs = gen_task.logs + session.commit() + finally: + session.close() + except Exception: + logger.warning("[task_id=%s] 日志持久化失败", task_id, exc_info=True) + + # ── 共享工具模块导入 ────────────────────────────────────────────────────────── from video_processing.dedup_helpers import create_video_record_and_dedup @@ -511,7 +534,7 @@ def generate_video(self, task_id: str) -> dict: """ from packages.domain import EditingMode - logger.info("开始生成视频任务: task_id=%s", task_id) + logger.info("[task_id=%s] [接收任务] 开始生成视频任务", task_id) # 从数据库加载任务信息 session = SessionLocal() @@ -523,7 +546,7 @@ def generate_video(self, task_id: str) -> dict: task_repo = SQLAlchemyGenerationTaskRepository(session) gen_task = task_repo.get(task_id) if gen_task is None: - logger.error("生成任务不存在: task_id=%s", task_id) + logger.error("[task_id=%s] [接收任务] 任务不存在", task_id) return {"status": "failed", "error": f"generation task {task_id} not found"} project_id = gen_task.project_id asset_library_id = gen_task.asset_library_id @@ -532,6 +555,16 @@ def generate_video(self, task_id: str) -> dict: mode = gen_task.strategy_id or "one_take" task_asset_ids = list(gen_task.asset_ids or []) batch_id = getattr(gen_task, "batch_id", "") or "" + + # 记录接收任务日志 + gen_task.append_log( + "接收任务", + f"模式={mode}, 模板={template_id}, 素材数={len(task_asset_ids)}", + mode=mode, + template_id=template_id, + asset_count=len(task_asset_ids), + ) + _flush_logs(task_id, gen_task) finally: session.close() @@ -558,12 +591,38 @@ def generate_video(self, task_id: str) -> dict: output_path = temp_path / output_name # 1. 从素材库/项目下载视频素材 + logger.info("[task_id=%s] [下载素材] 开始下载视频素材", task_id) + download_start = time.monotonic() downloaded_videos = _download_library_assets( temp_path, asset_library_id=asset_library_id, project_id=project_id, asset_ids=task_asset_ids or None, ) + download_elapsed = time.monotonic() - download_start + logger.info( + "[task_id=%s] [下载素材] 完成: 成功=%d个, 耗时=%.1fs", + task_id, + len(downloaded_videos), + download_elapsed, + ) + + # 重新加载 gen_task 以追加日志（session 已关闭） + _session = SessionLocal() + try: + _repo = SQLAlchemyGenerationTaskRepository(_session) + gen_task = _repo.get(task_id) + finally: + _session.close() + + if gen_task: + gen_task.append_log( + "下载素材", + f"成功下载 {len(downloaded_videos)} 个视频素材", + count=len(downloaded_videos), + duration=round(download_elapsed, 2), + ) + _flush_logs(task_id, gen_task) # 2. 下载配音（如有） audio_path: str | None = None @@ -571,6 +630,7 @@ def generate_video(self, task_id: str) -> dict: local_audio = temp_path / "voice.mp3" if _download_voice_asset(voice_library_id, local_audio): audio_path = str(local_audio) + logger.info("[task_id=%s] [下载配音] 配音下载成功", task_id) # 3. 渲染 if not downloaded_videos: @@ -588,7 +648,26 @@ def generate_video(self, task_id: str) -> dict: mode=editing_mode.value, ) + total_duration = sum(c.duration for c in virtual_clips) + logger.info( + "[task_id=%s] [剪辑计划] 片段数=%d, 总时长=%.1fs", + task_id, + len(virtual_clips), + total_duration, + ) + + if gen_task: + gen_task.append_log( + "剪辑计划", + f"片段数={len(virtual_clips)}, 总时长={total_duration:.1f}s", + segment_count=len(virtual_clips), + total_duration=round(total_duration, 2), + ) + _flush_logs(task_id, gen_task) + # 使用 UnifiedRenderService 渲染 + logger.info("[task_id=%s] [渲染] FFmpeg 渲染开始", task_id) + render_start = time.monotonic() render_service = UnifiedRenderService( plan=virtual_plan, clips=virtual_clips, @@ -599,6 +678,20 @@ def generate_video(self, task_id: str) -> dict: output_fps=int(OUTPUT_FPS), ) render_result = render_service.render() + render_elapsed = time.monotonic() - render_start + logger.info( + "[task_id=%s] [渲染] FFmpeg 渲染完成: 耗时=%.1fs", + task_id, + render_elapsed, + ) + + if gen_task: + gen_task.append_log( + "渲染", + f"FFmpeg 渲染完成, 耗时={render_elapsed:.1f}s", + duration=round(render_elapsed, 2), + ) + _flush_logs(task_id, gen_task) # 4. 如有配音，后处理混音 if audio_path: @@ -608,7 +701,7 @@ def generate_video(self, task_id: str) -> dict: # 混音成功，使用混音后的文件 output_path = final_path except Exception as mux_err: - logger.warning("音频混合失败，使用无音频版本: %s", mux_err) + logger.warning("[task_id=%s] [混音] 音频混合失败，使用无音频版本: %s", task_id, mux_err) output_path = render_result.output_path else: output_path = render_result.output_path @@ -617,9 +710,15 @@ def generate_video(self, task_id: str) -> dict: duration = probe_duration(output_path) # 5. 上传到 OSS — 失败必须抛异常，不能静默忽略 + logger.info("[task_id=%s] [OSS上传] 开始上传: size=%d", task_id, file_size) + upload_start = time.monotonic() file_url = upload_to_oss(output_path, storage_key) + upload_elapsed = time.monotonic() - upload_start if not file_url: # OSS 未配置或上传失败 + if gen_task: + gen_task.append_log("OSS上传", "上传失败", level="ERROR") + _flush_logs(task_id, gen_task) raise RuntimeError( f"OSS 上传失败: task_id={task_id}, storage_key={storage_key}, " f"output_path={output_path}" ) @@ -637,13 +736,33 @@ def generate_video(self, task_id: str) -> dict: logger.info( "URL 校验失败但 object_exists 确认文件存在，视为上传成功: storage_key=%s", key ) + if gen_task: + gen_task.append_log("OSS上传", "URL校验降级: object_exists确认存在", level="WARN") else: + if gen_task: + gen_task.append_log("OSS上传", "上传后URL不可访问", level="ERROR", file_url=file_url) + _flush_logs(task_id, gen_task) raise RuntimeError( f"OSS 上传后 URL 不可访问且 object_exists 失败: file_url={file_url}, " f"storage_key={storage_key}" ) - logger.info("OSS 上传成功: file_url=%s", file_url) + logger.info( + "[task_id=%s] [OSS上传] 成功: 耗时=%.1fs, file_url=%s", + task_id, + upload_elapsed, + file_url, + ) + + if gen_task: + gen_task.append_log( + "OSS上传", + f"上传成功, 大小={file_size}, 耗时={upload_elapsed:.1f}s", + file_size=file_size, + duration=round(upload_elapsed, 2), + file_url=file_url, + ) + _flush_logs(task_id, gen_task) # 6. 创建 GeneratedVideo 记录 + 查重 dedup_session = SessionLocal() @@ -665,7 +784,23 @@ def generate_video(self, task_id: str) -> dict: # 7. 标记任务为 completed _update_task_status(task_id, "mark_completed", result_count=video_count or 1) - logger.info("视频生成完成: task_id=%s duration=%.2fs file_size=%d", task_id, duration, file_size) + # 记录完成日志 + if gen_task: + gen_task.append_log( + "任务完成", + f"视频生成完成: 时长={duration:.2f}s, 大小={file_size}", + duration=round(duration, 2), + file_size=file_size, + video_count=video_count or 1, + ) + _flush_logs(task_id, gen_task) + + logger.info( + "[task_id=%s] [任务完成] duration=%.2fs file_size=%d", + task_id, + duration, + file_size, + ) return { "status": "completed", @@ -678,7 +813,27 @@ def generate_video(self, task_id: str) -> dict: "mode": editing_mode.value, } except Exception as error: - logger.error("Video generation failed: %s", error, exc_info=True) + logger.error("[task_id=%s] [任务失败] %s", task_id, error, exc_info=True) + + # 记录失败日志 + try: + _session = SessionLocal() + try: + _repo = SQLAlchemyGenerationTaskRepository(_session) + gen_task = _repo.get(task_id) + if gen_task: + gen_task.append_log( + "任务失败", + str(error), + level="ERROR", + error_type=type(error).__name__, + ) + _flush_logs(task_id, gen_task) + finally: + _session.close() + except Exception: + logger.warning("[task_id=%s] 记录失败日志异常", task_id, exc_info=True) + _update_task_status(task_id, "mark_failed", error_message=str(error)) return { "status": "failed", diff --git a/packages/adapters/sqlalchemy_impl/generation_task_repository.py b/packages/adapters/sqlalchemy_impl/generation_task_repository.py index b6cadb880..d736a7eb2 100644 --- a/packages/adapters/sqlalchemy_impl/generation_task_repository.py +++ b/packages/adapters/sqlalchemy_impl/generation_task_repository.py @@ -27,6 +27,7 @@ def _to_domain(model: GenerationTaskModel) -> GenerationTask: source_edit_plan_id=model.source_edit_plan_id or "", asset_select_mode=model.asset_select_mode or "", batch_id=model.batch_id or "", + logs=model.logs or "[]", created_at=model.created_at, ) @@ -56,6 +57,7 @@ class SQLAlchemyGenerationTaskRepository: source_edit_plan_id=task.source_edit_plan_id or None, asset_select_mode=task.asset_select_mode or "", batch_id=task.batch_id or "", + logs=task.logs, created_at=task.created_at, ) self.session.add(model) @@ -129,5 +131,6 @@ class SQLAlchemyGenerationTaskRepository: model.source_edit_plan_id = task.source_edit_plan_id or None model.asset_select_mode = task.asset_select_mode or "" model.batch_id = task.batch_id or "" + model.logs = task.logs self.session.commit() return task diff --git a/packages/adapters/sqlalchemy_impl/models.py b/packages/adapters/sqlalchemy_impl/models.py index 579ebf482..66660737a 100755 --- a/packages/adapters/sqlalchemy_impl/models.py +++ b/packages/adapters/sqlalchemy_impl/models.py @@ -257,6 +257,7 @@ class GenerationTaskModel(Base): asset_select_mode = Column(String(20), nullable=False, default="") batch_id = Column(String(36), nullable=False, default="", index=True) extra_meta = Column("metadata", JSON, nullable=False, default=dict) + logs = Column(Text, nullable=False, default="[]", server_default="[]") created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc)) diff --git a/packages/domain/generation_task.py b/packages/domain/generation_task.py index ca6b62a77..7222add07 100644 --- a/packages/domain/generation_task.py +++ b/packages/domain/generation_task.py @@ -8,6 +8,7 @@ from __future__ import annotations +import json import sys from dataclasses import dataclass, field from datetime import datetime, timezone @@ -85,6 +86,7 @@ class GenerationTask: created_by_user_id: str = "" asset_select_mode: str = "" batch_id: str = "" + logs: str = "[]" created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) @classmethod @@ -227,6 +229,43 @@ class GenerationTask: self.transition_to(GenerationTaskStatus.CANCELLED) self.completed_at = datetime.now(timezone.utc) + # ── 日志辅助 ──────────────────────────────────────────────────────────── + + _MAX_LOGS = 200 + + def append_log(self, stage: str, message: str, level: str = "INFO", **kwargs) -> None: + """追加一条结构化日志到 logs 字段。 + + Args: + stage: 阶段名称（如 "接收任务"、"下载素材"、"渲染"） + message: 日志消息 + level: 日志级别（INFO / WARN / ERROR） + **kwargs: 额外字段（如 asset_id、duration 等） + """ + try: + entries = json.loads(self.logs) if self.logs else [] + except (json.JSONDecodeError, TypeError): + entries = [] + entry = { + "ts": datetime.now(timezone.utc).isoformat(), + "level": level, + "stage": stage, + "message": message, + **kwargs, + } + entries.append(entry) + # 限制最多保留 _MAX_LOGS 条，防止字段过大 + if len(entries) > self._MAX_LOGS: + entries = entries[-self._MAX_LOGS :] + self.logs = json.dumps(entries, ensure_ascii=False) + + def get_logs(self) -> list[dict]: + """解析 logs 字段为 list[dict]。""" + try: + return json.loads(self.logs) if self.logs else [] + except (json.JSONDecodeError, TypeError): + return [] + def mark_pending_from_failed(self) -> None: """从失败状态重置为待处理（用于重试）。 diff --git a/tests/unit/test_generation_logging.py b/tests/unit/test_generation_logging.py new file mode 100644 index 000000000..13bdd5497 --- /dev/null +++ b/tests/unit/test_generation_logging.py @@ -0,0 +1,244 @@ +""" +一键生成链路日志最小集 单元测试 + +覆盖： +- GenerationTask.append_log() 正确追加结构化日志 +- GenerationTask.append_log() 超过 200 条时截断 +- GenerationTask.get_logs() 正确解析 JSON +- GenerationTask.get_logs() 异常 JSON 不抛异常 +- GenerationTaskResponse logs 字段 validator 解析 JSON 字符串 +- GenerationTaskResponse logs 字段 validator 处理非法输入 +- Worker 日志格式 [task_id=xxx] [阶段] 消息 +- _flush_logs 异常不抛出 +""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + +sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api")) +sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker")) + +from app.schemas.generation_task import GenerationTaskResponse + +from packages.domain.generation_task import GenerationTask, GenerationTaskStatus + + +def _make_task(**kwargs) -> GenerationTask: + """创建测试用 GenerationTask。""" + defaults = { + "id": "task-001", + "project_id": "proj-001", + "asset_library_id": "lib-001", + "strategy_id": "one_take", + "voice_library_id": "", + "template_id": "tpl-001", + "asset_ids": ["asset-1", "asset-2"], + "title_ids": [], + "voice_ids": [], + "status": GenerationTaskStatus.PENDING, + "progress": 0.0, + "result_count": 0, + "error_message": "", + "created_by_user_id": "user-001", + "source_edit_plan_id": "", + "asset_select_mode": "all", + "batch_id": "", + } + defaults.update(kwargs) + return GenerationTask(**defaults) + + +class TestAppendLog: + """GenerationTask.append_log() 单元测试。""" + + def test_append_single_log(self): + task = _make_task() + task.append_log("接收任务", "任务开始", mode="one_take") + + logs = task.get_logs() + assert len(logs) == 1 + entry = logs[0] + assert entry["level"] == "INFO" + assert entry["stage"] == "接收任务" + assert entry["message"] == "任务开始" + assert entry["mode"] == "one_take" + assert "ts" in entry + + def test_append_multiple_logs(self): + task = _make_task() + task.append_log("接收任务", "任务开始") + task.append_log("下载素材", "下载完成", count=3) + task.append_log("渲染", "渲染完成", duration=12.5) + + logs = task.get_logs() + assert len(logs) == 3 + assert logs[0]["stage"] == "接收任务" + assert logs[1]["stage"] == "下载素材" + assert logs[1]["count"] == 3 + assert logs[2]["stage"] == "渲染" + assert logs[2]["duration"] == 12.5 + + def test_append_log_with_error_level(self): + task = _make_task() + task.append_log("任务失败", "OSS上传失败", level="ERROR", error_type="RuntimeError") + + logs = task.get_logs() + assert len(logs) == 1 + assert logs[0]["level"] == "ERROR" + assert logs[0]["error_type"] == "RuntimeError" + + def test_append_log_truncates_at_200(self): + task = _make_task() + for i in range(250): + task.append_log("阶段", f"消息{i}") + + logs = task.get_logs() + assert len(logs) == 200 + # 保留最后 200 条 + assert logs[0]["message"] == "消息50" + assert logs[-1]["message"] == "消息249" + + def test_append_log_handles_corrupted_json(self): + task = _make_task(logs="not-valid-json") + task.append_log("接收任务", "任务开始") + + logs = task.get_logs() + assert len(logs) == 1 + assert logs[0]["message"] == "任务开始" + + def test_append_log_handles_empty_string(self): + task = _make_task(logs="") + task.append_log("接收任务", "任务开始") + + logs = task.get_logs() + assert len(logs) == 1 + + +class TestGetLogs: + """GenerationTask.get_logs() 单元测试。""" + + def test_get_logs_empty(self): + task = _make_task() + assert task.get_logs() == [] + + def test_get_logs_parses_json(self): + entries = [{"ts": "2026-01-01T00:00:00Z", "level": "INFO", "stage": "test", "message": "hello"}] + task = _make_task(logs=json.dumps(entries, ensure_ascii=False)) + logs = task.get_logs() + assert len(logs) == 1 + assert logs[0]["message"] == "hello" + + def test_get_logs_handles_invalid_json(self): + task = _make_task(logs="{broken") + assert task.get_logs() == [] + + def test_get_logs_handles_none(self): + task = _make_task(logs=None) + assert task.get_logs() == [] + + +class TestGenerationTaskResponseLogs: + """GenerationTaskResponse logs 字段 validator 测试。""" + + def _make_response_data(self, logs_value) -> dict: + return { + "id": "task-001", + "project_id": "proj-001", + "asset_library_id": "lib-001", + "strategy_id": "one_take", + "voice_library_id": "", + "template_id": "", + "asset_ids": [], + "title_ids": [], + "voice_ids": [], + "source_edit_plan_id": "", + "asset_select_mode": "all", + "batch_id": "", + "status": "completed", + "progress": 1.0, + "result_count": 1, + "error_message": "", + "logs": logs_value, + } + + def test_logs_json_string_parsed(self): + entries = [{"ts": "2026-01-01T00:00:00Z", "level": "INFO", "stage": "test", "message": "ok"}] + data = self._make_response_data(json.dumps(entries, ensure_ascii=False)) + resp = GenerationTaskResponse(**data) + assert isinstance(resp.logs, list) + assert len(resp.logs) == 1 + assert resp.logs[0]["message"] == "ok" + + def test_logs_list_passthrough(self): + entries = [{"ts": "2026-01-01T00:00:00Z", "level": "INFO", "stage": "test", "message": "ok"}] + data = self._make_response_data(entries) + resp = GenerationTaskResponse(**data) + assert resp.logs == entries + + def test_logs_invalid_json_returns_empty(self): + data = self._make_response_data("{broken") + resp = GenerationTaskResponse(**data) + assert resp.logs == [] + + def test_logs_empty_string_returns_empty(self): + data = self._make_response_data("") + resp = GenerationTaskResponse(**data) + assert resp.logs == [] + + def test_logs_default_empty(self): + data = self._make_response_data("[]") + resp = GenerationTaskResponse(**data) + assert resp.logs == [] + + +class TestWorkerLogFormat: + """Worker 日志格式 [task_id=xxx] [阶段] 消息 测试。""" + + def test_log_format_pattern(self): + """验证日志格式匹配 [task_id=xxx] [阶段] 消息。""" + import re + + task_id = "abc123" + stage = "下载素材" + message = "完成: 成功=3个, 耗时=1.5s" + formatted = f"[task_id={task_id}] [{stage}] {message}" + + pattern = r"^\[task_id=[\w-]+\] \[.+\] .+$" + assert re.match(pattern, formatted) + + def test_log_entries_contain_required_fields(self): + """验证 append_log 生成的条目包含所有必需字段。""" + task = _make_task() + task.append_log("OSS上传", "上传成功", file_size=1024000, duration=2.5) + + logs = task.get_logs() + entry = logs[0] + assert "ts" in entry + assert "level" in entry + assert "stage" in entry + assert "message" in entry + assert entry["file_size"] == 1024000 + assert entry["duration"] == 2.5 + + +class TestFlushLogs: + """_flush_logs 异常安全测试。""" + + def test_flush_logs_exception_not_raised(self): + """_flush_logs 在 DB 异常时不应抛出。""" + # 模拟 worker 环境 + sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker")) + + from worker_app.tasks.generation import _flush_logs + + task = _make_task() + task.append_log("测试", "消息") + + # Mock SessionLocal 抛异常 + with patch("worker_app.tasks.generation.SessionLocal", side_effect=RuntimeError("DB error")): + # 不应抛出 + _flush_logs("task-001", task) -- 2.54.0 From d32c86e71bbca74628e31bf4d1093281d2bbff12 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=94=A8=E6=88=B7CI=20Test?= Date: Fri, 10 Jul 2026 10:25:53 +0800 Subject: [PATCH 2/4] =?UTF-8?q?fix:=20=E6=9B=B4=E6=96=B0=20schema-metadata?= =?UTF-8?q?-snapshot=20=E6=94=AF=E6=8C=81=20logs=20=E5=AD=97=E6=AE=B5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/schema-metadata-snapshot.json | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/docs/schema-metadata-snapshot.json b/docs/schema-metadata-snapshot.json index b0a8f4e00..f500bc075 100644 --- a/docs/schema-metadata-snapshot.json +++ b/docs/schema-metadata-snapshot.json @@ -1549,6 +1549,14 @@ "type": "JSON", "unique": false }, + { + "index": false, + "name": "logs", + "nullable": false, + "primary_key": false, + "type": "TEXT", + "unique": false + }, { "index": false, "name": "created_at", -- 2.54.0 From 57923eeae63e4fa97ae8346cc619190c2fab9229 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=94=A8=E6=88=B7CI=20Test?= Date: Fri, 10 Jul 2026 10:39:51 +0800 Subject: [PATCH 3/4] =?UTF-8?q?fix:=20P2=20API=E5=BC=82=E5=B8=B8=E4=BF=A1?= =?UTF-8?q?=E6=81=AF=E6=B3=84=E9=9C=B2=20+=20P3-1=20=E9=80=90=E7=B4=A0?= =?UTF-8?q?=E6=9D=90=E4=B8=8B=E8=BD=BD=E6=97=A5=E5=BF=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P2: generation_tasks.py:260 异常detail改为通用消息，避免泄露内部错误 P3-1: _download_library_assets 添加逐素材日志(asset_id/耗时/大小/成功失败) 通过 gen_task.append_log 持久化到任务日志，前端可查 --- apps/api/app/api/routes/generation_tasks.py | 2 +- apps/worker/worker_app/tasks/generation.py | 58 +++++++++++++++++++-- 2 files changed, 55 insertions(+), 5 deletions(-) diff --git a/apps/api/app/api/routes/generation_tasks.py b/apps/api/app/api/routes/generation_tasks.py index 36083b5ce..36c11f709 100644 --- a/apps/api/app/api/routes/generation_tasks.py +++ b/apps/api/app/api/routes/generation_tasks.py @@ -259,7 +259,7 @@ def create_generation_task( ) except Exception as e: logger.error("[生成任务] 创建失败: %s", e, exc_info=True) - raise HTTPException(status_code=500, detail=f"创建生成任务失败: {e}") + raise HTTPException(status_code=500, detail="创建生成任务失败，请稍后重试或查看任务日志") items = [_to_generation_task_response(t) for t in created_tasks] return BatchGenerationTaskResponse(items=items, total=len(items)) diff --git a/apps/worker/worker_app/tasks/generation.py b/apps/worker/worker_app/tasks/generation.py index 58322695f..2d829ce66 100755 --- a/apps/worker/worker_app/tasks/generation.py +++ b/apps/worker/worker_app/tasks/generation.py @@ -343,6 +343,8 @@ def _download_library_assets( asset_ids: list[str] | None = None, video_extensions: tuple = (".mp4", ".mov", ".avi", ".mkv", ".webm"), strict: bool = True, + task_id: str = "", + gen_task=None, ) -> list[Path]: """下载视频素材 — 同时支持素材库模式和项目级模式。 @@ -442,19 +444,65 @@ def _download_library_assets( storage_key = asset.file_url if asset.file_url else None if not storage_key: failed_assets.append(f"{asset.name}({asset.id})") - logger.warning("素材缺少 file_url, 跳过: asset_id=%s name=%s", asset.id, asset.name) + logger.warning( + "[task_id=%s] 素材缺少 file_url, 跳过: asset_id=%s name=%s", task_id, asset.id, asset.name + ) + if gen_task: + gen_task.append_log( + "下载素材", + f"素材缺少file_url, 跳过", + level="WARN", + asset_id=asset.id, + asset_name=asset.name, + success=False, + file_size=0, + duration=0.0, + ) if strict: raise RuntimeError(f"素材缺少 file_url: asset_id={asset.id}, name={asset.name}") continue ext = Path(storage_key).suffix or ".mp4" local_file = temp_path / f"asset_{i:03d}_{asset.id}{ext}" - if download_asset(storage_key, local_file): + asset_start = time.monotonic() + download_ok = download_asset(storage_key, local_file) + asset_elapsed = time.monotonic() - asset_start + + if download_ok: + file_size = local_file.stat().st_size if local_file.exists() else 0 downloaded.append(local_file) - logger.info("Downloaded asset: %s -> %s", asset.name, local_file) + logger.info( + "[task_id=%s] Downloaded asset: %s -> %s (size=%d, time=%.1fs)", + task_id, + asset.name, + local_file, + file_size, + asset_elapsed, + ) + if gen_task: + gen_task.append_log( + "下载素材", + f"下载成功: {asset.name}", + asset_id=asset.id, + asset_name=asset.name, + success=True, + file_size=file_size, + duration=round(asset_elapsed, 2), + ) else: failed_assets.append(f"{asset.name}({asset.id})") - logger.warning("Failed to download asset: %s (id=%s)", asset.name, asset.id) + logger.warning("[task_id=%s] Failed to download asset: %s (id=%s)", task_id, asset.name, asset.id) + if gen_task: + gen_task.append_log( + "下载素材", + f"下载失败: {asset.name}", + level="WARN", + asset_id=asset.id, + asset_name=asset.name, + success=False, + file_size=0, + duration=round(asset_elapsed, 2), + ) if strict: raise RuntimeError(f"素材下载失败: asset_id={asset.id}, name={asset.name}") @@ -598,6 +646,8 @@ def generate_video(self, task_id: str) -> dict: asset_library_id=asset_library_id, project_id=project_id, asset_ids=task_asset_ids or None, + task_id=task_id, + gen_task=gen_task, ) download_elapsed = time.monotonic() - download_start logger.info( -- 2.54.0 From 7fa0906352dd729bd56e6e4d920c0cc8986f9564 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=94=A8=E6=88=B7CI=20Test?= Date: Fri, 10 Jul 2026 10:44:35 +0800 Subject: [PATCH 4/4] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=20flake8=20F541?= =?UTF-8?q?=20f-string=20=E6=97=A0=E5=8D=A0=E4=BD=8D=E7=AC=A6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 移除 generation.py:452 处多余的 f 前缀，消除 CI lint 报错。 --- apps/worker/worker_app/tasks/generation.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/worker/worker_app/tasks/generation.py b/apps/worker/worker_app/tasks/generation.py index 2d829ce66..af069d35a 100755 --- a/apps/worker/worker_app/tasks/generation.py +++ b/apps/worker/worker_app/tasks/generation.py @@ -450,7 +450,7 @@ def _download_library_assets( if gen_task: gen_task.append_log( "下载素材", - f"素材缺少file_url, 跳过", + "素材缺少file_url, 跳过", level="WARN", asset_id=asset.id, asset_name=asset.name, -- 2.54.0