- 整合dev环境接口到modal-dev.bowong.cc域名下

- 去除不必要的MediaCache类型，缓存状态整合进MediaSource类型中一起存储

src/cluster/config.py

@@ -11,6 +11,7 @@ class WorkerConfig(BaseSettings):
     modal_kv_name: Optional[str] = Field(default='media-cache', description="Modal视频缓存KV库")
     s3_region: Optional[str] = Field(default='ap-northeast-2', description="S3挂载桶的地域")
     s3_bucket_name: Optional[str] = Field(default='modal-media-cache', description="集群挂载的S3存储桶")
+    s3_mount_dir: str = Field(default='/mntS3', description="集群S3存储桶挂载在本地的根目录")
     cdn_endpoint: Optional[str] = Field(default="https://d2nj71io21vkj2.cloudfront.net",
                                         description="集群挂载S3存储桶的对应AWS Cloudfront CDN")
     environment: Optional[str] = Field(default="dev", description="Modal worker运行环境")
@@ -19,6 +20,7 @@ class WorkerConfig(BaseSettings):
 
 
 config = WorkerConfig(
+    app_name="bowong-ai-video",
     video_downloader_concurrency=10,
     ffmpeg_worker_concurrency=10,
     modal_kv_name="media-cache",

src/cluster/ffmpeg_worker/Utils/SentryUtils.py

@@ -37,3 +37,4 @@ class SentryUtils:
             return wrapper
 
         return decorator
+

src/cluster/ffmpeg_worker/Utils/VideoUtils.py

@@ -1,7 +1,26 @@
-from typing import Union, List, Tuple
+from typing import Union, List, Tuple, Optional
+
+import numpy as np
 from pydantic import BaseModel, ConfigDict, computed_field
 from ffmpeg import FFmpeg
 from ffmpeg.asyncio import FFmpeg as AsyncFFmpeg
+import soundfile as sf
+import pyloudnorm as pyln
+import noisereduce as nr
+from pedalboard import (
+    Pedalboard,
+    Compressor,
+    Limiter,
+    HighpassFilter,
+    LowpassFilter,
+    Gain,
+    Reverb,
+    Chorus,
+    Distortion,
+)
+from pedalboard.io import AudioFile
+
+from src.cluster.ffmpeg_worker.Utils.PathUtils import FileUtils
 
 
 class MediaStream(BaseModel):
@@ -84,3 +103,79 @@ class VideoUtils:
         )
         video_metadata = VideoMetadata.model_validate_json(await ffprobe.execute())
         return video_metadata.streams[0].width, video_metadata.streams[0].height
+
+    @staticmethod
+    def noise_reduce(media_path: str, noise_sample_path: Optional[str] = None) -> str:
+        samplerate = 44100
+        with AudioFile(media_path).resampled_to(float(samplerate)) as f:
+            audio = f.read(f.frames)
+
+        if noise_sample_path:
+            with AudioFile(noise_sample_path).resampled_to(float(samplerate)) as f:
+                noise_sample = f.read(f.frames)
+        else:
+            # 获取前2秒作为噪声样本
+            noise_sample_length = min(int(2 * samplerate), audio.shape[0])
+            noise_sample = audio[:noise_sample_length]
+
+        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,
+                                        n_fft=2048, win_length=1024, hop_length=512, n_jobs=1)
+
+        board = Pedalboard(
+            [
+                HighpassFilter(cutoff_frequency_hz=150),
+                LowpassFilter(cutoff_frequency_hz=8000),
+                Reverb(room_size=0.08, damping=0.7, wet_level=0.08,
+                       dry_level=0.92, width=0.4),
+                Chorus(rate_hz=0.7, depth=0.12, centre_delay_ms=3.0, mix=0.10),
+                Distortion(drive_db=3.0),
+                Compressor(threshold_db=-30, ratio=1.8, attack_ms=20, release_ms=200),
+                Compressor(threshold_db=-24, ratio=2.2, attack_ms=15, release_ms=180),
+                Compressor(threshold_db=-18, ratio=1.5, attack_ms=10, release_ms=150),
+                Gain(gain_db=4),
+                Limiter(threshold_db=-6, release_ms=200),
+            ]
+        )
+        # Convert to float32 if not already
+        reduced_noise = reduced_noise.astype(np.float32)
+        # Ensure audio is in the correct range (-1.0 to 1.0)
+        if np.abs(reduced_noise).max() > 1.0:
+            reduced_noise = reduced_noise / np.abs(reduced_noise).max()
+
+        processed_audio = board(reduced_noise, samplerate)
+        # 格式处理
+        if len(processed_audio.shape) == 1:
+            processed_audio = processed_audio.reshape(-1, 1)
+        elif len(processed_audio.shape) == 2:
+            if processed_audio.shape[0] < processed_audio.shape[1]:
+                processed_audio = processed_audio.T
+            if processed_audio.shape[1] > 2:
+                processed_audio = processed_audio[:, :2]
+        # 响度标准化
+        meter = pyln.Meter(samplerate)
+        min_samples = int(0.4 * samplerate)
+
+        if processed_audio.shape[0] < min_samples:
+            normalized_audio = processed_audio
+        else:
+            loudness = meter.integrated_loudness(processed_audio)
+            safety_factor = 0.7
+            processed_audio = processed_audio * safety_factor
+            normalized_audio = pyln.normalize.loudness(
+                processed_audio, loudness, -16.0
+            )
+
+            max_peak = np.max(np.abs(normalized_audio))
+            if max_peak > 0.85:
+                additional_safety_factor = 0.85 / max_peak
+                normalized_audio = normalized_audio * additional_safety_factor
+        processed_audio_path = FileUtils.file_path_extend(media_path, "nr")
+        sf.write(
+            processed_audio_path,
+            normalized_audio,
+            samplerate,
+            format="WAV",
+            subtype="PCM_16",
+        )
+        return processed_audio_path

src/cluster/ffmpeg_worker/worker.py

@@ -1,4 +1,5 @@
 import modal
+from modal import current_function_call_id
 
 ffmpeg_worker_image = (
     modal.Image.debian_slim(python_version="3.11")
@@ -7,9 +8,7 @@ ffmpeg_worker_image = (
                  'pedalboard==0.9.2', 'soundfile', 'pyloudnorm', 'noisereduce')
     .add_local_python_source("src.cluster.config", copy=True)
     .add_local_python_source("src.cluster.ffmpeg_worker", copy=True)
-    # .add_local_python_source("ffmpeg_worker", copy=True)
     .add_local_python_source("src.cluster.video_downloader.model", copy=True)
-    # .env({"PYTHONPATH": "/root/src"})
 )
 
 app = modal.App(image=ffmpeg_worker_image, )
@@ -25,27 +24,9 @@ with ffmpeg_worker_image.imports():
     from src.cluster.ffmpeg_worker.model import FFMpegSliceSegment
     from src.cluster.video_downloader.model import (MediaSources,
                                                     MediaSource,
-                                                    MediaCache,
-                                                    MediaProtocol,
-                                                    MediaCacheStatus)
+                                                    MediaProtocol)
     from src.cluster.web.model import SentryTransactionInfo
     import shutil, psutil, os, backoff, json, sentry_sdk
