From d0fee9c6e33f1da8c28048a314a3917c359eb2c7 Mon Sep 17 00:00:00 2001 From: "shuohigh@gmail.com" Date: Fri, 6 Jun 2025 17:00:41 +0800 Subject: [PATCH] =?UTF-8?q?fix=20:=20=E5=A2=9E=E5=8A=A0s3=E5=88=86?= =?UTF-8?q?=E7=89=87=E4=B8=8A=E4=BC=A0=E6=8E=A5=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .runtime.env | 2 - src/BowongModalFunctions/api.py | 1 - src/BowongModalFunctions/config.py | 4 -- .../models/media_model.py | 37 +++++++++- src/BowongModalFunctions/router/cache.py | 72 ++++++++++++++----- src/BowongModalFunctions/utils/KVCache.py | 9 +-- src/BowongModalFunctions/utils/SentryUtils.py | 5 +- src/BowongModalFunctions/utils/VideoUtils.py | 2 +- src/cluster/app.py | 7 +- src/cluster/video.py | 3 +- src/cluster/web.py | 1 + 11 files changed, 104 insertions(+), 39 deletions(-) diff --git a/.runtime.env b/.runtime.env index 6170374..f0442f9 100644 --- a/.runtime.env +++ b/.runtime.env @@ -4,5 +4,3 @@ S3_mount_dir=/mntS3 S3_bucket_name=modal-media-cache S3_region=ap-northeast-2 S3_cdn_endpoint=https://d2nj71io21vkj2.cloudfront.net -#CF_KV_namespace_id=f24d396e0daa418e89a1d7074b435c24 -CF_KV_namespace_id=527a61fea05543f2a49d62889ba868c5 \ No newline at end of file diff --git a/src/BowongModalFunctions/api.py b/src/BowongModalFunctions/api.py index 5251722..265868b 100644 --- a/src/BowongModalFunctions/api.py +++ b/src/BowongModalFunctions/api.py @@ -33,7 +33,6 @@ sentry_sdk.init(dsn="https://dab7b7ae652216282c89f029a76bb10a@sentry.bowongai.co ] ) modal_kv_cache = MediaSourceKVCache(kv_name=config.modal_kv_name, - cf_kv_id=config.CF_KV_namespace_id, environment=config.modal_environment, ) sentry_header_schema = { diff --git a/src/BowongModalFunctions/config.py b/src/BowongModalFunctions/config.py index b8c2ed7..1dc3dfd 100644 --- a/src/BowongModalFunctions/config.py +++ b/src/BowongModalFunctions/config.py @@ -14,10 +14,6 @@ class WorkerConfig(BaseSettings): S3_mount_dir: str = Field(default='/mntS3', description="集群S3存储桶挂载在本地的根目录") S3_cdn_endpoint: str = Field(default="https://d2nj71io21vkj2.cloudfront.net", description="集群挂载S3存储桶的对应AWS Cloudfront CDN") - - CF_KV_namespace_id: str = Field(default="527a61fea05543f2a49d62889ba868c5", - description="Cloudflare KV namespace ID") - modal_kv_name: str = Field(default='media-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集群名称") diff --git a/src/BowongModalFunctions/models/media_model.py b/src/BowongModalFunctions/models/media_model.py index fd13749..058ac00 100644 --- a/src/BowongModalFunctions/models/media_model.py +++ b/src/BowongModalFunctions/models/media_model.py @@ -7,7 +7,7 @@ from functools import cached_property from typing import List, Union, Optional, Any, Dict from urllib.parse import urlparse from pydantic import (BaseModel, Field, field_validator, ValidationError, - field_serializer, SerializationInfo, computed_field, FileUrl, Base64Str, Base64Bytes) + field_serializer, SerializationInfo, computed_field, FileUrl, Base64Str, Base64Bytes, ConfigDict) from pydantic.json_schema import JsonSchemaValue from ..config import WorkerConfig from ..utils.TimeUtils import TimeDelta @@ -270,7 +270,42 @@ class UploadPresignRequest(BaseModel): key: str = Field(description="上传文件的key", examples=['123/456/abc.mp4']) content_type: str = Field(description="上传对象的文件类型", examples=['video/mp4']) + class UploadPresignResponse(BaseModel): url: str = Field(description="就近加速的PUT上传地址") urn: str = Field(description="上传成功后获得的对应资源URN") expired_at: datetime = Field(description="上传地址签名过期时间戳") + + +class UploadMultipartPresignRequest(UploadPresignRequest): + parts_count: int = 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="上传地址签名过期时间戳") + + model_config = ConfigDict(json_schema_extra={ + "description": """ + 1. 本地按文件总大小分Chunk大小按urls链接内的顺序通过HTTP PUT请求上传文件分片; 并将上传完成后获得的返回头ETag值记录与PartNumber对应, PartNumber对应使用url在urls内的顺位, 以1开始 + 2. 所有分片上传完成后使用XML格式拼装出用于确认上传的body; 并通过HTTP POST complete_url确认上传, ContentType 需要确保为application/xml + + + 1 + "60575364b098a1a48765a28c3a48e0ef" + + + 2 + "a38691c31fd242faee5533c65b4501d7" + + + 3 + "7a7510cc83f98feea28f319567e4cf66" + + + 3. 如无法确认上传，可使用HTTP GET list_url debug当前分片上传状态，确认成功后list_url无法返回有效数据 + """ + }) diff --git a/src/BowongModalFunctions/router/cache.py b/src/BowongModalFunctions/router/cache.py index 8e4fa58..b01479a 100644 --- a/src/BowongModalFunctions/router/cache.py +++ b/src/BowongModalFunctions/router/cache.py @@ -10,6 +10,9 @@ import sentry_sdk from fastapi import APIRouter, Depends, UploadFile, HTTPException, File, Form from fastapi.responses import JSONResponse, RedirectResponse from starlette import status +import boto3 +from botocore.config import Config + from ..config import WorkerConfig from ..middleware.authorization import verify_token from ..models.media_model import (MediaSources, @@ -18,7 +21,8 @@ from ..models.media_model import (MediaSources, MediaCacheStatus, DownloadResult, UploadResultResponse, - UploadBase64Request, UploadPresignRequest, UploadPresignResponse + UploadBase64Request, UploadPresignRequest, UploadPresignResponse, + UploadMultipartPresignRequest, UploadMultipartPresignResponse ) from ..models.web_model import SentryTransactionInfo from ..utils.KVCache import MediaSourceKVCache @@ -26,12 +30,19 @@ from ..utils.SentryUtils import SentryUtils config = WorkerConfig() +client = boto3.client("s3", + aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID"), + aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY"), + region_name=config.S3_region, + endpoint_url="https://s3-accelerate.amazonaws.com", + config=Config( + s3={'addressing_style': 'virtual'}, + signature_version='s3v4', ) + ) + router = APIRouter(prefix="/cache", tags=['缓存'], ) -if not config.CF_KV_namespace_id: - raise ValueError("未配置Cloudflare KV namespace ID") modal_kv_cache = MediaSourceKVCache(kv_name=config.modal_kv_name, - cf_kv_id=config.CF_KV_namespace_id, environment=config.modal_environment) @@ -257,24 +268,11 @@ async def s3_upload_base64(body: UploadBase64Request) -> UploadResultResponse: return UploadResultResponse(media=media_source) -import boto3 -from botocore.config import Config - - @router.post('/upload-s3/simple/presign', summary="S3简单上传预签名", description="利用S3就近接入点上传", dependencies=[Depends(verify_token)]) async def s3_presign_upload(body: UploadPresignRequest) -> UploadPresignResponse: - client = boto3.client("s3", - aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID"), - aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY"), - region_name=config.S3_region, - endpoint_url="https://s3-accelerate.amazonaws.com", - config=Config( - s3={'addressing_style': 'virtual'}, - signature_version='s3v4', ) - ) expires_in = 3600 expired_at = datetime.datetime.now() + datetime.timedelta(seconds=expires_in) signed_url = client.generate_presigned_url("put_object", @@ -286,3 +284,43 @@ async def s3_presign_upload(body: UploadPresignRequest) -> UploadPresignResponse return UploadPresignResponse(url=signed_url, urn=f"s3://{config.S3_region}/{config.S3_bucket_name}/upload/{body.key}", expired_at=expired_at) + + +@router.post("/upload-s3/multipart/presign", + summary="S3分片上传预签名", + description="", dependencies=[Depends(verify_token)]) +async def s3_presign_upload_multipart(body: UploadMultipartPresignRequest) -> UploadMultipartPresignResponse: + chunk_count = body.parts_count + multipart_upload_response = client.create_multipart_upload(Bucket=config.S3_bucket_name, Key=body.key, + ContentType=body.content_type, ) + upload_id = multipart_upload_response.get("UploadId") + signed_urls = [] + expires_in = 3600 + expired_at = datetime.datetime.now() + datetime.timedelta(seconds=expires_in) + for i in range(chunk_count): + signed_url = client.generate_presigned_url("upload_part", + Params={ + 'Bucket': config.S3_bucket_name, + 'Key': body.key, + 'PartNumber': i + 1, + 'UploadId': upload_id, + }, ExpiresIn=expires_in) + signed_urls.append(signed_url) + + signed_completed_url = client.generate_presigned_url("complete_multipart_upload", + Params={ + 'Bucket': config.S3_bucket_name, + 'Key': body.key, + 'UploadId': upload_id, + }, ExpiresIn=expires_in) + signed_list_url = client.generate_presigned_url("list_parts", + Params={ + 'Bucket': config.S3_bucket_name, + 'Key': body.key, + 'UploadId': upload_id, + }, ExpiresIn=expires_in) + return UploadMultipartPresignResponse(urls=signed_urls, + urn=f"s3://{config.S3_region}/{config.S3_bucket_name}/upload/{body.key}", + expired_at=expired_at, + complete_url=signed_completed_url, + list_url=signed_list_url) diff --git a/src/BowongModalFunctions/utils/KVCache.py b/src/BowongModalFunctions/utils/KVCache.py index 5ecbe70..a1b8a54 100644 --- a/src/BowongModalFunctions/utils/KVCache.py +++ b/src/BowongModalFunctions/utils/KVCache.py @@ -11,16 +11,18 @@ from ..models.media_model import MediaSource # cf_kv_api_token = os.environ.get("CF_KV_API_TOKEN") # cf_kv_namespace_id = os.environ.get("CF_KV_NAMESPACE_ID") +# secrets = modal.Secret.from_name("cf-kv-secret") class KVCache: kv: modal.Dict - cf_kv_id: str + cf_kv_id: str = os.environ.get("CF_KV_NAMESPACE_ID") cf_account_id: str = os.environ.get("CF_ACCOUNT_ID") cf_kv_api_token: str = os.environ.get("CF_KV_API_TOKEN") - def __init__(self, kv_name: str, cf_kv_id: str, environment: str): - self.cf_kv_id = cf_kv_id + def __init__(self, kv_name: str, environment: str): + # self.cf_kv_id = cf_kv_id self.kv = modal.Dict.from_name(kv_name, environment_name=environment, create_if_missing=True) + logger.info(f"Using KV space : {self.cf_kv_id}") def batch_update_cloudflare_kv(self, caches: Dict[str, str]): with httpx.Client() as client: @@ -69,7 +71,6 @@ class KVCache: class MediaSourceKVCache(KVCache): - def get_cache(self, urn: str) -> Optional[MediaSource]: cache_json = self.kv.get(urn) if not cache_json: diff --git a/src/BowongModalFunctions/utils/SentryUtils.py b/src/BowongModalFunctions/utils/SentryUtils.py index 3e15b76..159c37b 100644 --- a/src/BowongModalFunctions/utils/SentryUtils.py +++ b/src/BowongModalFunctions/utils/SentryUtils.py @@ -64,11 +64,10 @@ class SentryUtils: return decorator @staticmethod - def webhook_handler(webhook: WebhookNotify, func_id: str): + def webhook_handler(webhook: WebhookNotify, func_id: str, raise_on_giveup: bool = False): def decorator(func): - @backoff.on_exception(exception=Exception, wait_gen=backoff.constant, - max_time=15, max_tries=5, raise_on_giveup=True) + max_time=15, max_tries=5, raise_on_giveup=raise_on_giveup) def webhook_with_retry(webhook: WebhookNotify, body: BaseFFMPEGTaskStatusResponse): if webhook.method == WebhookMethodEnum.POST: response = httpx.post(url=webhook.endpoint.__str__(), diff --git a/src/BowongModalFunctions/utils/VideoUtils.py b/src/BowongModalFunctions/utils/VideoUtils.py index 53b1b57..ca50a58 100644 --- a/src/BowongModalFunctions/utils/VideoUtils.py +++ b/src/BowongModalFunctions/utils/VideoUtils.py @@ -137,7 +137,7 @@ class VideoMetadata(BaseModel): audio = AudioStream.model_validate(stream) streams.append(audio) elif stream.get("codec_type") == 'video': - if stream.get("codec_name") in ("gif", "png", "mjpg", "jpeg", "webp"): + if stream.get("codec_name") in ("gif", "png", "mjpg", "mjpeg", "webp"): logger.info("Parsing image stream") image = ImageStream.model_validate(stream) streams.append(image) diff --git a/src/cluster/app.py b/src/cluster/app.py index eaa291f..e1365bd 100644 --- a/src/cluster/app.py +++ b/src/cluster/app.py @@ -3,15 +3,14 @@ from BowongModalFunctions.config import WorkerConfig from .video import app as media_app from .web import app as web_app from .ffmpeg_app import app as ffmpeg_app -from .comfyui_v1 import app as comfyui_v1_app -from .comfyui_v2 import app as comfyui_v2_app +# from .comfyui_v1 import app as comfyui_v1_app +# from .comfyui_v2 import app as comfyui_v2_app config = WorkerConfig() app = modal.App(config.modal_app_name, include_source=False, - secrets=[modal.Secret.from_name("cf-kv-secret", - environment_name=config.modal_environment)]) + secrets=[modal.Secret.from_name("cf-kv-secret")]) app.include(media_app) app.include(ffmpeg_app) diff --git a/src/cluster/video.py b/src/cluster/video.py index 5f7a0fd..35016e0 100644 --- a/src/cluster/video.py +++ b/src/cluster/video.py @@ -16,7 +16,7 @@ app = modal.App( image=downloader_image, include_source=False, secrets=[ - modal.Secret.from_name("cf-kv-secret", environment_name='dev'), + modal.Secret.from_name("cf-kv-secret"), ]) with downloader_image.imports(): @@ -55,7 +55,6 @@ with downloader_image.imports(): cf_kv_namespace_id = os.environ.get("CF_KV_NAMESPACE_ID") modal_kv_cache = MediaSourceKVCache(kv_name=config.modal_kv_name, - cf_kv_id=config.CF_KV_namespace_id, environment=config.modal_environment) diff --git a/src/cluster/web.py b/src/cluster/web.py index 20d3fe3..24a4ec5 100644 --- a/src/cluster/web.py +++ b/src/cluster/web.py @@ -16,6 +16,7 @@ app = modal.App( secrets=[ modal.Secret.from_name('google-secret'), modal.Secret.from_name('aws-s3-secret'), + modal.Secret.from_name('cf-kv-secret'), ], include_source=False)