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

staging

Browse Source
This commit is contained in:
shuohigh@gmail.com
2025-06-13 19:32:18 +08:00
parent c4d596a31c
commit 86daaf46b0
9 changed files with 68 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.runtime.env
View File

@@ -1,4 +1,4 @@
MODAL_ENVIRONMENT=test
MODAL_ENVIRONMENT=prod
modal_app_name=bowong-ai-video
S3_mount_dir=/mntS3
S3_bucket_name=modal-media-cache

2
src/BowongModalFunctions/models/ffmpeg_worker_model.py
View File

@@ -1,5 +1,5 @@
from typing import Union, Any
from pydantic import BaseModel, Field, computed_field, field_validator, model_validator
from pydantic import BaseModel, Field, computed_field, field_validator, model_validator, ConfigDict
from pydantic.json_schema import JsonSchemaValue
from ..utils.TimeUtils import TimeDelta

40
src/BowongModalFunctions/models/web_model.py
View File

@@ -4,9 +4,12 @@ from typing import List, Union, Optional, Dict, Any
import pydantic
from pydantic import BaseModel, Field, field_validator, ConfigDict, HttpUrl, computed_field
from pydantic.json_schema import JsonSchemaValue
from .ffmpeg_worker_model import FFMpegSliceSegment
from .media_model import MediaSource, MediaSources, MediaProtocol
from ..config import WorkerConfig
from ..utils.TimeUtils import TimeDelta
from ..utils.VideoUtils import VideoMetadata, FFMPEGSliceOptions
config = WorkerConfig()
@@ -374,6 +377,8 @@ class FFMPEGVideoLoopFillAudioResponse(BaseFFMPEGTaskStatusResponse):
class FFMPEGExtractFrameRequest(BaseFFMPEGTaskRequest):
video: MediaSource = Field(description="提取帧画面的来源")
seek_time: Optional[Union[str, int, float]] = Field(default=None, description="先跳转到视频对应时间再取首帧",
examples=["00:00:01.000", "5.4", "2"])
frame_index: int = Field(description="提取的第几帧, 从1开始，默认为1", default=1)
@field_validator('video', mode='before')
@@ -386,6 +391,24 @@ class FFMPEGExtractFrameRequest(BaseFFMPEGTaskRequest):
else:
raise TypeError(v)
@computed_field(description="格式处理后的Seek Time")
@property
def prased_seek_time(self) -> 'TimeDelta':
if self.seek_time is None:
return None
elif isinstance(self.seek_time, str):
return TimeDelta.from_format_string(self.seek_time)
elif isinstance(self.seek_time, int):
return TimeDelta(seconds=self.seek_time)
elif isinstance(self.seek_time, float):
return TimeDelta(seconds=self.seek_time)
else:
raise TypeError("不支持的时间类型")
model_config = {
"arbitrary_types_allowed": True
}
class FFMPEGExtractFrameStatusResponse(BaseFFMPEGTaskStatusResponse):
result: Optional[str] = Field(default=None, description="生成结果的URN")
@@ -429,7 +452,8 @@ class ComfyTaskRequest(BaseFFMPEGTaskRequest):
class FFMPEGStreamRecordRequest(BaseFFMPEGTaskRequest):
stream_source: str = Field(description="直播源地址")
first_segment_duration: int = Field(default=2, description="hls首个片段时长(秒), 首片段长度越小hls流能越快速开始播放")
first_segment_duration: int = Field(default=2,
description="hls首个片段时长(秒), 首片段长度越小hls流能越快速开始播放")
segment_duration: int = Field(default=10, description="hls片段时长(秒)")
recording_timeout: int = Field(default=300, description="hls流无内容后等待的时长(秒)")
monitor_timeout: int = Field(default=36000, description="录制监控最大时长(秒), 默认为10小时, 不可大于12小时",
@@ -439,10 +463,10 @@ class FFMPEGStreamRecordRequest(BaseFFMPEGTaskRequest):
class GeminiRequest(BaseFFMPEGTaskRequest):
media_hls_url: MediaSource = Field(default="", description="视频流录制HLS地址 hls://格式 需录制超过20分钟")
product_cover_grid_uri_list: List[str] = Field(description="商品封面网格拼图URI列表")
product_list: List[Union[str, dict]] = Field(description="商品名列表(时间倒序)"),
product_list: List[str] = Field(description="商品名列表(时间倒序)"),
start_time: str = Field(default="00:00:00.000", description="开始时间(hls)")
end_time: str = Field(default="00:20:00.000", description="结束时间(hls)")
options: FFMPEGSliceOptions = Field(default=FFMPEGSliceOptions() ,description="输出质量选项")
options: FFMPEGSliceOptions = Field(default=FFMPEGSliceOptions(), description="输出质量选项")
@field_validator('media_hls_url', mode='before')
@classmethod
@@ -458,6 +482,7 @@ class GeminiRequest(BaseFFMPEGTaskRequest):
else:
raise pydantic.ValidationError("media格式读取失败")
class GeminiResultResponse(BaseFFMPEGTaskStatusResponse):
result: str = Field(default="", description="推理出的json")
@@ -486,14 +511,17 @@ class MonitorLiveRoomProductRequest(BaseModel):
class LiveRoomProductCachesResponse(BaseModel):
status: int = Field(default=None, description="缓存状态/0-正常返回 1-直播已结束 2-IP风控 3-请求Tikhub API错误 4-内部错误")
status: int = Field(default=None,
description="缓存状态/0-正常返回 1-直播已结束 2-IP风控 3-请求Tikhub API错误 4-内部错误")
message: str = Field(default="", description="错误信息")
cache_json: str = Field(default="", description="缓存内容/Json文本")
class MakeGridGeminiRequest(BaseFFMPEGTaskRequest):
pic_info_list: List[Dict[str, str]] = Field(default=[], description="包含图片信息的字典列表，每个字典包含 \"title\" 和 \"cover\" 键")
pic_info_list: List[Dict[str, str]] = Field(default=[],
description="包含图片信息的字典列表，每个字典包含 \"title\" 和 \"cover\" 键")
image_size: int = Field(default=450, description="单个图片网格的尺寸/像素")
text_height: int = Field(default=40, description="文本框的高度/像素")
font_size: int = Field(default=18, description="文本尺寸/像素")
padding: int = Field(default=5, description="文本距离文本框边缘距离/像素")
separator: int = Field(default=12, description="分割线宽度/像素")
separator: int = Field(default=12, description="分割线宽度/像素")

2
src/BowongModalFunctions/router/ffmpeg.py
View File

@@ -199,7 +199,7 @@ async def video_extract_frame(body: FFMPEGExtractFrameRequest,
sentry_trace = None
if headers.x_trace_id and headers.x_baggage:
sentry_trace = SentryTransactionInfo(x_trace_id=headers.x_trace_id, x_baggage=headers.x_baggage)
fn_call = fn.spawn(media=body.video, frame_index=body.frame_index, sentry_trace=sentry_trace, webhook=body.webhook)
fn_call = fn.spawn(media=body.video,seek_time=body.prased_seek_time, frame_index=body.frame_index, sentry_trace=sentry_trace, webhook=body.webhook)
return ModalTaskResponse(success=True, taskId=fn_call.object_id)

4
src/BowongModalFunctions/utils/TimeUtils.py
View File

@@ -136,7 +136,7 @@ def merge_product_data(data, start_time_str, end_time_str):
for item in timeline:
start, end = parse_timeline_item(item)
# 比较起始时间与时间差
if (start - datetime.min) > duration:
if (start - datetime.strptime("00:00:00.000", '%H:%M:%S.%f')) > duration:
# 处理时间字符串
start_str = format_time(start)
end_str = format_time(end)
@@ -150,4 +150,4 @@ def merge_product_data(data, start_time_str, end_time_str):
product_dict[product]["timeline"] = merge_timeline_items(new_timeline)
# 返回合并后的列表
return list(product_dict.values())
return list(product_dict.values())

21
src/BowongModalFunctions/utils/VideoUtils.py
View File

@@ -386,7 +386,7 @@ class VideoUtils:
raise RuntimeError("输出是空文件")
else:
if not quiet:
if "SKIP" not in line:
if "Skip" not in line:
logger.warning(line)
return ffmpeg_cmd
@@ -731,14 +731,21 @@ class VideoUtils:
origin_time: datetime = playlist.segments[0].current_program_date_time
# 2. 解析TS片段URL
ts_urls: SegmentList[Segment] = SegmentList()
duration = 0
for segment in playlist.segments:
if not head:
head = 0
if not tail:
tail = 86400 # 使用24H时长替代♾️
if origin_time + timedelta(
seconds=head) <= segment.current_program_date_time <= origin_time + timedelta(seconds=tail):
duration += segment.duration
ts_urls.append(segment)
logger.info(f"{len(ts_urls)}")
# 3. 并行下载TS片段
tasks = []
playlist.segments = ts_urls
duration_delta = TimeDelta(seconds=duration)
logger.info(f"count : {len(playlist.segments)}, duration = {duration_delta.toFormatStr()}")
playlist.is_endlist = True
for url in ts_urls:
tasks.append(VideoUtils.async_download_file(url.absolute_uri, f"{temp_dir}/{url.uri}"))
@@ -773,7 +780,7 @@ class VideoUtils:
os.makedirs(os.path.dirname(output_path), exist_ok=True)
try:
local_m3u8_path, temp_dir = await VideoUtils.convert_m3u8_to_local_source(media_stream_url)
local_m3u8_path, temp_dir = await VideoUtils.convert_m3u8_to_local_source(media_stream_url=media_stream_url)
# 使用ffmpeg合并TS片段
ffmpeg_cmd = VideoUtils.async_ffmpeg_init()
ffmpeg_cmd.input(local_m3u8_path,
@@ -1181,7 +1188,8 @@ class VideoUtils:
return output_path, video_metadata
@staticmethod
async def ffmpeg_extract_frame_image(video_path: str, frame_index: int, output_path: Optional[str] = None) -> Tuple[
async def ffmpeg_extract_frame_image(video_path: str, frame_index: int, seek_time: Optional[TimeDelta] = None,
output_path: Optional[str] = None) -> Tuple[
str, VideoMetadata]:
"""
获取视频的第n帧输出为图片, 并返回图片相关的元数据
@@ -1191,7 +1199,10 @@ class VideoUtils:
output_path = FileUtils.file_path_change_extension(output_path, 'jpg')
os.makedirs(os.path.dirname(output_path), exist_ok=True)
ffmpeg_cmd = VideoUtils.async_ffmpeg_init()
ffmpeg_cmd.input(video_path)
if seek_time:
ffmpeg_cmd.input(video_path, ss=seek_time.total_seconds())
else:
ffmpeg_cmd.input(video_path)
ffmpeg_cmd.output(output_path, vframes=frame_index)
await ffmpeg_cmd.execute()
image_metadata = VideoUtils.ffprobe_media_metadata(output_path)

2
src/cluster/ffmpeg_apps/convert_stream.py
View File

@@ -13,7 +13,7 @@ with ffmpeg_worker_image.imports():
from modal import current_function_call_id
@app.function(timeout=1800, cloud="aws",
@app.function(timeout=3600 * 3, cloud="aws",
max_containers=config.ffmpeg_worker_concurrency,
cpu=(0.5, 64),
volumes={

10
src/cluster/ffmpeg_apps/extract_frame.py
View File

@@ -1,5 +1,6 @@
import modal
from ..ffmpeg_app import ffmpeg_worker_image, app, config, s3_mount, local_copy_to_s3, output_path_prefix
with ffmpeg_worker_image.imports():
@@ -8,6 +9,7 @@ with ffmpeg_worker_image.imports():
from BowongModalFunctions.utils.PathUtils import FileUtils
from BowongModalFunctions.utils.SentryUtils import SentryUtils
from BowongModalFunctions.utils.VideoUtils import VideoUtils
from BowongModalFunctions.utils.TimeUtils import TimeDelta
import sentry_sdk
from typing import Optional, Tuple
from modal import current_function_call_id
@@ -23,7 +25,7 @@ with ffmpeg_worker_image.imports():
),
}, )
@modal.concurrent(max_inputs=1)
async def ffmpeg_extract_frame(media: MediaSource, frame_index: int,
async def ffmpeg_extract_frame(media: MediaSource, frame_index: int, seek_time: Optional[TimeDelta] = None,
sentry_trace: Optional[SentryTransactionInfo] = None,
webhook: Optional[WebhookNotify] = None) -> Tuple[
FFMPEGResult, Optional[SentryTransactionInfo]]:
@@ -33,7 +35,8 @@ with ffmpeg_worker_image.imports():
sentry_trace_id=sentry_trace.x_trace_id if sentry_trace else None,
sentry_baggage=sentry_trace.x_baggage if sentry_trace else None)
@SentryUtils.webhook_handler(webhook=webhook, func_id=fn_id)
async def ffmpeg_process(media: MediaSource, frame_index: int, func_id: str) -> FFMPEGResult:
async def ffmpeg_process(media: MediaSource, frame_index: int, func_id: str,
seek_time: Optional[TimeDelta] = None) -> FFMPEGResult:
match media.protocol:
case MediaProtocol.hls:
video_path = f"{media.path}"
@@ -45,13 +48,14 @@ with ffmpeg_worker_image.imports():
raise NotImplementedError(f"暂不支持此协议")
output_path = f"{output_path_prefix}/{config.modal_environment}/extract_frame/{func_id}/output.jpg"
local_output, metadata = await VideoUtils.ffmpeg_extract_frame_image(video_path=video_path,
seek_time=seek_time,
frame_index=frame_index,
output_path=output_path)
s3_outputs = local_copy_to_s3([local_output])
return FFMPEGResult(urn=s3_outputs[0], metadata=metadata,
content_length=FileUtils.get_file_size(local_output), )
result = await ffmpeg_process(media=media, frame_index=frame_index, func_id=fn_id)
result = await ffmpeg_process(media=media, frame_index=frame_index, seek_time=seek_time, func_id=fn_id)
if not sentry_trace:
sentry_trace = SentryTransactionInfo(x_trace_id=sentry_sdk.get_traceparent(),
x_baggage=sentry_sdk.get_baggage())

5
src/cluster/ffmpeg_apps/slice_media.py
View File

@@ -75,6 +75,11 @@ with ffmpeg_worker_image.imports():
media_markers=markers,
options=options,
fn_id=fn_id)
case MediaProtocol.vod:
outputs = await ffmpeg_slice_process(media_source=media,
media_markers=markers,
options=options,
fn_id=fn_id)
case MediaProtocol.s3:
outputs = await ffmpeg_slice_process(media_source=media,
media_markers=markers,