-    from modal import current_function_call_id
-    import soundfile as sf
-    import pyloudnorm as pyln
-    import noisereduce as nr
-    from pedalboard import (
-        Pedalboard,
-        Compressor,
-        Limiter,
-        HighpassFilter,
-        LowpassFilter,
-        Gain,
-        Reverb,
-        Chorus,
-        Distortion,
-    )
-    from pedalboard.io import AudioFile
 
     sentry_sdk.init(
         dsn="https://75cca970bfcc3d45d24361e7c0f1833c@sentry.bowongai.com/4",
@@ -59,28 +40,10 @@ with ffmpeg_worker_image.imports():
     modal_kv = modal.Dict.from_name(config.modal_kv_name,
                                     environment_name=config.environment,
                                     create_if_missing=True)
-    s3_mount = "/mntS3"
+    s3_mount = config.s3_mount_dir
     output_path_prefix = "/mnt/outputs"
 
 
-    @sentry_sdk.trace
-    def get_cache_filepath(media: MediaSource) -> MediaCache:
-        """
-        获取缓存路径
-        :param media: 媒体文件
-        :return:
-        """
-        # 本地挂载缓存
-        if media.protocol == MediaProtocol.s3 and media.endpoint == config.s3_region and media.bucket == config.s3_bucket_name:
-            cache_data = MediaCache(status=MediaCacheStatus.ready, cache_filepath=f"{s3_mount}/{media.get_cdn_url()}")
-        else:
-            cache_data_json = modal_kv.get(media.cache_key)
-            if not cache_data_json:
-                raise FileNotFoundError(f"{media.cache_key} cache not found")
-            cache_data = MediaCache.model_validate_json(cache_data_json)
-        return cache_data
-
-
     @sentry_sdk.trace
     def capture_hardware_info() -> Tuple[Any, int, Any]:
         cpu_freq = psutil.cpu_freq()
@@ -145,7 +108,7 @@ with ffmpeg_worker_image.imports():
         # region='ap-northeast',
         max_containers=config.ffmpeg_worker_concurrency,
         volumes={
-            s3_mount: modal.CloudBucketMount(
+            f"/mntS3": modal.CloudBucketMount(
                 bucket_name=config.s3_bucket_name,
                 secret=modal.Secret.from_name("aws-s3-secret",
                                               environment_name=config.environment),
@@ -156,6 +119,8 @@ with ffmpeg_worker_image.imports():
                                    sentry_trace: Optional[SentryTransactionInfo] = None) -> Tuple[
         str, Optional[SentryTransactionInfo]]:
 
+        fn_id = current_function_call_id()
+
         @sentry_sdk.trace
         async def print_media_streams_info(media_paths: List[str]) -> List[Dict[str, Any]]:
             """
@@ -175,16 +140,17 @@ with ffmpeg_worker_image.imports():
                     continue
             return res
 
-        @sentry_sdk.trace
-        async def ffmpeg_process(media_sources: MediaSources, output_filepath: str) -> None:
-            ffmpeg_cmd = FFmpeg().option('y').option('hide_banner')
-
+        @SentryUtils.sentry_tracker(name="视频合并任务", op="ffmpeg.concat", 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)
+        async def ffmpeg_process(media_sources: MediaSources, output_filepath: str) -> str:
+            ffmpeg_cmd = ffmpeg_init()
             input_videos = []
             for media_source in media_sources.inputs:
-                cache = get_cache_filepath(media_source)
-                logger.info(cache.cache_filepath)
-                ffmpeg_cmd.input(cache.cache_filepath)
-                input_videos.append(cache.cache_filepath)
+                cache_filepath = f"{s3_mount}/{media_source.cache_filepath}"
+                logger.info(cache_filepath)
+                ffmpeg_cmd.input(cache_filepath)
+                input_videos.append(cache_filepath)
             target_width = 1080
             target_height = 1920
             # video_info = await print_media_streams_info(media_paths=input_videos)
@@ -232,67 +198,14 @@ with ffmpeg_worker_image.imports():
                 },
             )
 
-            @ffmpeg_cmd.on("start")
-            def on_start(arguments: list[str]):
-                try:
-                    filter_index = arguments.index("-filter_complex")
-                    filter_content = arguments[filter_index + 1]
-                    arguments[filter_index + 1] = f'"{filter_content}"'
-                    args = " ".join(arguments)
-                    logger.info(f"FFmpeg command:{args}")
-                    arguments[filter_index + 1] = filter_content
-                except ValueError:
-                    args = " ".join(arguments)
-                    logger.info(f"FFmpeg command:{args}")
-
-            @ffmpeg_cmd.on("progress")
-            def on_progress(progress):
-                logger.info(f"处理进度: {progress}")
-
-            @ffmpeg_cmd.on("completed")
-            def on_completed():
-                logger.info(f"FFMpeg task completed.")
-
-            @ffmpeg_cmd.on("stderr")
-            def on_stderr(line: str):
-                if line.startswith('Error'):
-                    logger.error(line)
-                    raise RuntimeError(line)
-                else:
-                    logger.warning(line)
-
             await ffmpeg_cmd.execute()
+            s3_outputs = local_copy_to_s3([output_filepath])
+            return s3_outputs[0]
 
-        fn_id = current_function_call_id()
-        cpu_freq, cpu_count, mem = capture_hardware_info()
-        total_mb = mem.total >> 20
-        available_mb = mem.available >> 20
-        logger.info((f"[{fn_id}] Current CPUs {cpu_count} * {cpu_freq.current / 1000:.2f} GHz"))
-        logger.info(f"Current Memories {available_mb} Mb / {total_mb} Mb | {mem.percent}% used")
-
-        if sentry_trace:
-            transaction = sentry_sdk.continue_trace(environ_or_headers={
-                "sentry-trace": sentry_trace.x_trace_id,
-                "baggage": sentry_trace.x_baggage,
-            })
-        else:
-            transaction = sentry_sdk.start_transaction(op='modal.function', name="Modal Function直接调用")
-            sentry_trace = SentryTransactionInfo(x_trace_id=sentry_sdk.get_traceparent(),
-                                                 x_baggage=sentry_sdk.get_baggage())
-
-        with transaction:
-            with transaction.start_child(op='ffmpeg.concat', name='视频合并任务') as span:
-                sentry_sdk.set_tag('fn.id', fn_id)
-                span.set_data('fn.id', fn_id)
-                span.set_data('cpu.count', cpu_count)
-                span.set_data('cpu.frequency', f"{cpu_freq.current / 1000:.2f} GHz")
-                span.set_data('memory.available', f"{total_mb} Mb")
-                output_path = f"{output_path_prefix}/concat/outputs/{fn_id}/output.mp4"
-                os.makedirs(os.path.dirname(output_path), exist_ok=True)
-                await ffmpeg_process(medias, output_filepath=output_path)
-                s3_outputs = local_copy_to_s3([output_path])
-
-        return s3_outputs[0], sentry_trace
+        output_path = f"{output_path_prefix}/concat/outputs/{fn_id}/output.mp4"
+        os.makedirs(os.path.dirname(output_path), exist_ok=True)
+        s3_output = await ffmpeg_process(medias, output_filepath=output_path)
+        return s3_output, sentry_trace
 
 
     @app.function(
@@ -300,7 +213,7 @@ with ffmpeg_worker_image.imports():
         cloud="aws",
         max_containers=config.ffmpeg_worker_concurrency,
         volumes={
-            s3_mount: modal.CloudBucketMount(
+            f"/mntS3": modal.CloudBucketMount(
                 bucket_name=config.s3_bucket_name,
                 secret=modal.Secret.from_name("aws-s3-secret",
                                               environment_name=config.environment),
@@ -313,14 +226,18 @@ with ffmpeg_worker_image.imports():
                                  webhook: Optional[str],  # todo : handle webhook callback
                                  sentry_trace: Optional[SentryTransactionInfo] = None) -> Tuple[
         List[str], Optional[SentryTransactionInfo]]:
+        fn_id = current_function_call_id()
 
-        @sentry_sdk.trace
+        @SentryUtils.sentry_tracker(name="视频切割任务", op="ffmpeg.slice", 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)
         async def ffmpeg_slice_process(media_source: MediaSource, media_markers: List[FFMpegSliceSegment],
                                        fn_id: str) -> List[str]:
-            cache = get_cache_filepath(media_source)
-            logger.info(cache.cache_filepath)
+            cache_filepath = f"{s3_mount}/{media_source.cache_filepath}"
+            logger.info(cache_filepath)
             outputs: List[str] = []
-            ffmpeg_cmd = FFmpeg().option('y').option('hide_banner').input(cache.cache_filepath)
+            ffmpeg_cmd = ffmpeg_init()
+            ffmpeg_cmd.input(cache_filepath)
             filter_complex: List[str] = []
             for index, marker in enumerate(media_markers):
                 filter_complex.extend(
@@ -332,7 +249,7 @@ with ffmpeg_worker_image.imports():
             ffmpeg_cmd.option('filter_complex', ';'.join(filter_complex))
 
             for i, marker in enumerate(media_markers):
-                filename = FileUtils.file_path_extend(os.path.basename(cache.cache_filepath), str(i))
+                filename = FileUtils.file_path_extend(os.path.basename(cache_filepath), str(i))
                 output_filepath = f"{output_path_prefix}/slice/outputs/{fn_id}/{filename}"
                 output_filepath = output_filepath.replace('//', '/')
                 workdir = os.path.dirname(output_filepath)
@@ -349,52 +266,26 @@ with ffmpeg_worker_image.imports():
                                   crf=16,
                                   r=30, )
 
-            @ffmpeg_cmd.on("start")
-            def on_start(arguments: list[str]):
-                try:
-                    filter_index = arguments.index("-filter_complex")
-                    filter_content = arguments[filter_index + 1]
-                    arguments[filter_index + 1] = f'"{filter_content}"'
-                    args = " ".join(arguments)
-                    logger.info(f"FFmpeg command:{args}")
-                    arguments[filter_index + 1] = filter_content
-                except ValueError:
-                    args = " ".join(arguments)
-                    logger.info(f"FFmpeg command:{args}")
-
-            @ffmpeg_cmd.on("progress")
-            def on_progress(progress):
-                logger.info(f"处理进度: {progress}")
-
-            @ffmpeg_cmd.on("completed")
-            def on_completed():
-                logger.info(f"FFMpeg task completed.")
-
-            @ffmpeg_cmd.on("stderr")
-            def on_stderr(line: str):
-                if line.startswith('Error'):
-                    logger.error(line)
-                    raise RuntimeError(line)
-                else:
-                    logger.warning(line)
-
             await ffmpeg_cmd.execute()
 
             s3_outputs = local_copy_to_s3(outputs)
 
             return s3_outputs
 
-        @sentry_sdk.trace
+        @SentryUtils.sentry_tracker(name="视频切割任务", op="ffmpeg.slice", 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)
         async def ffmpeg_hls_slice_process(media_source: MediaSource, media_markers: List[FFMpegSliceSegment],
                                            fn_id: str) -> List[str]:
-            hls_m3u8_url = media_source.url
+            hls_m3u8_url = media_source.path
             logger.info(hls_m3u8_url)
             outputs: List[str] = []
-            ffmpeg_cmd = FFmpeg().option('y').option('hide_banner').input(hls_m3u8_url,
-                                                                          protocol_whitelist="file,http,https,tcp,tls",
-                                                                          reconnect="1",  # 自动重连
-                                                                          reconnect_streamed="1",
-                                                                          reconnect_delay_max="5")
+            ffmpeg_cmd = ffmpeg_init()
+            ffmpeg_cmd.input(hls_m3u8_url,
+                             protocol_whitelist="file,http,https,tcp,tls",
+                             reconnect="1",  # 自动重连
+                             reconnect_streamed="1",
+                             reconnect_delay_max="5")
             filter_complex: List[str] = []
             for index, marker in enumerate(media_markers):
                 filter_complex.extend(
@@ -425,67 +316,17 @@ with ffmpeg_worker_image.imports():
                                   crf=16,
                                   r=30, )
 
-            @ffmpeg_cmd.on("start")
-            def on_start(arguments: list[str]):
-                try:
-                    filter_index = arguments.index("-filter_complex")
-                    filter_content = arguments[filter_index + 1]
-                    arguments[filter_index + 1] = f'"{filter_content}"'
-                    args = " ".join(arguments)
-                    logger.info(f"FFmpeg command:{args}")
-                    arguments[filter_index + 1] = filter_content
-                except ValueError:
-                    args = " ".join(arguments)
-                    logger.info(f"FFmpeg command:{args}")
-
-            @ffmpeg_cmd.on("progress")
-            def on_progress(progress):
-                logger.info(f"处理进度: {progress}")
-
-            @ffmpeg_cmd.on("completed")
-            def on_completed():
-                logger.info(f"FFMpeg task completed.")
-
-            @ffmpeg_cmd.on("stderr")
-            def on_stderr(line: str):
-                if line.startswith('Error'):
-                    logger.error(line)
-                    raise RuntimeError(line)
-                else:
-                    logger.warning(line)
-
             await ffmpeg_cmd.execute()
 
             s3_outputs = local_copy_to_s3(outputs)
 
             return s3_outputs
 
-        fn_id = current_function_call_id()
-        cpu_freq, cpu_count, mem = capture_hardware_info()
-        total_mb = mem.total >> 20
-        available_mb = mem.available >> 20
-        logger.info((f"[{fn_id}] Current CPUs {cpu_count} * {cpu_freq.current / 1000:.2f} GHz"))
-        logger.info(f"Current Memories {available_mb} Mb / {total_mb} Mb | {mem.percent}% used")
-
-        if sentry_trace:
-            transaction = sentry_sdk.continue_trace(environ_or_headers={"sentry-trace": sentry_trace.x_trace_id,
-                                                                        "baggage": sentry_trace.x_baggage, })
-        else:
-            transaction = sentry_sdk.start_transaction(op='modal.function', name="Modal Function直接调用")
-            sentry_trace = SentryTransactionInfo(x_trace_id=sentry_sdk.get_traceparent(),
-                                                 x_baggage=sentry_sdk.get_baggage())
-        with transaction:
-            with transaction.start_child(op='ffmpeg.slice', name='视频切割任务') as span:
-                sentry_sdk.set_tag('fn.id', fn_id)
-                span.set_data('fn.id', fn_id)
-                span.set_data('cpu.count', cpu_count)
-                span.set_data('cpu.frequency', f"{cpu_freq.current / 1000:.2f} GHz")
-                span.set_data('memory.available', f"{total_mb} Mb")
-                match media.protocol:
-                    case MediaProtocol.hls:
-                        outputs = await ffmpeg_hls_slice_process(media_source=media, media_markers=markers, fn_id=fn_id)
-                    case _:
-                        outputs = await ffmpeg_slice_process(media_source=media, media_markers=markers, fn_id=fn_id)
+        match media.protocol:
+            case MediaProtocol.hls:
+                outputs = await ffmpeg_hls_slice_process(media_source=media, media_markers=markers, fn_id=fn_id)
+            case _:
+                outputs = await ffmpeg_slice_process(media_source=media, media_markers=markers, fn_id=fn_id)
 
         return outputs, sentry_trace
 
@@ -495,7 +336,7 @@ with ffmpeg_worker_image.imports():
         cloud="aws",
         max_containers=config.ffmpeg_worker_concurrency,
         volumes={
-            s3_mount: modal.CloudBucketMount(
+            f"/mntS3": modal.CloudBucketMount(
                 bucket_name=config.s3_bucket_name,
                 secret=modal.Secret.from_name("aws-s3-secret",
                                               environment_name=config.environment),
@@ -509,16 +350,13 @@ with ffmpeg_worker_image.imports():
         Optional[str], Optional[SentryTransactionInfo]]:
 
         fn_id = current_function_call_id()
-        cpu_freq, cpu_count, mem = capture_hardware_info()
-        total_mb = mem.total >> 20
-        available_mb = mem.available >> 20
-        logger.info((f"[{fn_id}] Current CPUs {cpu_count} * {cpu_freq.current / 1000:.2f} GHz"))
-        logger.info(f"Current Memories {available_mb} Mb / {total_mb} Mb | {mem.percent}% used")
 
-        @sentry_sdk.trace
+        @SentryUtils.sentry_tracker(name="视频切割任务", op="ffmpeg.extract.audio", 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)
         async def ffmpeg_process(media: MediaSource, fn_id: str) -> str:
-            cache = get_cache_filepath(media)
-            ffprobe_cmd = FFmpeg('ffprobe').input(cache.cache_filepath,
+            cache_filepath = f"{s3_mount}/{media.cache_filepath}"
+            ffprobe_cmd = FFmpeg('ffprobe').input(cache_filepath,
                                                   v="quiet",
                                                   print_format="json",
                                                   select_streams="a",
@@ -561,36 +399,22 @@ with ffmpeg_worker_image.imports():
             output_path = f"{output_path_prefix}/extract_audio/outputs/{fn_id}/output.wav"
             os.makedirs(os.path.dirname(output_path), exist_ok=True)
             ffmpeg_cmd = ffmpeg_init()
-            ffmpeg_cmd.input(cache.cache_filepath).output(output_path,
-                                                          map="0:a",
-                                                          acodec="pcm_s16le",
-                                                          ar=44100,
-                                                          ac=1)
+            ffmpeg_cmd.input(cache_filepath).output(output_path,
+                                                    map="0:a",
+                                                    acodec="pcm_s16le",
+                                                    ar=44100,
+                                                    ac=1)
 
             await ffmpeg_cmd.execute()
             s3_outputs = local_copy_to_s3([output_path])
             return s3_outputs[0]
 
-        if sentry_trace:
-            transaction = sentry_sdk.continue_trace(environ_or_headers={"sentry-trace": sentry_trace.x_trace_id,
-                                                                        "baggage": sentry_trace.x_baggage, })
-        else:
-            transaction = sentry_sdk.start_transaction(op='modal.function', name="Modal Function直接调用")
-            sentry_trace = SentryTransactionInfo(x_trace_id=sentry_sdk.get_traceparent(),
-                                                 x_baggage=sentry_sdk.get_baggage())
-        with transaction:
-            with transaction.start_child(op='ffmpeg.extract.audio', name='视频切割任务') as span:
-                sentry_sdk.set_tag('fn.id', fn_id)
-                span.set_data('fn.id', fn_id)
-                span.set_data('cpu.count', cpu_count)
-                span.set_data('cpu.frequency', f"{cpu_freq.current / 1000:.2f} GHz")
-                span.set_data('memory.available', f"{total_mb} Mb")
-                match media_source.protocol:
-                    case MediaProtocol.hls:
-                        return None, sentry_trace
-                    case _:
-                        output = await ffmpeg_process(media_source, fn_id=fn_id)
-                        return output, sentry_trace
+        match media_source.protocol:
+            case MediaProtocol.hls:
+                return None, sentry_trace
+            case _:
+                output = await ffmpeg_process(media_source, fn_id=fn_id)
+                return output, sentry_trace
 
 
     async def ffmpeg_overlay_gif(media: MediaSource, gif: MediaSource):
@@ -602,7 +426,7 @@ with ffmpeg_worker_image.imports():
         cloud="aws",
         max_containers=config.ffmpeg_worker_concurrency,
         volumes={
-            s3_mount: modal.CloudBucketMount(
+            f"/mntS3": modal.CloudBucketMount(
                 bucket_name=config.s3_bucket_name,
                 secret=modal.Secret.from_name("aws-s3-secret",
                                               environment_name=config.environment),
@@ -622,8 +446,7 @@ with ffmpeg_worker_image.imports():
         async def ffmpeg_process(media: MediaSource, func_id: str, mirror_scale_down_size: int = 6,
                                  mirror_from_right: bool = True,
                                  mirror_position: tuple[float, float] = (40, 40)) -> str:
-            cache = get_cache_filepath(media)
-            media_filepath = cache.cache_filepath
+            media_filepath = f"{s3_mount}/{media.cache_filepath}"
             video_metadata = VideoUtils.ffprobe_video_format(media_filepath)
             mirror_x = (
                 f"main_w-overlay_w-{mirror_position[0]}"
@@ -669,7 +492,12 @@ with ffmpeg_worker_image.imports():
         raise NotImplementedError
 
 
-    async def ffmpeg_bgm_nosie_reduce(media: MediaSource, bgm: MediaSource, ):
+    async def ffmpeg_bgm_nosie_reduce(media: MediaSource, bgm: MediaSource, noise_sample: Optional[MediaSource] = None):
+        fn_id = current_function_call_id()
+
+        async def ffmpeg_process(video: MediaSource, bgm: MediaSource, noise_sample: Optional[MediaSource] = None):
+            raise NotImplementedError
+
         raise NotImplementedError
 
 

src/cluster/video_downloader/model.py

@@ -2,8 +2,11 @@ import os
 from datetime import datetime
 from enum import Enum
 from typing import List, Union, Optional, Any, Dict
-from pydantic import BaseModel, Field, field_validator, ValidationError, field_serializer, SerializationInfo
+from urllib.parse import urlparse
+from pydantic import BaseModel, Field, field_validator, ValidationError, field_serializer, SerializationInfo, \
+    computed_field
 from pydantic.json_schema import JsonSchemaValue
+from src.cluster.config import config
 
 
 class MediaProtocol(str, Enum):
@@ -20,57 +23,68 @@ class MediaCacheStatus(str, Enum):
     ready = "ready"
     deleted = "deleted"
     missing = "missing"
+    unknown = "unknown"
 
 
 class MediaSource(BaseModel):
-    url: str = Field()
-    protocol: MediaProtocol = Field()
-    endpoint: Optional[str] = Field()
-    bucket: Optional[str] = Field()
-    cache_key: Optional[str] = Field()
+    path: str = Field(description="媒体源的路径")
+    protocol: MediaProtocol = Field(description="媒体源的来源协议")
+    endpoint: Optional[str] = Field(description="媒体源来源的终端地址，根据使用协议的不同，会有没有endpoint的情况")
+    bucket: Optional[str] = Field(description="媒体源所使用的存储桶(s3)/SubAppId(vod)")
+    urn: Optional[str] = Field(description="媒体源的唯一指定标识")
+    status: MediaCacheStatus = Field(default=MediaCacheStatus.unknown, description="媒体源在Modal集群挂载的缓存状态")
+    expired_at: Optional[datetime] = Field(description="缓存过期时间点", default=None)
+    downloader_id: Optional[str] = Field(description="正在处理下载的Downloader ID", default=None)
+    progress: int = Field(description="缓存进度", default=0)
+
+    # cache_filepath: Optional[str] = Field(description="s3挂载路径下的缓存相对路径", default=None)
 
     @classmethod
-    def from_str(cls, mediaUrl: str) -> 'MediaSource':
-        if mediaUrl.startswith('http://') or mediaUrl.startswith('https://'):
-            return MediaSource(url=mediaUrl,
+    def from_str(cls, media_url: str) -> 'MediaSource':
+        if media_url.startswith('http://') or media_url.startswith('https://'):
+            parsed_url = urlparse(media_url)
+            path_str = f"{parsed_url.path}?{parsed_url.query}" if parsed_url.query else parsed_url.path
+            return MediaSource(path=path_str,
                                protocol=MediaProtocol.http,
-                               endpoint=None,
+                               endpoint=parsed_url.netloc,  # domain of http url
                                bucket=None,
-                               cache_key=mediaUrl)
-        elif mediaUrl.startswith('s3://'):  # s3://{endpoint}/{bucket}/{url}
-            paths = mediaUrl[5:].split('/')
-            return MediaSource(url='/'.join(paths[2:]),
+                               urn=media_url)
+        elif media_url.startswith('s3://'):  # s3://{endpoint}/{bucket}/{url}
+            paths = media_url[5:].split('/')
+            return MediaSource(path='/'.join(paths[2:]),
                                protocol=MediaProtocol.s3,
                                endpoint=paths[0],
                                bucket=paths[1],
-                               cache_key=mediaUrl)
-        elif mediaUrl.startswith('vod://'):  # vod://{endpoint}/{subAppId}/{fileId}
-            paths = mediaUrl[6:].split('/')
+                               urn=media_url)
+        elif media_url.startswith('vod://'):  # vod://{endpoint}/{subAppId}/{fileId}
+            paths = media_url[6:].split('/')
             # 兼容有文件类型后缀和没有文件类型后缀的格式
             url = paths[2] if '.' in os.path.basename(paths[2]) else paths[2] + ".mp4"
-            return MediaSource(url=url,
+            return MediaSource(path=url,
                                protocol=MediaProtocol.vod,
                                bucket=paths[1],
                                endpoint=paths[0],
-                               cache_key=mediaUrl)
-        elif mediaUrl.startswith('cos://'):  # cos://{endpoint}/{bucket}/{url}
-            paths = mediaUrl[6:].split('/')
-            return MediaSource(url='/'.join(paths[2:]),
+                               urn=media_url)
+        elif media_url.startswith('cos://'):  # cos://{endpoint}/{bucket}/{url}
+            paths = media_url[6:].split('/')
+            return MediaSource(path='/'.join(paths[2:]),
                                protocol=MediaProtocol.cos,
                                endpoint=paths[0],
                                bucket=paths[1],
-                               cache_key=mediaUrl)
-        elif mediaUrl.startswith('hls://'):
+                               urn=media_url)
+        elif media_url.startswith('hls://'):
             # hls://merge-local-1324682537.cos.ap-shanghai.myqcloud.com/streams/1264/30322.m3u8
-            paths = mediaUrl[6:]
-            return MediaSource(url=f"https://{paths}",
+            paths = media_url[6:]
+            return MediaSource(path=f"https://{paths}",
                                protocol=MediaProtocol.hls,
                                endpoint=None,
                                bucket=None,
-                               cache_key=mediaUrl
+                               urn=media_url
                                )
         else:
-            raise ValidationError("mediaUrl必须以http[s]、s3或vod协议开头")
+            available_schemas = [member.value for member in MediaProtocol]
+            available_schemas_str = ','.join(available_schemas)
+            raise ValidationError(f"mediaUrl必须以{available_schemas_str}协议开头")
 
     @classmethod
     def __get_pydantic_json_schema__(cls, core_schema: Any, handler: Any) -> JsonSchemaValue:
@@ -85,16 +99,44 @@ class MediaSource(BaseModel):
             ]
         }
 
+    @computed_field(description="s3挂载路径下的缓存相对路径")
+    @property
+    def cache_filepath(self) -> str:
+        match self.protocol:
+            case MediaProtocol.s3:
+                # 本地挂载缓存
+                if self.protocol == MediaProtocol.s3 and self.endpoint == config.s3_region and self.bucket == config.s3_bucket_name:
+                    return f"{self.path}"
+                else:
+                    return f"{self.protocol.value}/{self.endpoint}/{self.bucket}/{self.path}"
+            case MediaProtocol.http:
+                clean_path = self.path.split('?')[0] if '?' in self.path else self.path
+                return f"{self.protocol.value}/{self.endpoint}/{clean_path}"
+            case _:
+                return f"{self.protocol.value}/{self.endpoint}/{self.bucket}/{self.path}"
+
+    @field_serializer('expired_at')
+    def serialize_datetime(self, value: Optional[datetime], info: SerializationInfo) -> Optional[str]:
+        if value:
+            return value.isoformat()
+        else:
+            return None
+
     def get_cdn_url(self) -> str:
         if self.protocol == MediaProtocol.s3:
-            return f"{self.url}"
+            return f"{self.path}"
         elif self.protocol == MediaProtocol.http:
-            return f"{self.protocol.value}/{self.url}"
-        return f"{self.protocol.value}/{self.endpoint}/{self.bucket}/{self.url}"
+            return f"{self.protocol.value}/{self.path}"
+        return f"{self.protocol.value}/{self.endpoint}/{self.bucket}/{self.path}"
 
-
-class CachedMediaSource(MediaSource):
-    cached_s3_path: str = Field(description="s3上的相对缓存路径")
+    def __str__(self):
+        match self.protocol:
+            case MediaProtocol.http:
+                return f"{self.protocol.value}://{self.path}"
+            case MediaProtocol.hls:
+                return f"{self.protocol.value}://{self.path[8:]}"  # strip "https://" from url
+            case _:
+                return f"{self.protocol.value}://{self.endpoint}/{self.bucket}/{self.path}"
 
 
 class MediaSources(BaseModel):
@@ -126,23 +168,8 @@ class MediaSources(BaseModel):
     }
 
 
-class MediaCache(BaseModel):
-    status: MediaCacheStatus = Field(description="缓存被处理的状态")
-    progress: float = Field(description="缓存下载进度0-1", default=0)
-    expired_at: Optional[datetime] = Field(description="缓存过期时间点", default=None)
-    downloader_id: Optional[str] = Field(description="正在处理下载的Downloader ID", default=None)
-    cache_filepath: Optional[str] = Field(description="缓存的文件地址", default=None)
-
-    @field_serializer('expired_at')
-    def serialize_datetime(self, value: Optional[datetime], info: SerializationInfo) -> Optional[str]:
-        if value:
-            return value.isoformat()
-        else:
-            return None
-
-
 class CacheResult(BaseModel):
-    caches: Dict[str, MediaCache] = Field(description="Cache ID")
+    caches: Dict[str, MediaSource] = Field(description="Cache ID")
 
 
 class DownloadResult(BaseModel):

src/cluster/video_downloader/worker.py

@@ -14,11 +14,12 @@ downloader_image = (
 with downloader_image.imports():
     import os, httpx, crcmod
     from tqdm import tqdm
-    from typing import Tuple, Dict
+    from typing import Tuple, Dict, List
     from loguru import logger
     from modal import current_function_call_id
     from src.cluster.config import config
-    from src.cluster.video_downloader.model import MediaSource, MediaCacheStatus, MediaCache, MediaProtocol
+    from src.cluster.web.Utils.KVCache import KVCache
+    from src.cluster.video_downloader.model import MediaSource, MediaCacheStatus, MediaProtocol
     from src.cluster.web.model import SentryTransactionInfo
     from datetime import datetime, UTC, timedelta
     from tencentcloud.common.credential import Credential
@@ -45,9 +46,11 @@ with downloader_image.imports():
     cf_kv_api_token = os.environ.get("CF_KV_API_TOKEN")
     cf_kv_namespace_id = os.environ.get("CF_KV_NAMESPACE_ID")
 
+    modal_kv_cache = KVCache(kv_name=config.modal_kv_name, environment=config.environment)
+
 
     @sentry_sdk.trace
-    def batch_update_cloudflare_kv(caches: Dict[str, MediaCache]):
+    def batch_update_cloudflare_kv(caches: List[MediaSource]):
         with httpx.Client() as client:
             try:
                 response = client.put(
@@ -56,10 +59,10 @@ with downloader_image.imports():
                     json=[
                         {
                             "based64": False,
-                            "key": mediaKey,
-                            "value": cache_data.model_dump_json(),
+                            "key": cache.urn,
+                            "value": cache.model_dump_json(),
                         }
-                        for (mediaKey, cache_data) in caches.items()
+                        for cache in caches
                     ]
                 )
                 response.raise_for_status()
@@ -75,13 +78,13 @@ with downloader_image.imports():
 
 
     @sentry_sdk.trace
-    def batch_remove_cloudflare_kv(caches: Dict[str, MediaCache]):
+    def batch_remove_cloudflare_kv(caches: List[MediaSource]):
         with httpx.Client() as client:
             try:
                 response = client.post(
                     f"https://api.cloudflare.com/client/v4/accounts/{cf_account_id}/storage/kv/namespaces/{cf_kv_namespace_id}/bulk/delete",
                     headers={"Authorization": f"Bearer {cf_kv_api_token}"},
-                    json=[mediaKey for (mediaKey, cache_data) in caches.items()]
+                    json=[cache.urn for cache in caches]
                 )
                 response.raise_for_status()
             except httpx.RequestError as e:
@@ -107,7 +110,7 @@ with downloader_image.imports():
                          },
                          secrets=[modal.Secret.from_name("tencent-cloud-secret", environment_name=config.environment)])
     @modal.concurrent(max_inputs=10)
-    async def cache_submit(media: MediaSource, sentry_trace: SentryTransactionInfo) -> MediaCache:
+    async def cache_submit(media: MediaSource, sentry_trace: SentryTransactionInfo) -> MediaSource:
         def vod_init():
             tencent_secret_id = os.environ["VOD_SECRET_ID"]
             tencent_secret_key = os.environ["VOD_SECRET_KEY"]
@@ -119,18 +122,18 @@ with downloader_image.imports():
             request = vod_request_models.DescribeMediaInfosRequest()
             request.SubAppId = int(media.bucket)
             # 兼容fileId带文件类型的格式和不带文件类型的格式
-            request.FileIds = [media.url.split('.')[0] if '.' in media.url else media.url]
+            request.FileIds = [media.path.split('.')[0] if '.' in media.path else media.path]
             response = vod_client.DescribeMediaInfos(request)
             if len(response.MediaInfoSet) > 0:
                 media_info = response.MediaInfoSet[0].BasicInfo
                 logger.info(f"VOD info = {media_info}")
                 file_extension = media_info.Type
-                cache_dir = f"/mntS3/{media.protocol.value}/{media.endpoint}/{media.bucket}"
-                cache_file = f"{media.url}.{file_extension}"
+                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:
                 raise FileNotFoundError(
-                    f"FileId : {media.url} not found in SubAppId: {media.bucket} at {media.endpoint}")
+                    f"FileId : {media.path} not found in SubAppId: {media.bucket} at {media.endpoint}")
 
         def vod_download(media: MediaSource, on_progress_update: callable(float) = None) -> str:
             cache_dir, cache_file, url = vod_info(media)
@@ -226,8 +229,7 @@ with downloader_image.imports():
                     progress_bar.close()
 
         vod_client = vod_init()
-        modal_kv = modal.Dict.from_name(config.modal_kv_name, environment_name=config.environment,
-                                        create_if_missing=True)
+        modal_kv = modal_kv_cache
         fn_id = current_function_call_id()
 
         with sentry_sdk.continue_trace(environ_or_headers={"sentry-trace": sentry_trace.x_trace_id,
@@ -245,56 +247,50 @@ with downloader_image.imports():
                 receive_span.set_data("messaging.message.id", fn_id)
                 receive_span.set_data("messaging.destination.name", "video-downloader.cache_submit")
                 receive_span.set_data("messaging.message.retry.count", 0)
-                receive_span.set_data("cache.key", media.cache_key)
+                receive_span.set_data("cache.key", media.urn)
                 with receive_span.start_child(name="处理缓存视频任务", op="queue.process") as process_span:
                     process_span.set_data("messaging.message.id", fn_id)
                     process_span.set_data("messaging.destination.name", "video-downloader.cache_submit")
                     process_span.set_data("messaging.message.retry.count", 0)
-                    process_span.set_data("cache.key", media.cache_key)
+                    process_span.set_data("cache.key", media.urn)
                     volume_cache_path = None
-                    media_cache_downloading = MediaCache(status=MediaCacheStatus.downloading,
-                                                         downloader_id=fn_id)
                     match media.protocol:
                         case MediaProtocol.vod:
-                            # def on_progress_callback(progress: float):
-                            #     media_cache_downloading.progress = progress
-                            #     caches_data = {media.cache_key: media_cache_downloading}
-                            #     modal_kv.put(media.cache_key, media_cache_downloading.model_dump_json())
-                            #     batch_update_cloudflare_kv(caches_data)
-
                             try:
                                 volume_cache_path = vod_download(media)
                                 process_span.set_status("success")
                             except Exception as e:
                                 logger.exception(e)
-                                media_cache_downloading.status = MediaCacheStatus.failed
-                                caches = {f"{media.cache_key}": media_cache_downloading}
-                                modal_kv.put(media.cache_key, media_cache_downloading)
-                                batch_update_cloudflare_kv(caches)
+                                media.status = MediaCacheStatus.failed
+                                modal_kv.set_cache(media)
+                                batch_update_cloudflare_kv([media])
                                 process_span.set_status("failed")
                         case MediaProtocol.http:
                             try:
-
-                                download_large_file(url=url, output_path=local_cache_filepath)
+                                cache_filepath = f"{config.s3_mount_dir}/{media.cache_filepath}"
+                                download_large_file(url=media.__str__(), output_path=cache_filepath)
                             except Exception as e:
                                 logger.exception(e)
-                                media_cache_downloading.status = MediaCacheStatus.failed
-                                caches = {f"{media.cache_key}": media_cache_downloading}
-                                modal_kv.put(media.cache_key, media_cache_downloading)
-                                batch_update_cloudflare_kv(caches)
+                                media.status = MediaCacheStatus.failed
+                                modal_kv.set_cache(media)
+                                batch_update_cloudflare_kv([media])
                                 process_span.set_status("failed")
+                        case MediaProtocol.s3:
+                            # 本地挂载缓存
+                            if media.protocol == MediaProtocol.s3 and media.endpoint == config.s3_region and media.bucket == config.s3_bucket_name:
+                                volume_cache_path = f"{config.s3_mount_dir}/{media.cache_filepath}"
+                            else:
+                                logger.error("protocol not yet supported")
                         case _:
                             process_span.set_status("failed")
-                            raise NotImplementedError("protocol not yet supported")
-                    media_cache_ready = MediaCache(
-                        downloader_id=fn_id,
-                        cache_filepath=volume_cache_path,
-                        status=MediaCacheStatus.ready if volume_cache_path else MediaCacheStatus.failed,
-                        progress=1 if volume_cache_path else 0,
-                        expired_at=datetime.now(UTC) + timedelta(days=7) if volume_cache_path else None, )
-                    modal_kv.put(media.cache_key, media_cache_ready.model_dump_json())
-                    batch_update_cloudflare_kv({media.cache_key: media_cache_ready})
-        return media_cache_ready
+                            logger.error(f"protocol not yet supported")
+                    media.downloader_id = fn_id
+                    media.status = MediaCacheStatus.ready if volume_cache_path else MediaCacheStatus.failed
+                    media.progress = 1 if volume_cache_path else 0
+                    media.expired_at = datetime.now(UTC) + timedelta(days=7) if volume_cache_path else None
+                    modal_kv.set_cache(media)
+                    batch_update_cloudflare_kv([media])
+        return media
 
 
     @worker_app.function(cpu=1, timeout=300,
@@ -306,16 +302,7 @@ with downloader_image.imports():
                              ),
                          })
     @modal.concurrent(max_inputs=10)
-    async def cache_delete(cache: MediaCache) -> MediaCache:
-        sentry_sdk.add_breadcrumb({
-            "MODAL_CLOUD_PROVIDER": os.environ.get('MODAL_CLOUD_PROVIDER', 'unknown'),
-            "MODAL_ENVIRONMENT": os.environ.get('MODAL_ENVIRONMENT', 'unknown'),
-            "MODAL_IMAGE_ID": os.environ.get('MODAL_IMAGE_ID', 'unknown'),
-            "MODAL_IS_REMOTE": os.environ.get('MODAL_IS_REMOTE', 'unknown'),
-            "MODAL_REGION": os.environ.get('MODAL_REGION', 'unknown'),
-            "MODAL_TASK_ID": os.environ.get('MODAL_TASK_ID', 'unknown'),
-            "MODAL_IDENTITY_TOKEN": os.environ.get('MODAL_IDENTITY_TOKEN', 'unknown'),
-        })
+    async def cache_delete(cache: MediaSource) -> MediaSource:
         if os.path.exists(cache.cache_filepath):
             os.remove(cache.cache_filepath)
             cache.status = MediaCacheStatus.deleted

src/cluster/web/Utils/KVCache.py

@@ -0,0 +1,33 @@
+from typing import Optional
+
+import modal
+
+from src.cluster.video_downloader.model import MediaSource
+
+
+class KVCache:
+    kv: modal.Dict
+
+    def __init__(self, kv_name: str, environment: str):
+        self.kv = modal.Dict.from_name(kv_name, environment_name=environment, create_if_missing=True)
+
+    def get_cache(self, urn: str) -> Optional[MediaSource]:
+        cache_json = self.kv.get(urn)
+        if not cache_json:
+            return None
+        return MediaSource.model_validate_json(cache_json)
+
+    def set_cache(self, media: MediaSource):
+        cache_json = media.model_dump_json()
+        self.kv.put(media.urn, cache_json)
+
+    def clear(self):
+        self.kv.clear()
+
+    def pop(self, urn: str, raise_exception: bool = True) -> Optional[MediaSource]:
+        cache_json = self.kv.pop(urn)
+        if not cache_json:
+            if raise_exception:
+                raise KeyError("URN错误，资源不存在")
+            return None
+        return MediaSource.model_validate_json(cache_json)

src/cluster/web/model.py

@@ -124,5 +124,6 @@ class FFMPEGCornerMirrorRequest(BaseFFMPEGTaskRequest):
         else:
             raise TypeError(v)
 
+
 class FFMPEGCornerMirrorTaskStatusResponse(BaseFFMPEGTaskStatusResponse):
     result: Optional[str] = Field(default=None, description="任务运行结果")

src/cluster/web/worker.py

@@ -1,64 +1,74 @@
 import modal
+from src.cluster.config import config
 
 fastapi_image = (
     modal.Image
     .debian_slim(python_version="3.11")
     .pip_install("fastapi[standard]", "sentry-sdk[fastapi]",
-                 'loguru', 'pydantic', 'pydantic_settings', 'scalar-fastapi')
-    .add_local_python_source("src.cluster.video_downloader.model", copy=True)
-    .add_local_python_source("src.cluster.web.model", copy=True)
+                 'loguru', 'pydantic', 'pydantic_settings', 'scalar-fastapi', 'psutil')
     .add_local_python_source("src.cluster.config", copy=True)
+    .add_local_python_source("src.cluster.web", copy=True)
+    .add_local_python_source("src.cluster.video_downloader", copy=True)
+    .add_local_python_source("src.cluster.ffmpeg_worker", copy=True)
 )
 
-with fastapi_image.imports():
-    import os
-    import httpx
-    from typing import Dict, List, Annotated, Tuple, Any, Optional
-    from modal import current_function_call_id
-    from modal.call_graph import InputStatus
-    from loguru import logger
-    from fastapi import FastAPI, Response, Depends, Header
-    from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
-    from fastapi.responses import JSONResponse, RedirectResponse
-    from fastapi.exceptions import HTTPException
-    from starlette import status
-    from scalar_fastapi import get_scalar_api_reference
-    import sentry_sdk
-    from sentry_sdk.integrations.loguru import LoguruIntegration, LoggingLevels
-    from sentry_sdk.integrations.fastapi import FastApiIntegration
-    from src.cluster.config import config
-    from src.cluster.video_downloader.model import (MediaSource, MediaSources, MediaCacheStatus, MediaCache,
-                                                    CacheResult, DownloadResult)
-    from src.cluster.web.model import (TaskStatus, ErrorCode,
-                                       SentryTransactionInfo,
-                                       SentryTransactionHeader,
-                                       ModalTaskResponse,
-                                       FFMPEGSliceRequest,
-                                       FFMPEGConcatRequest,
-                                       FFMPEGExtractAudioRequest,
-                                       FFMPEGCornerMirrorRequest,
-                                       FFMPEGSliceTaskStatusResponse,
-                                       FFMPEGConcatTaskStatusResponse,
-                                       FFMPEGExtractAudioTaskStatusResponse,
-                                       FFMPEGCornerMirrorTaskStatusResponse)
-
-    fastapi_app = modal.App(image=fastapi_image)
+fastapi_app = modal.App(image=fastapi_image, )
 
 
-    @fastapi_app.function(scaledown_window=60,
-                          secrets=[
-                              modal.Secret.from_name("cf-kv-secret", environment_name=config.environment),
-                          ])
-    @modal.concurrent(max_inputs=100)
-    @modal.asgi_app()
-    def fastapi_webapp():
+@fastapi_app.function(scaledown_window=60,
+                      secrets=[
+                          modal.Secret.from_name("cf-kv-secret", environment_name=config.environment),
+                      ])
+@modal.concurrent(max_inputs=100)
+@modal.asgi_app(custom_domains=["modal-dev.bowong.cc"])
+def fastapi_webapp():
+    with fastapi_image.imports():
+        import os
+        import httpx
+        import asyncio
+        from typing import List, Annotated, Tuple, Any, Optional
+        from modal import current_function_call_id
+        from modal.call_graph import InputStatus
+        from loguru import logger
+        from fastapi import FastAPI, Response, Depends, Header
+        from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+        from fastapi.responses import JSONResponse, RedirectResponse
+        from fastapi.exceptions import HTTPException
+        from starlette import status
+        from scalar_fastapi import get_scalar_api_reference
+        import sentry_sdk
+        from sentry_sdk.integrations.loguru import LoguruIntegration, LoggingLevels
+        from sentry_sdk.integrations.fastapi import FastApiIntegration
+        from src.cluster.web.Utils.KVCache import KVCache
+        from src.cluster.ffmpeg_worker.Utils.SentryUtils import SentryUtils
+        from src.cluster.video_downloader.model import (MediaSource,
+                                                        MediaSources,
+                                                        MediaCacheStatus,
+                                                        CacheResult,
+                                                        DownloadResult)
+        from src.cluster.web.model import (TaskStatus, ErrorCode,
+                                           SentryTransactionInfo,
+                                           SentryTransactionHeader,
+                                           ModalTaskResponse,
+                                           FFMPEGSliceRequest,
+                                           FFMPEGConcatRequest,
+                                           FFMPEGExtractAudioRequest,
+                                           FFMPEGCornerMirrorRequest,
+                                           FFMPEGSliceTaskStatusResponse,
+                                           FFMPEGConcatTaskStatusResponse,
+                                           FFMPEGExtractAudioTaskStatusResponse,
+                                           FFMPEGCornerMirrorTaskStatusResponse)
+
         bearer_scheme = HTTPBearer()
 
         web_app = FastAPI(title="Modal worker API",
                           summary="Modal Worker的API, 包括缓存视频, 发起生产任务等",
                           servers=[
-                              {'url': 'https://bowongai-dev--video-downloader-fastapi-webapp.modal.run',
-                               'description': 'modal dev环境测试服务'}])
+                              # {'url': 'https://bowongai-dev--video-downloader-fastapi-webapp.modal.run',
+                              #  'description': 'modal dev环境测试服务'},
+                              {'url': 'https://modal-dev.bowong.cc',
+                               'description': 'modal dev环境测试服务'},
+                          ])
 
         async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(bearer_scheme)):
             token = credentials.credentials
@@ -87,8 +97,7 @@ with fastapi_image.imports():
                             FastApiIntegration()
                         ]
                         )
-        media_cache_kv = modal.Dict.from_name('media-cache', environment_name=config.environment,
-                                              create_if_missing=True)
+        modal_kv_cache = KVCache(kv_name=config.modal_kv_name, environment=config.environment)
         cf_account_id = os.environ.get("CF_ACCOUNT_ID")
         cf_kv_api_token = os.environ.get("CF_KV_API_TOKEN")
         cf_kv_namespace_id = os.environ.get("CF_KV_NAMESPACE_ID")
@@ -109,7 +118,7 @@ with fastapi_image.imports():
         }
 
         @sentry_sdk.trace
-        def batch_update_cloudflare_kv(caches: Dict[str, MediaCache]):
+        def batch_update_cloudflare_kv(caches: List[MediaSource]):
             with httpx.Client() as client:
                 try:
                     response = client.put(
@@ -118,10 +127,10 @@ with fastapi_image.imports():
                         json=[
                             {
                                 "based64": False,
-                                "key": mediaKey,
-                                "value": cache_data.model_dump_json(),
+                                "key": cache.urn,
+                                "value": cache.model_dump_json(),
                             }
-                            for (mediaKey, cache_data) in caches.items()
+                            for cache in caches
                         ]
                     )
                     response.raise_for_status()
@@ -215,62 +224,74 @@ with fastapi_image.imports():
         @sentry_sdk.trace
         async def cache(medias: MediaSources) -> CacheResult:
             fn_id = current_function_call_id()
-            caches: Dict[str, MediaCache] = {}
-            parent = sentry_sdk.get_current_span()
+            caches: MediaSources
             sentry_trace = SentryTransactionInfo(x_trace_id=sentry_sdk.get_traceparent(),
                                                  x_baggage=sentry_sdk.get_baggage())
-            for media in medias.inputs:
-                with parent.start_child(name="同步视频缓存", op="cache.get") as cache_span:
-                    cache_span.set_data("runner_id", fn_id)
-                    cache_span.set_data("cache.key", [media.cache_key])
-                    video_cache_status_json = media_cache_kv.get(media.cache_key)
-                    video_cache: MediaCache
-                    cache_hit: bool = False
 
-                    if not video_cache_status_json:
+            @SentryUtils.sentry_tracker(name="同步视频缓存", op="cache.get", fn_id=fn_id,
+                                        sentry_trace_id=None, sentry_baggage=None)
+            async def cache_handler(media: MediaSource):
+                cache_span = sentry_sdk.get_current_span()
+                cache_span.set_data("runner_id", fn_id)
+                cache_span.set_data("cache.key", [media.urn])
+                cached_media = modal_kv_cache.get_cache(media.urn)
+                cache_hit: bool = False
+                if not cached_media:
+                    # start new download task
+                    with cache_span.start_child(name="视频缓存任务入队",
+                                                op="queue.publish") as queue_publish_span:
+                        fn = modal.Function.from_name(config.app_name, 'cache_submit')
+                        fn_task = fn.spawn(media, sentry_trace)
+                        queue_publish_span.set_data("cache.key", media.urn)
+                        queue_publish_span.set_data("messaging.message.id", fn_task.object_id)
+                        queue_publish_span.set_data("messaging.destination.name",
+                                                    "video-downloader.cache_submit")
+                        queue_publish_span.set_data("messaging.message.body.size", 0)
+                        # video_cache = MediaCache(status=MediaCacheStatus.downloading,
+                        #                          downloader_id=fn_task.object_id)
+                        media.status = MediaCacheStatus.downloading
+                        media.downloader_id = fn_task.object_id
+                        # video_cache_status_json = video_cache.model_dump_json()
+                        modal_kv_cache.set_cache(media)
+                else:
+                    media = cached_media
+                # video_cache = MediaCache.model_validate_json(video_cache_status_json)
+                match media.status:
+                    case MediaCacheStatus.ready:
+                        cache_hit = True
+                    case MediaCacheStatus.downloading:  # 下载任务已经在进行
+                        cache_hit = True
+                    case _:
                         # start new download task
                         with cache_span.start_child(name="视频缓存任务入队",
                                                     op="queue.publish") as queue_publish_span:
                             fn = modal.Function.from_name(config.app_name, 'cache_submit')
                             fn_task = fn.spawn(media, sentry_trace)
-                            queue_publish_span.set_data("cache.key", media.cache_key)
+                            queue_publish_span.set_data("cache.key", media.urn)
                             queue_publish_span.set_data("messaging.message.id", fn_task.object_id)
                             queue_publish_span.set_data("messaging.destination.name",
                                                         "video-downloader.cache_submit")
                             queue_publish_span.set_data("messaging.message.body.size", 0)
-                            video_cache = MediaCache(status=MediaCacheStatus.downloading,
-                                                     downloader_id=fn_task.object_id)
-                            video_cache_status_json = video_cache.model_dump_json()
-                            media_cache_kv.put(media.cache_key, video_cache_status_json)
+                            media.status = MediaCacheStatus.downloading
+                            media.downloader_id = fn_task.object_id
+                            # video_cache = MediaCache(status=MediaCacheStatus.downloading,
+                            #                          downloader_id=fn_task.object_id)
+                            # video_cache_status_json = video_cache.model_dump_json()
+                            modal_kv_cache.set_cache(media)
+                            cache_hit = False
 
-                    video_cache = MediaCache.model_validate_json(video_cache_status_json)
-                    match video_cache.status:
-                        case MediaCacheStatus.ready:
-                            cache_hit = True
-                        case MediaCacheStatus.downloading:  # 下载任务已经在进行
-                            cache_hit = True
-                        case _:
-                            # start new download task
-                            with cache_span.start_child(name="视频缓存任务入队",
-                                                        op="queue.publish") as queue_publish_span:
-                                fn = modal.Function.from_name(config.app_name, 'cache_submit')
-                                fn_task = fn.spawn(media, sentry_trace)
-                                queue_publish_span.set_data("cache.key", media.cache_key)
-                                queue_publish_span.set_data("messaging.message.id", fn_task.object_id)
-                                queue_publish_span.set_data("messaging.destination.name",
-                                                            "video-downloader.cache_submit")
-                                queue_publish_span.set_data("messaging.message.body.size", 0)
-                                video_cache = MediaCache(status=MediaCacheStatus.downloading,
-                                                         downloader_id=fn_task.object_id)
-                                video_cache_status_json = video_cache.model_dump_json()
-                                media_cache_kv.put(media.cache_key, video_cache_status_json)
+                # caches[media.urn] = video_cache
+                # logger.info(f"Media cache hit ? {cache_hit}")
+                cache_span.set_data("cache.hit", cache_hit)
+                return media
 
-                    caches[media.cache_key] = video_cache
-                    logger.info(f"Media cache hit ? {cache_hit}")
-                    cache_span.set_data("cache.hit", cache_hit)
-            batch_update_cloudflare_kv(caches)
-            return CacheResult(caches=caches)
-            # return JSONResponse(content={"caches": jsonable_encoder(caches)})
+            async with asyncio.TaskGroup() as group:
+                tasks = [group.create_task(cache_handler(media)) for media in medias.inputs]
+
+            cache_task_result = [task.result() for task in tasks]
+
+            batch_update_cloudflare_kv(cache_task_result)
+            return CacheResult(caches={media.urn: media for media in cache_task_result})
 
         @web_app.post("/cache/download",
                       tags=["缓存"],
@@ -303,7 +324,7 @@ with fastapi_image.imports():
         async def purge_kv_all():
             parent = sentry_sdk.get_current_span()
             span = parent.start_child(name="清除缓存KV", op="purge.flush")
-            media_cache_kv.clear()
+            modal_kv_cache.clear()
             span.set_data("cache.success", True)
             span.finish()
             return JSONResponse(content={"success": True})
@@ -316,8 +337,8 @@ with fastapi_image.imports():
         async def purge_kv(medias: MediaSources):
             try:
                 for media in medias.inputs:
-                    media_cache_kv.pop(media.cache_key)
-                keys = [media.cache_key for media in medias.inputs]
+                    modal_kv_cache.pop(media.urn)
+                keys = [media.urn for media in medias.inputs]
                 batch_remove_cloudflare_kv(keys)
                 return JSONResponse(content={"success": True, "keys": keys})
             except Exception as e:
@@ -329,20 +350,22 @@ with fastapi_image.imports():
                       description="清除指定的所有缓存(包括KV记录和S3存储文件)",
                       dependencies=[Depends(verify_token)])
         async def purge_media(medias: MediaSources):
-            caches: Dict[str, MediaCache] = {}
-            for media in medias.inputs:
-                try:
-                    cache_data_json = media_cache_kv.pop(media.cache_key)
-                    cache_data = MediaCache.model_validate_json(cache_data_json)
-                    fn = modal.Function.from_name(config.app_name, "cache_delete",
-                                                  environment_name=config.environment)
-                    deleted_cache = await fn.remote.aio(cache_data)
-                    caches[media.cache_key] = deleted_cache
-                except KeyError:
-                    logger.warning("cache key not found")
-                    caches[media.cache_key] = MediaCache(status=MediaCacheStatus.missing)
-                    continue
-            keys = [key for key in caches.keys()]
+            fn_id = current_function_call_id()
+            fn = modal.Function.from_name(config.app_name, "cache_delete", environment_name=config.environment)
+
+            @SentryUtils.sentry_tracker(name="清除媒体源缓存", op="cache.purge", fn_id=fn_id,
+                                        sentry_trace_id=None, sentry_baggage=None)
+            async def purge_handle(media: MediaSource):
+                cache_media = modal_kv_cache.pop(media.urn)
+                if cache_media:
+                    deleted_cache: MediaSource = await fn.remote.aio(cache_media)
+                    return deleted_cache.urn
+                return None
+
+            async with asyncio.TaskGroup() as group:
+                tasks = [group.create_task(purge_handle(media)) for media in medias.inputs]
+
+            keys = [task.result() for task in tasks]
             batch_remove_cloudflare_kv(keys)
             return JSONResponse(content={"success": True, "keys": keys})
 

src/deploy.py

@@ -2,4 +2,6 @@ import modal.cli.run
 from cluster.config import config
 
 if __name__ == '__main__':
-    modal.cli.run.deploy(app_ref='.\\src\\cluster\\app.py', env=config.environment, use_module_mode=False)
+    # modal.cli.run.deploy(app_ref='src.cluster.app', env=config.environment, use_module_mode=True)
+    # todo : 通过python命令部署不成功，换https://modal.com/docs/guide/continuous-deployment推荐的方式通过命令行部署
+    modal.cli.run.deploy(app_ref='src\\cluster\\app.py', env=config.environment, use_module_mode=False)