新增添加特效的接口

docs/add_effects.md

@@ -0,0 +1,120 @@
+# 添加特效接口文档
+
+## 接口概述
+向现有剪映草稿中添加视频特效轨道和特效片段。支持批量添加多种视频特效，包括边框特效、滤镜特效、动态特效等，为视频内容增加丰富的视觉效果。
+
+## 接口信息
+- **请求方式**: POST
+- **接口路径**: `/v1/add_effects`
+- **Content-Type**: `application/json`
+
+## 请求参数
+
+### 请求体 (Body)
+
+| 参数名 | 类型 | 必填 | 默认值 | 描述 |
+|--------|------|------|--------|------|
+| draft_url | string | 是 | "" | 目标草稿的完整URL |
+| effect_infos | string | 是 | "" | JSON字符串格式的特效信息数组 |
+
+#### effect_infos 字段格式
+
+`effect_infos` 是一个JSON字符串，包含特效信息数组，每个特效对象包含以下字段：
+
+```json
+[
+    {
+        "effect_title": "录制边框 III",  // 特效名称/标题，必选参数
+        "start": 0,                     // 特效开始时间（微秒），必选参数  
+        "end": 5000000                  // 特效结束时间（微秒），必选参数
+    }
+]
+```
+
+**字段说明**:
+- `effect_title`: 特效名称，必须是系统中已存在的特效名称
+- `start`: 特效开始时间，单位为微秒，必须大于等于0
+- `end`: 特效结束时间，单位为微秒，必须大于start
+
+### 常见特效名称
+
+| 特效类型 | 特效名称示例 |
+|----------|--------------|
+| 边框特效 | "录制边框 III", "简约边框", "霓虹边框" |
+| 滤镜特效 | "复古滤镜", "黑白滤镜", "暖色调" |
+| 动态特效 | "粒子效果", "光晕效果", "闪烁特效" |
+| 转场特效 | "淡入淡出", "推拉门", "马赛克转场" |
+
+## 请求示例
+
+```bash
+curl -X POST "http://localhost:8000/v1/add_effects" \
+-H "Content-Type: application/json" \
+-d '{
+  "draft_url": "https://ts.fyshark.com/#/cozeToJianyin?drafId=7427078525303048221",
+  "effect_infos": "[{\"effect_title\": \"录制边框 III\", \"start\": 0, \"end\": 5000000}, {\"effect_title\": \"复古滤镜\", \"start\": 2000000, \"end\": 7000000}]"
+}'
+```
+
+## 响应格式
+
+### 成功响应
+
+```json
+{
+  "code": 0,
+  "message": "Success",
+  "data": {
+    "draft_url": "https://ts.fyshark.com/#/cozeToJianyin?drafId=7427078525303048221",
+    "track_id": "effect_track_123",
+    "effect_ids": ["effect_001", "effect_002"],
+    "segment_ids": ["seg_001", "seg_002"]
+  }
+}
+```
+
+**响应字段说明**:
+- `draft_url`: 草稿URL
+- `track_id`: 创建的特效轨道ID
+- `effect_ids`: 添加的特效ID列表
+- `segment_ids`: 创建的特效片段ID列表
+
+### 错误响应
+
+```json
+{
+  "code": 2020,
+  "message": "无效的特效信息，请检查effect_infos字段值是否正确"
+}
+```
+
+## 错误码说明
+
+| 错误码 | 错误信息 | 描述 |
+|--------|----------|------|
+| 2001 | 无效的草稿URL | 草稿URL格式错误或草稿不存在 |
+| 2020 | 无效的特效信息 | effect_infos格式错误或字段值无效 |
+| 2021 | 特效添加失败 | 添加特效过程中发生错误 |
+| 2022 | 特效未找到 | 指定的特效名称不存在 |
+
+## 使用说明
+
+1. **时间单位**: 所有时间参数使用微秒（μs）为单位，1秒 = 1,000,000微秒
+2. **时间范围**: 确保 `start < end`，且时间范围合理
+3. **特效名称**: 特效名称必须完全匹配系统中预定义的特效名称，区分大小写
+4. **批量添加**: 支持一次添加多个特效，按数组顺序依次处理
+5. **轨道管理**: 系统会自动创建特效轨道，无需手动管理
+
+## 注意事项
+
+- 特效会按照指定的时间范围添加到视频中
+- 如果特效名称不存在，会返回特效未找到错误
+- 特效的时间范围不能超出视频总时长
+- 建议在添加特效前确保草稿中已有视频内容
+- 特效效果可能因版本差异而有所不同
+
+## 相关接口
+
+- [创建草稿](./create_draft.md) - 创建新的剪映草稿
+- [添加视频](./add_videos.md) - 向草稿添加视频内容
+- [保存草稿](./save_draft.md) - 保存草稿更改

