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

fix : 增加s3分片上传接口

Browse Source
This commit is contained in:
shuohigh@gmail.com
2025-06-06 17:00:41 +08:00
parent 45bb33e67a
commit d0fee9c6e3
11 changed files with 104 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.runtime.env
View File

@@ -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

1
src/BowongModalFunctions/api.py
View File

@@ -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 = {

4
src/BowongModalFunctions/config.py
View File

@@ -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集群名称")

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

@@ -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
<CompleteMultipartUpload>
<Part>
<PartNumber>1</PartNumber>
<ETag>"60575364b098a1a48765a28c3a48e0ef"</ETag>
</Part>
<Part>
<PartNumber>2</PartNumber>
<ETag>"a38691c31fd242faee5533c65b4501d7"</ETag>
</Part>
<Part>
<PartNumber>3</PartNumber>
<ETag>"7a7510cc83f98feea28f319567e4cf66"</ETag>
</Part>
</CompleteMultipartUpload>
3. 如无法确认上传，可使用HTTP GET list_url debug当前分片上传状态，确认成功后list_url无法返回有效数据
"""
})

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

@@ -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)

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

@@ -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:

5
src/BowongModalFunctions/utils/SentryUtils.py
View File

@@ -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__(),

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

@@ -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)

7
src/cluster/app.py
View File

@@ -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)

3
src/cluster/video.py
View File

@@ -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)

1
src/cluster/web.py
View File

@@ -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)