news 2025/12/16 0:11:54

【剪映小助手源码精讲】第30章 素材获取服务

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
【剪映小助手源码精讲】第30章 素材获取服务

第30章 素材获取服务

30.1 概述

素材获取服务是剪映小助手的基础功能模块，主要负责获取各种媒体素材的信息，包括音频时长、图片动画效果、文字动画效果等。该服务通过分析媒体文件的内容，为视频编辑提供必要的素材信息支持。服务支持多种媒体格式，采用异步处理方式，确保高效和稳定的性能表现。

30.2 音频时长获取服务

30.2.1 核心实现

音频时长获取服务的核心实现位于src/service/get_audio_duration.py文件中：

asyncdefget_audio_duration(request:GetAudioDurationRequest)->GetAudioDurationResponse:"""获取音频时长"""logger.info(f"获取音频时长:{request.mp3_url}")# 参数验证ifnotrequest.mp3_url:raiseValueError("音频URL不能为空")# 下载音频文件audio_file=awaitdownload_audio_file(str(request.mp3_url))ifnotaudio_file:raiseAUDIO_DOWNLOAD_FAILEDtry:# 获取音频时长duration=awaitextract_audio_duration(audio_file)logger.info(f"音频时长获取成功:{duration}微秒")returnGetAudioDurationResponse(duration=duration)finally:# 清理临时文件ifos.path.exists(audio_file):os.remove(audio_file)logger.debug(f"清理临时文件:{audio_file}")

30.2.2 音频文件下载

下载音频文件到临时目录：

asyncdefdownload_audio_file(mp3_url:str)->str:"""下载音频文件"""try:# 创建临时文件temp_dir=tempfile.gettempdir()file_extension=get_file_extension(mp3_url)temp_file=os.path.join(temp_dir,f"audio_{uuid.uuid4().hex}.{file_extension}")logger.info(f"开始下载音频文件:{mp3_url}")# 下载文件asyncwithaiohttp.ClientSession()assession:timeout=aiohttp.ClientTimeout(total=30)asyncwithsession.get(mp3_url,timeout=timeout)asresponse:ifresponse.status!=200:raiseException(f"下载失败，状态码:{response.status}")# 写入临时文件withopen(temp_file,'wb')asf:asyncforchunkinresponse.content.iter_chunked(8192):f.write(chunk)# 验证文件大小file_size=os.path.getsize(temp_file)iffile_size==0:raiseException("下载的文件为空")logger.info(f"音频文件下载成功:{temp_file}, 大小:{file_size}bytes")returntemp_fileexceptExceptionase:logger.error(f"音频文件下载失败:{str(e)}")ifos.path.exists(temp_file):os.remove(temp_file)raiseException(f"音频文件下载失败:{str(e)}")

30.2.3 音频时长提取

使用ffprobe提取音频时长：

asyncdefextract_audio_duration(audio_file:str)->int:"""提取音频时长"""try:# 构建ffprobe命令cmd=['ffprobe','-v','quiet','-print_format','json','-show_format','-show_streams',audio_file]logger.info(f"执行ffprobe命令:{' '.join(cmd)}")# 执行命令process=awaitasyncio.create_subprocess_exec(*cmd,stdout=asyncio.subprocess.PIPE,stderr=asyncio.subprocess.PIPE)stdout,stderr=awaitprocess.communicate()ifprocess.returncode!=0:error_msg=stderr.decode('utf-8',errors='ignore')logger.error(f"ffprobe执行失败:{error_msg}")# 尝试备用方法returnawaitextract_duration_fallback(audio_file)# 解析JSON输出try:probe_data=json.loads(stdout.decode('utf-8'))exceptjson.JSONDecodeErrorase:logger.error(f"ffprobe输出解析失败:{str(e)}")returnawaitextract_duration_fallback(audio_file)# 查找音频流duration=Noneforstreaminprobe_data.get('streams',[]):ifstream.get('codec_type')=='audio':# 优先使用流的时长if'duration'instream:duration=float(stream['duration'])break# 如果没有流时长，使用格式时长elif'duration'inprobe_data.get('format',{}):duration=float(probe_data['format']['duration'])breakifdurationisNone:logger.error("未找到音频时长信息")returnawaitextract_duration_fallback(audio_file)# 转换为微秒duration_microseconds=int(duration*1000000)logger.info(f"音频时长:{duration}秒 ={duration_microseconds}微秒")returnduration_microsecondsexceptExceptionase:logger.error(f"音频时长提取失败:{str(e)}")raiseException(f"音频时长提取失败:{str(e)}")