exceptions.py

@@ -31,6 +31,9 @@ class CustomError(Enum):
     INVALID_KEYFRAME_PROPERTY = (2017, "无效的关键帧属性类型", "Invalid keyframe property type.")
     INVALID_CAPTION_INFO = (2018, "无效的字幕信息，请检查captions字段值是否正确", "Invalid caption information, please check if the value of the captions field is correct.")
     CAPTION_ADD_FAILED = (2019, "字幕添加失败", "Caption addition failed")
+    INVALID_EFFECT_INFO = (2020, "无效的特效信息，请检查effect_infos字段值是否正确", "Invalid effect information, please check if the value of the effect_infos field is correct.")
+    EFFECT_ADD_FAILED = (2021, "特效添加失败", "Effect addition failed")
+    EFFECT_NOT_FOUND = (2022, "特效未找到，请检查特效名称是否正确", "Effect not found, please check if the effect name is correct.")
 
     # ===== 系统错误码 (9000-9999) =====
     INTERNAL_SERVER_ERROR = (9998, "系统内部错误", "Internal server error")

src/router/v1.py

@@ -6,6 +6,7 @@ from src.schemas.add_images import AddImagesResponse
 from src.schemas.add_sticker import AddStickerResponse
 from src.schemas.add_keyframes import AddKeyframesResponse
 from src.schemas.add_captions import AddCaptionsResponse
+from src.schemas.add_effects import AddEffectsResponse
 from src.schemas.save_draft import SaveDraftResponse
 from src.schemas.create_draft import CreateDraftResponse
 from fastapi import APIRouter, Request, Depends
@@ -16,6 +17,7 @@ from src.schemas.add_images import AddImagesRequest, AddImagesResponse
 from src.schemas.add_sticker import AddStickerRequest, AddStickerResponse
 from src.schemas.add_keyframes import AddKeyframesRequest, AddKeyframesResponse
 from src.schemas.add_captions import AddCaptionsRequest, AddCaptionsResponse
+from src.schemas.add_effects import AddEffectsRequest, AddEffectsResponse
 from src.schemas.save_draft import SaveDraftRequest, SaveDraftResponse
 from src.schemas.gen_video import GenVideoRequest, GenVideoResponse
 from src.schemas.get_draft import GetDraftRequest, GetDraftResponse
@@ -185,6 +187,25 @@ def add_captions(acr: AddCaptionsRequest) -> AddCaptionsResponse:
         segment_ids=segment_ids
     )
 
