From b0f954c55d7527249235458d9b48df2d312b7cd0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=82=96=E5=AE=87=E8=BF=AA?= Date: Tue, 24 Jun 2025 18:02:29 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9ENakama=E7=99=BB=E5=BD=95?= =?UTF-8?q?=E6=8E=A5=E5=8F=A3=EF=BC=8C=E5=90=8C=E6=97=B6=E6=94=AF=E6=8C=81?= =?UTF-8?q?=E4=BD=BF=E7=94=A8Nakama=20JWT=E8=B0=83=E7=94=A8API?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 新增Nakama登录接口，同时支持使用Nakama JWT调用API * 将tikhub_api和api分离 * 拆分models为按功能划分,以便管理依赖 --------- Merge request URL: https://g-ldyi2063.coding.net/p/dev/d/modalDeploy/git/merge/4865?initial=true Co-authored-by: shuohigh@gmail.com --- src/BowongModalFunctions/api.py | 29 +- .../middleware/authorization.py | 24 +- .../models/cache_tasks/__init__.py | 0 .../{media_model.py => cache_tasks/models.py} | 250 +++++---- .../models/enums/__init__.py | 0 .../models/enums/models.py | 47 ++ .../models/ffmpeg_tasks/__init__.py | 0 .../models/ffmpeg_tasks/models.py | 310 ++++++++++++ .../models/ffmpeg_worker_model.py | 93 ---- .../models/requests/__init__.py | 0 .../{web_model.py => requests/models.py} | 476 ++++-------------- .../models/responses/__init__.py | 0 .../models/responses/models.py | 182 +++++++ .../models/settings/__init__.py | 0 .../{config.py => models/settings/cluster.py} | 4 +- src/BowongModalFunctions/router/cache.py | 35 +- src/BowongModalFunctions/router/comfyui.py | 11 +- src/BowongModalFunctions/router/ffmpeg.py | 23 +- src/BowongModalFunctions/router/google.py | 18 +- src/BowongModalFunctions/router/task.py | 23 +- src/BowongModalFunctions/utils/HTTPUtils.py | 5 +- src/BowongModalFunctions/utils/KVCache.py | 3 +- src/BowongModalFunctions/utils/ModalUtils.py | 13 +- src/BowongModalFunctions/utils/SentryUtils.py | 5 +- src/BowongModalFunctions/utils/TimeUtils.py | 24 +- src/BowongModalFunctions/utils/VideoUtils.py | 179 +------ src/Douyin_TikTok_Download_API/tikhub_api.py | 22 + src/cluster/app.py | 2 +- src/cluster/comfyui_v1.py | 4 +- src/cluster/comfyui_v2.py | 4 +- src/cluster/ffmpeg_app.py | 2 +- src/cluster/ffmpeg_apps/bgm_nosie_reduce.py | 4 +- src/cluster/ffmpeg_apps/concat_medias.py | 8 +- src/cluster/ffmpeg_apps/convert_stream.py | 8 +- src/cluster/ffmpeg_apps/corner_mirror.py | 4 +- src/cluster/ffmpeg_apps/extract_audio.py | 4 +- src/cluster/ffmpeg_apps/extract_frame.py | 5 +- src/cluster/ffmpeg_apps/loop_fill.py | 4 +- src/cluster/ffmpeg_apps/overlay_gif.py | 4 +- src/cluster/ffmpeg_apps/slice_media.py | 10 +- .../ffmpeg_apps/stream_record_as_hls.py | 5 +- src/cluster/ffmpeg_apps/subtitle_apply.py | 4 +- src/cluster/ffmpeg_apps/zoom_loop.py | 4 +- src/cluster/video.py | 4 +- src/cluster/video_apps/cache_delete.py | 3 +- src/cluster/video_apps/cache_submit.py | 5 +- src/cluster/video_apps/hls_slice_inference.py | 11 +- src/cluster/video_apps/make_grid_upload.py | 3 +- .../monitor_live_room_product_trigger.py | 3 +- src/cluster/web.py | 10 +- tests/ffmpeg_test_case.py | 4 +- tests/pydantic_model_validation_test.py | 2 +- 52 files changed, 969 insertions(+), 928 deletions(-) create mode 100644 src/BowongModalFunctions/models/cache_tasks/__init__.py rename src/BowongModalFunctions/models/{media_model.py => cache_tasks/models.py} (57%) create mode 100644 src/BowongModalFunctions/models/enums/__init__.py create mode 100644 src/BowongModalFunctions/models/enums/models.py create mode 100644 src/BowongModalFunctions/models/ffmpeg_tasks/__init__.py create mode 100644 src/BowongModalFunctions/models/ffmpeg_tasks/models.py delete mode 100644 src/BowongModalFunctions/models/ffmpeg_worker_model.py create mode 100644 src/BowongModalFunctions/models/requests/__init__.py rename src/BowongModalFunctions/models/{web_model.py => requests/models.py} (52%) create mode 100644 src/BowongModalFunctions/models/responses/__init__.py create mode 100644 src/BowongModalFunctions/models/responses/models.py create mode 100644 src/BowongModalFunctions/models/settings/__init__.py rename src/BowongModalFunctions/{config.py => models/settings/cluster.py} (92%) create mode 100644 src/Douyin_TikTok_Download_API/tikhub_api.py diff --git a/src/BowongModalFunctions/api.py b/src/BowongModalFunctions/api.py index f77ad21..1159bad 100644 --- a/src/BowongModalFunctions/api.py +++ b/src/BowongModalFunctions/api.py @@ -7,8 +7,7 @@ from fastapi.middleware.cors import CORSMiddleware from .utils.KVCache import MediaSourceKVCache from .router import ffmpeg, cache, comfyui, google, task -from Douyin_TikTok_Download_API.app.api import router as tikhub -from .config import WorkerConfig +from .models.settings.cluster import WorkerConfig config = WorkerConfig() @@ -23,14 +22,9 @@ web_app = FastAPI(title="Modal worker API", ]) web_app_tikhub = FastAPI(title="Modal Tikhub API", - version="1.0.0", - summary="Modal Tikhub API, 包含抖音Web API", - servers=[ - { - 'url': f'https://bowongai-{config.modal_environment}--{config.modal_app_name}-fastapi-webapp-tikhub.modal.run', - 'description': '当前Modal Tikhub API接口endpoint' - } - ]) + version="1.0.0", + summary="Modal Tikhub API, 包含抖音Web API", + ) sentry_sdk.init(dsn="https://dab7b7ae652216282c89f029a76bb10a@sentry.bowongai.com/2", send_default_pii=True, @@ -73,14 +67,6 @@ web_app.add_middleware( allow_headers=["*"], ) -web_app_tikhub.add_middleware( - CORSMiddleware, - allow_origins=["*"], - allow_credentials=True, - allow_methods=["*"], - allow_headers=["*"], -) - @web_app.middleware("http") @web_app.middleware("https") @@ -94,10 +80,6 @@ async def alias_middleware(request: Request, call_next): async def scalar(): return get_scalar_api_reference(openapi_url='/openapi.json', title="Modal worker web endpoint") -@web_app_tikhub.get("/scalar", include_in_schema=False) -async def scalar(): - return get_scalar_api_reference(openapi_url='/openapi.json', title="Modal worker web endpoint") - web_app.include_router(ffmpeg.router) web_app.include_router(cache.router) @@ -106,6 +88,3 @@ web_app.include_router(cache.router) web_app.include_router(google.router) web_app.include_router(task.router) - -# Tikhub API -web_app_tikhub.include_router(tikhub.router) diff --git a/src/BowongModalFunctions/middleware/authorization.py b/src/BowongModalFunctions/middleware/authorization.py index 9c1edc0..934d732 100644 --- a/src/BowongModalFunctions/middleware/authorization.py +++ b/src/BowongModalFunctions/middleware/authorization.py @@ -1,7 +1,14 @@ +import httpx from fastapi import Depends, HTTPException from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer +from pydantic import EmailStr from starlette import status +from ..models.responses.models import NakamaJWTResponse +from ..models.settings.cluster import WorkerConfig + +config = WorkerConfig() + bearer_scheme = HTTPBearer(scheme_name="bearer", description="固定API Token的鉴权方式", auto_error=True) @@ -18,6 +25,17 @@ async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(beare def is_valid_token(token: str) -> bool: - # 这里实现具体的token验证逻辑 - # 例如：验证JWT token，检查数据库中的token等 - return token == "bowong7777" + # + if token == config.api_server_token: + return True + else: + response = httpx.get(url=f"{config.nakama_client_endpoint}/v2/account", + headers={"Authorization": f"Bearer {token}"}) + return response.is_success + + +def nakama_login(email: EmailStr, password: str) -> NakamaJWTResponse: + response = httpx.post(url=f"{config.nakama_client_endpoint}/v2/account/authenticate/email", + json={"email": email.__str__(), "password": password, }) + response.raise_for_status() + return NakamaJWTResponse.model_validate_json(response.text) diff --git a/src/BowongModalFunctions/models/cache_tasks/__init__.py b/src/BowongModalFunctions/models/cache_tasks/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/BowongModalFunctions/models/media_model.py b/src/BowongModalFunctions/models/cache_tasks/models.py similarity index 57% rename from src/BowongModalFunctions/models/media_model.py rename to src/BowongModalFunctions/models/cache_tasks/models.py index 36b73e4..886fb9e 100644 --- a/src/BowongModalFunctions/models/media_model.py +++ b/src/BowongModalFunctions/models/cache_tasks/models.py @@ -1,41 +1,22 @@ +import json import os import re from datetime import datetime -from enum import Enum from functools import cached_property -from typing import List, Union, Optional, Any, Dict +from typing import Optional, Union, Any, List from urllib.parse import urlparse -from pydantic import (BaseModel, Field, field_validator, ValidationError, - field_serializer, SerializationInfo, computed_field, FileUrl, Base64Str, Base64Bytes, ConfigDict) + +from pydantic import BaseModel, Field, field_validator, model_validator, ValidationError, computed_field, \ + field_serializer, Base64Bytes, RootModel from pydantic.json_schema import JsonSchemaValue -from ..config import WorkerConfig -from ..utils.TimeUtils import TimeDelta -from ..utils.VideoUtils import VideoMetadata +from pydantic_core.core_schema import SerializationInfo + +from ..enums.models import MediaProtocol, MediaCacheStatus, CacheOperationType +from ..settings.cluster import WorkerConfig +from ..ffmpeg_tasks.models import VideoMetadata config = WorkerConfig() -s3_region = config.S3_region -s3_bucket_name = config.S3_bucket_name -s3_mount_point = config.S3_mount_dir - - -class MediaProtocol(str, Enum): - http = "http" - s3 = "s3" - vod = "vod" - cos = "cos" - hls = "hls" - gs = "gs" - - -class MediaCacheStatus(str, Enum): - downloading = "downloading" - failed = "failed" - ready = "ready" - deleted = "deleted" - missing = "missing" - unknown = "unknown" - class MediaSource(BaseModel): path: str = Field(description="媒体源的路径") @@ -63,7 +44,7 @@ class MediaSource(BaseModel): elif media_url.startswith('s3://'): pattern = r"^s3://[a-z]{2}-[a-z]+-\d/.*$" if not re.match(pattern, media_url): - media_url = media_url.replace("s3://", f"s3://{s3_region}/") + media_url = media_url.replace("s3://", f"s3://{config.S3_region}/") # s3://{endpoint}/{bucket}/{url} paths = media_url[5:].split('/') @@ -139,7 +120,7 @@ class MediaSource(BaseModel): match self.protocol: case MediaProtocol.s3: # 本地挂载缓存 - if self.protocol == MediaProtocol.s3 and self.endpoint == s3_region and self.bucket == s3_bucket_name: + 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}" @@ -169,7 +150,7 @@ class MediaSource(BaseModel): @computed_field(description="本地挂载地址") @property def local_mount_path(self) -> str: - return f"{s3_mount_point}/{self.cache_filepath}" + return f"{config.S3_mount_dir}/{self.cache_filepath}" @field_serializer('expired_at') def serialize_datetime(self, value: Optional[datetime], info: SerializationInfo) -> Optional[str]: @@ -200,59 +181,56 @@ class MediaSource(BaseModel): return f"{self.protocol.value}://{self.endpoint}/{self.bucket}/{self.path}" -class MediaSources(BaseModel): - inputs: List[MediaSource] = Field(examples=[ - [ - "vod://ap-shanghai/1500034234/1397757910405340824.mp4", - "vod://ap-shanghai/1500034234/1397757910403699452.mp4", - "s3://ap-northeast-2/modal-media-cache/concat/outputs/fc-01JTPV5FCNA74CKX3N3214XJPD/output.mp4" - ] - ], description="支持多种协议['vod://', 's3://'], 计划支持['cos://', http://]") +class CacheTask(BaseModel): + type: CacheOperationType = Field(description="操作类型") + source: MediaSource = Field(description="源媒体URN") + target: Optional[MediaSource] = Field(description="目标媒体URN") - @field_validator('inputs', mode='before') + @field_validator('source', mode='before') @classmethod - def parse_inputs(cls, v: Union[str, MediaSource]) -> List[MediaSource]: - if not v: - raise ValidationError([ - { - 'loc': ('inputs',), - 'msg': "inputs为空", - 'type': 'value_error', - 'input': v - } - ], MediaSources) - result = [] - for item in v: - if isinstance(item, str): - result.append(MediaSource.from_str(item)) - elif isinstance(item, MediaSource): - result.append(item) + def parse_source(cls, v: Union[str, MediaSource]) -> MediaSource: + if isinstance(v, str): + media_source = MediaSource.from_str(v) + if media_source.protocol == MediaProtocol.s3: + return media_source else: - raise ValidationError([ - { - 'loc': ('inputs',), - 'msg': "inputs元素类型错误: 必须是字符串", - 'type': 'value_error', - 'input': v - } - ], MediaSources) - return result + raise ValidationError('media只支持s3格式的urn') + elif isinstance(v, MediaSource): + return v + else: + raise ValidationError("media格式读取失败") - model_config = { - "arbitrary_types_allowed": True - } + @field_validator('target', mode='before') + @classmethod + def parse_target(cls, v: Union[str, MediaSource]) -> Optional[MediaSource]: + if v is None: + return None + if isinstance(v, str): + media_source = MediaSource.from_str(v) + if media_source.protocol == MediaProtocol.s3: + return media_source + else: + raise ValidationError('media只支持s3格式的urn') + elif isinstance(v, MediaSource): + return v + else: + raise ValidationError("media格式读取失败") + + @model_validator(mode="after") + def parse_model(self): + match self.type: + case CacheOperationType.copy: + if self.target is None: + raise ValidationError("使用copy行为时必填target URN") + case CacheOperationType.move: + if self.target is None: + raise ValidationError("使用move行为时必填target URN") + case _: + return self -class CacheResult(BaseModel): - caches: Dict[str, MediaSource] = Field(description="Cache ID") - - -class DownloadResult(BaseModel): - urls: List[str] = Field(description="下载链接") - - -class UploadResultResponse(BaseModel): - media: MediaSource = Field(description="上传完成的媒体资源") +class CacheTaskResult(CacheTask): + success: bool = Field(default=False, description="执行成功") class Base64File(BaseModel): @@ -260,49 +238,103 @@ class Base64File(BaseModel): filename: str = Field(description="文件名, 包含文件类型后缀") content_type: str = Field(description="文件数据类型") - # @computed_field(description="Base64解码后的二进制内容") - # @property - # def content(self) -> Base64Bytes: - # return base64.b64decode(self.raw_content) - @computed_field(description="文件字节大小") @property def size(self) -> int: - # b64_str = self.raw_content.strip().replace('\n', '').replace('\r', '') - # padding = b64_str.count('=', -2) # 只统计末尾的 '=' - # size = len(b64_str) return len(self.raw_content) -class UploadBase64Request(BaseModel): - file: Base64File = Field(description="上传的文件") - prefix: Optional[str] = Field(description="文件存在的前缀目录", default=None) +class LiveProduct(BaseModel): + title: str = Field(default="", description="商品标题") + leaf_category: str = Field(default="", description="商品分类") + shop_id: int = Field(default=0, description="店铺ID") + product_id: str = Field(default="", description="商品ID") + cover: str = Field(default="", description="商品封面图链接") + detail_url: str = Field(default="", description="商品详情链接") -class UploadPresignRequest(BaseModel): - key: str = Field(description="上传文件的key", examples=['123/456/abc.mp4']) - content_type: str = Field(description="上传对象的文件类型", examples=['video/mp4']) +class LiveProductCaches(BaseModel): + room_id: str = Field(default="", description="直播间room_id/Room room_id") + author_id: str = Field(default="", description="作者id/Author id") + update_time: str = Field(default=0, description="商品列表更新时间") + count: int = Field(default=0, description="缓存直播间商品数量") + product_list: List[LiveProduct] = Field(default=None, description="缓存直播间商品列表") -class UploadPresignResponse(BaseModel): - url: str = Field(description="就近加速的PUT上传地址") - urn: str = Field(description="上传成功后获得的对应资源URN") - expired_at: datetime = Field(description="上传地址签名过期时间戳") +class VisualFeatures(BaseModel): + """ + 代表产品的视觉特征 + """ + color: str = Field( + description="商品详细颜色配色等色彩特征" + ) + pattern: str = Field( + description="商品详细图案纹理布料等材质特征, 材料可以根据产品名称和图像确定" + ) + style: str = Field( + default=None, + description="商品详细款式风格设计等有辨识度的款式特征" + ) -class UploadMultipartPresignRequest(UploadPresignRequest): - parts_count: int = Field(description="分片数量") +class ProductInfo(BaseModel): + """ + 表示从图像中提取单个产品的信息 + """ + image_order: int = Field( + description="图像出现的顺序" + ) + image_name: str = Field( + description="图像上显示的原始文本" + ) + matched_product: str = Field( + description="匹配的标准产品名称（必须与产品列表中的标准产品名称保持一致），或者如果没有匹配，则空" + ) + match_confidence: int = Field( + ge=0, + le=100, + description="产品匹配的置信度范围从0到100" + ) + visual_features: VisualFeatures = Field( + description="产品的详细视觉特征" + ) -class UploadMultipartPresignResponse(BaseModel): - urls: List[str] = Field(description="就近加速的PUT分片上传地址") - list_url: str = Field(description="用于确认分片上传状态的请求地址") - complete_url: str = Field(description="用于确认完成分片上传的请求地址") - urn: str = Field(description="上传成功后获得的对应资源URN") - expired_at: datetime = Field(description="上传地址签名过期时间戳") +class GeminiFirstStageResponseModel(RootModel): + """ + 从图像中提取的产品信息列表 + """ + root: List[ProductInfo] -class MediaCopyRequest(BaseModel): - class MediaCopyTask(BaseModel): - source: MediaSource = Field(description="源媒体") - destination: MediaSource = Field(description="") +class ProductTimeline(BaseModel): + """ + 表示单个商品及其在视频时间线中的出现信息。 + """ + product: str = Field(description="标准商品名称") + timeline: List[str] = Field( + description="该商品在视频中出现的时间段列表。每个字符串表示一个时间段，格式为 '开始时间 - 结束时间 (描述)'" + ) + + +class GeminiFirstStagePromptVariables(BaseModel): + product_list: List[str] = Field(description="商品名列表") + + @computed_field(description="xml格式排列的商品列表") + @property + def product_list_xml(self) -> str: + xml_items = [f" {product}" for product in self.product_list] + xml_string = "\n".join(xml_items) + return f"\n{xml_string}\n " + + +class GeminiSecondStagePromptVariables(BaseModel): + product_json_list: List[dict] = Field(description="识别出的商品JSON列表") + + @computed_field(description="xml格式排列的识别出的商品JSON列表") + @property + def product_json_list_xml(self) -> str: + xml_items = [f" {json.dumps(product, ensure_ascii=False)}" for product in + self.product_json_list] + xml_string = "\n".join(xml_items) + return f"\n{xml_string}\n " diff --git a/src/BowongModalFunctions/models/enums/__init__.py b/src/BowongModalFunctions/models/enums/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/BowongModalFunctions/models/enums/models.py b/src/BowongModalFunctions/models/enums/models.py new file mode 100644 index 0000000..d31d994 --- /dev/null +++ b/src/BowongModalFunctions/models/enums/models.py @@ -0,0 +1,47 @@ +from enum import Enum + + +class MediaProtocol(str, Enum): + http = "http" + s3 = "s3" + vod = "vod" + cos = "cos" + hls = "hls" + gs = "gs" + + +class MediaCacheStatus(str, Enum): + downloading = "downloading" + failed = "failed" + ready = "ready" + deleted = "deleted" + missing = "missing" + unknown = "unknown" + + +class TaskStatus(str, Enum): + running = "running" + failed = "failed" + success = "success" + expired = "expired" + + +class CacheOperationType(str, Enum): + copy = "copy" + move = "move" + delete = "delete" + + +class ErrorCode(int, Enum): + SUCCESS = 0 + PARAM_ERROR = 10001 + NOT_FOUND = 10002 + UNAUTHORIZED = 10003 + FORBIDDEN = 10004 + BUSINESS_ERROR = 10005 + SYSTEM_ERROR = 99999 + + +class WebhookMethodEnum(str, Enum): + GET = "get" + POST = "post" diff --git a/src/BowongModalFunctions/models/ffmpeg_tasks/__init__.py b/src/BowongModalFunctions/models/ffmpeg_tasks/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/BowongModalFunctions/models/ffmpeg_tasks/models.py b/src/BowongModalFunctions/models/ffmpeg_tasks/models.py new file mode 100644 index 0000000..8a4201a --- /dev/null +++ b/src/BowongModalFunctions/models/ffmpeg_tasks/models.py @@ -0,0 +1,310 @@ +import json +from typing import Union, Any, Optional, List, Dict + +from loguru import logger +from pydantic import BaseModel, Field, computed_field, field_validator, model_validator, ConfigDict, HttpUrl +from pydantic.json_schema import JsonSchemaValue + +from BowongModalFunctions.models.enums.models import WebhookMethodEnum +from BowongModalFunctions.models.settings.cluster import WorkerConfig +from BowongModalFunctions.utils.TimeUtils import TimeDelta + +config = WorkerConfig() + + +class FFMpegSliceSegment(BaseModel): + start: TimeDelta = Field( + description="视频切割的开始时间点秒数, 可为浮点小数(精确到小数点后3位,毫秒级)或者为标准格式的时间戳") + end: TimeDelta = Field( + description="视频切割的结束时间点秒数, 可为浮点小数(精确到小数点后3位,毫秒级)或者标准格式的时间戳") + + @computed_field + @property + def duration(self) -> TimeDelta: + return self.end - self.start + + @field_validator('start', mode='before') + @classmethod + def parse_start(cls, v: Union[float, str, TimeDelta]): + if isinstance(v, float): + if v < 0.0: + raise ValueError("开始时间点不能小于0") + return TimeDelta(seconds=v) + elif isinstance(v, int): + if v < 0: + raise ValueError("开始时间点不能小于0") + return TimeDelta(seconds=v) + elif isinstance(v, str): + timedelta = TimeDelta.from_format_string(v) + if timedelta.total_seconds() < 0.0: + raise ValueError("开始时间点不能小于0") + return timedelta + elif isinstance(v, TimeDelta): + if v.total_seconds() < 0.0: + raise ValueError("开始时间点不能小于0") + return v + else: + raise TypeError(v) + + @field_validator('end', mode='before') + @classmethod + def parse_end(cls, v: Union[float, TimeDelta]): + if isinstance(v, float): + if v < 0.0: + raise ValueError("结束时间点不能小于0") + return TimeDelta(seconds=v) + elif isinstance(v, int): + if v < 0: + raise ValueError("结束时间点不能小于0") + return TimeDelta(seconds=v) + elif isinstance(v, str): + timedelta = TimeDelta.from_format_string(v) + if timedelta.total_seconds() < 0.0: + raise ValueError("结束时间点不能小于0") + return timedelta + elif isinstance(v, TimeDelta): + if v.total_seconds() < 0.0: + raise ValueError("结束时间点不能小于0") + return v + else: + raise TypeError(v) + + @model_validator(mode='after') + def validate_end_after_start(self) -> 'FFMpegSliceSegment': + if self.end <= self.start: + raise ValueError("end time must be greater than start time") + return self + + @classmethod + def __get_pydantic_json_schema__(cls, core_schema: Any, handler: Any) -> JsonSchemaValue: + # Override the schema to represent it as a string + return { + "type": "object", + "properties": { + "start": { + "type": "number", + "examples": [5, 10.5, '00:00:10.500'], + }, + "end": { + "type": "number", + "examples": [8, 12.5, '00:00:12.500'], + } + }, + "required": [ + "start", + "end" + ] + } + + model_config = { + "arbitrary_types_allowed": True + } + + +class FFMPEGSliceOptions(BaseModel): + limit_size: Optional[int] = Field(default=None, description="不超过指定文件(字节)大小, 默认为空不限制输出大小") + bit_rate: Optional[int] = Field(default=None, description="指定输出视频的比特率, 不能与limit_size同时设置") + + crf: int = Field(default=16, description="输出视频的质量") + fps: int = Field(default=30, description="输出视频的FPS") + width: int = Field(default=None, description="输出视频的宽(像素)") + height: int = Field(default=None, description="输出视频的高(像素)") + + @computed_field(description="解析为字符表达式的文件大小") + @property + def pretty_limit_size(self) -> Optional[str]: + if self.limit_size is not None: + for unit in ['B', 'KB', 'MB', 'GB', 'TB']: + if self.limit_size < 1024.0: + return f"{self.limit_size:.2f}{unit}" + self.limit_size /= 1024.0 + return f"{self.limit_size:.2f}PB" + else: + return None + + @computed_field(description="解析为字符表达式的比特率") + @property + def pretty_bit_rate(self) -> Optional[str]: + if self.bit_rate is not None: + return f"{self.bit_rate}k" + + +class MediaStream(BaseModel): + duration: float = Field(0, description="时长") + codec_name: str + tags: Optional[Any] = Field(None) + + model_config = ConfigDict(extra='allow') + + +class HLSMediaVideoStream(BaseModel): + stream_type: str = "video" + codec_name: str + codec_type: str + width: int + height: int + avg_frame_rate: str + tags: Optional[Any] = Field(None) + duration: Optional[float] = Field(None) + + @computed_field + @property + def video_frame_rate(self) -> float: + numerator, denominator = map(int, self.avg_frame_rate.split('/')) + if denominator != 0: + return numerator / denominator + return 0 + + +class HLSMediaAudioStream(BaseModel): + stream_type: str = "audio" + sample_rate: str + channels: int + channel_layout: str + start_time: str + tags: Optional[Any] = Field(None) + + +class AudioStream(MediaStream): + stream_type: str = Field("audio") + codec_type: str + sample_rate: str + channels: int + tags: Optional[Any] = Field(None) + + model_config = ConfigDict(extra='allow') + + +class VideoStream(MediaStream): + stream_type: str = "video" + width: int + height: int + bit_rate: int + avg_frame_rate: str + + @computed_field + @property + def video_bitrate(self) -> str: + return str(int(self.bit_rate / 1000)) + 'k' + + @computed_field + @property + def video_frame_rate(self) -> float: + numerator, denominator = map(int, self.avg_frame_rate.split('/')) + if denominator != 0: + return numerator / denominator + return 0 + + model_config = ConfigDict(extra='allow') + + +class ImageStream(MediaStream): + stream_type: str = "image" + width: int + height: int + + model_config = ConfigDict(extra='allow') + + +class SubtitleStreamTags(BaseModel): + language: str = Field(description="内嵌字幕的语言") + + model_config = ConfigDict(extra='allow') + + +class SubtitleStream(MediaStream): + stream_type: str = "subtitle" + tags: SubtitleStreamTags + + model_config = ConfigDict(extra='allow') + + +class VideoFormat(BaseModel): + filename: str + format_name: str + start_time: float = Field(0, description="起始时间") + size: int + bit_rate: Optional[int] = Field(None, description="文件比特率") + duration: float = Field(3600 * 12, description="文件时长") + + +class VideoMetadata(BaseModel): + streams: List[ + Union[ImageStream, AudioStream, VideoStream, HLSMediaAudioStream, HLSMediaVideoStream, SubtitleStream]] = Field( + description="媒体包含的数据轨道") + format: Optional[VideoFormat] = Field(None) + + @field_validator('streams', mode='before') + def parse_streams(cls, value): + streams = [] + if isinstance(value, List): + for stream in value: + if isinstance(stream, Dict): + logger.info(f"Parsing stream : {json.dumps(stream, ensure_ascii=False)}") + if stream.get("codec_type") == 'audio': + if stream.get("duration") is None: + logger.info("Parsing audio stream") + hls_audio = HLSMediaAudioStream.model_validate(stream) + streams.append(hls_audio) + else: + logger.info("Parsing hls audio stream") + audio = AudioStream.model_validate(stream) + streams.append(audio) + elif stream.get("codec_type") == 'video': + if stream.get("codec_name") in ("gif", "png", "mjpg", "mjpeg", "webp"): + logger.info("Parsing image stream") + image = ImageStream.model_validate(stream) + streams.append(image) + else: + if stream.get("duration") is None: + logger.info("Parsing hls video stream") + hls_video = HLSMediaVideoStream.model_validate(stream) + streams.append(hls_video) + else: + logger.info("Parsing video stream") + video = VideoStream.model_validate(stream) + streams.append(video) + elif stream.get("codec_type") == 'subtitle': + logger.info("Parsing subtitle stream") + subtitle_stream = SubtitleStream.model_validate(stream) + streams.append(subtitle_stream) + return streams + else: + raise TypeError + + model_config = ConfigDict(extra='allow') + + +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): + x_trace_id: str = Field(description="Sentry Transaction ID") + x_baggage: str = Field(description="Sentry Transaction baggage") + + +class WebhookNotify(BaseModel): + endpoint: HttpUrl = Field(description="Webhook回调端点", examples=["https://webhook.example.com?query=123"]) + method: WebhookMethodEnum = Field( + description="Webhook回调请求方法, 使用POST方法时body与查询ffmpeg任务所获得的json body格式一致") + headers: Optional[Dict[str, str]] = Field(description="Webhook回调附带的Headers", default=None) + + model_config = ConfigDict(json_schema_extra={ + "description": "Webhook返回值与查询ffmpeg任务所获得的json格式一致" + }) + + +class FFMPEGResult(BaseModel): + urn: str = Field(description="FFMPEG任务结果urn") + content_length: int = Field(description="媒体资源文件字节大小(Byte)") + metadata: VideoMetadata = Field(description="媒体元数据") + + @computed_field(description="可通过CDN访问的资源链接") + @property + def url(self) -> str: + prefix = f"s3://{config.S3_region}/{config.S3_bucket_name}" + if not self.urn.startswith(prefix): + raise ValueError("无法转换非s3前缀协议和非当前挂载点的s3协议") + return self.urn.replace(prefix, f"{config.S3_cdn_endpoint}") diff --git a/src/BowongModalFunctions/models/ffmpeg_worker_model.py b/src/BowongModalFunctions/models/ffmpeg_worker_model.py deleted file mode 100644 index ed28886..0000000 --- a/src/BowongModalFunctions/models/ffmpeg_worker_model.py +++ /dev/null @@ -1,93 +0,0 @@ -from typing import Union, Any -from pydantic import BaseModel, Field, computed_field, field_validator, model_validator -from pydantic.json_schema import JsonSchemaValue -from ..utils.TimeUtils import TimeDelta - - -class FFMpegSliceSegment(BaseModel): - start: TimeDelta = Field( - description="视频切割的开始时间点秒数, 可为浮点小数(精确到小数点后3位,毫秒级)或者为标准格式的时间戳") - end: TimeDelta = Field( - description="视频切割的结束时间点秒数, 可为浮点小数(精确到小数点后3位,毫秒级)或者标准格式的时间戳") - - @computed_field - @property - def duration(self) -> TimeDelta: - return self.end - self.start - - @field_validator('start', mode='before') - @classmethod - def parse_start(cls, v: Union[float, str, TimeDelta]): - if isinstance(v, float): - if v < 0.0: - raise ValueError("开始时间点不能小于0") - return TimeDelta(seconds=v) - elif isinstance(v, int): - if v < 0: - raise ValueError("开始时间点不能小于0") - return TimeDelta(seconds=v) - elif isinstance(v, str): - timedelta = TimeDelta.from_format_string(v) - if timedelta.total_seconds() < 0.0: - raise ValueError("开始时间点不能小于0") - return timedelta - elif isinstance(v, TimeDelta): - if v.total_seconds() < 0.0: - raise ValueError("开始时间点不能小于0") - return v - else: - raise TypeError(v) - - @field_validator('end', mode='before') - @classmethod - def parse_end(cls, v: Union[float, TimeDelta]): - if isinstance(v, float): - if v < 0.0: - raise ValueError("结束时间点不能小于0") - return TimeDelta(seconds=v) - elif isinstance(v, int): - if v < 0: - raise ValueError("结束时间点不能小于0") - return TimeDelta(seconds=v) - elif isinstance(v, str): - timedelta = TimeDelta.from_format_string(v) - if timedelta.total_seconds() < 0.0: - raise ValueError("结束时间点不能小于0") - return timedelta - elif isinstance(v, TimeDelta): - if v.total_seconds() < 0.0: - raise ValueError("结束时间点不能小于0") - return v - else: - raise TypeError(v) - - @model_validator(mode='after') - def validate_end_after_start(self) -> 'FFMpegSliceSegment': - if self.end <= self.start: - raise ValueError("end time must be greater than start time") - return self - - @classmethod - def __get_pydantic_json_schema__(cls, core_schema: Any, handler: Any) -> JsonSchemaValue: - # Override the schema to represent it as a string - return { - "type": "object", - "properties": { - "start": { - "type": "number", - "examples": [5, 10.5, '00:00:10.500'], - }, - "end": { - "type": "number", - "examples": [8, 12.5, '00:00:12.500'], - } - }, - "required": [ - "start", - "end" - ] - } - - model_config = { - "arbitrary_types_allowed": True - } diff --git a/src/BowongModalFunctions/models/requests/__init__.py b/src/BowongModalFunctions/models/requests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/BowongModalFunctions/models/web_model.py b/src/BowongModalFunctions/models/requests/models.py similarity index 52% rename from src/BowongModalFunctions/models/web_model.py rename to src/BowongModalFunctions/models/requests/models.py index c9e2f5e..9c16611 100644 --- a/src/BowongModalFunctions/models/web_model.py +++ b/src/BowongModalFunctions/models/requests/models.py @@ -1,160 +1,14 @@ -import json import uuid -from datetime import datetime -from enum import Enum -from typing import List, Union, Optional, Dict, Any +from typing import List, Union, Optional, Dict import pydantic -from pydantic import BaseModel, Field, field_validator, ConfigDict, HttpUrl, computed_field, RootModel, root_validator, \ - model_validator +from pydantic import BaseModel, Field, ConfigDict, field_validator, ValidationError, model_validator, computed_field, \ + EmailStr -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() - - -class TaskStatus(str, Enum): - running = "running" - failed = "failed" - success = "success" - expired = "expired" - - -class CacheOperationType(str, Enum): - copy = "copy" - move = "move" - delete = "delete" - - -class ErrorCode(int, Enum): - SUCCESS = 0 - PARAM_ERROR = 10001 - NOT_FOUND = 10002 - UNAUTHORIZED = 10003 - FORBIDDEN = 10004 - BUSINESS_ERROR = 10005 - SYSTEM_ERROR = 99999 - - -class WebhookMethodEnum(str, Enum): - GET = "get" - POST = "post" - - -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): - x_trace_id: str = Field(description="Sentry Transaction ID") - x_baggage: str = Field(description="Sentry Transaction baggage") - - -class FFMPEGSliceTaskStatusRequest(BaseModel): - taskId: str = Field(description="任务Id") - - -class ModalTaskResponse(BaseModel): - success: bool = Field(description="任务接受成功") - taskId: str = Field(description="任务Id") - - -class CacheDeleteTaskResponse(BaseModel): - success: bool = Field(description="运行成功") - keys: List[str] = Field(description="成功从KV和S3删除掉的URN") - non_kv_keys: List[str] = Field(alias="nonKVKeys", serialization_alias="nonKVKeys", - description="成功从S3删除的URN, 不存在于KV中") - not_found_keys: List[str] = Field(alias="notFoundKeys", serialization_alias="notFoundKeys", - description="不存在的URN") - - -class RecordingTaskResponse(BaseModel): - success: bool = Field(description="任务接受成功") - taskId: str = Field(description="任务Id") - manifest: str = Field(description="播放地址") - manifest_urn: str = Field(description="播放列表URN") - - -class WebhookNotify(BaseModel): - endpoint: HttpUrl = Field(description="Webhook回调端点", examples=["https://webhook.example.com?query=123"]) - method: WebhookMethodEnum = Field( - description="Webhook回调请求方法, 使用POST方法时body与查询ffmpeg任务所获得的json body格式一致") - headers: Optional[Dict[str, str]] = Field(description="Webhook回调附带的Headers", default=None) - - model_config = ConfigDict(json_schema_extra={ - "description": "Webhook返回值与查询ffmpeg任务所获得的json格式一致" - }) - - -class FFMPEGResult(BaseModel): - urn: str = Field(description="FFMPEG任务结果urn") - content_length: int = Field(description="媒体资源文件字节大小(Byte)") - metadata: VideoMetadata = Field(description="媒体元数据") - - @computed_field(description="可通过CDN访问的资源链接") - @property - def url(self) -> str: - prefix = f"s3://{config.S3_region}/{config.S3_bucket_name}" - if not self.urn.startswith(prefix): - raise ValueError("无法转换非s3前缀协议和非当前挂载点的s3协议") - return self.urn.replace(prefix, f"{config.S3_cdn_endpoint}") - - -class CacheTask(BaseModel): - type: CacheOperationType = Field(description="操作类型") - source: MediaSource = Field(description="源媒体URN") - target: Optional[MediaSource] = Field(description="目标媒体URN") - - @field_validator('source', mode='before') - @classmethod - def parse_source(cls, v: Union[str, MediaSource]) -> MediaSource: - if isinstance(v, str): - media_source = MediaSource.from_str(v) - if media_source.protocol == MediaProtocol.s3: - return media_source - else: - raise pydantic.ValidationError('media只支持s3格式的urn') - elif isinstance(v, MediaSource): - return v - else: - raise pydantic.ValidationError("media格式读取失败") - - @field_validator('target', mode='before') - @classmethod - def parse_target(cls, v: Union[str, MediaSource]) -> Optional[MediaSource]: - if v is None: - return None - if isinstance(v, str): - media_source = MediaSource.from_str(v) - if media_source.protocol == MediaProtocol.s3: - return media_source - else: - raise pydantic.ValidationError('media只支持s3格式的urn') - elif isinstance(v, MediaSource): - return v - else: - raise pydantic.ValidationError("media格式读取失败") - - @model_validator(mode="after") - def parse_model(self): - match self.type: - case CacheOperationType.copy: - if self.target is None: - raise pydantic.ValidationError("使用copy行为时必填target URN") - case CacheOperationType.move: - if self.target is None: - raise pydantic.ValidationError("使用move行为时必填target URN") - case _: - return self - - -class CacheTaskResult(CacheTask): - success: bool = Field(default=False, description="执行成功") +from ..cache_tasks.models import CacheTask, MediaSource, Base64File +from ..enums.models import MediaProtocol +from ..ffmpeg_tasks.models import WebhookNotify, FFMpegSliceSegment, FFMPEGSliceOptions +from ...utils.TimeUtils import TimeDelta class ClusterCacheBatchRequest(BaseModel): @@ -163,48 +17,77 @@ class ClusterCacheBatchRequest(BaseModel): model_config = ConfigDict() -class ClusterCacheBatchResponse(BaseModel): - results: List[CacheTaskResult] = Field(description="批量操作任务结果") +class MediaSourcesRequest(BaseModel): + inputs: List[MediaSource] = Field(examples=[ + [ + "vod://ap-shanghai/1500034234/1397757910405340824.mp4", + "vod://ap-shanghai/1500034234/1397757910403699452.mp4", + "s3://ap-northeast-2/modal-media-cache/concat/outputs/fc-01JTPV5FCNA74CKX3N3214XJPD/output.mp4" + ] + ], description="支持多种协议['vod://', 's3://'], 计划支持['cos://', http://]") + + @field_validator('inputs', mode='before') + @classmethod + def parse_inputs(cls, v: Union[str, MediaSource]) -> List[MediaSource]: + if not v: + raise ValidationError([ + { + 'loc': ('inputs',), + 'msg': "inputs为空", + 'type': 'value_error', + 'input': v + } + ], MediaSourcesRequest) + result = [] + for item in v: + if isinstance(item, str): + result.append(MediaSource.from_str(item)) + elif isinstance(item, MediaSource): + result.append(item) + else: + raise ValidationError([ + { + 'loc': ('inputs',), + 'msg': "inputs元素类型错误: 必须是字符串", + 'type': 'value_error', + 'input': v + } + ], MediaSourcesRequest) + return result + + model_config = { + "arbitrary_types_allowed": True + } + + +class UploadBase64Request(BaseModel): + file: Base64File = Field(description="上传的文件") + prefix: Optional[str] = Field(description="文件存在的前缀目录", default=None) + + +class UploadPresignRequest(BaseModel): + key: str = Field(description="上传文件的key", examples=['123/456/abc.mp4']) + content_type: str = Field(description="上传对象的文件类型", examples=['video/mp4']) + + +class UploadMultipartPresignRequest(UploadPresignRequest): + parts_count: int = Field(description="分片数量") + + +class MediaCopyRequest(BaseModel): + class MediaCopyTask(BaseModel): + source: MediaSource = Field(description="源媒体") + destination: MediaSource = Field(description="") + + +class FFMPEGSliceTaskStatusRequest(BaseModel): + taskId: str = Field(description="任务Id") class BaseFFMPEGTaskRequest(BaseModel): webhook: Optional[WebhookNotify] = Field(description="Task webhook", default=None) -class BaseFFMPEGTaskStatusResponse(BaseModel): - taskId: str = Field(description="任务Id") - status: TaskStatus = Field(description="任务运行状态") - error: Optional[str] = Field(description="任务错误原因", default=None) - code: Optional[int] = Field(description="任务错误原因代码", default=None) - task_type: str = Field(description="任务类型", default="unknown") - results: Optional[List[Union[FFMPEGResult, Any]]] = Field(description="任务运行结果", default=None) - - model_config = ConfigDict(extra='ignore') - - -class FFMPEGConvertStreamRequest(BaseFFMPEGTaskRequest): - media: MediaSource = Field(description="待转换的媒体源") - options: FFMPEGSliceOptions = Field(default_factory=FFMPEGSliceOptions, description="输出质量选项") - - @field_validator('media', mode='before') - @classmethod - def parse_inputs(cls, v: Union[str, MediaSource]): - if isinstance(v, str): - media_source = MediaSource.from_str(v) - if media_source.protocol == MediaProtocol.hls: - return media_source - else: - raise pydantic.ValidationError('media只支持hls格式的urn') - elif isinstance(v, MediaSource): - return v - else: - raise pydantic.ValidationError("media格式读取失败") - - -class FFMPEGConvertStreamResponse(BaseFFMPEGTaskStatusResponse): - result: Optional[str] = Field(default=None, description="任务运行结果") - - class FFMPEGSliceRequest(BaseFFMPEGTaskRequest): media: MediaSource = Field(description="待切割的媒体源") markers: List[FFMpegSliceSegment] = Field(description="按照时间顺序排序过的切割标记数组") @@ -221,16 +104,8 @@ class FFMPEGSliceRequest(BaseFFMPEGTaskRequest): raise TypeError(v) -class FFMPEGSliceTaskStatusResponse(BaseFFMPEGTaskStatusResponse): - result: Optional[List[str]] = Field(default=None, description="任务运行结果") - - class FFMPEGConcatRequest(BaseFFMPEGTaskRequest): - medias: MediaSources = Field(description="待合并的媒体源") - - -class FFMPEGConcatTaskStatusResponse(BaseFFMPEGTaskStatusResponse): - result: Optional[str] = Field(default=None, description="任务运行结果") + medias: MediaSourcesRequest = Field(description="待合并的媒体源") class FFMPEGExtractAudioRequest(BaseFFMPEGTaskRequest): @@ -247,10 +122,6 @@ class FFMPEGExtractAudioRequest(BaseFFMPEGTaskRequest): raise TypeError(v) -class FFMPEGExtractAudioTaskStatusResponse(BaseFFMPEGTaskStatusResponse): - result: Optional[str] = Field(default=None, description="任务运行结果") - - class FFMPEGCornerMirrorRequest(BaseFFMPEGTaskRequest): media: MediaSource = Field(description="需要处理的媒体源") @@ -265,10 +136,6 @@ class FFMPEGCornerMirrorRequest(BaseFFMPEGTaskRequest): raise TypeError(v) -class FFMPEGCornerMirrorTaskStatusResponse(BaseFFMPEGTaskStatusResponse): - result: Optional[str] = Field(default=None, description="任务运行结果") - - class FFMPEGBgmRequest(BaseFFMPEGTaskRequest): media: MediaSource = Field(description="需要处理的媒体源") bgm_media: MediaSource = Field(description="添加的BGM媒体源", alias="bgmMedia") @@ -296,10 +163,6 @@ class FFMPEGBgmRequest(BaseFFMPEGTaskRequest): raise TypeError(v) -class FFMPEGBgmTaskStatusResponse(BaseFFMPEGTaskStatusResponse): - result: Optional[str] = Field(default=None, description="生成结果的URN") - - class FFMPEGZoomLoopRequest(BaseFFMPEGTaskRequest): media: MediaSource = Field(description="需要处理的媒体源") duration: float = Field(description="放大缩小一个循环的持续时间秒数", default=6.0) @@ -316,10 +179,6 @@ class FFMPEGZoomLoopRequest(BaseFFMPEGTaskRequest): raise TypeError(v) -class FFMPEGZoomLoopTaskStatusResponse(BaseFFMPEGTaskStatusResponse): - result: Optional[str] = Field(default=None, description="生成结果的URN") - - class FFMPEGOverlayGifRequest(BaseFFMPEGTaskRequest): media: MediaSource = Field(description="需要处理的媒体源") gif: MediaSource = Field(description="叠加的特效gif") @@ -348,10 +207,6 @@ class FFMPEGOverlayGifRequest(BaseFFMPEGTaskRequest): raise TypeError(v) -class FFMPEGOverlayGifTaskStatusResponse(BaseFFMPEGTaskStatusResponse): - result: Optional[str] = Field(default=None, description="生成结果的URN") - - class FFMPEGSubtitleOverlayRequest(BaseFFMPEGTaskRequest): media: MediaSource = Field(description="需要处理的媒体源") subtitle: Optional[MediaSource] = Field(default=None, description="需要叠加的字幕文件") @@ -422,10 +277,6 @@ class FFMPEGSubtitleOverlayRequest(BaseFFMPEGTaskRequest): return self -class FFMPEGSubtitleTaskStatusResponse(BaseFFMPEGTaskStatusResponse): - result: Optional[str] = Field(default=None, description="生成结果的URN") - - class FFMPEGMixBgmWithNoiseReduceRequest(BaseFFMPEGTaskRequest): media: MediaSource = Field(description="需要处理的媒体源") bgm: MediaSource = Field(description="需要添加的BGM媒体源") @@ -467,19 +318,11 @@ class FFMPEGMixBgmWithNoiseReduceRequest(BaseFFMPEGTaskRequest): raise TypeError(v) -class FFMPEGMixBgmWithNoiseReduceStatusResponse(BaseFFMPEGTaskStatusResponse): - result: Optional[str] = Field(default=None, description="生成结果的URN") - - class FFMPEGVideoLoopFillAudioRequest(BaseFFMPEGTaskRequest): video: MediaSource = Field(description="用来填充的视频素材") audio: MediaSource = Field(description="被填充的音频素材") -class FFMPEGVideoLoopFillAudioResponse(BaseFFMPEGTaskStatusResponse): - result: Optional[str] = Field(default=None, description="生成结果的URN") - - class FFMPEGExtractFrameRequest(BaseFFMPEGTaskRequest): video: MediaSource = Field(description="提取帧画面的来源") seek_time: Optional[Union[str, int, float]] = Field(default=None, description="先跳转到视频对应时间再取首帧", @@ -515,23 +358,6 @@ class FFMPEGExtractFrameRequest(BaseFFMPEGTaskRequest): } -class FFMPEGExtractFrameStatusResponse(BaseFFMPEGTaskStatusResponse): - result: Optional[str] = Field(default=None, description="生成结果的URN") - - -class ModalTaskCancelResponse(BaseModel): - success: bool = Field(description="成功取消任务") - error: Optional[str] = Field(default=None, description="失败原因") - - -class ComfyTaskStatusResponse(BaseModel): - taskId: str = Field(description="任务Id") - status: TaskStatus = Field(description="任务运行状态") - error: Optional[str] = Field(description="任务错误原因", default=None) - code: Optional[int] = Field(description="任务错误原因代码", default=None) - result: Optional[str] = Field(description="任务运行结果", default=None) - - class ComfyTaskRequest(BaseFFMPEGTaskRequest): video_path: MediaSource = Field( default=None, description="视频源") @@ -590,40 +416,12 @@ class GeminiRequest(BaseFFMPEGTaskRequest): raise pydantic.ValidationError("media格式读取失败") -class GeminiResultResponse(BaseFFMPEGTaskStatusResponse): - result: str = Field(default="", description="推理出的json") - - -class LiveProduct(BaseModel): - title: str = Field(default="", description="商品标题") - leaf_category: str = Field(default="", description="商品分类") - shop_id: int = Field(default=0, description="店铺ID") - product_id: str = Field(default="", description="商品ID") - cover: str = Field(default="", description="商品封面图链接") - detail_url: str = Field(default="", description="商品详情链接") - - -class LiveProductCaches(BaseModel): - room_id: str = Field(default="", description="直播间room_id/Room room_id") - author_id: str = Field(default="", description="作者id/Author id") - update_time: str = Field(default=0, description="商品列表更新时间") - count: int = Field(default=0, description="缓存直播间商品数量") - product_list: List[LiveProduct] = Field(default=None, description="缓存直播间商品列表") - - class MonitorLiveRoomProductRequest(BaseModel): cookie: str = Field(default="YOUR_COOKIE", description="用户网页版抖音Cookie/Your web version of Douyin Cookie") room_id: str = Field(default="", description="直播间room_id/Room room_id") author_id: str = Field(default="", description="作者id/Author id") -class LiveRoomProductCachesResponse(BaseModel): - 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\" 键") @@ -634,115 +432,25 @@ class MakeGridGeminiRequest(BaseFFMPEGTaskRequest): separator: int = Field(default=12, description="分割线宽度/像素") -class GoogleUploadResponse(BaseModel): - kind: str - id: str - self_link: HttpUrl = Field(alias='selfLink') - media_link: HttpUrl = Field(alias='mediaLink') - name: str - bucket: str - generation: str - metageneration: str - content_type: str = Field(alias='contentType') - storage_class: str = Field(alias='storageClass') - size: int - md5_hash: str = Field(alias='md5Hash') - crc32c: str - etag: str - time_created: datetime = Field(alias='timeCreated') - updated: datetime - time_storage_class_updated: datetime = Field(alias='timeStorageClassUpdated') - time_finalized: datetime = Field(alias='timeFinalized') +class FFMPEGConvertStreamRequest(BaseFFMPEGTaskRequest): + media: MediaSource = Field(description="待转换的媒体源") + options: FFMPEGSliceOptions = Field(default_factory=FFMPEGSliceOptions, description="输出质量选项") - @computed_field(description="适用于Google内部服务的URN") - @property - def urn(self) -> str: - return ("gs://" + self.id).replace(f"/{self.generation}", "") - - model_config = ConfigDict(populate_by_name=True) + @field_validator('media', mode='before') + @classmethod + def parse_inputs(cls, v: Union[str, MediaSource]): + if isinstance(v, str): + media_source = MediaSource.from_str(v) + if media_source.protocol == MediaProtocol.hls: + return media_source + else: + raise pydantic.ValidationError('media只支持hls格式的urn') + elif isinstance(v, MediaSource): + return v + else: + raise pydantic.ValidationError("media格式读取失败") -class VisualFeatures(BaseModel): - """ - 代表产品的视觉特征 - """ - color: str = Field( - description="商品详细颜色配色等色彩特征" - ) - pattern: str = Field( - description="商品详细图案纹理布料等材质特征, 材料可以根据产品名称和图像确定" - ) - style: str = Field( - default=None, - description="商品详细款式风格设计等有辨识度的款式特征" - ) - - -class ProductInfo(BaseModel): - """ - 表示从图像中提取单个产品的信息 - """ - image_order: int = Field( - description="图像出现的顺序" - ) - image_name: str = Field( - description="图像上显示的原始文本" - ) - matched_product: str = Field( - description="匹配的标准产品名称（必须与产品列表中的标准产品名称保持一致），或者如果没有匹配，则空" - ) - match_confidence: int = Field( - ge=0, - le=100, - description="产品匹配的置信度范围从0到100" - ) - visual_features: VisualFeatures = Field( - description="产品的详细视觉特征" - ) - - -class GeminiFirstStageResponseModel(RootModel): - """ - 从图像中提取的产品信息列表 - """ - root: List[ProductInfo] - - -class ProductTimeline(BaseModel): - """ - 表示单个商品及其在视频时间线中的出现信息。 - """ - product: str = Field(description="标准商品名称") - timeline: List[str] = Field( - description="该商品在视频中出现的时间段列表。每个字符串表示一个时间段，格式为 '开始时间 - 结束时间 (描述)'" - ) - - -class GeminiSecondStageResponseModel(RootModel): - """ - 表示一个包含多个商品及其时间线信息的列表。 - """ - root: List[ProductTimeline] - - -class GeminiFirstStagePromptVariables(BaseModel): - product_list: List[str] = Field(description="商品名列表") - - @computed_field(description="xml格式排列的商品列表") - @property - def product_list_xml(self) -> str: - xml_items = [f" {product}" for product in self.product_list] - xml_string = "\n".join(xml_items) - return f"\n{xml_string}\n " - - -class GeminiSecondStagePromptVariables(BaseModel): - product_json_list: List[dict] = Field(description="识别出的商品JSON列表") - - @computed_field(description="xml格式排列的识别出的商品JSON列表") - @property - def product_json_list_xml(self) -> str: - xml_items = [f" {json.dumps(product, ensure_ascii=False)}" for product in - self.product_json_list] - xml_string = "\n".join(xml_items) - return f"\n{xml_string}\n " +class NakamaLogin(BaseModel): + email: EmailStr = Field(description="Bowong Echo(Nakama)的登录账号") + password: str = Field(description="Bowong Echo(Nakama)的登录密码") diff --git a/src/BowongModalFunctions/models/responses/__init__.py b/src/BowongModalFunctions/models/responses/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/BowongModalFunctions/models/responses/models.py b/src/BowongModalFunctions/models/responses/models.py new file mode 100644 index 0000000..9b0e4b3 --- /dev/null +++ b/src/BowongModalFunctions/models/responses/models.py @@ -0,0 +1,182 @@ +from datetime import datetime +from typing import List, Dict, Optional, Union, Any + +from pydantic import BaseModel, Field, ConfigDict, HttpUrl, computed_field, RootModel + +from ..cache_tasks.models import MediaSource, CacheTaskResult, ProductTimeline +from ..enums.models import TaskStatus +from ..ffmpeg_tasks.models import FFMPEGResult + + +class DownloadResult(BaseModel): + urls: List[str] = Field(description="下载链接") + + +class CacheResultResponse(BaseModel): + caches: Dict[str, MediaSource] = Field(description="Cache ID") + + +class UploadResultResponse(BaseModel): + media: MediaSource = Field(description="上传完成的媒体资源") + + +class UploadPresignResponse(BaseModel): + url: str = Field(description="就近加速的PUT上传地址") + urn: str = Field(description="上传成功后获得的对应资源URN") + expired_at: datetime = Field(description="上传地址签名过期时间戳") + + +class UploadMultipartPresignResponse(BaseModel): + urls: List[str] = Field(description="就近加速的PUT分片上传地址") + list_url: str = Field(description="用于确认分片上传状态的请求地址") + complete_url: str = Field(description="用于确认完成分片上传的请求地址") + urn: str = Field(description="上传成功后获得的对应资源URN") + expired_at: datetime = Field(description="上传地址签名过期时间戳") + + +class ModalTaskResponse(BaseModel): + success: bool = Field(description="任务接受成功") + taskId: str = Field(description="任务Id") + + +class CacheDeleteTaskResponse(BaseModel): + success: bool = Field(description="运行成功") + keys: List[str] = Field(description="成功从KV和S3删除掉的URN") + non_kv_keys: List[str] = Field(alias="nonKVKeys", serialization_alias="nonKVKeys", + description="成功从S3删除的URN, 不存在于KV中") + not_found_keys: List[str] = Field(alias="notFoundKeys", serialization_alias="notFoundKeys", + description="不存在的URN") + + +class RecordingTaskResponse(BaseModel): + success: bool = Field(description="任务接受成功") + taskId: str = Field(description="任务Id") + manifest: str = Field(description="播放地址") + manifest_urn: str = Field(description="播放列表URN") + + +class ClusterCacheBatchResponse(BaseModel): + results: List[CacheTaskResult] = Field(description="批量操作任务结果") + + +class BaseFFMPEGTaskStatusResponse(BaseModel): + taskId: str = Field(description="任务Id") + status: TaskStatus = Field(description="任务运行状态") + error: Optional[str] = Field(description="任务错误原因", default=None) + code: Optional[int] = Field(description="任务错误原因代码", default=None) + task_type: str = Field(description="任务类型", default="unknown") + results: Optional[List[Union[FFMPEGResult, Any]]] = Field(description="任务运行结果", default=None) + + model_config = ConfigDict(extra='ignore') + + +class FFMPEGConvertStreamResponse(BaseFFMPEGTaskStatusResponse): + result: Optional[str] = Field(default=None, description="任务运行结果") + + +class FFMPEGSliceTaskStatusResponse(BaseFFMPEGTaskStatusResponse): + result: Optional[List[str]] = Field(default=None, description="任务运行结果") + + +class FFMPEGConcatTaskStatusResponse(BaseFFMPEGTaskStatusResponse): + result: Optional[str] = Field(default=None, description="任务运行结果") + + +class FFMPEGExtractAudioTaskStatusResponse(BaseFFMPEGTaskStatusResponse): + result: Optional[str] = Field(default=None, description="任务运行结果") + + +class FFMPEGCornerMirrorTaskStatusResponse(BaseFFMPEGTaskStatusResponse): + result: Optional[str] = Field(default=None, description="任务运行结果") + + +class FFMPEGBgmTaskStatusResponse(BaseFFMPEGTaskStatusResponse): + result: Optional[str] = Field(default=None, description="生成结果的URN") + + +class FFMPEGZoomLoopTaskStatusResponse(BaseFFMPEGTaskStatusResponse): + result: Optional[str] = Field(default=None, description="生成结果的URN") + + +class FFMPEGOverlayGifTaskStatusResponse(BaseFFMPEGTaskStatusResponse): + result: Optional[str] = Field(default=None, description="生成结果的URN") + + +class FFMPEGSubtitleTaskStatusResponse(BaseFFMPEGTaskStatusResponse): + result: Optional[str] = Field(default=None, description="生成结果的URN") + + +class FFMPEGMixBgmWithNoiseReduceStatusResponse(BaseFFMPEGTaskStatusResponse): + result: Optional[str] = Field(default=None, description="生成结果的URN") + + +class FFMPEGVideoLoopFillAudioResponse(BaseFFMPEGTaskStatusResponse): + result: Optional[str] = Field(default=None, description="生成结果的URN") + + +class FFMPEGExtractFrameStatusResponse(BaseFFMPEGTaskStatusResponse): + result: Optional[str] = Field(default=None, description="生成结果的URN") + + +class ModalTaskCancelResponse(BaseModel): + success: bool = Field(description="成功取消任务") + error: Optional[str] = Field(default=None, description="失败原因") + + +class ComfyTaskStatusResponse(BaseModel): + taskId: str = Field(description="任务Id") + status: TaskStatus = Field(description="任务运行状态") + error: Optional[str] = Field(description="任务错误原因", default=None) + code: Optional[int] = Field(description="任务错误原因代码", default=None) + result: Optional[str] = Field(description="任务运行结果", default=None) + + +class GeminiResultResponse(BaseFFMPEGTaskStatusResponse): + result: str = Field(default="", description="推理出的json") + + +class LiveRoomProductCachesResponse(BaseModel): + 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 GoogleUploadResponse(BaseModel): + kind: str + id: str + self_link: HttpUrl = Field(alias='selfLink') + media_link: HttpUrl = Field(alias='mediaLink') + name: str + bucket: str + generation: str + metageneration: str + content_type: str = Field(alias='contentType') + storage_class: str = Field(alias='storageClass') + size: int + md5_hash: str = Field(alias='md5Hash') + crc32c: str + etag: str + time_created: datetime = Field(alias='timeCreated') + updated: datetime + time_storage_class_updated: datetime = Field(alias='timeStorageClassUpdated') + time_finalized: datetime = Field(alias='timeFinalized') + + @computed_field(description="适用于Google内部服务的URN") + @property + def urn(self) -> str: + return ("gs://" + self.id).replace(f"/{self.generation}", "") + + model_config = ConfigDict(populate_by_name=True) + + +class GeminiSecondStageResponseModel(RootModel): + """ + 表示一个包含多个商品及其时间线信息的列表。 + """ + root: List[ProductTimeline] + + +class NakamaJWTResponse(BaseModel): + token: str = Field(description="Nakama JWT Token") + refresh_token: str = Field(description="Nakama JWT Refresh Token") diff --git a/src/BowongModalFunctions/models/settings/__init__.py b/src/BowongModalFunctions/models/settings/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/BowongModalFunctions/config.py b/src/BowongModalFunctions/models/settings/cluster.py similarity index 92% rename from src/BowongModalFunctions/config.py rename to src/BowongModalFunctions/models/settings/cluster.py index c167d32..aebe21f 100644 --- a/src/BowongModalFunctions/config.py +++ b/src/BowongModalFunctions/models/settings/cluster.py @@ -1,4 +1,5 @@ from typing import Optional, Any + from pydantic import Field, HttpUrl from pydantic_settings import BaseSettings, SettingsConfigDict @@ -20,7 +21,6 @@ class WorkerConfig(BaseSettings): modal_product_kv_name: str = Field(default='live-product-cache', description="Modal抖音直播间商品缓存KV库") modal_environment: str = Field(default="dev", description="Modal worker运行环境") modal_app_name: str = Field(default='bowong-ai-video', description="Modal App集群名称") - modal_is_local: bool = Field(default=False, description="本地开发环境") comfyui_s3_input: Optional[str] = Field(default="comfyui-input", description="ComfyUI input S3文件夹名") comfyui_s3_output: Optional[str] = Field(default="comfyui-output", description="ComfyUI output S3文件夹名") @@ -29,8 +29,10 @@ class WorkerConfig(BaseSettings): description="Langfuse public key") langfuse_secret_key: str = Field(default="sk-lf-dd20cb0b-ef2e-49f6-80f0-b2d9cff1bb11", description="Langfuse secret key") + nakama_client_endpoint: str = Field(default="http://43.143.58.201:7350", description="Nakama client endpoint") api_version: str = Field(default="0.1.7", description="API接口版本") + api_server_token: str = Field(default="bowong7777", description="固定的API调用Bearer Token") modal_config: Any = SettingsConfigDict(json_schema_extra={ "description": "可通过本地环境变量加载对应Field, 不区分大小写, Modal创建App的Image时可通过dotenv加载指定.env文件写入到Docker Image的系统变量", diff --git a/src/BowongModalFunctions/router/cache.py b/src/BowongModalFunctions/router/cache.py index f2adddd..2f648b7 100644 --- a/src/BowongModalFunctions/router/cache.py +++ b/src/BowongModalFunctions/router/cache.py @@ -13,22 +13,17 @@ from fastapi.responses import JSONResponse, RedirectResponse from starlette import status import boto3 from botocore.config import Config -from typing_inspection.typing_objects import target -from ..config import WorkerConfig +from ..models.settings.cluster import WorkerConfig from ..middleware.authorization import verify_token -from ..models.media_model import (MediaSources, - CacheResult, - MediaSource, - MediaCacheStatus, - DownloadResult, - UploadResultResponse, - UploadBase64Request, UploadPresignRequest, UploadPresignResponse, - UploadMultipartPresignRequest, UploadMultipartPresignResponse, MediaProtocol - ) -from ..models.web_model import SentryTransactionInfo, MonitorLiveRoomProductRequest, ModalTaskResponse, \ - LiveRoomProductCachesResponse, CacheDeleteTaskResponse, ClusterCacheBatchRequest, CacheOperationType, \ - ClusterCacheBatchResponse, CacheTaskResult +from ..models.responses.models import DownloadResult, CacheResultResponse, UploadResultResponse, UploadPresignResponse, \ + UploadMultipartPresignResponse, CacheDeleteTaskResponse, ClusterCacheBatchResponse, LiveRoomProductCachesResponse +from ..models.enums.models import MediaCacheStatus, CacheOperationType +from ..models.ffmpeg_tasks.models import SentryTransactionInfo +from ..models.requests.models import ClusterCacheBatchRequest, MediaSourcesRequest, UploadBase64Request, \ + UploadPresignRequest, \ + UploadMultipartPresignRequest, MonitorLiveRoomProductRequest +from ..models.cache_tasks.models import CacheTaskResult, MediaSource from ..utils.KVCache import MediaSourceKVCache, LiveProductKVCache from ..utils.SentryUtils import SentryUtils @@ -57,9 +52,9 @@ modal_kv_product_cache = LiveProductKVCache(kv_name=config.modal_product_kv_name summary="缓存视频文件", description="异步缓存视频文件到S3存储桶和Modal Dict(KV)", dependencies=[Depends(verify_token)]) -async def cache(medias: MediaSources) -> CacheResult: +async def cache(medias: MediaSourcesRequest) -> CacheResultResponse: fn_id = current_function_call_id() - caches: MediaSources + caches: MediaSourcesRequest sentry_trace = SentryTransactionInfo(x_trace_id=sentry_sdk.get_traceparent(), x_baggage=sentry_sdk.get_baggage()) @@ -123,14 +118,14 @@ async def cache(medias: MediaSources) -> CacheResult: cache_task_result_list.append(result) modal_kv_cache.batch_update_cloudflare_kv(cache_task_result_dict) - return CacheResult(caches={media.urn: media for media in cache_task_result_list}) + return CacheResultResponse(caches={media.urn: media for media in cache_task_result_list}) @router.delete("/", summary="清除指定的所有缓存", description="清除指定的所有缓存(包括KV记录和S3存储文件)", dependencies=[Depends(verify_token)]) -async def purge_media_kv_file(medias: MediaSources) -> CacheDeleteTaskResponse: +async def purge_media_kv_file(medias: MediaSourcesRequest) -> CacheDeleteTaskResponse: fn_id = current_function_call_id() fn = modal.Function.from_name(config.modal_app_name, "cache_delete", environment_name=config.modal_environment) @@ -182,7 +177,7 @@ async def purge_media_kv_file(medias: MediaSources) -> CacheDeleteTaskResponse: description="获取已缓存的视频下载地址", dependencies=[Depends(verify_token)]) @sentry_sdk.trace -async def download_caches(medias: MediaSources) -> DownloadResult: +async def download_caches(medias: MediaSourcesRequest) -> DownloadResult: cdn_endpoint = config.S3_cdn_endpoint urls = [] for media in medias.inputs: @@ -218,7 +213,7 @@ async def purge_kv_all(): summary="删除对应的KV记录", description="删除请求中对应的视频缓存记录", dependencies=[Depends(verify_token)]) -async def purge_kv(medias: MediaSources): +async def purge_kv(medias: MediaSourcesRequest): try: for media in medias.inputs: modal_kv_cache.pop(media.urn) diff --git a/src/BowongModalFunctions/router/comfyui.py b/src/BowongModalFunctions/router/comfyui.py index e107ae5..e7fa5d5 100644 --- a/src/BowongModalFunctions/router/comfyui.py +++ b/src/BowongModalFunctions/router/comfyui.py @@ -5,13 +5,14 @@ from fastapi import APIRouter, Depends, Header from fastapi.responses import Response from starlette import status -from ..config import WorkerConfig +from ..models.settings.cluster import WorkerConfig from ..middleware.authorization import verify_token -from ..models.media_model import MediaSource +from ..models.cache_tasks.models import MediaSource from ..utils.ModalUtils import ModalUtils -from ..models.web_model import (SentryTransactionHeader, - ModalTaskResponse, SentryTransactionInfo, - ComfyTaskRequest, ComfyTaskStatusResponse, TaskStatus) +from ..models.responses.models import ModalTaskResponse, ComfyTaskStatusResponse +from ..models.requests.models import ComfyTaskRequest +from ..models.ffmpeg_tasks.models import SentryTransactionHeader, SentryTransactionInfo +from ..models.enums.models import TaskStatus config = WorkerConfig() diff --git a/src/BowongModalFunctions/router/ffmpeg.py b/src/BowongModalFunctions/router/ffmpeg.py index 8b37f33..672ad98 100644 --- a/src/BowongModalFunctions/router/ffmpeg.py +++ b/src/BowongModalFunctions/router/ffmpeg.py @@ -6,24 +6,15 @@ from fastapi.responses import Response from loguru import logger from starlette import status -from ..config import WorkerConfig +from ..models.settings.cluster import WorkerConfig from ..middleware.authorization import verify_token from ..utils.ModalUtils import ModalUtils -from ..models.web_model import (FFMPEGSliceRequest, SentryTransactionHeader, - ModalTaskResponse, SentryTransactionInfo, - FFMPEGConcatRequest, - FFMPEGExtractAudioRequest, - FFMPEGCornerMirrorRequest, - FFMPEGOverlayGifRequest, - FFMPEGZoomLoopRequest, - FFMPEGSubtitleOverlayRequest, - FFMPEGMixBgmWithNoiseReduceRequest, - FFMPEGVideoLoopFillAudioRequest, - FFMPEGConvertStreamRequest, - FFMPEGExtractFrameRequest, - BaseFFMPEGTaskStatusResponse, - FFMPEGStreamRecordRequest, - RecordingTaskResponse) +from ..models.responses.models import ModalTaskResponse, RecordingTaskResponse, BaseFFMPEGTaskStatusResponse +from ..models.requests.models import FFMPEGSliceRequest, FFMPEGConcatRequest, FFMPEGExtractAudioRequest, \ + FFMPEGCornerMirrorRequest, FFMPEGZoomLoopRequest, FFMPEGOverlayGifRequest, FFMPEGSubtitleOverlayRequest, \ + FFMPEGMixBgmWithNoiseReduceRequest, FFMPEGVideoLoopFillAudioRequest, FFMPEGExtractFrameRequest, \ + FFMPEGStreamRecordRequest, FFMPEGConvertStreamRequest +from ..models.ffmpeg_tasks.models import SentryTransactionHeader, SentryTransactionInfo config = WorkerConfig() diff --git a/src/BowongModalFunctions/router/google.py b/src/BowongModalFunctions/router/google.py index 62b681c..6c66109 100644 --- a/src/BowongModalFunctions/router/google.py +++ b/src/BowongModalFunctions/router/google.py @@ -13,12 +13,13 @@ from pydantic import computed_field from starlette import status from starlette.responses import JSONResponse -from BowongModalFunctions.config import WorkerConfig -from BowongModalFunctions.middleware.authorization import verify_token -from BowongModalFunctions.models.web_model import SentryTransactionInfo, GeminiResultResponse, GeminiRequest, \ - ModalTaskResponse, MakeGridGeminiRequest, GoogleUploadResponse -from BowongModalFunctions.utils.ModalUtils import ModalUtils -from BowongModalFunctions.utils.HTTPUtils import GoogleAuthUtils +from ..models.settings.cluster import WorkerConfig +from ..middleware.authorization import verify_token +from ..models.responses.models import ModalTaskResponse, GeminiResultResponse, GoogleUploadResponse +from ..models.requests.models import GeminiRequest, MakeGridGeminiRequest +from ..models.ffmpeg_tasks.models import SentryTransactionInfo +from ..utils.ModalUtils import ModalUtils +from ..utils.HTTPUtils import GoogleAuthUtils config = WorkerConfig() @@ -191,10 +192,7 @@ async def make_grid_gemini_upload(data: MakeGridGeminiRequest, headers: Annotate @router.post('/inference_gemini', summary="使用Gemini推理hls视频流指定时间段的打点情况") -async def inference_gemini( - data: GeminiRequest, - headers: Annotated[BundleHeaders, Header()], -) -> ModalTaskResponse: +async def inference_gemini(data: GeminiRequest, headers: Annotated[BundleHeaders, Header()], ) -> ModalTaskResponse: google_api_key = headers.x_google_api_key or os.environ.get("GOOGLE_API_KEY") if not google_api_key: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing Google API Key") diff --git a/src/BowongModalFunctions/router/task.py b/src/BowongModalFunctions/router/task.py index 16b577a..a4445ee 100644 --- a/src/BowongModalFunctions/router/task.py +++ b/src/BowongModalFunctions/router/task.py @@ -1,14 +1,18 @@ import modal -from fastapi import APIRouter, Depends +from fastapi import APIRouter, Depends, HTTPException from loguru import logger +from starlette import status -from ..middleware.authorization import verify_token -from ..models.web_model import ModalTaskCancelResponse +from ..middleware.authorization import verify_token, nakama_login +from ..models.requests.models import NakamaLogin +from ..models.responses.models import ModalTaskCancelResponse, NakamaJWTResponse -router = APIRouter(prefix="/task", tags=["tasks"], dependencies=[Depends(verify_token)]) +router = APIRouter(prefix="/task", tags=["tasks"]) -@router.get("/cancel/{task_id}", summary="终止任务", description="终止任务, 无论是正在排队还是真正运行") +@router.get("/cancel/{task_id}", summary="终止任务", + description="终止任务, 无论是正在排队还是真正运行", + dependencies=[Depends(verify_token)]) async def task_cancel(task_id: str) -> ModalTaskCancelResponse: try: fn_call = modal.FunctionCall.from_id(task_id) @@ -18,3 +22,12 @@ async def task_cancel(task_id: str) -> ModalTaskCancelResponse: logger.exception(e) return ModalTaskCancelResponse(success=False, error=e.message if hasattr(e, 'message') else str(e)) + + +@router.post("/oauth", summary="登录以获取Nakama JWT") +async def nakama_oauth(body: NakamaLogin) -> NakamaJWTResponse: + try: + return nakama_login(body.email, body.password) + except Exception as e: + logger.exception(e) + raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="账号无法登录") diff --git a/src/BowongModalFunctions/utils/HTTPUtils.py b/src/BowongModalFunctions/utils/HTTPUtils.py index c01d2a6..1d9fdfe 100644 --- a/src/BowongModalFunctions/utils/HTTPUtils.py +++ b/src/BowongModalFunctions/utils/HTTPUtils.py @@ -12,7 +12,7 @@ from loguru import logger import aiofiles from pathlib import Path from pydantic import BaseModel, computed_field, Field, field_validator -from BowongModalFunctions.models.web_model import GoogleUploadResponse +from ..models.responses.models import GoogleUploadResponse class HTTPDownloadUtils: @@ -192,7 +192,8 @@ class GoogleAuthUtils: return GoogleAuthUtils.GoogleAuthResponse.model_validate_json(response.text) @staticmethod - async def google_upload_file(file_stream: Union[BinaryIO ,bytes], content_type: str, google_api_key: str, bucket_name: str, + async def google_upload_file(file_stream: Union[BinaryIO, bytes], content_type: str, google_api_key: str, + bucket_name: str, filename: str) -> GoogleUploadResponse: safe_filename = quote(filename) diff --git a/src/BowongModalFunctions/utils/KVCache.py b/src/BowongModalFunctions/utils/KVCache.py index db6b2eb..375da42 100644 --- a/src/BowongModalFunctions/utils/KVCache.py +++ b/src/BowongModalFunctions/utils/KVCache.py @@ -4,8 +4,7 @@ import httpx import modal from loguru import logger from .VideoUtils import VideoUtils -from ..models.media_model import MediaSource -from ..models.web_model import LiveProductCaches +from ..models.cache_tasks.models import MediaSource, LiveProductCaches class KVCache: diff --git a/src/BowongModalFunctions/utils/ModalUtils.py b/src/BowongModalFunctions/utils/ModalUtils.py index 0a8ae13..caf151c 100644 --- a/src/BowongModalFunctions/utils/ModalUtils.py +++ b/src/BowongModalFunctions/utils/ModalUtils.py @@ -1,11 +1,12 @@ -from typing import Tuple, Optional, Any, List +from typing import Optional, Any, List import modal from loguru import logger -from modal.call_graph import InputStatus +from modal.call_graph import InputStatus, InputInfo from pydantic import BaseModel, Field -from BowongModalFunctions.models.web_model import TaskStatus, SentryTransactionInfo, ErrorCode +from ..models.ffmpeg_tasks.models import SentryTransactionInfo +from ..models.enums.models import TaskStatus, ErrorCode class ModalTaskInfo(BaseModel): @@ -25,16 +26,16 @@ class ModalUtils: :param task_id: modal 任务id :return: (TaskStaus 任务运行状态, function_name 任务名字, errorCode 错误代码, errorReason 错误原因, results 任务结果, sentryTransactionInfo Sentry跟踪信息) """ - task = None + task: Optional[InputInfo] = None try: fn_task = modal.FunctionCall.from_id(task_id) - call_graph = fn_task.get_call_graph() root = call_graph[0] if len(root.children) == 0: return ModalTaskInfo(function_name="unknown", error_code=ErrorCode.NOT_FOUND, status=TaskStatus.expired, error_reason="NOT_FOUND") - task = root.children[0] + # task: Optional[InputInfo] = root.children[0] + task: Optional[InputInfo] = next(iter(root.children), None) if not task.function_call_id == task_id: return ModalTaskInfo(function_name=task.function_name, status=TaskStatus.expired, error_code=ErrorCode.NOT_FOUND, error_reason="NOT_FOUND") diff --git a/src/BowongModalFunctions/utils/SentryUtils.py b/src/BowongModalFunctions/utils/SentryUtils.py index 5e2703b..1816c6d 100644 --- a/src/BowongModalFunctions/utils/SentryUtils.py +++ b/src/BowongModalFunctions/utils/SentryUtils.py @@ -8,8 +8,9 @@ import sentry_sdk from loguru import logger import functools -from BowongModalFunctions.models.web_model import WebhookNotify, WebhookMethodEnum, BaseFFMPEGTaskStatusResponse, \ - TaskStatus, ErrorCode, FFMPEGResult +from ..models.responses.models import BaseFFMPEGTaskStatusResponse +from ..models.ffmpeg_tasks.models import WebhookNotify, FFMPEGResult +from ..models.enums.models import TaskStatus, ErrorCode, WebhookMethodEnum class SentryUtils: diff --git a/src/BowongModalFunctions/utils/TimeUtils.py b/src/BowongModalFunctions/utils/TimeUtils.py index 4ed99bb..51ed5fe 100644 --- a/src/BowongModalFunctions/utils/TimeUtils.py +++ b/src/BowongModalFunctions/utils/TimeUtils.py @@ -1,8 +1,6 @@ import re from datetime import timedelta, datetime -from loguru import logger - class TimeDelta(timedelta): @classmethod @@ -23,38 +21,38 @@ def parse_time(time_str): """解析时间字符串为datetime对象""" parsed_time = None - if re.match(r"^\d{2}:\d{2}:\d{2}\.\d{3}$",time_str): + if re.match(r"^\d{2}:\d{2}:\d{2}\.\d{3}$", time_str): # 先尝试完整格式 HH:MM:SS.fff parsed_time = datetime.strptime(time_str, '%H:%M:%S.%f') - elif re.match(r"^\d{2}:\d{2}:\d{2}:\d{3}$",time_str): + elif re.match(r"^\d{2}:\d{2}:\d{2}:\d{3}$", time_str): # 如果失败，尝试 HH:MM:SS:fff 格式 parsed_time = datetime.strptime(time_str, '%H:%M:%S:%f') - elif re.match(r"^\d{2}:\d{2}\.\d{3}$",time_str): + elif re.match(r"^\d{2}:\d{2}\.\d{3}$", time_str): # 如果失败，尝试 MM:SS.fff 格式 dt = datetime.strptime(time_str, '%M:%S.%f') # 将小时设为0，只保留分钟和秒 parsed_time = datetime.combine(dt.date(), dt.time().replace(hour=0)) - elif re.match(r"^\d{2}\.\d{2}:\d{3}$",time_str): + elif re.match(r"^\d{2}\.\d{2}:\d{3}$", time_str): # 如果失败，尝试 MM.SS:fff 格式 dt = datetime.strptime(time_str, '%M.%S:%f') # 将小时设为0，只保留分钟和秒 parsed_time = datetime.combine(dt.date(), dt.time().replace(hour=0)) - elif re.match(r"^\d{2}\.\d{2}\.\d{3}$",time_str): + elif re.match(r"^\d{2}\.\d{2}\.\d{3}$", time_str): # 如果失败，尝试 MM.SS.fff 格式 dt = datetime.strptime(time_str, '%M.%S.%f') # 将小时设为0，只保留分钟和秒 parsed_time = datetime.combine(dt.date(), dt.time().replace(hour=0)) - elif re.match(r"^\d{2}:\d{2}:\d{3}$",time_str): + elif re.match(r"^\d{2}:\d{2}:\d{3}$", time_str): # 如果失败，尝试 MM:SS:fff 格式 dt = datetime.strptime(time_str, '%M:%S:%f') # 将小时设为0，只保留分钟和秒 parsed_time = datetime.combine(dt.date(), dt.time().replace(hour=0)) - elif re.match(r"^\d{2}:\d{2}$",time_str): + elif re.match(r"^\d{2}:\d{2}$", time_str): # 如果失败，尝试 MM:SS:fff 格式 dt = datetime.strptime(time_str, '%M:%S') # 将小时设为0，只保留分钟和秒 parsed_time = datetime.combine(dt.date(), dt.time().replace(hour=0)) - elif re.match(r"^\d{2}\.\d{2}$",time_str): + elif re.match(r"^\d{2}\.\d{2}$", time_str): # 如果失败，尝试 MM:SS:fff 格式 dt = datetime.strptime(time_str, '%M.%S') # 将小时设为0，只保留分钟和秒 @@ -136,7 +134,7 @@ def merge_product_data(data, start_time_str, end_time_str, merge_diff=5): hours, remainder = divmod(total_seconds, 3600) minutes, seconds = divmod(remainder, 60) microseconds = duration.microseconds - max_time_str = f"{int(hours):02d}:{int(minutes):02d}:{int(seconds):02d}.{microseconds//1000:03d}" + max_time_str = f"{int(hours):02d}:{int(minutes):02d}:{int(seconds):02d}.{microseconds // 1000:03d}" product_dict = {} @@ -155,7 +153,7 @@ def merge_product_data(data, start_time_str, end_time_str, merge_diff=5): start, end = parse_timeline_item(item) # 比较起始时间与时间差 start_str = format_time(start) - if (start - datetime.strptime("00:00:00.000",'%H:%M:%S.%f')) > duration and not start_str.startswith("00"): + if (start - datetime.strptime("00:00:00.000", '%H:%M:%S.%f')) > duration and not start_str.startswith("00"): new_start_str = convert_time(start_str) else: new_start_str = start_str @@ -163,7 +161,7 @@ def merge_product_data(data, start_time_str, end_time_str, merge_diff=5): new_start_str = max_time_str end_str = format_time(end) - if (end - datetime.strptime("00:00:00.000",'%H:%M:%S.%f')) > duration and not end_str.startswith("00"): + if (end - datetime.strptime("00:00:00.000", '%H:%M:%S.%f')) > duration and not end_str.startswith("00"): new_end_str = convert_time(end_str) else: new_end_str = end_str diff --git a/src/BowongModalFunctions/utils/VideoUtils.py b/src/BowongModalFunctions/utils/VideoUtils.py index 24cbaca..4c0ef4d 100644 --- a/src/BowongModalFunctions/utils/VideoUtils.py +++ b/src/BowongModalFunctions/utils/VideoUtils.py @@ -5,7 +5,7 @@ from datetime import datetime, timedelta import aiofiles import aiohttp -from typing import Union, List, Tuple, Optional, Dict, Any +from typing import List, Tuple, Optional, Any import m3u8 import numpy as np @@ -14,7 +14,6 @@ import math from aiohttp import ClientTimeout from .TimeUtils import TimeDelta -from pydantic import BaseModel, ConfigDict, computed_field, Field, field_validator from ffmpeg import FFmpeg from ffmpeg.asyncio import FFmpeg as AsyncFFmpeg import soundfile as sf @@ -35,181 +34,7 @@ from pedalboard.io import AudioFile from loguru import logger from .PathUtils import FileUtils -from ..models.ffmpeg_worker_model import FFMpegSliceSegment - - -class FFMPEGSliceOptions(BaseModel): - limit_size: Optional[int] = Field(default=None, description="不超过指定文件(字节)大小, 默认为空不限制输出大小") - bit_rate: Optional[int] = Field(default=None, description="指定输出视频的比特率, 不能与limit_size同时设置") - - crf: int = Field(default=16, description="输出视频的质量") - fps: int = Field(default=30, description="输出视频的FPS") - width: int = Field(default=None, description="输出视频的宽(像素)") - height: int = Field(default=None, description="输出视频的高(像素)") - - @computed_field(description="解析为字符表达式的文件大小") - @property - def pretty_limit_size(self) -> Optional[str]: - if self.limit_size is not None: - for unit in ['B', 'KB', 'MB', 'GB', 'TB']: - if self.limit_size < 1024.0: - return f"{self.limit_size:.2f}{unit}" - self.limit_size /= 1024.0 - return f"{self.limit_size:.2f}PB" - else: - return None - - @computed_field(description="解析为字符表达式的比特率") - @property - def pretty_bit_rate(self) -> Optional[str]: - if self.bit_rate is not None: - return f"{self.bit_rate}k" - - -class MediaStream(BaseModel): - duration: float = Field(0, description="时长") - codec_name: str - tags: Optional[Any] = Field(None) - - model_config = ConfigDict(extra='allow') - - -class HLSMediaVideoStream(BaseModel): - stream_type: str = "video" - codec_name: str - codec_type: str - width: int - height: int - avg_frame_rate: str - tags: Optional[Any] = Field(None) - duration: Optional[float] = Field(None) - - @computed_field - @property - def video_frame_rate(self) -> float: - numerator, denominator = map(int, self.avg_frame_rate.split('/')) - if denominator != 0: - return numerator / denominator - return 0 - - -class HLSMediaAudioStream(BaseModel): - stream_type: str = "audio" - sample_rate: str - channels: int - channel_layout: str - start_time: str - tags: Optional[Any] = Field(None) - - -class AudioStream(MediaStream): - stream_type: str = Field("audio") - codec_type: str - sample_rate: str - channels: int - tags: Optional[Any] = Field(None) - - model_config = ConfigDict(extra='allow') - - -class VideoStream(MediaStream): - stream_type: str = "video" - width: int - height: int - bit_rate: int - avg_frame_rate: str - - @computed_field - @property - def video_bitrate(self) -> str: - return str(int(self.bit_rate / 1000)) + 'k' - - @computed_field - @property - def video_frame_rate(self) -> float: - numerator, denominator = map(int, self.avg_frame_rate.split('/')) - if denominator != 0: - return numerator / denominator - return 0 - - model_config = ConfigDict(extra='allow') - - -class ImageStream(MediaStream): - stream_type: str = "image" - width: int - height: int - - model_config = ConfigDict(extra='allow') - - -class SubtitleStreamTags(BaseModel): - language: str = Field(description="内嵌字幕的语言") - - model_config = ConfigDict(extra='allow') - - -class SubtitleStream(MediaStream): - stream_type: str = "subtitle" - tags: SubtitleStreamTags - - model_config = ConfigDict(extra='allow') - - -class VideoFormat(BaseModel): - filename: str - format_name: str - start_time: float = Field(0, description="起始时间") - size: int - bit_rate: Optional[int] = Field(None, description="文件比特率") - duration: float = Field(3600 * 12, description="文件时长") - - -class VideoMetadata(BaseModel): - streams: List[ - Union[ImageStream, AudioStream, VideoStream, HLSMediaAudioStream, HLSMediaVideoStream, SubtitleStream]] = Field( - description="媒体包含的数据轨道") - format: Optional[VideoFormat] = Field(None) - - @field_validator('streams', mode='before') - def parse_streams(cls, value): - streams = [] - if isinstance(value, List): - for stream in value: - if isinstance(stream, Dict): - logger.info(f"Parsing stream : {json.dumps(stream, ensure_ascii=False)}") - if stream.get("codec_type") == 'audio': - if stream.get("duration") is None: - logger.info("Parsing audio stream") - hls_audio = HLSMediaAudioStream.model_validate(stream) - streams.append(hls_audio) - else: - logger.info("Parsing hls audio stream") - audio = AudioStream.model_validate(stream) - streams.append(audio) - elif stream.get("codec_type") == 'video': - if stream.get("codec_name") in ("gif", "png", "mjpg", "mjpeg", "webp"): - logger.info("Parsing image stream") - image = ImageStream.model_validate(stream) - streams.append(image) - else: - if stream.get("duration") is None: - logger.info("Parsing hls video stream") - hls_video = HLSMediaVideoStream.model_validate(stream) - streams.append(hls_video) - else: - logger.info("Parsing video stream") - video = VideoStream.model_validate(stream) - streams.append(video) - elif stream.get("codec_type") == 'subtitle': - logger.info("Parsing subtitle stream") - subtitle_stream = SubtitleStream.model_validate(stream) - streams.append(subtitle_stream) - return streams - else: - raise TypeError - - model_config = ConfigDict(extra='allow') +from ..models.ffmpeg_tasks.models import FFMpegSliceSegment, FFMPEGSliceOptions, VideoStream, VideoMetadata class VideoUtils: diff --git a/src/Douyin_TikTok_Download_API/tikhub_api.py b/src/Douyin_TikTok_Download_API/tikhub_api.py new file mode 100644 index 0000000..2d0101e --- /dev/null +++ b/src/Douyin_TikTok_Download_API/tikhub_api.py @@ -0,0 +1,22 @@ +from fastapi import FastAPI +from fastapi.middleware.cors import CORSMiddleware +from scalar_fastapi import get_scalar_api_reference +from Douyin_TikTok_Download_API.app.api import router +web_app_tikhub = FastAPI(title="Modal Tikhub API", + version="1.0.0", + summary="Modal Tikhub API, 包含抖音Web API", + ) + +web_app_tikhub.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], +) + +@web_app_tikhub.get("/scalar", include_in_schema=False) +async def scalar(): + return get_scalar_api_reference(openapi_url='/openapi.json', title="Modal worker web endpoint") + +web_app_tikhub.include_router(router.router) \ No newline at end of file diff --git a/src/cluster/app.py b/src/cluster/app.py index e1365bd..ae35187 100644 --- a/src/cluster/app.py +++ b/src/cluster/app.py @@ -1,5 +1,5 @@ import modal -from BowongModalFunctions.config import WorkerConfig +from BowongModalFunctions.models.settings.cluster import WorkerConfig from .video import app as media_app from .web import app as web_app from .ffmpeg_app import app as ffmpeg_app diff --git a/src/cluster/comfyui_v1.py b/src/cluster/comfyui_v1.py index 27e2d50..548af29 100644 --- a/src/cluster/comfyui_v1.py +++ b/src/cluster/comfyui_v1.py @@ -75,8 +75,8 @@ with comfyui_image.imports(): from typing import Tuple, Any, Dict, Optional import sentry_sdk from modal import current_function_call_id - from BowongModalFunctions.config import WorkerConfig - from BowongModalFunctions.models.web_model import SentryTransactionInfo, WebhookNotify + from BowongModalFunctions.models.settings.cluster import WorkerConfig + from BowongModalFunctions.models.ffmpeg_tasks.models import SentryTransactionInfo, WebhookNotify from BowongModalFunctions.utils.SentryUtils import SentryUtils from sentry_sdk.integrations.loguru import LoguruIntegration diff --git a/src/cluster/comfyui_v2.py b/src/cluster/comfyui_v2.py index d445b5f..ee5558e 100644 --- a/src/cluster/comfyui_v2.py +++ b/src/cluster/comfyui_v2.py @@ -56,10 +56,10 @@ with comfyui_latentsync_1_5_image.imports(): import sentry_sdk import psutil from loguru import logger - from BowongModalFunctions.config import WorkerConfig + from BowongModalFunctions.models.settings.cluster import WorkerConfig from modal import current_function_call_id from sentry_sdk.integrations.loguru import LoguruIntegration - from BowongModalFunctions.models.web_model import SentryTransactionInfo, WebhookNotify + from BowongModalFunctions.models.ffmpeg_tasks.models import SentryTransactionInfo, WebhookNotify from BowongModalFunctions.utils.SentryUtils import SentryUtils config = WorkerConfig() diff --git a/src/cluster/ffmpeg_app.py b/src/cluster/ffmpeg_app.py index 2e73cc6..1a15c82 100644 --- a/src/cluster/ffmpeg_app.py +++ b/src/cluster/ffmpeg_app.py @@ -19,7 +19,7 @@ with ffmpeg_worker_image.imports(): import shutil, backoff, sentry_sdk from typing import List from loguru import logger - from BowongModalFunctions.config import WorkerConfig + from BowongModalFunctions.models.settings.cluster import WorkerConfig config = WorkerConfig() diff --git a/src/cluster/ffmpeg_apps/bgm_nosie_reduce.py b/src/cluster/ffmpeg_apps/bgm_nosie_reduce.py index 797907c..603b83f 100644 --- a/src/cluster/ffmpeg_apps/bgm_nosie_reduce.py +++ b/src/cluster/ffmpeg_apps/bgm_nosie_reduce.py @@ -3,8 +3,8 @@ 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(): - from BowongModalFunctions.models.media_model import MediaSource - from BowongModalFunctions.models.web_model import SentryTransactionInfo, WebhookNotify, FFMPEGResult + from BowongModalFunctions.models.cache_tasks.models import MediaSource + from BowongModalFunctions.models.ffmpeg_tasks.models import SentryTransactionInfo, WebhookNotify, FFMPEGResult from BowongModalFunctions.utils.PathUtils import FileUtils from BowongModalFunctions.utils.SentryUtils import SentryUtils from BowongModalFunctions.utils.VideoUtils import VideoUtils diff --git a/src/cluster/ffmpeg_apps/concat_medias.py b/src/cluster/ffmpeg_apps/concat_medias.py index 85fa93b..6226e0e 100644 --- a/src/cluster/ffmpeg_apps/concat_medias.py +++ b/src/cluster/ffmpeg_apps/concat_medias.py @@ -3,8 +3,8 @@ 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(): - from BowongModalFunctions.models.media_model import MediaSources - from BowongModalFunctions.models.web_model import SentryTransactionInfo, WebhookNotify, FFMPEGResult + from BowongModalFunctions.models.requests.models import MediaSourcesRequest + from BowongModalFunctions.models.ffmpeg_tasks.models import SentryTransactionInfo, WebhookNotify, FFMPEGResult from BowongModalFunctions.utils.PathUtils import FileUtils from BowongModalFunctions.utils.SentryUtils import SentryUtils from BowongModalFunctions.utils.VideoUtils import VideoUtils @@ -26,7 +26,7 @@ with ffmpeg_worker_image.imports(): ), }, ) @modal.concurrent(max_inputs=1) - async def ffmpeg_concat_medias(medias: MediaSources, + async def ffmpeg_concat_medias(medias: MediaSourcesRequest, sentry_trace: Optional[SentryTransactionInfo] = None, webhook: Optional[WebhookNotify] = None) -> Tuple[ FFMPEGResult, Optional[SentryTransactionInfo]]: @@ -36,7 +36,7 @@ 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_sources: MediaSources, output_filepath: str) -> FFMPEGResult: + async def ffmpeg_process(media_sources: MediaSourcesRequest, output_filepath: str) -> FFMPEGResult: input_videos = [f"{s3_mount}/{media_source.cache_filepath}" for media_source in media_sources.inputs] local_output_path, metadata = await VideoUtils.ffmpeg_concat_medias(media_paths=input_videos, output_path=output_filepath) diff --git a/src/cluster/ffmpeg_apps/convert_stream.py b/src/cluster/ffmpeg_apps/convert_stream.py index a7653f3..ab620e1 100644 --- a/src/cluster/ffmpeg_apps/convert_stream.py +++ b/src/cluster/ffmpeg_apps/convert_stream.py @@ -3,11 +3,13 @@ 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(): - from BowongModalFunctions.models.media_model import MediaSource, MediaProtocol - from BowongModalFunctions.models.web_model import SentryTransactionInfo, WebhookNotify, FFMPEGResult + from BowongModalFunctions.models.cache_tasks.models import MediaSource + from BowongModalFunctions.models.enums.models import MediaProtocol from BowongModalFunctions.utils.PathUtils import FileUtils from BowongModalFunctions.utils.SentryUtils import SentryUtils - from BowongModalFunctions.utils.VideoUtils import VideoUtils, FFMPEGSliceOptions + from BowongModalFunctions.utils.VideoUtils import VideoUtils + from BowongModalFunctions.models.ffmpeg_tasks.models import FFMPEGSliceOptions, SentryTransactionInfo, WebhookNotify, \ + FFMPEGResult import sentry_sdk from typing import Optional, Tuple from modal import current_function_call_id diff --git a/src/cluster/ffmpeg_apps/corner_mirror.py b/src/cluster/ffmpeg_apps/corner_mirror.py index 8a2957f..af4d87c 100644 --- a/src/cluster/ffmpeg_apps/corner_mirror.py +++ b/src/cluster/ffmpeg_apps/corner_mirror.py @@ -3,8 +3,8 @@ 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(): - from BowongModalFunctions.models.media_model import MediaSource - from BowongModalFunctions.models.web_model import SentryTransactionInfo, WebhookNotify, FFMPEGResult + from BowongModalFunctions.models.cache_tasks.models import MediaSource + from BowongModalFunctions.models.ffmpeg_tasks.models import SentryTransactionInfo, WebhookNotify, FFMPEGResult from BowongModalFunctions.utils.PathUtils import FileUtils from BowongModalFunctions.utils.SentryUtils import SentryUtils from BowongModalFunctions.utils.VideoUtils import VideoUtils diff --git a/src/cluster/ffmpeg_apps/extract_audio.py b/src/cluster/ffmpeg_apps/extract_audio.py index f98bac6..a2402a1 100644 --- a/src/cluster/ffmpeg_apps/extract_audio.py +++ b/src/cluster/ffmpeg_apps/extract_audio.py @@ -3,8 +3,8 @@ 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(): - from BowongModalFunctions.models.media_model import MediaSource - from BowongModalFunctions.models.web_model import SentryTransactionInfo, WebhookNotify, FFMPEGResult + from BowongModalFunctions.models.cache_tasks.models import MediaSource + from BowongModalFunctions.models.ffmpeg_tasks.models import SentryTransactionInfo, WebhookNotify, FFMPEGResult from BowongModalFunctions.utils.PathUtils import FileUtils from BowongModalFunctions.utils.SentryUtils import SentryUtils from BowongModalFunctions.utils.VideoUtils import VideoUtils diff --git a/src/cluster/ffmpeg_apps/extract_frame.py b/src/cluster/ffmpeg_apps/extract_frame.py index 73bbfc4..ee7d138 100644 --- a/src/cluster/ffmpeg_apps/extract_frame.py +++ b/src/cluster/ffmpeg_apps/extract_frame.py @@ -4,8 +4,9 @@ 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(): - from BowongModalFunctions.models.media_model import MediaSource, MediaProtocol - from BowongModalFunctions.models.web_model import SentryTransactionInfo, WebhookNotify, FFMPEGResult + from BowongModalFunctions.models.cache_tasks.models import MediaSource + from BowongModalFunctions.models.enums.models import MediaProtocol + from BowongModalFunctions.models.ffmpeg_tasks.models import SentryTransactionInfo, WebhookNotify, FFMPEGResult from BowongModalFunctions.utils.PathUtils import FileUtils from BowongModalFunctions.utils.SentryUtils import SentryUtils from BowongModalFunctions.utils.VideoUtils import VideoUtils diff --git a/src/cluster/ffmpeg_apps/loop_fill.py b/src/cluster/ffmpeg_apps/loop_fill.py index b4c8453..7f7a922 100644 --- a/src/cluster/ffmpeg_apps/loop_fill.py +++ b/src/cluster/ffmpeg_apps/loop_fill.py @@ -3,8 +3,8 @@ 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(): - from BowongModalFunctions.models.media_model import MediaSource - from BowongModalFunctions.models.web_model import SentryTransactionInfo, WebhookNotify, FFMPEGResult + from BowongModalFunctions.models.cache_tasks.models import MediaSource + from BowongModalFunctions.models.ffmpeg_tasks.models import SentryTransactionInfo, WebhookNotify, FFMPEGResult from BowongModalFunctions.utils.PathUtils import FileUtils from BowongModalFunctions.utils.SentryUtils import SentryUtils from BowongModalFunctions.utils.VideoUtils import VideoUtils diff --git a/src/cluster/ffmpeg_apps/overlay_gif.py b/src/cluster/ffmpeg_apps/overlay_gif.py index 1d61b5f..5922aae 100644 --- a/src/cluster/ffmpeg_apps/overlay_gif.py +++ b/src/cluster/ffmpeg_apps/overlay_gif.py @@ -3,8 +3,8 @@ 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(): - from BowongModalFunctions.models.media_model import MediaSource - from BowongModalFunctions.models.web_model import SentryTransactionInfo, WebhookNotify, FFMPEGResult + from BowongModalFunctions.models.cache_tasks.models import MediaSource + from BowongModalFunctions.models.ffmpeg_tasks.models import SentryTransactionInfo, WebhookNotify, FFMPEGResult from BowongModalFunctions.utils.PathUtils import FileUtils from BowongModalFunctions.utils.SentryUtils import SentryUtils from BowongModalFunctions.utils.VideoUtils import VideoUtils diff --git a/src/cluster/ffmpeg_apps/slice_media.py b/src/cluster/ffmpeg_apps/slice_media.py index eccb2a8..dcfcd01 100644 --- a/src/cluster/ffmpeg_apps/slice_media.py +++ b/src/cluster/ffmpeg_apps/slice_media.py @@ -3,12 +3,14 @@ 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(): - from BowongModalFunctions.models.ffmpeg_worker_model import FFMpegSliceSegment - from BowongModalFunctions.models.media_model import MediaSources, MediaProtocol, MediaSource - from BowongModalFunctions.models.web_model import SentryTransactionInfo, WebhookNotify, FFMPEGResult + from BowongModalFunctions.models.ffmpeg_tasks.models import FFMpegSliceSegment, FFMPEGSliceOptions, \ + SentryTransactionInfo, WebhookNotify, FFMPEGResult + from BowongModalFunctions.models.requests.models import MediaSourcesRequest + from BowongModalFunctions.models.cache_tasks.models import MediaSource + from BowongModalFunctions.models.enums.models import MediaProtocol from BowongModalFunctions.utils.PathUtils import FileUtils from BowongModalFunctions.utils.SentryUtils import SentryUtils - from BowongModalFunctions.utils.VideoUtils import VideoUtils, FFMPEGSliceOptions + from BowongModalFunctions.utils.VideoUtils import VideoUtils import sentry_sdk from loguru import logger from typing import Optional, Tuple, List diff --git a/src/cluster/ffmpeg_apps/stream_record_as_hls.py b/src/cluster/ffmpeg_apps/stream_record_as_hls.py index 7bc91e9..255cea0 100644 --- a/src/cluster/ffmpeg_apps/stream_record_as_hls.py +++ b/src/cluster/ffmpeg_apps/stream_record_as_hls.py @@ -6,8 +6,9 @@ import modal from ..ffmpeg_app import ffmpeg_worker_image, app, config, s3_mount with ffmpeg_worker_image.imports(): - from BowongModalFunctions.models.web_model import SentryTransactionInfo, WebhookNotify, FFMPEGResult, \ - WebhookMethodEnum, BaseFFMPEGTaskStatusResponse, TaskStatus + from BowongModalFunctions.models.responses.models import BaseFFMPEGTaskStatusResponse + from BowongModalFunctions.models.ffmpeg_tasks.models import SentryTransactionInfo, WebhookNotify, FFMPEGResult + from BowongModalFunctions.models.enums.models import TaskStatus, WebhookMethodEnum from BowongModalFunctions.utils.PathUtils import FileUtils from BowongModalFunctions.utils.VideoUtils import VideoUtils import sentry_sdk diff --git a/src/cluster/ffmpeg_apps/subtitle_apply.py b/src/cluster/ffmpeg_apps/subtitle_apply.py index 2666565..f53b0c3 100644 --- a/src/cluster/ffmpeg_apps/subtitle_apply.py +++ b/src/cluster/ffmpeg_apps/subtitle_apply.py @@ -3,8 +3,8 @@ 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(): - from BowongModalFunctions.models.media_model import MediaSource - from BowongModalFunctions.models.web_model import SentryTransactionInfo, WebhookNotify, FFMPEGResult + from BowongModalFunctions.models.cache_tasks.models import MediaSource + from BowongModalFunctions.models.ffmpeg_tasks.models import SentryTransactionInfo, WebhookNotify, FFMPEGResult from BowongModalFunctions.utils.PathUtils import FileUtils from BowongModalFunctions.utils.SentryUtils import SentryUtils from BowongModalFunctions.utils.VideoUtils import VideoUtils diff --git a/src/cluster/ffmpeg_apps/zoom_loop.py b/src/cluster/ffmpeg_apps/zoom_loop.py index 49a83f7..c95d759 100644 --- a/src/cluster/ffmpeg_apps/zoom_loop.py +++ b/src/cluster/ffmpeg_apps/zoom_loop.py @@ -3,8 +3,8 @@ 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(): - from BowongModalFunctions.models.media_model import MediaSource - from BowongModalFunctions.models.web_model import SentryTransactionInfo, WebhookNotify, FFMPEGResult + from BowongModalFunctions.models.cache_tasks.models import MediaSource + from BowongModalFunctions.models.ffmpeg_tasks.models import SentryTransactionInfo, WebhookNotify, FFMPEGResult from BowongModalFunctions.utils.PathUtils import FileUtils from BowongModalFunctions.utils.SentryUtils import SentryUtils from BowongModalFunctions.utils.VideoUtils import VideoUtils diff --git a/src/cluster/video.py b/src/cluster/video.py index 1468fe3..06b70c7 100644 --- a/src/cluster/video.py +++ b/src/cluster/video.py @@ -28,9 +28,9 @@ with downloader_image.imports(): from typing import List from loguru import logger - from BowongModalFunctions.config import WorkerConfig + from BowongModalFunctions.models.settings.cluster import WorkerConfig from BowongModalFunctions.utils.KVCache import MediaSourceKVCache, LiveProductKVCache - from BowongModalFunctions.models.media_model import MediaSource + from BowongModalFunctions.models.cache_tasks.models import MediaSource config = WorkerConfig() diff --git a/src/cluster/video_apps/cache_delete.py b/src/cluster/video_apps/cache_delete.py index 4ceb408..bf0919b 100644 --- a/src/cluster/video_apps/cache_delete.py +++ b/src/cluster/video_apps/cache_delete.py @@ -4,7 +4,8 @@ from ..video import downloader_image, app, config with downloader_image.imports(): import os - from BowongModalFunctions.models.media_model import MediaSource, MediaCacheStatus + from BowongModalFunctions.models.cache_tasks.models import MediaSource + from BowongModalFunctions.models.enums.models import MediaCacheStatus @app.function(cpu=1, timeout=300, diff --git a/src/cluster/video_apps/cache_submit.py b/src/cluster/video_apps/cache_submit.py index 6100f88..fa1594c 100644 --- a/src/cluster/video_apps/cache_submit.py +++ b/src/cluster/video_apps/cache_submit.py @@ -15,8 +15,9 @@ with downloader_image.imports(): from tencentcloud.vod.v20180717.vod_client import VodClient from tencentcloud.vod.v20180717 import models as vod_request_models - from BowongModalFunctions.models.media_model import MediaSource, MediaCacheStatus, MediaProtocol - from BowongModalFunctions.models.web_model import SentryTransactionInfo + from BowongModalFunctions.models.cache_tasks.models import MediaSource + from BowongModalFunctions.models.enums.models import MediaProtocol, MediaCacheStatus + from BowongModalFunctions.models.ffmpeg_tasks.models import SentryTransactionInfo @app.function(cpu=1, timeout=1800, diff --git a/src/cluster/video_apps/hls_slice_inference.py b/src/cluster/video_apps/hls_slice_inference.py index e6d9959..0770657 100644 --- a/src/cluster/video_apps/hls_slice_inference.py +++ b/src/cluster/video_apps/hls_slice_inference.py @@ -18,12 +18,13 @@ with downloader_image.imports(): from fastapi import HTTPException from starlette import status from BowongModalFunctions.utils.SentryUtils import SentryUtils - from BowongModalFunctions.models.media_model import MediaSource - from BowongModalFunctions.models.web_model import SentryTransactionInfo, WebhookNotify, GeminiFirstStagePromptVariables, \ - GeminiSecondStagePromptVariables, GeminiFirstStageResponseModel, GeminiSecondStageResponseModel - from BowongModalFunctions.models.ffmpeg_worker_model import FFMpegSliceSegment + from BowongModalFunctions.models.cache_tasks.models import MediaSource, GeminiFirstStageResponseModel, \ + GeminiFirstStagePromptVariables, GeminiSecondStagePromptVariables + from BowongModalFunctions.models.responses.models import GeminiSecondStageResponseModel + from BowongModalFunctions.models.ffmpeg_tasks.models import FFMpegSliceSegment, FFMPEGSliceOptions, \ + SentryTransactionInfo, WebhookNotify from BowongModalFunctions.utils.TimeUtils import TimeDelta, merge_product_data - from BowongModalFunctions.utils.VideoUtils import FFMPEGSliceOptions, VideoUtils + from BowongModalFunctions.utils.VideoUtils import VideoUtils @app.function(cpu=(0.5, 64), timeout=1800, diff --git a/src/cluster/video_apps/make_grid_upload.py b/src/cluster/video_apps/make_grid_upload.py index 6742f35..bb0a6b3 100644 --- a/src/cluster/video_apps/make_grid_upload.py +++ b/src/cluster/video_apps/make_grid_upload.py @@ -18,7 +18,8 @@ with downloader_image.imports(): from modal import current_function_call_id from BowongModalFunctions.utils.SentryUtils import SentryUtils - from BowongModalFunctions.models.web_model import SentryTransactionInfo + from BowongModalFunctions.models.ffmpeg_tasks.models import SentryTransactionInfo + @app.function(cpu=(0.5, 16), max_containers=config.video_downloader_concurrency, diff --git a/src/cluster/video_apps/monitor_live_room_product_trigger.py b/src/cluster/video_apps/monitor_live_room_product_trigger.py index c19890a..9543dbf 100644 --- a/src/cluster/video_apps/monitor_live_room_product_trigger.py +++ b/src/cluster/video_apps/monitor_live_room_product_trigger.py @@ -8,7 +8,8 @@ with downloader_image.imports(): from datetime import datetime, timedelta, timezone from httpx import Timeout - from BowongModalFunctions.models.web_model import LiveProduct, LiveProductCaches + from BowongModalFunctions.models.cache_tasks.models import LiveProduct, LiveProductCaches + @app.function(max_containers=config.video_downloader_concurrency, timeout=130) @modal.concurrent(max_inputs=50) diff --git a/src/cluster/web.py b/src/cluster/web.py index e61bbf9..2276955 100644 --- a/src/cluster/web.py +++ b/src/cluster/web.py @@ -22,8 +22,9 @@ app = modal.App( include_source=False) with fastapi_image.imports(): - from BowongModalFunctions.api import web_app, web_app_tikhub - from BowongModalFunctions.config import WorkerConfig + from BowongModalFunctions.api import web_app + from BowongModalFunctions.models.settings.cluster import WorkerConfig + from Douyin_TikTok_Download_API.tikhub_api import web_app_tikhub config = WorkerConfig() @@ -44,11 +45,10 @@ with fastapi_image.imports(): def fastapi_webapp(): return web_app + # 短保持时间保证IP不停更换 @app.function(scaledown_window=2, - secrets=[ - modal.Secret.from_name("cf-kv-secret"), - ], + secrets=[modal.Secret.from_name("cf-kv-secret"), ], cloud="aws", volumes={ config.S3_mount_dir: modal.CloudBucketMount( diff --git a/tests/ffmpeg_test_case.py b/tests/ffmpeg_test_case.py index 41e73bd..113c5fa 100644 --- a/tests/ffmpeg_test_case.py +++ b/tests/ffmpeg_test_case.py @@ -5,8 +5,8 @@ from typing import List, Optional import httpx from loguru import logger from tqdm import tqdm -from BowongModalFunctions.models.ffmpeg_worker_model import FFMpegSliceSegment -from BowongModalFunctions.utils.VideoUtils import VideoUtils, FFMPEGSliceOptions +from BowongModalFunctions.models.ffmpeg_tasks.models import FFMpegSliceSegment, FFMPEGSliceOptions +from BowongModalFunctions.utils.VideoUtils import VideoUtils class FFMPEGTestCase(unittest.IsolatedAsyncioTestCase): diff --git a/tests/pydantic_model_validation_test.py b/tests/pydantic_model_validation_test.py index 87b2149..5808b56 100644 --- a/tests/pydantic_model_validation_test.py +++ b/tests/pydantic_model_validation_test.py @@ -3,7 +3,7 @@ import unittest import httpx from loguru import logger -from BowongModalFunctions.models.web_model import FFMPEGSubtitleOverlayRequest +from BowongModalFunctions.models.requests.models import FFMPEGSubtitleOverlayRequest class PydanticModelTestCase(unittest.TestCase):