comi-logic/modalDeploy
8
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

更新2个ffmpeg worker方法和web调用接口

Browse Source
This commit is contained in:
shuohigh@gmail.com
2025-05-07 15:08:30 +08:00
parent 8e5cb27f32
commit a05e153717
10 changed files with 1657 additions and 260 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
pyproject.toml
View File

@@ -5,11 +5,15 @@ description = "管理Modal worker的工程"
readme = "README.md"
requires-python = ">=3.11"
dependencies = [
"backoff>=2.2.1",
"boto3>=1.37.37",
"cos-python-sdk-v5>=1.9.36",
"crcmod>=1.7",
"fastapi[standard]>=0.115.12",
"httpx>=0.28.1",
"loguru>=0.7.3",
"modal>=0.74.57",
"psutil>=7.0.0",
"pydantic>=2.11.3",
"pydantic-settings>=2.9.1",
"python-ffmpeg>=2.0.12",

7
src/cluster/app.py
View File

@@ -2,12 +2,13 @@ import modal
from config import config
from video_downloader.worker import worker_app
from web.worker import fastapi_app
from ffmpeg_worker.worker import ffmpeg_worker_app
from ffmpeg_worker.worker import app as ffmpeg_app
app = modal.App('video-downloader',
app = modal.App(config.app_name,
include_source=False,
secrets=[modal.Secret.from_name("cf-kv-secret",
environment_name=config.environment)])
app.include(fastapi_app)
app.include(worker_app)
app.include(ffmpeg_worker_app)
app.include(ffmpeg_app)

5
src/cluster/config.py
View File

@@ -4,10 +4,12 @@ from pydantic_settings import BaseSettings, SettingsConfigDict
class WorkerConfig(BaseSettings):
app_name: str = Field(default='video-downloader', description="Modal App集群名称")
video_downloader_concurrency: Optional[int] = Field(default=10, description="处理缓存任务的并行数")
ffmpeg_worker_concurrency: Optional[int] = Field(default=10, description="处理视频合成任务的并行数")
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存储桶")
cdn_endpoint: Optional[str] = Field(default="https://d2nj71io21vkj2.cloudfront.net",
description="集群挂载S3存储桶的对应AWS Cloudfront CDN")
@@ -20,6 +22,7 @@ config = WorkerConfig(
video_downloader_concurrency=10,
ffmpeg_worker_concurrency=10,
modal_kv_name="media-cache",
s3_region='ap-northeast-2',
s3_bucket_name="modal-media-cache",
environment="dev"
)
)

8
src/cluster/ffmpeg_worker/Utils/PathUtils.py
View File

@@ -10,3 +10,11 @@ class FileUtils:
filenames[0] = f"{filenames[0]}_{extend}"
extend_filename = '.'.join(filenames)
return os.path.join(media_dir, extend_filename)
@staticmethod
def file_path_replace_root_prefix(media_path: str, prefix: str, depth: int = 1) -> str:
media_dirs = media_path.split('/')
if depth >= len(media_dirs):
raise IndexError("Depth is out of range")
media_dir_prefix = prefix + '/'.join(media_dirs[depth:])
return media_dir_prefix

335
src/cluster/ffmpeg_worker/worker.py
View File

@@ -1,25 +1,33 @@
import os.path
import modal
from src.cluster.config import config
ffmpeg_worker_image = (
modal.Image.debian_slim(python_version="3.11")
.apt_install('ffmpeg')
.pip_install('sentry-sdk', 'loguru', 'pydantic', 'pydantic_settings', 'httpx', 'python-ffmpeg')
.pip_install('sentry-sdk', 'loguru', 'pydantic', 'pydantic_settings', 'httpx', 'python-ffmpeg', 'psutil', 'backoff')
.add_local_python_source("src.cluster.config", copy=True)
.add_local_python_source("src.cluster.ffmpeg_worker.model", copy=True)
.add_local_python_source("src.cluster.video_downloader.model", copy=True)
)
ffmpeg_worker_app = modal.App("ffmpeg_worker_app", image=ffmpeg_worker_image, )
ffmpeg_worker_app.set_description("FFMPEG worker app")
app = modal.App(image=ffmpeg_worker_image, )
with ffmpeg_worker_image.imports():
from typing import List
from src.cluster.config import config
from typing import List, Optional, Tuple, Dict, Any
from loguru import logger
import sentry_sdk
from src.cluster.video_downloader.model import MediaSource, MediaCache
from src.cluster.ffmpeg_worker.model import FFMpegSliceSegment, TimeDelta
from ffmpeg.asyncio import FFmpeg
from src.cluster.ffmpeg_worker.Utils.PathUtils import FileUtils
from src.cluster.video_downloader.model import (MediaSources,
MediaSource,
MediaCache,
MediaProtocol,
MediaCacheStatus)
from src.cluster.web.model import SentryTransactionInfo
from src.cluster.ffmpeg_worker.model import FFMpegSliceSegment
import shutil, psutil, os, backoff, json, sentry_sdk
from psutil import svmem
from psutil._common import scpufreq
from modal import current_function_call_id
sentry_sdk.init(
dsn="https://75cca970bfcc3d45d24361e7c0f1833c@sentry.bowongai.com/4",
@@ -33,62 +41,54 @@ 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"
output_path_prefix = "/mnt/outputs"
@sentry_sdk.trace
def get_cache_filepath(media: MediaSource) -> MediaCache:
cache_data_json = modal_kv.get(media.cache_key)
if cache_data_json:
"""
获取缓存路径
:param media: 媒体文件
:return:
"""
# 本地挂载缓存
if media.protocol == MediaProtocol.s3 and media.endpoint == config.s3_region and media.bucket == config.bucket:
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
raise FileNotFoundError(f"{media.cache_key} cache not found")
return cache_data
@ffmpeg_worker_app.function(
cpu=6, timeout=1800, memory=(2048, 6144),
# cloud="aws",
# region='ap-northeast',
max_containers=config.ffmpeg_worker_concurrency,
volumes={
"/mntS3": modal.CloudBucketMount(
bucket_name=config.s3_bucket_name,
secret=modal.Secret.from_name("aws-s3-secret", environment_name=config.environment),
),
}, )
@modal.concurrent(max_inputs=1)
def ffmpeg_concat_medias():
return {}
@sentry_sdk.trace
def capture_hardware_info() -> Tuple[scpufreq, int, svmem]:
cpu_freq = psutil.cpu_freq()
cpu_count = psutil.cpu_count()
mem = psutil.virtual_memory()
return (cpu_freq, cpu_count, mem)
@ffmpeg_worker_app.function(
cpu=12, timeout=900, memory=(2048, 4096),
cloud="aws",
max_containers=config.ffmpeg_worker_concurrency,
volumes={
"/mntS3": modal.CloudBucketMount(
bucket_name=config.s3_bucket_name,
secret=modal.Secret.from_name("aws-s3-secret", environment_name=config.environment),
),
},
)
@modal.concurrent(max_inputs=1)
async def ffmpeg_slice_media(media: MediaSource, markers: List[FFMpegSliceSegment], ):
from src.cluster.ffmpeg_worker.Utils.PathUtils import FileUtils
from ffmpeg.asyncio import FFmpeg
cache = get_cache_filepath(media)
@sentry_sdk.trace
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)
outputs: List[str] = []
ffmpeg_cmd = FFmpeg().option('y').option('hide_banner').input(cache.cache_filepath)
filter_complex: List[str] = []
for index, marker in enumerate(markers):
for index, marker in enumerate(media_markers):
segment_out_index = f"cut{index}"
filter_complex.append(
f"[0]trim=start={marker.start.total_seconds()}:end={marker.end.total_seconds()},setpts=PTS-STARTPTS[{segment_out_index}]")
ffmpeg_cmd.option('filter_complex', ';'.join(filter_complex))
for i, marker in enumerate(markers):
output_filepath = FileUtils.file_path_extend(cache.cache_filepath, str(i))
# f"{marker.start.toFormatStr()}_{marker.end.toFormatStr()}")
for i, marker in enumerate(media_markers):
filename = FileUtils.file_path_extend(os.path.basename(cache.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)
os.makedirs(workdir, exist_ok=True)
outputs.append(output_filepath)
@@ -97,7 +97,7 @@ with ffmpeg_worker_image.imports():
reset_timestamps="1",
sc_threshold="0",
g="1",
force_key_frames="expr:gte(t, n_forced * 1)",
force_key_frames="expr:gte(t,n_forced*1)",
vcodec="libx264",
crf=16,
r=30, )
@@ -124,9 +124,236 @@ with ffmpeg_worker_image.imports():
logger.info(f"FFMpeg task completed.")
@ffmpeg_cmd.on("stderr")
def on_stderr(line):
logger.warning(line)
def on_stderr(line: str):
if line.startswith('Error'):
logger.error(line)
raise RuntimeError(line)
else:
logger.warning(line)
await ffmpeg_cmd.execute()
return outputs
s3_outputs = local_copy_to_s3(outputs)
return s3_outputs
@backoff.on_exception(backoff.constant, exception=Exception, max_tries=5, max_time=60, raise_on_giveup=True)
def local_copy_to_s3(local_outputs: List[str]) -> List[str]:
s3_outputs = []
for output in local_outputs:
out_s3 = output.replace(output_path_prefix, s3_mount)
out_s3_dir = os.path.dirname(out_s3)
logger.info(out_s3_dir)
os.makedirs(out_s3_dir, exist_ok=True)
shutil.copy(output, out_s3)
s3_outputs.append(
out_s3.replace(s3_mount + '/', f"s3://{config.s3_region}/{config.s3_bucket_name}/"))
return s3_outputs
@app.function(
name="concat_medias",
cpu=12, timeout=1800, memory=(2048, 6144),
cloud="aws",
# region='ap-northeast',
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.environment),
),
}, )
@modal.concurrent(max_inputs=1)
async def ffmpeg_concat_medias(medias: MediaSources,
sentry_trace: Optional[SentryTransactionInfo] = None) -> Tuple[
str, Optional[SentryTransactionInfo]]:
@sentry_sdk.trace
async def print_media_streams_info(media_paths: List[str]) -> List[Dict[str, Any]]:
"""
打印多个媒体文件的视频和音频流详细信息
:param media_paths: 媒体文件路径列表
"""
res = []
for path in media_paths:
try:
# 使用ffprobe探测媒体文件信息
probe_info_byte = await FFmpeg(executable="ffprobe").input(path, print_format="json",
show_streams=None).execute()
probe_info = json.loads(probe_info_byte)
res.append(probe_info)
except Exception as e:
logger.exception(e)
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')
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)
video_info = print_media_streams_info(media_paths=input_videos)
filter_complex = []
for i in range(len(input_videos)):
info = video_info[i]
streams = info.get("streams", [])
video_stream = streams[0]
# audio_stream = streams[1]
video_duration = video_stream.get("duration")
# audio_duration = audio_stream.get("duration")
'''
filter_complex.extend(
[
# 先缩放到统一分辨率，然后设置帧率和格式
f"[{i}:v]scale={target_width}:{target_height}:force_original_aspect_ratio=decrease,"
f"pad={target_width}:{target_height}:(ow-iw)/2:(oh-ih)/2,"
f"fps=30,format=yuv420p[v{i}]",
f"[{i}:a]aformat=sample_fmts=fltp:sample_rates=44100:channel_layouts=stereo[a{i}]",
]
)
'''
filter_complex.extend(
[
f"[{i}:v]trim=0:{video_duration},setpts=PTS-STARTPTS[v{i}]",
f"[{i}:a]atrim=0:{video_duration},asetpts=PTS-STARTPTS[a{i}]",
]
)
# 3. 准备处理后的视频流和音频流的连接字符串
video_streams = "".join(f"[v{i}]" for i in range(len(input_videos)))
audio_streams = "".join(f"[a{i}]" for i in range(len(input_videos)))
# 4. 使用concat过滤器合并视频和音频
filter_complex.extend(
[
f"{video_streams}concat=n={len(input_videos)}:v=1:a=0[vconcated]",
f"{audio_streams}concat=n={len(input_videos)}:v=0:a=1[aconcated]",
]
)
@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)
ffmpeg_cmd.output(
output_filepath,
{
"filter_complex": ";".join(filter_complex),
"map": ["[vconcated]", "[aconcated]"],
"vcodec": "libx264",
"crf": 16,
"r": 30,
"acodec": "aac",
"ar": 44100,
"ac": 2,
"ab": "192k",
},
)
await ffmpeg_cmd.execute()
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:
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"
await ffmpeg_process(medias, output_filepath=output_path)
s3_outputs = local_copy_to_s3([output_path])
return s3_outputs[0], sentry_trace
@app.function(
name="slice_media",
cpu=12, timeout=900, memory=(2048, 4096),
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.environment),
),
},
)
@modal.concurrent(max_inputs=1)
async def ffmpeg_slice_media(media: MediaSource,
markers: List[FFMpegSliceSegment],
sentry_trace: Optional[SentryTransactionInfo] = None) -> Tuple[
List[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")
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:
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")
outputs = await ffmpeg_slice_process(media_source=media, media_markers=markers, fn_id=fn_id)
return outputs, sentry_trace

11
src/cluster/video_downloader/model.py
View File

@@ -59,6 +59,13 @@ class MediaSource(BaseModel):
]
}
def get_cdn_url(self) -> str:
if self.protocol == MediaProtocol.vod:
return f"{self.protocol.value}/{self.endpoint}/{self.bucket}/{self.url}.mp4"
elif self.protocol == MediaProtocol.s3:
return f"{self.url}"
return f"{self.protocol.value}/{self.endpoint}/{self.bucket}/{self.url}"
class MediaSources(BaseModel):
inputs: List[MediaSource] = Field(examples=[
@@ -101,3 +108,7 @@ class MediaCache(BaseModel):
class CacheResult(BaseModel):
caches: Dict[str, MediaCache] = Field(description="Cache ID")
class DownloadResult(BaseModel):
urls: List[str] = Field(description="下载链接")

17
src/cluster/video_downloader/worker.py
View File

@@ -1,26 +1,23 @@
import modal
from src.cluster.config import config
downloader_image = (
modal.Image
.debian_slim(python_version="3.11")
.pip_install('httpx', 'cos-python-sdk-v5', 'loguru', 'pydantic', 'pydantic_settings', 'sentry-sdk[loguru]',
'tqdm', 'crcmod', 'tencentcloud-sdk-python-common', 'tencentcloud-sdk-python-vod')
.add_local_python_source("src.cluster.config", copy=True)
.add_local_python_source("src.cluster.video_downloader.model", copy=True)
.add_local_python_source("src.cluster.web.model", copy=True)
.add_local_python_source("src.cluster.config", copy=True)
)
worker_app = modal.App("video-downloader-worker", image=downloader_image, secrets=[
modal.Secret.from_name("cf-kv-secret", environment_name=config.environment),
])
with downloader_image.imports():
import os, httpx, crcmod
from tqdm import tqdm
from typing import Tuple, Dict
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.model import SentryTransactionInfo
from datetime import datetime, UTC, timedelta
@@ -28,6 +25,11 @@ with downloader_image.imports():
from tencentcloud.vod.v20180717.vod_client import VodClient
from tencentcloud.vod.v20180717 import models as vod_request_models
import sentry_sdk
from sentry_sdk.integrations.loguru import LoguruIntegration
worker_app = modal.App(image=downloader_image, secrets=[
modal.Secret.from_name("cf-kv-secret", environment_name=config.environment),
])
sentry_sdk.init(dsn="https://85632fdcd62f699c2f88af6ca489e9ec@sentry.bowongai.com/3",
send_default_pii=True,
@@ -35,6 +37,7 @@ with downloader_image.imports():
profiles_sample_rate=1.0,
add_full_stack=True,
shutdown_timeout=2,
integrations=[LoguruIntegration()],
environment=config.environment,
)
@@ -217,8 +220,8 @@ with downloader_image.imports():
create_if_missing=True)
fn_id = current_function_call_id()
with sentry_sdk.continue_trace(environ_or_headers={"sentry-trace": sentry_trace.trace_id,
"baggage": sentry_trace.baggage, }) as transaction:
with sentry_sdk.continue_trace(environ_or_headers={"sentry-trace": sentry_trace.x_trace_id,
"baggage": sentry_trace.x_baggage, }) as transaction:
transaction.set_context("runtime_environment", {
"MODAL_CLOUD_PROVIDER": os.environ.get('MODAL_CLOUD_PROVIDER', 'unknown'),
"MODAL_ENVIRONMENT": os.environ.get('MODAL_ENVIRONMENT', 'unknown'),

32
src/cluster/web/model.py
View File

@@ -1,15 +1,18 @@
from enum import Enum
from typing import List, Union, Any, Optional
from pydantic import BaseModel, Field, field_validator
from pydantic import BaseModel, Field, field_validator, ConfigDict
from src.cluster.ffmpeg_worker.model import FFMpegSliceSegment
from src.cluster.video_downloader.model import MediaSource
from src.cluster.video_downloader.model import MediaSource, MediaSources
class SentryTransactionHeader(BaseModel):
x_trace_id: Optional[str] = Field(description="Sentry Transaction ID", default=None)
x_baggage: Optional[str] = Field(description="Sentry Transaction baggage", default=None)
class SentryTransactionInfo(BaseModel):
trace_id: str = Field(description="Sentry Transaction ID")
baggage: str = Field(description="Sentry Transaction baggage")
x_trace_id: str = Field(description="Sentry Transaction ID")
x_baggage: str = Field(description="Sentry Transaction baggage")
class FFMPEGSliceRequest(BaseModel):
@@ -27,12 +30,12 @@ class FFMPEGSliceRequest(BaseModel):
raise TypeError(v)
class FFMPEGSliceResponse(BaseModel):
success: bool = Field(description="任务创建成功与否")
class FFMPEGSliceTaskStatusRequest(BaseModel):
taskId: str = Field(description="任务Id")
class FFMPEGSliceTaskStatusRequest(BaseModel):
class ModalTaskResponse(BaseModel):
success: bool = Field(description="任务接受成功")
taskId: str = Field(description="任务Id")
@@ -43,7 +46,16 @@ class TaskStatus(str, Enum):
expired = "expired"
class FFMPEGSliceTaskStatusResponse(BaseModel):
class BaseFFMPEGTaskStatusResponse(BaseModel):
taskId: str = Field(description="任务Id")
status: TaskStatus = Field(description="任务运行状态")
error: Optional[str] = Field(description="任务错误原因", default=None)
model_config = ConfigDict(extra='ignore')
class FFMPEGSliceTaskStatusResponse(BaseFFMPEGTaskStatusResponse):
result: Optional[List[str]] = Field(default=None, description="任务运行结果")
class FFMPEGConcatTaskStatusResponse(BaseFFMPEGTaskStatusResponse):
result: Optional[str] = Field(default=None, description="任务运行结果")

229
src/cluster/web/worker.py
View File

@@ -1,5 +1,4 @@
import modal
from src.cluster.config import config
fastapi_image = (
modal.Image
@@ -11,9 +10,31 @@ fastapi_image = (
.add_local_python_source("src.cluster.config", copy=True)
)
fastapi_app = modal.App("video-downloader-web", image=fastapi_image)
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
from sentry_sdk.integrations.loguru import 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 SentryTransactionInfo, SentryTransactionHeader, FFMPEGSliceRequest, \
FFMPEGSliceTaskStatusResponse, FFMPEGConcatTaskStatusResponse, TaskStatus, ModalTaskResponse
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),
@@ -21,26 +42,6 @@ with fastapi_image.imports():
@modal.concurrent(max_inputs=100)
@modal.asgi_app()
def fastapi_webapp():
import os
import httpx
from typing import Dict, List
from modal import current_function_call_id
from loguru import logger
from fastapi import FastAPI, Request, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi.responses import JSONResponse
from fastapi.exceptions import HTTPException
from fastapi.encoders import jsonable_encoder
from starlette import status
from scalar_fastapi import get_scalar_api_reference
import sentry_sdk
from sentry_sdk.integrations.loguru import LoguruIntegration
from sentry_sdk.integrations.loguru import LoggingLevels
from sentry_sdk.integrations.fastapi import FastApiIntegration
from src.cluster.video_downloader.model import MediaSources, MediaCacheStatus, MediaCache, CacheResult
from src.cluster.web.model import SentryTransactionInfo, FFMPEGSliceRequest, FFMPEGSliceResponse, \
FFMPEGSliceTaskStatusRequest, FFMPEGSliceTaskStatusResponse, TaskStatus
bearer_scheme = HTTPBearer()
web_app = FastAPI(title="Modal worker API",
@@ -82,6 +83,21 @@ with fastapi_image.imports():
cf_kv_api_token = os.environ.get("CF_KV_API_TOKEN")
cf_kv_namespace_id = os.environ.get("CF_KV_NAMESPACE_ID")
sentry_header_schema = {
"x-trace-id": {
"description": "Sentry Transaction ID",
"schema": {
"type": "string",
}
},
"x-baggage": {
"description": "Sentry Transaction baggage",
"schema": {
"type": "string",
}
}
}
@sentry_sdk.trace
def batch_update_cloudflare_kv(caches: Dict[str, MediaCache]):
with httpx.Client() as client:
@@ -129,11 +145,53 @@ with fastapi_image.imports():
logger.error(f"An unexpected error occurred: {str(e)}")
raise e
@sentry_sdk.trace
async def get_modal_task_status(task_id: str) -> Tuple[
TaskStatus, Optional[str], Optional[Any], Optional[SentryTransactionInfo]]:
"""
:param task_id:
:return: (TaskStaus, errorReason, results, sentryTransactionInfo)
"""
try:
fn_task = modal.FunctionCall.from_id(task_id)
call_graph = fn_task.get_call_graph()
task = call_graph[0]
match task.status:
case InputStatus.PENDING:
return TaskStatus.running, None, None, None
case InputStatus.SUCCESS:
try:
results, sentry_trace = fn_task.get(timeout=2)
return TaskStatus.success, None, results, sentry_trace
except modal.exception.OutputExpiredError:
return TaskStatus.expired, None, None, None
except TimeoutError:
return TaskStatus.running, None, None, None
except Exception as e:
logger.error(e)
return TaskStatus.failed, None, None, None
case _:
error = 'FAILURE'
match task.status:
case InputStatus.INIT_FAILURE:
error = 'INIT_FAILURE'
case InputStatus.TERMINATED:
error = 'TERMINATED'
case InputStatus.TIMEOUT:
error = 'TIMEOUT'
return TaskStatus.failed, error, None, None
except Exception as e:
logger.exception(e)
raise RuntimeError("Modal task status check failed")
@web_app.get("/scalar", include_in_schema=False)
async def scalar():
return get_scalar_api_reference(openapi_url=web_app.openapi_schema, title="Modal worker web endpoint")
@web_app.post("/cache",
tags=["缓存"],
summary="缓存视频文件",
description="异步缓存视频文件到S3存储桶和Modal Dict(KV)",
dependencies=[Depends(verify_token)])
@@ -142,8 +200,8 @@ with fastapi_image.imports():
fn_id = current_function_call_id()
caches: Dict[str, MediaCache] = {}
parent = sentry_sdk.get_current_span()
sentry_trace = SentryTransactionInfo(trace_id=sentry_sdk.get_traceparent(),
baggage=sentry_sdk.get_baggage())
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)
@@ -156,7 +214,7 @@ with fastapi_image.imports():
# start new download task
with cache_span.start_child(name="视频缓存任务入队",
op="queue.publish") as queue_publish_span:
fn = modal.Function.from_name('video-downloader', 'cache_submit')
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)
@@ -178,7 +236,7 @@ with fastapi_image.imports():
# start new download task
with cache_span.start_child(name="视频缓存任务入队",
op="queue.publish") as queue_publish_span:
fn = modal.Function.from_name('video-downloader', 'cache_submit')
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)
@@ -197,7 +255,31 @@ with fastapi_image.imports():
return CacheResult(caches=caches)
# return JSONResponse(content={"caches": jsonable_encoder(caches)})
@web_app.post("/cache/download",
tags=["缓存"],
summary="批量获取下载地址",
description="获取已缓存的视频下载地址",
dependencies=[Depends(verify_token)])
@sentry_sdk.trace
async def download_caches(medias: MediaSources) -> DownloadResult:
cdn_endpoint = config.cdn_endpoint
urls = []
for media in medias.inputs:
urls.append(f"{cdn_endpoint}/{media.get_cdn_url()}")
return DownloadResult(urls=urls)
@web_app.get("/cache/download",
tags=["缓存"],
summary="下载已缓存的视频",
description="通过CDN下载已缓存的视频文件")
@sentry_sdk.trace
async def download_cache(media: str) -> RedirectResponse:
cdn_endpoint = config.cdn_endpoint
media = MediaSource.from_str(media)
return RedirectResponse(url=f"{cdn_endpoint}/{media.get_cdn_url()}", status_code=status.HTTP_302_FOUND)
@web_app.delete("/cache/kv",
tags=["缓存"],
summary="清除KV记录",
description="清除当前环境下KV缓存过的所有数据(S3存储桶内的文件会保留)",
dependencies=[Depends(verify_token)])
@@ -210,6 +292,7 @@ with fastapi_image.imports():
return JSONResponse(content={"success": True})
@web_app.post("/cache/kv",
tags=["缓存"],
summary="删除对应的KV记录",
description="删除请求中对应的视频缓存记录",
dependencies=[Depends(verify_token)])
@@ -224,6 +307,7 @@ with fastapi_image.imports():
return JSONResponse(content={"success": False, "error": str(e)})
@web_app.post("/cache/media",
tags=["缓存"],
summary="清除指定的所有缓存",
description="清除指定的所有缓存(包括KV记录和S3存储文件)",
dependencies=[Depends(verify_token)])
@@ -233,7 +317,7 @@ with fastapi_image.imports():
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("video-downloader", "cache_delete",
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
@@ -245,37 +329,70 @@ with fastapi_image.imports():
batch_remove_cloudflare_kv(keys)
return JSONResponse(content={"success": True, "keys": keys})
@web_app.post("/merge",
summary="发起合成任务",
description="开发中",
@web_app.post("/ffmpeg/slice",
tags=["发起任务"],
summary="发起切割任务",
description="依据打点信息切出多个片段",
dependencies=[Depends(verify_token)])
async def merge_media(request: Request):
body_json = await request.json()
return JSONResponse(content={"success": False, "message": "Not Implemented"})
async def slice_media(request: FFMPEGSliceRequest,
headers: Annotated[SentryTransactionHeader, Header()]) -> ModalTaskResponse:
@web_app.post("/ffmpeg/slice", summary="发起切割任务", description="依据打点信息切出多个片段",
dependencies=[Depends(verify_token)])
async def slice_media(request: FFMPEGSliceRequest):
fn = modal.Function.from_name("video-downloader", "ffmpeg_slice_media", environment_name=config.environment)
fn_call = fn.spawn(request.media, request.markers)
return FFMPEGSliceResponse(success=True, taskId=fn_call.object_id)
fn = modal.Function.from_name(config.app_name, "ffmpeg_slice_media", environment_name=config.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(request.media, request.markers, sentry_trace)
return ModalTaskResponse(success=True, taskId=fn_call.object_id)
@web_app.get("/ffmpeg/slice/{task_id}", summary="查询切割任务状态/结果", description="根据任务Id查询运行状态",
@web_app.get("/ffmpeg/slice/{task_id}",
tags=["查询任务"],
summary="查询切割任务状态/结果",
description="根据任务Id查询运行状态",
responses={
status.HTTP_200_OK: {
"description": "",
"headers": sentry_header_schema
},
},
dependencies=[Depends(verify_token)])
async def slice_media(task_id: str):
try:
fn_task = modal.FunctionCall.from_id(task_id)
try:
results: List[str] = fn_task.get(timeout=2)
return FFMPEGSliceTaskStatusResponse(taskId=task_id, status=TaskStatus.success, result=results)
except modal.exception.OutputExpiredError:
return FFMPEGSliceTaskStatusResponse(taskId=task_id, status=TaskStatus.expired)
except TimeoutError:
return FFMPEGSliceTaskStatusResponse(taskId=task_id, status=TaskStatus.running)
except Exception as e:
logger.error(e)
return FFMPEGSliceTaskStatusResponse(taskId=task_id, status=TaskStatus.failed)
except Exception as e:
return JSONResponse(content={"success": False, "message": "任务Id不存在"}, status_code=400)
async def slice_media(task_id: str, response: Response) -> FFMPEGSliceTaskStatusResponse:
task_status, reason, results, transaction = await 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 FFMPEGSliceTaskStatusResponse(taskId=task_id, status=TaskStatus.failed, error=reason)
@web_app.post("/ffmpeg/concat",
tags=["发起任务"],
summary="发起合并任务",
description="依据AI分析的结果发起合并任务",
dependencies=[Depends(verify_token)])
async def concat_media(medias: MediaSources,
headers: Annotated[SentryTransactionHeader, Header()]) -> ModalTaskResponse:
fn = modal.Function.from_name(config.app_name, "ffmpeg_concat_medias",
environment_name=config.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(medias, sentry_trace)
return ModalTaskResponse(success=True, taskId=fn_call.object_id)
@web_app.get("/ffmpeg/concat/{task_id}",
tags=["查询任务"],
summary="获取合并任务结果",
description="获取合并任务的处理状态和结果",
responses={
status.HTTP_200_OK: {
"description": "",
"headers": sentry_header_schema
},
},
dependencies=[Depends(verify_token)])
async def concat_media_status(task_id: str, response: Response) -> FFMPEGConcatTaskStatusResponse:
task_status, reason, results, transaction = await 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 FFMPEGConcatTaskStatusResponse(taskId=task_id, status=task_status, error=reason, result=results)
return web_app

1269
uv.lock generated
View File

File diff suppressed because it is too large Load Diff