预发前多处bug修复

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "BowongModalFunctions"
-version = "0.1.4"
+version = "0.1.5"
 authors = [
     { name = "Yudi Xiao", email = "xyd@bowong.ai" },
 ]
@@ -38,8 +38,6 @@ classifiers = [
     "Operating System :: OS Independent",
     "Development Status :: 2 - Pre-Alpha"
 ]
-license = "MIT"
-license-files = ["LICEN[CS]E*"]
 
 [project.optional-dependencies]
 cli = [

src/BowongModalFunctions/api.py

@@ -4,24 +4,20 @@ import sentry_sdk
 from sentry_sdk.integrations.loguru import LoguruIntegration, LoggingLevels
 from sentry_sdk.integrations.fastapi import FastApiIntegration
 from fastapi.middleware.cors import CORSMiddleware
+
 from .utils.KVCache import KVCache
-from .router import ffmpeg, cache, comfyui, google
+from .router import ffmpeg, cache, comfyui, google, task
 from .config import WorkerConfig
 
 config = WorkerConfig()
 
 web_app = FastAPI(title="Modal worker API",
-                  version="0.1.2",
+                  version="0.1.5",
                   summary="Modal Worker的API, 包括缓存视频, 发起生产任务等",
                   servers=[
-                      {'url': f'https://bowongai-dev--{config.modal_app_name}-fastapi-webapp.modal.run',
+                      {
-                       'description': 'modal 开发环境服务'},
+                          'url': f'https://bowongai-{config.modal_environment}--{config.modal_app_name}-fastapi-webapp.modal.run',
-                      {'url': 'https://modal-dev.bowong.cc',
+                          'description': '当前Modal API接口endpoint'
-                       'description': 'modal 开发环境服务独立域名'},
-                      {'url': f'https://bowongai-test--{config.modal_app_name}-fastapi-webapp.modal.run',
-                       'description': 'modal 测试环境服务'},
-                      {'url': f'https://bowongai-main--{config.modal_app_name}-fastapi-webapp.modal.run',
-                       'description': 'modal 生产环境服务'
                       }
                   ])
 
@@ -83,3 +79,4 @@ web_app.include_router(ffmpeg.router)
 web_app.include_router(cache.router)
 web_app.include_router(comfyui.router)
 web_app.include_router(google.router)
+web_app.include_router(task.router)

src/BowongModalFunctions/config.py

@@ -4,7 +4,7 @@ from pydantic_settings import BaseSettings, SettingsConfigDict
 
 class WorkerConfig(BaseSettings):
     video_downloader_concurrency: int = Field(default=10, description="处理缓存任务的并行数")
-    ffmpeg_worker_concurrency: int = Field(default=10, description="处理视频合成任务的并行数")
+    ffmpeg_worker_concurrency: int = Field(default=100, description="处理视频合成任务的并行数")
     local_output_mount: str = Field(default="/mnt/outputs", description="本地结果挂载路径前缀")
 
     S3_region: str = Field(default='ap-northeast-2', description="S3挂载桶的地域")

src/BowongModalFunctions/models/web_model.py

@@ -87,9 +87,11 @@ class FFMPEGConvertStreamRequest(BaseFFMPEGTaskRequest):
         else:
             raise pydantic.ValidationError("media格式读取失败")
 
+
 class FFMPEGConvertStreamResponse(BaseFFMPEGTaskStatusResponse):
     result: Optional[str] = Field(default=None, description="任务运行结果")
 
+
 class FFMPEGSliceRequest(BaseFFMPEGTaskRequest):
     media: MediaSource = Field(description="待切割的媒体源")
     markers: List[FFMpegSliceSegment] = Field(description="切割标记数组")
@@ -159,7 +161,7 @@ class FFMPEGBgmRequest(BaseFFMPEGTaskRequest):
 
     @field_validator('media', mode='before')
     @classmethod
-    def parse_inputs(cls, v: Union[str, MediaSource]):
+    def parse_media_input(cls, v: Union[str, MediaSource]):
         if isinstance(v, str):
             return MediaSource.from_str(v)
         elif isinstance(v, MediaSource):
@@ -169,7 +171,7 @@ class FFMPEGBgmRequest(BaseFFMPEGTaskRequest):
 
     @field_validator('bgm_media', mode='before')
     @classmethod
-    def parse_inputs(cls, v: Union[str, MediaSource]):
+    def parse_bgm_input(cls, v: Union[str, MediaSource]):
         if isinstance(v, str):
             media = MediaSource.from_str(v)
             if not media.file_extension in ['wav', 'mp3']:
@@ -210,17 +212,17 @@ class FFMPEGOverlayGifRequest(BaseFFMPEGTaskRequest):
 
     @field_validator('media', mode='before')
     @classmethod
-    def parse_inputs(cls, v: Union[str, MediaSource]):
+    def parse_media_input(cls, v: Union[str, MediaSource]):
         if isinstance(v, str):
             return MediaSource.from_str(v)
         elif isinstance(v, MediaSource):
             return v
         else:
-            raise TypeError(v)
+            raise TypeError(f"解析{type(v)}类型不支持")
 
     @field_validator('gif', mode='before')
     @classmethod
-    def parse_inputs(cls, v: Union[str, MediaSource]):
+    def parse_fig_input(cls, v: Union[str, MediaSource]):
         if isinstance(v, str):
             media = MediaSource.from_str(v)
             if not media.file_extension == 'gif':
@@ -243,7 +245,7 @@ class FFMPEGSubtitleOverlayRequest(BaseFFMPEGTaskRequest):
 
     @field_validator('media', mode='before')
     @classmethod
-    def parse_inputs(cls, v: Union[str, MediaSource]):
+    def parse_media_input(cls, v: Union[str, MediaSource]):
         if isinstance(v, str):
             return MediaSource.from_str(v)
         elif isinstance(v, MediaSource):
@@ -253,7 +255,7 @@ class FFMPEGSubtitleOverlayRequest(BaseFFMPEGTaskRequest):
 
     @field_validator('subtitle', mode='before')
     @classmethod
-    def parse_inputs(cls, v: Union[str, MediaSource]):
+    def parse_subtitle_input(cls, v: Union[str, MediaSource]):
         if isinstance(v, str):
             media = MediaSource.from_str(v)
             if not media.file_extension in ['ass']:
@@ -266,7 +268,7 @@ class FFMPEGSubtitleOverlayRequest(BaseFFMPEGTaskRequest):
 
     @field_validator('fonts', mode='before')
     @classmethod
-    def parse_inputs(cls, v: Union[str, MediaSource]) -> List[MediaSource]:
+    def parse_font_input(cls, v: Union[str, MediaSource]) -> List[MediaSource]:
         if not v:
             raise pydantic.ValidationError("fonts输入为空")
         result = []
@@ -294,7 +296,7 @@ class FFMPEGMixBgmWithNoiseReduceRequest(BaseFFMPEGTaskRequest):
 
     @field_validator('media', mode='before')
     @classmethod
-    def parse_inputs(cls, v: Union[str, MediaSource]):
+    def parse_media_input(cls, v: Union[str, MediaSource]):
         if isinstance(v, str):
             return MediaSource.from_str(v)
         elif isinstance(v, MediaSource):
@@ -304,7 +306,7 @@ class FFMPEGMixBgmWithNoiseReduceRequest(BaseFFMPEGTaskRequest):
 
     @field_validator('bgm', mode='before')
     @classmethod
-    def parse_inputs(cls, v: Union[str, MediaSource]):
+    def parse_bgm_input(cls, v: Union[str, MediaSource]):
         if isinstance(v, str):
             return MediaSource.from_str(v)
         elif isinstance(v, MediaSource):
@@ -314,7 +316,7 @@ class FFMPEGMixBgmWithNoiseReduceRequest(BaseFFMPEGTaskRequest):
 
     @field_validator('noise_sample', mode='before')
     @classmethod
-    def parse_inputs(cls, v: Union[None, str, MediaSource]):
+    def parse_sample_input(cls, v: Union[None, str, MediaSource]):
         if not v:
             return None
         if isinstance(v, str):
@@ -338,6 +340,29 @@ class FFMPEGVideoLoopFillAudioResponse(BaseFFMPEGTaskStatusResponse):
     result: Optional[str] = Field(default=None, description="生成结果的URN")
 
 
+class FFMPEGExtractFrameRequest(BaseFFMPEGTaskRequest):
+    video: MediaSource = Field(description="提取帧画面的来源")
+    frame_index: int = Field(description="提取的第几帧, 从1开始，默认为1", default=1)
+
+    @field_validator('video', mode='before')
+    @classmethod
+    def parse_video_input(cls, v: Union[str, MediaSource]):
+        if isinstance(v, str):
+            return MediaSource.from_str(v)
+        elif isinstance(v, MediaSource):
+            return v
+        else:
+            raise TypeError(v)
+
+class FFMPEGExtractFrameStatusResponse(BaseFFMPEGTaskStatusResponse):
+    result: Optional[str] = Field(default=None, description="生成结果的URN")
+
+
+class ModalTaskCancelResponse(BaseModel):
+    success: bool = Field(description="成功取消任务")
+    error: Optional[str] = Field(default=None, description="失败原因")
+
+
 class ComfyTaskStatusResponse(BaseModel):
     taskId: str = Field(description="任务Id")
     status: TaskStatus = Field(description="任务运行状态")

src/BowongModalFunctions/router/ffmpeg.py

@@ -19,6 +19,8 @@ from ..models.web_model import (FFMPEGSliceRequest, SentryTransactionHeader,
                                 FFMPEGMixBgmWithNoiseReduceRequest,
                                 FFMPEGVideoLoopFillAudioRequest,
                                 FFMPEGConvertStreamRequest,
+                                FFMPEGExtractFrameRequest,
+                                FFMPEGExtractFrameStatusResponse,
                                 FFMPEGConvertStreamResponse,
                                 FFMPEGSliceTaskStatusResponse,
                                 FFMPEGConcatTaskStatusResponse,
@@ -387,3 +389,37 @@ async def video_loop_fill_audio_status(task_id: str, response: Response) -> FFMP
         response.headers["x-baggage"] = transaction.x_baggage
     return FFMPEGVideoLoopFillAudioResponse(taskId=task_id, status=task_status, code=code,
                                             error=reason, result=result)
+
+
+@router.post("/extract-frame",
+             tags=["发起任务"],
+             summary="发起视频截取帧画面任务",
+             dependencies=[Depends(verify_token)])
+async def video_extract_frame(body: FFMPEGExtractFrameRequest,
+                              headers: Annotated[SentryTransactionHeader, Header()]) -> ModalTaskResponse:
+    fn = modal.Function.from_name(config.modal_app_name, "ffmpeg_extract_frame",
+                                  environment_name=config.modal_environment)
+    sentry_trace = None
+    if headers.x_trace_id and headers.x_baggage:
+        sentry_trace = SentryTransactionInfo(x_trace_id=headers.x_trace_id, x_baggage=headers.x_baggage)
+    fn_call = fn.spawn(media=body.video, frame_index=body.frame_index, sentry_trace=sentry_trace, webhook=body.webhook)
+    return ModalTaskResponse(success=True, taskId=fn_call.object_id)
+
+
+@router.get("/extract-frame/{task_id}",
+            tags=["查询任务"],
+            summary="查询视频截取帧画面任务状态",
+            responses={
+                status.HTTP_200_OK: {
+                    "description": "",
+                    "headers": sentry_header_schema
+                },
+            },
+            dependencies=[Depends(verify_token)])
+async def video_extract_frame_status(task_id: str, response: Response) -> FFMPEGExtractFrameStatusResponse:
+    task_status, code, reason, result, transaction = await ModalUtils.get_modal_task_status(task_id)
+    if transaction:
+        response.headers["x-trace-id"] = transaction.x_trace_id
+        response.headers["x-baggage"] = transaction.x_baggage
+    return FFMPEGExtractFrameStatusResponse(taskId=task_id, status=task_status, code=code,
+                                            error=reason, result=result)

src/BowongModalFunctions/router/google.py

@@ -12,7 +12,7 @@ from BowongModalFunctions.config import WorkerConfig
 
 config = WorkerConfig
 
-router = APIRouter(prefix="/google", tags=["google"])
+router = APIRouter(prefix="/google", tags=["Google"])
 
 
 class GoogleAPIKeyHeaders(BaseModel):

src/BowongModalFunctions/router/task.py

@@ -0,0 +1,22 @@
+import modal
+from fastapi import APIRouter, Depends
+from loguru import logger
+
+from ..middleware.authorization import verify_token
+from ..models.web_model import ModalTaskCancelResponse
+
+router = APIRouter(prefix="/task")
+
+
+@router.get("/cancel/{task_id}", tags=["取消任务"],
+            summary="终止任务, 无论是正在排队还是真正运行",
+            dependencies=[Depends(verify_token)])
+async def task_cancel(task_id: str) -> ModalTaskCancelResponse:
+    try:
+        fn_call = modal.FunctionCall.from_id(task_id)
+        fn_call.cancel()
+        return ModalTaskCancelResponse(success=True)
+    except Exception as e:
+        logger.exception(e)
+        return ModalTaskCancelResponse(success=False,
+                                       error=e.message if hasattr(e, 'message') else str(e))

src/BowongModalFunctions/utils/ModalUtils.py

@@ -54,3 +54,12 @@ class ModalUtils:
         except Exception as e:
             logger.exception(e)
             return TaskStatus.failed, ErrorCode.NOT_FOUND.value, "NOT_FOUND", None, None
+
+    @staticmethod
+    async def cancel_modal_task(task_id : str):
+        try:
+            fn_task = modal.FunctionCall.from_id(task_id)
+            fn_task.cancel()
+        except Exception as e:
+            logger.exception(e)
+            return TaskStatus.failed, ErrorCode.NOT_FOUND.value, "NOT_FOUND", None, None

src/BowongModalFunctions/utils/VideoUtils.py

@@ -65,8 +65,17 @@ class VideoStream(MediaStream):
     model_config = ConfigDict(extra='allow')
 
 
+class ImageStream(MediaStream):
+    stream_type: str = "image"
+    width: int
+    height: int
+    pix_fmt: str
+
+    model_config = ConfigDict(extra='allow')
+
+
 class VideoMetadata(BaseModel):
-    streams: List[Union[VideoStream, AudioStream]]
+    streams: List[Union[VideoStream, ImageStream, AudioStream]]
 
     model_config = ConfigDict(extra='allow')
 
@@ -137,10 +146,10 @@ class VideoUtils:
         return video_metadata.streams[0].width, video_metadata.streams[0].height
 
     @staticmethod
-    def noise_reduce(media_path: str, noise_sample_path: Optional[str] = None,
+    def noise_reduce(audio_path: str, noise_sample_path: Optional[str] = None,
-                     output_path: Optional[str] = None) -> Tuple[str, VideoStream]:
+                     output_path: Optional[str] = None) -> str:
         samplerate = 44100
-        with AudioFile(media_path).resampled_to(float(samplerate)) as f:
+        with AudioFile(audio_path).resampled_to(float(samplerate)) as f:
             audio = f.read(f.frames)
 
         if noise_sample_path:
@@ -152,7 +161,7 @@ class VideoUtils:
             noise_sample = audio[:noise_sample_length]
 
         if not output_path:
-            output_path = FileUtils.file_path_extend(media_path, "nr")
+            output_path = FileUtils.file_path_extend(audio_path, "nr")
 
         reduced_noise = nr.reduce_noise(y=audio, y_noise=noise_sample, sr=samplerate,
                                         stationary=True, prop_decrease=0.75, n_std_thresh_stationary=1.5,
@@ -579,10 +588,12 @@ class VideoUtils:
         return output_path, video_metadata
 
     @staticmethod
-    async def ffmpeg_mix_bgm_with_noise_reduce(media_path: str, bgm_audio_path: str,
+    async def ffmpeg_mix_bgm_with_noise_reduce(media_path: str,
+                                               bgm_audio_path: str,
                                                video_volume: float = 1.4,
                                                music_volume: float = 0.1,
                                                noise_sample_path: Optional[str] = None,
+                                               temp_audio_path: Optional[str] = None,
                                                output_path: Optional[str] = None) -> Tuple[str, VideoMetadata]:
         """
         先对待处理的视频音轨降噪，再将降噪后的结果添加BGM，最终输出降噪过且混合BGM的视频；
@@ -592,6 +603,7 @@ class VideoUtils:
         :param video_volume: 最终输出的视频音量系数
         :param music_volume: 最终输出的BGM音量系数
         :param noise_sample_path: 降噪使用的噪音样本，如不指定将使用源视频的前2秒作为样本（可选）
+        :param temp_audio_path: 指定暂存音频的路径（可选）
         :param output_path: 指定输出视频的路径（可选）
         :return: 最终输出视频的路径， 最终输出视频时长
         """
@@ -599,19 +611,27 @@ class VideoUtils:
         if not output_path:
             output_path = FileUtils.file_path_extend(media_path, "bgm_nr")
         os.makedirs(os.path.dirname(output_path), exist_ok=True)
-        nr_audio_path = VideoUtils.noise_reduce(media_path=media_path, noise_sample_path=noise_sample_path)
+        if not temp_audio_path:
+            temp_audio_path = FileUtils.file_path_extend(media_path, "temp")
+            temp_audio_path = FileUtils.file_path_change_extension(temp_audio_path, "wav")
+        media_audio, metadata = await VideoUtils.ffmpeg_extract_audio_async(media_path=media_path,
+                                                                            output_path=temp_audio_path)
+        logger.info(f"media_audio = {media_audio}, metadata = {metadata}")
+        nr_audio_path = VideoUtils.noise_reduce(audio_path=media_audio, noise_sample_path=noise_sample_path)
+        logger.info(f"nr_audio_path = {nr_audio_path}")
         video_metadata = VideoUtils.ffprobe_video_format(media_path)
         origin_audio_duration = VideoUtils.ffprobe_audio_duration(nr_audio_path)
         bgm_duration = VideoUtils.ffprobe_audio_duration(bgm_audio_path)
         loops_needed = math.ceil(origin_audio_duration.total_seconds() / bgm_duration.total_seconds())
-
+        logger.info(
+            f"{bgm_duration.total_seconds()}s的BGM循环{loops_needed}次, 填充{origin_audio_duration.total_seconds()}s的视频长度")
         ffmpeg_cmd = VideoUtils.async_ffmpeg_init()
         ffmpeg_cmd.input(media_path)
         ffmpeg_cmd.input(nr_audio_path)
         ffmpeg_cmd.input(bgm_audio_path)
         filter_complex = [
             f"[1:a]volume={video_volume}[a1]",
-            f"[2:a]aloop=loop={loops_needed}:size={bgm_duration.total_seconds()},volume={music_volume}[a2]"
+            f"[2:a]aloop=loop={loops_needed}:size={bgm_duration.total_seconds()},volume={music_volume}[a2]",
             "[a1][a2]amix=inputs=2:duration=first[audio]"
         ]
         ffmpeg_cmd.output(output_path,
@@ -650,7 +670,7 @@ class VideoUtils:
         ]
         ffmpeg_cmd = VideoUtils.async_ffmpeg_init()
         ffmpeg_cmd.input(media_path)
-        ffmpeg_cmd.input(overlay_gif_path)
+        ffmpeg_cmd.input(overlay_gif_path, stream_loop=-1) # 使用stream_loop让GIF循环直到视频结束
         ffmpeg_cmd.output(output_path,
                           options={"filter_complex": ";".join(filter_complex), },
                           map=["[v]", "0:a"],
@@ -808,3 +828,20 @@ class VideoUtils:
         await ffmpeg_cmd.execute()
         video_metadata = VideoUtils.ffprobe_media_metadata(output_path)
         return output_path, video_metadata
+
+    @staticmethod
+    async def ffmpeg_extract_frame_image(video_path: str, frame_index: int, output_path: Optional[str] = None) -> Tuple[
+        str, VideoMetadata]:
+        """
+        获取视频的第n帧输出为图片, 并返回图片相关的元数据
+        """
+        if not output_path:
+            output_path = FileUtils.file_path_extend(video_path, 'cover')
+            output_path = FileUtils.file_path_change_extension(output_path, 'jpg')
+        os.makedirs(os.path.dirname(output_path), exist_ok=True)
+        ffmpeg_cmd = VideoUtils.async_ffmpeg_init()
+        ffmpeg_cmd.input(video_path)
+        ffmpeg_cmd.output(output_path, vframes=frame_index)
+        await ffmpeg_cmd.execute()
+        image_metadata = VideoUtils.ffprobe_media_metadata(output_path)
+        return output_path, image_metadata

src/cluster/ffmpeg_app.py

@@ -377,10 +377,12 @@ with ffmpeg_worker_image.imports():
                                     sentry_baggage=sentry_trace.x_baggage if sentry_trace else None)
         @SentryUtils.webhook_handler(webhook=webhook, func_id=fn_id)
         async def ffmpeg_process(video: MediaSource, bgm: MediaSource, func_id: str,
+                                 video_volume: float = 1.4, music_volume: float = 0.1,
                                  noise_sample: Optional[MediaSource] = None) -> FFMPEGResult:
             local_input_filepath = f"{s3_mount}/{video.cache_filepath}"
             bgm_filepath = f"{s3_mount}/{bgm.cache_filepath}"
             noise_sample_path = f"{s3_mount}/{noise_sample.cache_filepath}" if noise_sample else None
+            temp_audio_path = f"{output_path_prefix}/{config.modal_environment}/bgm_nosie_reduce/outputs/{func_id}/input.wav"
             output_path = f"{output_path_prefix}/{config.modal_environment}/bgm_nosie_reduce/outputs/{func_id}/output.mp4"
             local_output_filepath, metadata = await VideoUtils.ffmpeg_mix_bgm_with_noise_reduce(
                 media_path=local_input_filepath,
@@ -388,6 +390,7 @@ with ffmpeg_worker_image.imports():
                 video_volume=video_volume,
                 music_volume=music_volume,
                 noise_sample_path=noise_sample_path,
+                temp_audio_path=temp_audio_path,
                 output_path=output_path
             )
 
@@ -395,7 +398,7 @@ with ffmpeg_worker_image.imports():
             return FFMPEGResult(urn=s3_outputs[0], metadata=metadata,
                                 content_length=os.path.getsize(local_output_filepath), )
 
-        result = await ffmpeg_process(video=media, bgm=bgm, video_volume=video_volume,
+        result = await ffmpeg_process(video=media, bgm=bgm, video_volume=video_volume, func_id=fn_id,
                                       music_volume=music_volume, noise_sample=noise_sample)
 
         if not sentry_trace:
@@ -521,3 +524,38 @@ with ffmpeg_worker_image.imports():
                                                  x_baggage=sentry_sdk.get_baggage())
 
         return output.urn, sentry_trace
+
+
+    @app.function(timeout=600, cloud="aws",
+                  max_containers=config.ffmpeg_worker_concurrency,
+                  volumes={
+                      s3_mount: modal.CloudBucketMount(
+                          bucket_name=config.S3_bucket_name,
+                          secret=modal.Secret.from_name("aws-s3-secret",
+                                                        environment_name=config.modal_environment),
+                      ),
+                  }, )
+    @modal.concurrent(max_inputs=1)
+    async def ffmpeg_extract_frame(media: MediaSource, frame_index: int,
+                                   sentry_trace: Optional[SentryTransactionInfo] = None,
+                                   webhook: Optional[WebhookNotify] = None):
+        fn_id = current_function_call_id()
+
+        @SentryUtils.sentry_tracker(name="获取指定的视频帧画面输出为图片", op="ffmpeg.extract.frame", fn_id=fn_id,
+                                    sentry_trace_id=sentry_trace.x_trace_id if sentry_trace else None,
+                                    sentry_baggage=sentry_trace.x_baggage if sentry_trace else None)
+        @SentryUtils.webhook_handler(webhook=webhook, func_id=fn_id)
+        async def ffmpeg_process(media: MediaSource, frame_index: int, func_id: str) -> FFMPEGResult:
+            video_path = f"{s3_mount}/{media.cache_filepath}"
+            output_path = f"{output_path_prefix}/{config.modal_environment}/extract_frame/{func_id}/output.jpg"
+            local_output, metadata = await VideoUtils.ffmpeg_extract_frame_image(video_path=video_path,
+                                                                                 frame_index=frame_index,
+                                                                                 output_path=output_path)
+            s3_outputs = local_copy_to_s3([local_output])
+            return FFMPEGResult(urn=s3_outputs[0], metadata=metadata, content_length=os.path.getsize(local_output), )
+
+        output = await ffmpeg_process(media=media, frame_index=frame_index, func_id=fn_id)
+        if not sentry_trace:
+            sentry_trace = SentryTransactionInfo(x_trace_id=sentry_sdk.get_traceparent(),
+                                                 x_baggage=sentry_sdk.get_baggage())
+        return output.urn, sentry_trace

src/cluster/video.py

@@ -134,7 +134,7 @@ with downloader_image.imports():
                 media_info = response.MediaInfoSet[0].BasicInfo
                 logger.info(f"VOD info = {media_info}")
                 file_extension = media_info.Type
-                cache_dir = f"/{config.S3_mount_dir}/{media.protocol.value}/{media.endpoint}/{media.bucket}"
+                cache_dir = f"{config.S3_mount_dir}/{media.protocol.value}/{media.endpoint}/{media.bucket}"
                 cache_file = media.path if '.' in media.path else f"{media.path}.{file_extension}"
                 return cache_dir, cache_file, media_info.MediaUrl
             else:
@@ -178,6 +178,7 @@ with downloader_image.imports():
                                     if local_crc64 == remote_crc64:
                                         logger.success("File size verification passed!")
                                     return
+                            os.makedirs(os.path.dirname(output_path), exist_ok=True)
 
                     logger.info(f"Downloading {url}...")
                     # 发起流式请求