@@ -1,7 +1,7 @@
import asyncio
import datetime
import os
from typing import Annotated , Optional
from typing import Annotated , Optional , List , Tuple
import modal
from loguru import logger
@@ -22,10 +22,10 @@ from ..models.media_model import (MediaSources,
DownloadResult ,
UploadResultResponse ,
UploadBase64Request , UploadPresignRequest , UploadPresignResponse ,
UploadMultipartPresignRequest , UploadMultipartPresignResponse
UploadMultipartPresignRequest , UploadMultipartPresignResponse , MediaProtocol
)
from . . models . web_model import SentryTransactionInfo , MonitorLiveRoomProductRequest , ModalTaskResponse , \
LiveRoomProductCachesResponse
LiveRoomProductCachesResponse , CacheDeleteTaskResponse
from . . utils . KVCache import MediaSourceKVCache , LiveProductKVCache
from . . utils . SentryUtils import SentryUtils
@@ -127,29 +127,55 @@ async def cache(medias: MediaSources) -> CacheResult:
summary = " 清除指定的所有缓存 " ,
description = " 清除指定的所有缓存(包括KV记录和S3存储文件) " ,
dependencies = [ Depends ( verify_token ) ] )
async def purge_media_kv_file ( medias : MediaSources ) :
async def purge_media_kv_file ( medias : MediaSources ) - > CacheDeleteTaskResponse :
fn_id = current_function_call_id ( )
fn = modal . Function . from_name ( config . modal_app_name , " cache_delete " , environment_name = config . modal_environment )
@SentryUtils.sentry_tracker ( name = " 清除媒体源缓存 " , op = " cache.purge " , fn_id = fn_id ,
sentry_trace_id = None , sentry_baggage = None )
async def purge_handle ( media : MediaSource ) :
cache_media = modal_kv_cache . pop ( media . urn )
if cache_media :
deleted_cache : MediaSource = await fn . remote . aio ( cache_media )
return deleted_cache . urn
return None
async def purge_handle ( media : MediaSource ) - > Tuple [ Optional [ str ] , int ] :
try :
cache_media = modal_kv_cache . pop ( media . urn )
if cache_media :
deleted_cache : MediaSource = await fn . remote . aio ( cache_media )
return deleted_cache . urn , 1
except KeyError as e :
logger . exception ( e )
if media . local_exists :
deleted_cache : MediaSource = await fn . remote . aio ( media )
if deleted_cache . status == MediaCacheStatus . missing :
logger . warning ( f " 不存在s3挂载文件 { deleted_cache . urn } " )
return None , 0
else :
logger . warning ( f " s3挂载文件 { deleted_cache . urn } 已删除 " )
return deleted_cache . urn , 0
return media . urn , - 1
async with asyncio . TaskGroup ( ) as group :
tasks = [ group . create_task ( purge_handle ( media ) ) for media in medias . inputs ]
keys = [task . result ( ) for task in tasks ]
keys : List [ str ] = [ ]
non_kv_keys : List [ str ] = [ ]
error_keys : List [ str ] = [ ]
for task in tasks :
urn , task_status = task . result ( )
if urn :
if task_status == 1 : # 成功从kv和s3删除
keys . append ( urn )
elif task_status == 2 : # 只从s3删除
non_kv_keys . append ( urn )
else :
error_keys . append ( urn )
# keys = [task.result() for task in tasks]
modal_kv_cache . batch_remove_cloudflare_kv ( keys )
return JSONResponse ( content = { " success" : True , " keys " : keys } )
return CacheDeleteTaskResponse ( success= True , keys = keys , nonKVKeys = non_kv_keys , notFoundKeys = error_keys )
# return JSONResponse(content={"success": True, "keys": keys, "nonKVKeys": non_kv_keys, "errorKeys": error_keys})
@router.post ( " /download " ,
summary = " 批量获取下载地址 " ,
summary = " 批量获取下载地址, 返回的MediaSource类自带CDN访问URL, 不需要另外请求获取 " ,
deprecated = True ,
description = " 获取已缓存的视频下载地址 " ,
dependencies = [ Depends ( verify_token ) ] )
@sentry_sdk.trace
@@ -163,7 +189,8 @@ async def download_caches(medias: MediaSources) -> DownloadResult:
@router.get ( " /download " ,
summary = " 下载已缓存的视频 " ,
description = " 通过CDN下载已缓存的视频文件 " )
deprecated = True ,
description = " 通过CDN下载已缓存的视频文件, 不在提供通过此接口下载, 请使用CDN URL直接下载文件 " )
@sentry_sdk.trace
async def download_cache ( media : str ) - > RedirectResponse :
cdn_endpoint = config . S3_cdn_endpoint
@@ -199,32 +226,6 @@ async def purge_kv(medias: MediaSources):
return JSONResponse ( content = { " success " : False , " error " : str ( e ) } )
@router.post ( " /media " ,
summary = " 清除指定的所有缓存 " ,
description = " 清除指定的所有缓存(包括KV记录和S3存储文件), 将要被淘汰, 使用DELETE /cache/替代 " ,
deprecated = True ,
dependencies = [ Depends ( verify_token ) ] )
async def purge_media ( medias : MediaSources ) :
fn_id = current_function_call_id ( )
fn = modal . Function . from_name ( config . modal_app_name , " cache_delete " , environment_name = config . modal_environment )
@SentryUtils.sentry_tracker ( name = " 清除媒体源缓存 " , op = " cache.purge " , fn_id = fn_id ,
sentry_trace_id = None , sentry_baggage = None )
async def purge_handle ( media : MediaSource ) :
cache_media = modal_kv_cache . pop ( media . urn )
if cache_media :
deleted_cache : MediaSource = await fn . remote . aio ( cache_media )
return deleted_cache . urn
return None
async with asyncio . TaskGroup ( ) as group :
tasks = [ group . create_task ( purge_handle ( media ) ) for media in medias . inputs ]
keys = [ task . result ( ) for task in tasks ]
modal_kv_cache . batch_remove_cloudflare_kv ( keys )
return JSONResponse ( content = { " success " : True , " keys " : keys } )
@router.post ( " /upload-s3 " ,
summary = " 上传文件到S3 " ,
description = " 上传文件到S3的文件必须小于200M " ,