diff --git a/.runtime.env b/.runtime.env index a31361e..f766917 100644 --- a/.runtime.env +++ b/.runtime.env @@ -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 diff --git a/src/BowongModalFunctions/models/ffmpeg_worker_model.py b/src/BowongModalFunctions/models/ffmpeg_worker_model.py index ed28886..310173f 100644 --- a/src/BowongModalFunctions/models/ffmpeg_worker_model.py +++ b/src/BowongModalFunctions/models/ffmpeg_worker_model.py @@ -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 diff --git a/src/BowongModalFunctions/models/web_model.py b/src/BowongModalFunctions/models/web_model.py index fe414bd..b61eeae 100644 --- a/src/BowongModalFunctions/models/web_model.py +++ b/src/BowongModalFunctions/models/web_model.py @@ -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="分割线宽度/像素") \ No newline at end of file + separator: int = Field(default=12, description="分割线宽度/像素") diff --git a/src/BowongModalFunctions/router/ffmpeg.py b/src/BowongModalFunctions/router/ffmpeg.py index 28e62fc..8217d50 100644 --- a/src/BowongModalFunctions/router/ffmpeg.py +++ b/src/BowongModalFunctions/router/ffmpeg.py @@ -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) diff --git a/src/BowongModalFunctions/utils/TimeUtils.py b/src/BowongModalFunctions/utils/TimeUtils.py index b417f1a..bac4f32 100644 --- a/src/BowongModalFunctions/utils/TimeUtils.py +++ b/src/BowongModalFunctions/utils/TimeUtils.py @@ -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()) \ No newline at end of file + return list(product_dict.values()) diff --git a/src/BowongModalFunctions/utils/VideoUtils.py b/src/BowongModalFunctions/utils/VideoUtils.py index 76dc68e..f22a32b 100644 --- a/src/BowongModalFunctions/utils/VideoUtils.py +++ b/src/BowongModalFunctions/utils/VideoUtils.py @@ -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) diff --git a/src/cluster/ffmpeg_apps/convert_stream.py b/src/cluster/ffmpeg_apps/convert_stream.py index f500cc9..d202731 100644 --- a/src/cluster/ffmpeg_apps/convert_stream.py +++ b/src/cluster/ffmpeg_apps/convert_stream.py @@ -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={ diff --git a/src/cluster/ffmpeg_apps/extract_frame.py b/src/cluster/ffmpeg_apps/extract_frame.py index e00418d..73bbfc4 100644 --- a/src/cluster/ffmpeg_apps/extract_frame.py +++ b/src/cluster/ffmpeg_apps/extract_frame.py @@ -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()) diff --git a/src/cluster/ffmpeg_apps/slice_media.py b/src/cluster/ffmpeg_apps/slice_media.py index 8fde0b4..abec939 100644 --- a/src/cluster/ffmpeg_apps/slice_media.py +++ b/src/cluster/ffmpeg_apps/slice_media.py @@ -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,