30.2.4 备用时长提取方法

当ffprobe失败时的备用方法：

asyncdefextract_duration_fallback(audio_file:str)->int:"""备用时长提取方法"""try:# 使用ffprobe的简化模式cmd=['ffprobe','-v','error','-show_entries','format=duration','-of','default=noprint_wrappers=1:nokey=1',audio_file]process=awaitasyncio.create_subprocess_exec(*cmd,stdout=asyncio.subprocess.PIPE,stderr=asyncio.subprocess.PIPE)stdout,stderr=awaitprocess.communicate()ifprocess.returncode==0:duration=float(stdout.decode('utf-8').strip())duration_microseconds=int(duration*1000000)logger.info(f"备用方法获取音频时长:{duration_microseconds}微秒")returnduration_microsecondselse:raiseException(f"备用方法也失败:{stderr.decode()}")exceptExceptionase:logger.error(f"备用时长提取方法失败:{str(e)}")raiseException(f"无法获取音频时长:{str(e)}")

30.3 图片动画获取服务

30.3.1 图片入场动画

获取图片的入场动画效果：

asyncdefget_image_animations(request:GetImageAnimationsRequest)->GetImageAnimationsResponse:"""获取图片出入场动画"""logger.info(f"获取图片动画，类型:{request.type}, 模式:{request.mode}")# 参数验证ifnotrequest.type:raiseValueError("动画类型不能为空")# 获取动画列表animations=awaitload_image_animations(request.type,request.mode)# 过滤和排序filtered_animations=filter_animations(animations,request)returnGetImageAnimationsResponse(effects=json.dumps([anim.dict()foraniminfiltered_animations],ensure_ascii=False))

30.3.2 动画数据加载

从素材库加载动画数据：

asyncdefload_image_animations(animation_type:str,mode:int)->List[ImageAnimationItem]:"""加载图片动画数据"""try:# 构建查询条件filters={'type':animation_type,'material_type':'sticker','platform':'all'}# 根据模式过滤ifmode==1:# VIPfilters['is_vip']=Trueelifmode==2:# 免费filters['is_free']=True# 从数据库或缓存获取动画数据animation_data=awaitget_animation_data('image',filters)# 转换为模型对象animations=[]fordatainanimation_data:animation=ImageAnimationItem(resource_id=data['resource_id'],type=data['type'],category_id=data['category_id'],category_name=data['category_name'],duration=data['duration'],id=data['id'],name=data['name'],icon_url=data['icon_url'],material_type=data.get('material_type','sticker'),panel=data.get('panel',''),path=data.get('path',''),platform=data.get('platform','all'))animations.append(animation)logger.info(f"加载到{len(animations)}个图片动画")returnanimationsexceptExceptionase:logger.error(f"加载图片动画失败:{str(e)}")raiseException(f"加载图片动画失败:{str(e)}")

30.4 文字动画获取服务

30.4.1 文字出入场动画

获取文字的出入场动画效果：

asyncdefget_text_animations(request:GetTextAnimationsRequest)->GetTextAnimationsResponse:"""获取文字出入场动画"""logger.info(f"获取文字动画，类型:{request.type}, 模式:{request.mode}")# 参数验证ifnotrequest.type:raiseValueError("动画类型不能为空")# 获取动画列表animations=awaitload_text_animations(request.type,request.mode)# 过滤和排序filtered_animations=filter_animations(animations,request)returnGetTextAnimationsResponse(effects=json.dumps([anim.dict()foraniminfiltered_animations],ensure_ascii=False))

30.4.2 文字动画特性

文字动画的特殊处理：

defprocess_text_animation(animation:TextAnimationItem)->TextAnimationItem:"""处理文字动画的特殊属性"""# 文字动画通常需要更短的持续时间ifanimation.duration>2000000:# 超过2秒animation.duration=1500000# 调整为1.5秒# 设置文字动画的默认缓动函数ifnotanimation.path:animation.path="ease_in_out"# 根据动画类型调整参数ifanimation.type=="in":# 入场动画从透明到不透明animation.start=0elifanimation.type=="out":# 出场动画从不透明到透明animation.start=500000# 延迟0.5秒开始elifanimation.type=="loop":# 循环动画持续进行animation.start=0returnanimation