+@router.post(path="/add_effects", response_model=AddEffectsResponse)
+def add_effects(aer: AddEffectsRequest) -> AddEffectsResponse:
+    """
+    向剪映草稿添加特效 (v1版本)
+    """
+
+    # 调用service层处理业务逻辑
+    draft_url, track_id, effect_ids, segment_ids = service.add_effects(
+        draft_url=aer.draft_url,
+        effect_infos=aer.effect_infos
+    )
+
+    return AddEffectsResponse(
+        draft_url=draft_url,
+        track_id=track_id,
+        effect_ids=effect_ids,
+        segment_ids=segment_ids
+    )
+
 @router.get(path="/get_draft", response_model=GetDraftResponse)
 def get_draft(params: Annotated[GetDraftRequest, Depends()]) -> GetDraftResponse:
     """

src/schemas/add_effects.py

@@ -0,0 +1,23 @@
+from pydantic import BaseModel, Field
+from typing import List
+
+
+class AddEffectsRequest(BaseModel):
+    """添加特效请求参数"""
+    draft_url: str = Field(default="", description="草稿URL")
+    effect_infos: str = Field(default="", description="特效信息列表, 用JSON字符串表示")
+
+
+class EffectItem(BaseModel):
+    """单个特效信息"""
+    effect_title: str = Field(..., description="特效名称/标题")
+    start: int = Field(..., description="特效开始时间（微秒）")
+    end: int = Field(..., description="特效结束时间（微秒）")
+
+
+class AddEffectsResponse(BaseModel):
+    """添加特效响应参数"""
+    draft_url: str = Field(default="", description="草稿URL")
+    track_id: str = Field(default="", description="特效轨道ID")
+    effect_ids: List[str] = Field(default=[], description="特效ID列表")
+    segment_ids: List[str] = Field(default=[], description="特效片段ID列表")

src/service/__init__.py

@@ -5,8 +5,9 @@ from .add_images import add_images
 from .add_sticker import add_sticker
 from .add_keyframes import add_keyframes
 from .add_captions import add_captions
+from .add_effects import add_effects
 from .save_draft import save_draft
 from .gen_video import gen_video
 from .get_draft import get_draft
 
-__all__ = ["create_draft", "add_videos", "add_audios", "add_images", "add_sticker", "add_keyframes", "add_captions", "save_draft", "gen_video", "get_draft"]
+__all__ = ["create_draft", "add_videos", "add_audios", "add_images", "add_sticker", "add_keyframes", "add_captions", "add_effects", "save_draft", "gen_video", "get_draft"]

src/service/add_effects.py

@@ -0,0 +1,256 @@
+import json
+from typing import List, Dict, Any, Tuple, Optional, Union
+
+from src.utils.logger import logger
+from src.pyJianYingDraft import ScriptFile, TrackType, EffectSegment, Timerange
+from src.pyJianYingDraft.metadata import VideoSceneEffectType, VideoCharacterEffectType
+from src.utils.draft_cache import DRAFT_CACHE
+from exceptions import CustomException, CustomError
+from src.utils import helper
+
+
+def add_effects(
+    draft_url: str,
+    effect_infos: str
+) -> Tuple[str, str, List[str], List[str]]:
+    """
+    添加特效到剪映草稿的业务逻辑
+    
+    Args:
+        draft_url: 草稿URL
+        effect_infos: 特效信息列表的JSON字符串，格式如下：
+            [
+                {
+                    "effect_title": "录制边框 III",  # 特效名称/标题，必选参数
+                    "start": 0,  # 特效开始时间（微秒），必选参数
+                    "end": 5000000  # 特效结束时间（微秒），必选参数
+                }
+            ]
+    
+    Returns:
+        draft_url: 草稿URL
+        track_id: 特效轨道ID
+        effect_ids: 特效ID列表
+        segment_ids: 特效片段ID列表
+    
+    Raises:
+        CustomException: 特效添加失败
+    """
+    logger.info(f"add_effects started, draft_url: {draft_url}, effects count: {len(json.loads(effect_infos) if effect_infos else [])}")
+
+    # 1. 提取草稿ID
+    draft_id = helper.get_url_param(draft_url, "draft_id")
+    if (not draft_id) or (draft_id not in DRAFT_CACHE):
+        logger.error(f"Invalid draft_url or draft not found in cache: {draft_url}")
+        raise CustomException(CustomError.INVALID_DRAFT_URL)
+
+    # 2. 解析特效信息
+    effect_items = parse_effects_data(json_str=effect_infos)
+    if len(effect_items) == 0:
+        logger.info(f"No effect info provided, draft_id: {draft_id}")
+        raise CustomException(CustomError.INVALID_EFFECT_INFO)
+
+    logger.info(f"Parsed {len(effect_items)} effect items")
+
+    # 3. 从缓存中获取草稿
+    script: ScriptFile = DRAFT_CACHE[draft_id]
+
+    # 4. 添加特效轨道
+    track_name = f"effect_track_{helper.gen_unique_id()}"
+    script.add_track(track_type=TrackType.effect, track_name=track_name)
+    logger.info(f"Added effect track: {track_name}")
+
+    # 5. 遍历特效信息，添加特效到草稿中的指定轨道，收集片段ID
+    segment_ids = []
+    effect_ids = []
+    for i, effect in enumerate(effect_items):
+        try:
+            logger.info(f"Processing effect {i+1}/{len(effect_items)}, title: {effect['effect_title']}")
+            
+            segment_id, effect_id = add_effect_to_draft(
+                script, track_name, effect=effect
+            )
+            segment_ids.append(segment_id)
+            effect_ids.append(effect_id)
+            logger.info(f"Added effect {i+1}/{len(effect_items)}, segment_id: {segment_id}")
+        except Exception as e:
+            logger.error(f"Failed to add effect {i+1}/{len(effect_items)}, error: {str(e)}")
+            raise
+
+    # 6. 保存草稿
+    script.save()
+    logger.info(f"Draft saved successfully")
+
+    # 7. 获取当前特效轨道ID
+    track_id = ""
+    for key in script.tracks.keys():
+        if script.tracks[key].name == track_name:
+            track_id = script.tracks[key].track_id
+            break
+    logger.info(f"Effect track created, draft_id: {draft_id}, track_id: {track_id}")
+
+    logger.info(f"add_effects completed successfully - draft_id: {draft_id}, track_id: {track_id}, effects_added: {len(effect_items)}")
+    
+    return draft_url, track_id, effect_ids, segment_ids
+
+
+def add_effect_to_draft(
+    script: ScriptFile,
+    track_name: str,
+    effect: dict
+) -> Tuple[str, str]:
+    """
+    向剪映草稿中添加单个特效
+    
+    Args:
+        script: 草稿文件对象
+        track_name: 特效轨道名称
+        effect: 特效信息字典，包含以下字段：
+            effect_title: 特效名称/标题
+            start: 特效开始时间（微秒）
+            end: 特效结束时间（微秒）
+    
+    Returns:
+        segment_id: 片段ID
+        effect_id: 特效ID（material_id）
+    
+    Raises:
+        CustomException: 添加特效失败
+    """
+    try:
+        # 1. 查找特效类型
+        effect_type = find_effect_type_by_name(effect['effect_title'])
+        if effect_type is None:
+            logger.error(f"Effect type not found for title: {effect['effect_title']}")
+            raise CustomException(CustomError.EFFECT_NOT_FOUND)
+        
+        # 2. 创建时间范围
+        effect_duration = effect['end'] - effect['start']
+        timerange = Timerange(start=effect['start'], duration=effect_duration)
+        
+        # 3. 创建特效片段
+        effect_segment = EffectSegment(
+            effect_type=effect_type,
+            target_timerange=timerange
+        )
+        
+        logger.info(f"Created effect segment, effect_id: {effect_segment.effect_inst.global_id}")
+        logger.info(f"Effect segment details - start: {effect['start']}, duration: {effect_duration}, title: {effect['effect_title']}")
+
+        # 4. 向指定轨道添加片段
+        script.add_segment(effect_segment, track_name)
+
+        return effect_segment.segment_id, effect_segment.effect_inst.global_id
+        
+    except CustomException:
+        logger.error(f"Add effect to draft failed, effect: {effect}")
+        raise
+    except Exception as e:
+        logger.error(f"Add effect to draft failed, error: {str(e)}")
+        raise CustomException(CustomError.EFFECT_ADD_FAILED)
+
+
+def find_effect_type_by_name(effect_title: str) -> Optional[Union[VideoSceneEffectType, VideoCharacterEffectType]]:
+    """
+    根据特效名称查找对应的特效类型
+    
+    Args:
+        effect_title: 特效名称/标题
+    
+    Returns:
+        对应的特效类型枚举，如果未找到则返回None
+    """
+    logger.info(f"Searching for effect type with title: {effect_title}")
+    
+    # 搜索VideoSceneEffectType中的特效
+    for effect_type in VideoSceneEffectType:
+        if effect_type.value.name == effect_title:
+            logger.info(f"Found scene effect: {effect_title}")
+            return effect_type
+    
+    # 搜索VideoCharacterEffectType中的特效
+    for effect_type in VideoCharacterEffectType:
+        if effect_type.value.name == effect_title:
+            logger.info(f"Found character effect: {effect_title}")
+            return effect_type
+    
+    logger.warning(f"Effect type not found for title: {effect_title}")
+    return None
+
+
+def parse_effects_data(json_str: str) -> List[Dict[str, Any]]:
+    """
+    解析特效数据的JSON字符串，验证必选字段和数值范围
+    
+    Args:
+        json_str: 包含特效数据的JSON字符串，格式如下：
+        [
+            {
+                "effect_title": "录制边框 III",  # [必选] 特效名称/标题
+                "start": 0,  # [必选] 特效开始时间（微秒）
+                "end": 5000000  # [必选] 特效结束时间（微秒）
+            }
+        ]
+    
+    Returns:
+        包含特效对象的数组，每个对象都验证过格式和范围
+    
+    Raises:
+        CustomException: 当JSON格式错误或缺少必选字段时抛出
+    """
+    try:
+        # 解析JSON字符串
+        data = json.loads(json_str)
+    except json.JSONDecodeError as e:
+        logger.error(f"JSON parse error: {e.msg}")
+        raise CustomException(CustomError.INVALID_EFFECT_INFO, f"JSON parse error: {e.msg}")
+    
+    # 确保输入是列表
+    if not isinstance(data, list):
+        logger.error("effect_infos should be a list")
+        raise CustomException(CustomError.INVALID_EFFECT_INFO, "effect_infos should be a list")
+    
+    result = []
+    
+    for i, item in enumerate(data):
+        if not isinstance(item, dict):
+            logger.error(f"the {i}th item should be a dict")
+            raise CustomException(CustomError.INVALID_EFFECT_INFO, f"the {i}th item should be a dict")
+        
+        # 检查必选字段
+        required_fields = ["effect_title", "start", "end"]
+        missing_fields = [field for field in required_fields if field not in item]
+        
+        if missing_fields:
+            logger.error(f"the {i}th item is missing required fields: {', '.join(missing_fields)}")
+            raise CustomException(CustomError.INVALID_EFFECT_INFO, f"the {i}th item is missing required fields: {', '.join(missing_fields)}")
+        
+        # 创建处理后的对象
+        processed_item = {
+            "effect_title": str(item["effect_title"]),
+            "start": item["start"],
+            "end": item["end"]
+        }
+        
+        # 验证数值类型和范围
+        if not isinstance(processed_item["start"], (int, float)) or processed_item["start"] < 0:
+            logger.error(f"the {i}th item has invalid start time: {processed_item['start']}")
+            raise CustomException(CustomError.INVALID_EFFECT_INFO, f"the {i}th item has invalid start time")
+        
+        if not isinstance(processed_item["end"], (int, float)) or processed_item["end"] <= processed_item["start"]:
+            logger.error(f"the {i}th item has invalid end time: {processed_item['end']}")
+            raise CustomException(CustomError.INVALID_EFFECT_INFO, f"the {i}th item has invalid end time")
+        
+        # 验证特效名称
+        if len(processed_item["effect_title"].strip()) == 0:
+            logger.error(f"the {i}th item has invalid effect_title: {processed_item['effect_title']}")
+            raise CustomException(CustomError.INVALID_EFFECT_INFO, f"the {i}th item has invalid effect_title")
+        
+        # 将时间转换为整数（微秒）
+        processed_item["start"] = int(processed_item["start"])
+        processed_item["end"] = int(processed_item["end"])
+        
+        result.append(processed_item)
+    
+    logger.info(f"Successfully parsed {len(result)} effect items")
+    return result