Method generate_video_extend

src/services/flow_client.py:3421–3572

视频续写,基于已生成的视频延伸7秒 Args: at: Access Token project_id: 项目ID prompt: 续写提示词 model_key: veo_3_1_extend_portrait / veo_3_1_extend 等 aspect_ratio: 视频宽高比 video_media_id: 源视频的 mediaGenerationId user_paygate_tier: 用户等

(
        self,
        at: str,
        project_id: str,
        prompt: str,
        model_key: str,
        aspect_ratio: str,
        video_media_id: str,
        user_paygate_tier: str = "PAYGATE_TIER_ONE",
        token_id: Optional[int] = None,
        token_video_concurrency: Optional[int] = None,
    )

    # ========== 视频续写 (Video Extend) ==========

    async def generate_video_extend(
        self,
        at: str,
        project_id: str,
        prompt: str,
        model_key: str,
        aspect_ratio: str,
        video_media_id: str,
        user_paygate_tier: str = "PAYGATE_TIER_ONE",
        token_id: Optional[int] = None,
        token_video_concurrency: Optional[int] = None,
    ) -> dict:
        """视频续写,基于已生成的视频延伸7秒

        Args:
            at: Access Token
            project_id: 项目ID
            prompt: 续写提示词
            model_key: veo_3_1_extend_portrait / veo_3_1_extend 等
            aspect_ratio: 视频宽高比
            video_media_id: 源视频的 mediaGenerationId
            user_paygate_tier: 用户等级

        Returns:
            同 generate_video_text (operations 列表)
        """
        url = f"{self.api_base_url}/video:batchAsyncGenerateVideoExtendVideo"

        # 403/reCAPTCHA 重试逻辑 - reCAPTCHA evaluation failed 时允许更高重试上限
        max_retries = self._resolve_generation_retry_budget(config.flow_max_retries)
        last_error = None
        retry_attempt = 0
        session_id = self._generate_session_id()
        workflow_id = str(uuid.uuid4())
        batch_id = str(uuid.uuid4())
        video_context_warmed = False
        while retry_attempt < max_retries:
            launch_gate_acquired = False
            launch_ok, _, _ = await self._acquire_video_launch_gate(
                token_id=token_id,
                token_video_concurrency=token_video_concurrency,
            )
            if not launch_ok:
                last_error = Exception("Video launch queue wait timeout")
                raise last_error

            launch_gate_acquired = True
            try:
                recaptcha_token, browser_id = await self._get_recaptcha_token(
                    project_id,
                    action="VIDEO_GENERATION",
                    token_id=token_id
                )
            finally:
                if launch_gate_acquired:
                    await self._release_video_launch_gate(token_id)
            if not recaptcha_token:
                last_error = Exception("Failed to obtain reCAPTCHA token")