30.5 数据结构定义

30.5.1 音频时长数据结构

classGetAudioDurationRequest(BaseModel):"""获取音频时长请求参数"""mp3_url:HttpUrl=Field(...,description="音频文件URL，支持mp3、wav、m4a等常见音频格式")classConfig:json_schema_extra={"example":{"mp3_url":"https://www.soundjay.com/misc/sounds/bell-ringing-05.wav"}}classGetAudioDurationResponse(BaseModel):"""获取音频时长响应参数"""duration:int=Field(...,description="音频时长，单位：微秒",ge=0)

30.5.2 图片动画数据结构

classGetImageAnimationsRequest(BaseModel):"""获取图片出入场动画的请求模型"""mode:int=Field(default=0,description="动画模式：0=所有，1=VIP，2=免费")type:Literal["in","out","loop"]=Field(...,description="动画类型：in=入场，out=出场，loop=循环")classImageAnimationItem(BaseModel):"""单个图片动画项的数据模型"""resource_id:str=Field(...,description="动画资源ID")type:str=Field(...,description="动画类型")category_id:str=Field(...,description="动画分类ID")category_name:str=Field(...,description="动画分类名称")duration:int=Field(...,description="动画时长（微秒）")id:str=Field(...,description="动画唯一标识ID")name:str=Field(...,description="动画名称")request_id:str=Field(default="",description="请求ID")start:int=Field(default=0,description="动画开始时间")icon_url:str=Field(...,description="动画图标URL")material_type:str=Field(default="sticker",description="素材类型")panel:str=Field(default="",description="面板信息")path:str=Field(default="",description="路径信息")platform:str=Field(default="all",description="支持平台")classGetImageAnimationsResponse(BaseModel):"""获取图片出入场动画的响应模型"""effects:str=Field(...,description="图片出入场动画数组的JSON字符串")

30.5.3 文字动画数据结构

classGetTextAnimationsRequest(BaseModel):"""获取文字出入场动画的请求模型"""mode:int=Field(default=0,description="动画模式：0=所有，1=VIP，2=免费")type:Literal["in","out","loop"]=Field(...,description="动画类型：in=入场，out=出场，loop=循环")classTextAnimationItem(BaseModel):"""单个文字动画项的数据模型"""resource_id:str=Field(...,description="动画资源ID")type:str=Field(...,description="动画类型")category_id:str=Field(...,description="动画分类ID")category_name:str=Field(...,description="动画分类名称")duration:int=Field(...,description="动画时长（微秒）")id:str=Field(...,description="动画唯一标识ID")name:str=Field(...,description="动画名称")request_id:str=Field(default="",description="请求ID")start:int=Field(default=0,description="动画开始时间")icon_url:str=Field(...,description="动画图标URL")material_type:str=Field(default="sticker",description="素材类型")panel:str=Field(default="",description="面板信息")path:str=Field(default="",description="路径信息")platform:str=Field(default="all",description="支持平台")classGetTextAnimationsResponse(BaseModel):"""获取文字出入场动画的响应模型"""effects:str=Field(...,description="文字出入场动画数组的JSON字符串")

30.6 异常处理

素材获取服务定义了完善的异常处理机制：

# 音频下载失败AUDIO_DOWNLOAD_FAILED=HTTPException(status_code=400,detail="音频文件下载失败")# 音频时长提取失败AUDIO_DURATION_EXTRACTION_FAILED=HTTPException(status_code=500,detail="音频时长提取失败")# 动画数据加载失败ANIMATION_DATA_LOAD_FAILED=HTTPException(status_code=500,detail="动画数据加载失败")# 不支持的音频格式UNSUPPORTED_AUDIO_FORMAT=HTTPException(status_code=400,detail="不支持的音频格式")

30.7 API接口定义

30.7.1 音频时长接口

@router.post("/getAudioDuration",response_model=GetAudioDurationResponse)asyncdefget_audio_duration_endpoint(request:GetAudioDurationRequest):"""获取音频时长"""try:returnawaitget_audio_duration(request)exceptExceptionase:logger.error(f"获取音频时长失败:{str(e)}")raiseAUDIO_DURATION_EXTRACTION_FAILED

