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

fix : 修复直播切分视频接口

Browse Source
This commit is contained in:
shuohigh@gmail.com
2025-06-09 18:33:43 +08:00
parent e01ea2df6f
commit c56f87eeef
8 changed files with 1364 additions and 1337 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
pyproject.toml
View File

@@ -28,7 +28,7 @@ dependencies = [
"boto3>=1.37.37",
"psutil>=7.0.0",
"scalar-fastapi>=1.0.3",
"modal>=0.76.3",
"modal>=1.0.3",
"python-dotenv>=1.1.0",
"python-multipart>=0.0.20",
"m3u8>=6.0.0",

22
src/BowongModalFunctions/models/media_model.py
View File

@@ -286,25 +286,3 @@ class UploadMultipartPresignResponse(BaseModel):
complete_url: str = Field(description="用于确认完成分片上传的请求地址")
urn: str = Field(description="上传成功后获得的对应资源URN")
expired_at: datetime = Field(description="上传地址签名过期时间戳")
model_config = ConfigDict(json_schema_extra={
"description": """
1. 本地按文件总大小分Chunk大小按urls链接内的顺序通过HTTP PUT请求上传文件分片; 并将上传完成后获得的返回头ETag值记录与PartNumber对应, PartNumber对应使用url在urls内的顺位, 以1开始
2. 所有分片上传完成后使用XML格式拼装出用于确认上传的body; 并通过HTTP POST complete_url确认上传, ContentType 需要确保为application/xml
<CompleteMultipartUpload>
<Part>
<PartNumber>1</PartNumber>
<ETag>"60575364b098a1a48765a28c3a48e0ef"</ETag>
</Part>
<Part>
<PartNumber>2</PartNumber>
<ETag>"a38691c31fd242faee5533c65b4501d7"</ETag>
</Part>
<Part>
<PartNumber>3</PartNumber>
<ETag>"7a7510cc83f98feea28f319567e4cf66"</ETag>
</Part>
</CompleteMultipartUpload>
3. 如无法确认上传，可使用HTTP GET list_url debug当前分片上传状态，确认成功后list_url无法返回有效数据
"""
})

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

@@ -123,7 +123,7 @@ class FFMPEGConvertStreamResponse(BaseFFMPEGTaskStatusResponse):
class FFMPEGSliceRequest(BaseFFMPEGTaskRequest):
media: MediaSource = Field(description="待切割的媒体源")
markers: List[FFMpegSliceSegment] = Field(description="切割标记数组")
markers: List[FFMpegSliceSegment] = Field(description="按照时间顺序排序过的切割标记数组")
@field_validator('media', mode='before')
@classmethod
@@ -428,7 +428,7 @@ class FFMPEGStreamRecordRequest(BaseFFMPEGTaskRequest):
stream_source: str = Field(description="直播源地址")
segment_duration: int = Field(default=5, description="hls片段时长(秒)")
recording_timeout: int = Field(default=300, description="hls流无内容后等待的时长(秒)")
monitor_timeout: int = Field(default=36000, description="录制监控最大时长(秒), 默认为10小时")
monitor_timeout: int = Field(default=36000, description="录制监控最大时长(秒), 默认为10小时, 不可大于12小时", le=43200)
class GeminiRequest(BaseFFMPEGTaskRequest):

31
src/BowongModalFunctions/router/cache.py
View File

@@ -24,7 +24,8 @@ from ..models.media_model import (MediaSources,
UploadBase64Request, UploadPresignRequest, UploadPresignResponse,
UploadMultipartPresignRequest, UploadMultipartPresignResponse
)
from ..models.web_model import SentryTransactionInfo, MonitorLiveRoomProductRequest, ModalTaskResponse, LiveRoomProductCachesResponse
from ..models.web_model import SentryTransactionInfo, MonitorLiveRoomProductRequest, ModalTaskResponse, \
LiveRoomProductCachesResponse
from ..utils.KVCache import MediaSourceKVCache, LiveProductKVCache
from ..utils.SentryUtils import SentryUtils
@@ -270,13 +271,14 @@ async def s3_upload_base64(body: UploadBase64Request) -> UploadResultResponse:
media_source.downloader_id = fn_id
return UploadResultResponse(media=media_source)
@router.post("/monitor_live_room_product_trigger",
summary="触发监控直播间商品信息并缓存",
description="触发监控直播间商品信息并缓存, 如果直播结束清除缓存, 触发间隔请控制在60s以上",
dependencies=[Depends(verify_token)])
async def monitor_live_room_product(body:MonitorLiveRoomProductRequest) -> LiveRoomProductCachesResponse:
fn = modal.Function.from_name(config.modal_app_name, "monitor_live_room_product_trigger", environment_name=config.modal_environment)
async def monitor_live_room_product(body: MonitorLiveRoomProductRequest) -> LiveRoomProductCachesResponse:
fn = modal.Function.from_name(config.modal_app_name, "monitor_live_room_product_trigger",
environment_name=config.modal_environment)
status = await fn.remote.aio(body.cookie, body.room_id, body.author_id)
if status == 0:
product_list = modal_kv_product_cache.get_cache(body.room_id)
@@ -290,6 +292,7 @@ async def monitor_live_room_product(body:MonitorLiveRoomProductRequest) -> LiveR
else:
return LiveRoomProductCachesResponse(status=4, message="内部错误")
@router.post('/upload-s3/simple/presign',
summary="S3简单上传预签名",
description="利用S3就近接入点上传",
@@ -310,7 +313,25 @@ async def s3_presign_upload(body: UploadPresignRequest) -> UploadPresignResponse
@router.post("/upload-s3/multipart/presign",
summary="S3分片上传预签名",
description="", dependencies=[Depends(verify_token)])
description="""
1. 本地按文件总大小分Chunk大小按urls链接内的顺序通过HTTP PUT请求上传文件分片; 并将上传完成后获得的返回头ETag值记录与PartNumber对应, PartNumber对应使用url在urls内的顺位, 以1开始
\n2. 所有分片上传完成后使用XML格式拼装出用于确认上传的body; 并通过HTTP POST complete_url确认上传, ContentType 需要确保为application/xml\n\n
<CompleteMultipartUpload>
<Part>
<PartNumber>1</PartNumber>
<ETag>"60575364b098a1a48765a28c3a48e0ef"</ETag>
</Part>
<Part>
<PartNumber>2</PartNumber>
<ETag>"a38691c31fd242faee5533c65b4501d7"</ETag>
</Part>
<Part>
<PartNumber>3</PartNumber>
<ETag>"7a7510cc83f98feea28f319567e4cf66"</ETag>
</Part>
</CompleteMultipartUpload>
\n3. 如无法确认上传，可使用HTTP GET list_url debug当前分片上传状态，确认成功后list_url无法返回有效数据
""", dependencies=[Depends(verify_token)])
async def s3_presign_upload_multipart(body: UploadMultipartPresignRequest) -> UploadMultipartPresignResponse:
chunk_count = body.parts_count
multipart_upload_response = client.create_multipart_upload(Bucket=config.S3_bucket_name, Key=body.key,

10
src/BowongModalFunctions/utils/KVCache.py
View File

@@ -7,7 +7,6 @@ from .VideoUtils import VideoUtils
from ..models.media_model import MediaSource
from ..models.web_model import LiveProductCaches
# secrets = modal.Secret.from_name("cf-kv-secret")
class KVCache:
kv: modal.Dict
@@ -18,7 +17,10 @@ class KVCache:
def __init__(self, kv_name: str, environment: str):
# self.cf_kv_id = cf_kv_id
self.kv = modal.Dict.from_name(kv_name, environment_name=environment, create_if_missing=True)
logger.info(f"Using KV space : {self.cf_kv_id}")
if self.cf_kv_id:
logger.info(f"Using KV space : {self.cf_kv_id}")
else:
logger.warning(f"CF_KV_NAMESPACE_ID为空, 如果是本地使用modal deploy时触发此警告可忽略")
def batch_update_cloudflare_kv(self, caches: Dict[str, str]):
with httpx.Client() as client:
@@ -102,8 +104,10 @@ class MediaSourceKVCache(KVCache):
return None
return MediaSource.model_validate_json(cache_json)
class LiveProductKVCache(KVCache):
cf_kv_id: str = os.environ.get("CF_PRODUCT_KV_NAMESPACE_ID")
def get_cache(self, room_id: str) -> Optional[LiveProductCaches]:
cache_json = self.kv.get(room_id)
if not cache_json:
@@ -125,4 +129,4 @@ class LiveProductKVCache(KVCache):
if raise_exception:
raise KeyError("ROOM_ID错误，资源不存在")
return None
return LiveProductCaches.model_validate_json(cache_json)
return LiveProductCaches.model_validate_json(cache_json)

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

@@ -1,6 +1,8 @@
import asyncio
import re
import tempfile
from datetime import datetime, timedelta
import aiofiles
import aiohttp
from typing import Union, List, Tuple, Optional, Dict, Any
@@ -357,6 +359,8 @@ class VideoUtils:
if line.startswith('Error') and ".m3u8" not in line:
logger.error(line)
raise RuntimeError(line)
elif "Output file is empty" in line:
raise RuntimeError("输出是空文件")
else:
if not quiet:
logger.warning(line)
@@ -479,12 +483,20 @@ class VideoUtils:
playlist = m3u8.load(media_path)
stream_total_duration: float = sum(segment.duration for segment in playlist.segments)
seek_head = media_markers[0].start.total_seconds()
seek_tail = media_markers[-1].end.total_seconds()
duration = seek_tail - seek_head
logger.info(f"Only using {seek_head}s --> {seek_tail}s = {duration}s")
ffmpeg_cmd = VideoUtils.async_ffmpeg_init()
# ffmpeg_cmd.option('loglevel', 'debug')
ffmpeg_cmd.input(media_path,
ss=seek_head,
t=duration,
protocol_whitelist="file,http,https,tcp,tls",
reconnect="1", # 自动重连
reconnect_streamed="1",
reconnect_delay_max="5")
filter_complex: List[str] = []
temp_outputs: List[str] = []
@@ -529,8 +541,9 @@ class VideoUtils:
@staticmethod
async def ffmpeg_slice_stream_media_multithread(media_path: str,
media_markers: List[FFMpegSliceSegment],
output_path: Optional[str] = None) -> List[Tuple[str, VideoMetadata]]:
media_markers: List[FFMpegSliceSegment],
output_path: Optional[str] = None) -> List[
Tuple[str, VideoMetadata]]:
"""
按时间分段切割HLS视频流_预先多线程下载所有ts
:param media_path: hls manifest URL
@@ -539,9 +552,22 @@ class VideoUtils:
:return: 输出片段的本地路径， 输出片段时长
"""
import m3u8
local_m3u8_path, temp_dir = await VideoUtils.convert_m3u8_to_local_source(media_path)
playlist = m3u8.load(local_m3u8_path)
stream_total_duration: float = sum(segment.duration for segment in playlist.segments)
seek_head = media_markers[0].start.total_seconds()
seek_tail = media_markers[-1].end.total_seconds()
duration = seek_tail - seek_head
logger.info(f"Only using {seek_head}s --> {seek_tail}s = {duration}s")
local_m3u8_path, temp_dir = await VideoUtils.convert_m3u8_to_local_source(media_path, head=seek_head,
tail=seek_tail)
logger.info(f"local_playlist: {local_m3u8_path}")
# playlist = m3u8.load(f"file://{local_m3u8_path}")
# stream_total_duration: float = sum(segment.duration for segment in playlist.segments)
stream_total_duration = duration
for segment in media_markers:
segment.start = segment.start - timedelta(seconds=seek_head)
segment.end = segment.end - timedelta(seconds=seek_head)
ffmpeg_cmd = VideoUtils.async_ffmpeg_init()
ffmpeg_cmd.input(local_m3u8_path,
@@ -639,42 +665,43 @@ class VideoUtils:
logger.warning(f"Retrying downloading {url} to {output_path} Remain Times: {t}")
@staticmethod
async def convert_m3u8_to_local_source(media_stream_url:str) -> tuple[str, str]:
async def convert_m3u8_to_local_source(media_stream_url: str,
head: Optional[float] = None,
tail: Optional[float] = None) -> tuple[str, str]:
"""
转换m3u8为本地来源
"""
# 创建临时目录存储TS片段
temp_dir = tempfile.mkdtemp()
from m3u8 import SegmentList, Segment
try:
# 1. 下载m3u8文件
m3u8_content = await VideoUtils.async_download_file(media_stream_url)
playlist = m3u8.load(media_stream_url)
# duration = (tail - head) if head else None
origin_time: datetime = playlist.segments[0].current_program_date_time
# 2. 解析TS片段URL
base_url = media_stream_url.rsplit('/', 1)[0] + '/'
ts_urls = re.findall(r'(.*?\.ts\?.*)', m3u8_content)
ts_urls = [url if url.startswith('http') else base_url + url for url in ts_urls]
ts_urls: SegmentList[Segment] = SegmentList()
for segment in playlist.segments:
if origin_time + timedelta(
seconds=head) <= segment.current_program_date_time <= origin_time + timedelta(seconds=tail):
ts_urls.append(segment)
logger.info(f"{len(ts_urls)}")
# 3. 并行下载TS片段
tasks = []
for i, url in enumerate(ts_urls):
ts_path = os.path.join(temp_dir, f"segment_{i}.ts")
tasks.append(VideoUtils.async_download_file(url, ts_path))
playlist.segments = ts_urls
playlist.is_endlist = True
for url in ts_urls:
tasks.append(VideoUtils.async_download_file(url.absolute_uri, f"{temp_dir}/{url.uri}"))
await asyncio.gather(*tasks)
# 4. 修改m3u8文件指向本地TS片段
local_m3u8_path = os.path.join(temp_dir, "local.m3u8")
local_m3u8_content = m3u8_content
for i in range(len(tasks)):
local_ts_path = os.path.join(temp_dir, f"segment_{i}.ts")
local_m3u8_content = local_m3u8_content.replace(ts_urls[i], local_ts_path)
playlist.dump(local_m3u8_path)
async with aiofiles.open(local_m3u8_path, 'w') as f:
await f.write(local_m3u8_content)
return local_m3u8_path, temp_dir
except Exception as e:
logger.exception(f"下载TS转换M3U8失败 {e}")
logger.exception(e)
raise Exception(f"下载TS转换M3U8失败 {e}")
@staticmethod
@@ -691,7 +718,8 @@ class VideoUtils:
@staticmethod
async def ffmpeg_convert_stream_media_multithread(media_stream_url: str,
output_path: Optional[str] = None) -> tuple[str, VideoMetadata] | None:
output_path: Optional[str] = None) -> tuple[
str, VideoMetadata] | None:
if not output_path:
output_path = FileUtils.file_path_extend(media_stream_url, "convert")
if not output_path.endswith(".mp4"):
@@ -699,6 +727,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)
# 使用ffmpeg合并TS片段
ffmpeg_cmd = VideoUtils.async_ffmpeg_init()