30.7.2 图片动画接口

@router.post("/getImageAnimations",response_model=GetImageAnimationsResponse)asyncdefget_image_animations_endpoint(request:GetImageAnimationsRequest):"""获取图片出入场动画"""try:returnawaitget_image_animations(request)exceptExceptionase:logger.error(f"获取图片动画失败:{str(e)}")raiseANIMATION_DATA_LOAD_FAILED

30.7.3 文字动画接口

@router.post("/getTextAnimations",response_model=GetTextAnimationsResponse)asyncdefget_text_animations_endpoint(request:GetTextAnimationsRequest):"""获取文字出入场动画"""try:returnawaitget_text_animations(request)exceptExceptionase:logger.error(f"获取文字动画失败:{str(e)}")raiseANIMATION_DATA_LOAD_FAILED

30.8 使用示例

30.8.1 音频时长请求示例

{"mp3_url":"https://example.com/audio/background-music.mp3"}

30.8.2 音频时长响应示例

{"duration":180000000}

30.8.3 动画获取请求示例

{"type":"in","mode":0}

30.8.4 动画获取响应示例

{"effects":"[{\"resource_id\":\"anim_fade_in\",\"type\":\"in\",\"category_id\":\"basic\",\"category_name\":\"基础\",\"duration\":1000000,\"id\":\"fade_in_001\",\"name\":\"淡入\",\"icon_url\":\"https://example.com/icons/fade_in.png\",\"material_type\":\"sticker\",\"platform\":\"all\"}]"}

30.9 性能优化

素材获取服务采用了多种性能优化策略：

30.9.1 缓存优化

# 使用Redis缓存动画数据importaioredisclassAnimationCache:def__init__(self):self.redis=Noneasyncdefinit_cache(self):self.redis=awaitaioredis.create_redis_pool('redis://localhost:6379',encoding='utf-8')asyncdefget_animation_data(self,key:str):"""获取缓存的动画数据"""cached_data=awaitself.redis.get(key)ifcached_data:returnjson.loads(cached_data)returnNoneasyncdefset_animation_data(self,key:str,data:dict,expire:int=3600):"""设置动画数据缓存"""awaitself.redis.setex(key,expire,json.dumps(data))

30.9.2 连接池优化

# 使用连接池管理HTTP连接classHttpConnectionPool:def__init__(self):self.connector=aiohttp.TCPConnector(limit=100,limit_per_host=30,ttl_dns_cache=300,use_dns_cache=True,keepalive_timeout=30)self.session=aiohttp.ClientSession(connector=self.connector)asyncdefdownload_file(self,url:str,timeout:int=30):"""下载文件"""timeout_config=aiohttp.ClientTimeout(total=timeout)asyncwithself.session.get(url,timeout=timeout_config)asresponse:ifresponse.status==200:returnawaitresponse.read()else:raiseException(f"下载失败:{response.status}")

30.9.3 异步处理优化

# 使用Semaphore限制并发数classAsyncLimiter:def__init__(self,max_concurrent:int=10):self.semaphore=asyncio.Semaphore(max_concurrent)asyncdefacquire(self):awaitself.semaphore.acquire()defrelease(self):self.semaphore.release()asyncdef__aenter__(self):awaitself.acquire()returnselfasyncdef__aexit__(self,exc_type,exc_val,exc_tb):self.release()# 使用示例limiter=AsyncLimiter(max_concurrent=5)asyncdeflimited_download(url:str):asyncwithlimiter:returnawaitdownload_file(url)

30.10 扩展性设计

素材获取服务具有良好的扩展性：

  • 媒体格式扩展：易于添加新的媒体格式支持
  • 动画类型扩展：支持动态添加新的动画类型
  • 数据源扩展：支持从多个数据源获取素材信息
  • 缓存策略扩展：支持自定义缓存策略和过期时间

附录

代码仓库地址：

  • GitHub:https://github.com/Hommy-master/capcut-mate
  • Gitee:https://gitee.com/taohongmin-gitee/capcut-mate

接口文档地址：

  • API文档地址:https://docs.jcaigc.cn
版权声明: 本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若内容造成侵权/违法违规/事实不符，请联系邮箱：809451989@qq.com进行投诉反馈，一经查实，立即删除！