17
src/cluster/ffmpeg_app.py
View File

@@ -105,7 +105,7 @@ with ffmpeg_worker_image.imports():
@app.function(
timeout=900,
cloud="aws",
cpu=(0.5,64),
cpu=(0.5, 64),
max_containers=config.ffmpeg_slice_worker_concurrency,
volumes={
s3_mount: modal.CloudBucketMount(
@@ -146,13 +146,10 @@ with ffmpeg_worker_image.imports():
async def ffmpeg_hls_slice_process(media_source: MediaSource,
media_markers: List[FFMpegSliceSegment],
fn_id: str) -> List[FFMPEGResult]:
# if media_source.protocol is MediaProtocol.s3 and media_source.urn.endswith(".m3u8"):
# hls_m3u8_url = f"{s3_mount}/{media_source.cache_filepath}"
# else:
hls_m3u8_url = media_source.path
segments = await VideoUtils.ffmpeg_slice_stream_media(media_path=hls_m3u8_url,
media_markers=media_markers,
output_path=f"{output_path_prefix}/{config.modal_environment}/slice/outputs/{fn_id}/output.mp4")
segments = await VideoUtils.ffmpeg_slice_stream_media_multithread(media_path=hls_m3u8_url,
media_markers=media_markers,
output_path=f"{output_path_prefix}/{config.modal_environment}/slice/outputs/{fn_id}/output.mp4")
return [FFMPEGResult(urn=local_copy_to_s3([segment[0]])[0], metadata=segment[1],
content_length=FileUtils.get_file_size(segment[0])) for segment in segments]
@@ -472,7 +469,7 @@ with ffmpeg_worker_image.imports():
@app.function(timeout=1800, cloud="aws",
max_containers=config.ffmpeg_worker_concurrency,
cpu=(0.5,64),
cpu=(0.5, 64),
volumes={
s3_mount: modal.CloudBucketMount(
bucket_name=config.S3_bucket_name,
@@ -566,7 +563,7 @@ with ffmpeg_worker_image.imports():
hls_recording_mount_point = "/mnt/stream_records"
@app.function(timeout=43200, # 最长处理12h的录制任务
@app.function(timeout=43200 + 300, # 最长处理12h的录制任务 + 5分钟的清理缓存工作
cloud="aws",
# todo: 暂时不限制最大同时存在的录制数量
# max_containers=config.ffmpeg_worker_concurrency,
@@ -631,8 +628,6 @@ with ffmpeg_worker_image.imports():
self.webhook_on_start()
except Exception as e:
logger.exception(e)
else:
logger.info("[Start] no webhook")
@backoff.on_exception(exception=Exception, wait_gen=backoff.constant,
max_time=5, max_tries=5, raise_on_giveup=True)

2538
uv.lock generated
View File

File diff suppressed because it is too large Load Diff