Google Veo2 Video - ComfyUI ネイティブノードドキュメント

class VeoVideoGenerationNode(ComfyNodeABC):
    """
    Google の Veo API を使用して、テキストプロンプトから動画を生成します。

    このノードは、テキスト記述およびオプションの画像入力から動画を作成でき、
    アスペクト比や再生時間などのパラメーターを制御できます。
    """

    @classmethod
    def INPUT_TYPES(s):
        return {
            "required": {
                "prompt": (
                    IO.STRING,
                    {
                        "multiline": True,
                        "default": "",
                        "tooltip": "動画の内容を記述するテキスト",
                    },
                ),
                "aspect_ratio": (
                    IO.COMBO,
                    {
                        "options": ["16:9", "9:16"],
                        "default": "16:9",
                        "tooltip": "出力動画のアスペクト比",
                    },
                ),
            },
            "optional": {
                "negative_prompt": (
                    IO.STRING,
                    {
                        "multiline": True,
                        "default": "",
                        "tooltip": "動画内で回避すべき内容を示すネガティブプロンプト",
                    },
                ),
                "duration_seconds": (
                    IO.INT,
                    {
                        "default": 5,
                        "min": 5,
                        "max": 8,
                        "step": 1,
                        "display": "number",
                        "tooltip": "出力動画の再生時間（秒単位）",
                    },
                ),
                "enhance_prompt": (
                    IO.BOOLEAN,
                    {
                        "default": True,
                        "tooltip": "AI を用いてプロンプトを強化するかどうか",
                    }
                ),
                "person_generation": (
                    IO.COMBO,
                    {
                        "options": ["ALLOW", "BLOCK"],
                        "default": "ALLOW",
                        "tooltip": "動画内での人物生成を許可するかどうか",
                    },
                ),
                "seed": (
                    IO.INT,
                    {
                        "default": 0,
                        "min": 0,
                        "max": 0xFFFFFFFF,
                        "step": 1,
                        "display": "number",
                        "control_after_generate": True,
                        "tooltip": "動画生成用のシード値（0 の場合はランダム）",
                    },
                ),
                "image": (IO.IMAGE, {
                    "default": None,
                    "tooltip": "動画生成を補助するための参照画像（任意）",
                }),
            },
            "hidden": {
                "auth_token": "AUTH_TOKEN_COMFY_ORG",
            },
        }

    RETURN_TYPES = (IO.VIDEO,)
    FUNCTION = "generate_video"
    CATEGORY = "api node/video/Veo"
    DESCRIPTION = "Google の Veo API を使用して、テキストプロンプトから動画を生成します"
    API_NODE = True

    def generate_video(
        self,
        prompt,
        aspect_ratio="16:9",
        negative_prompt="",
        duration_seconds=5,
        enhance_prompt=True,
        person_generation="ALLOW",
        seed=0,
        image=None,
        auth_token=None,
    ):
        # リクエスト用のインスタンスを準備
        instances = []

        instance = {
            "prompt": prompt
        }

        # 画像が提供されている場合、追加
        if image is not None:
            image_base64 = convert_image_to_base64(image)
            if image_base64:
                instance["image"] = {
                    "bytesBase64Encoded": image_base64,
                    "mimeType": "image/png"
                }

        instances.append(instance)

        # パラメーター辞書を作成
        parameters = {
            "aspectRatio": aspect_ratio,
            "personGeneration": person_generation,
            "durationSeconds": duration_seconds,
            "enhancePrompt": enhance_prompt,
        }

        # 提供されている場合、オプションパラメーターを追加
        if negative_prompt:
            parameters["negativePrompt"] = negative_prompt
        if seed > 0:
            parameters["seed"] = seed

        # 動画生成を開始するための初期リクエスト
        initial_operation = SynchronousOperation(
            endpoint=ApiEndpoint(
                path="/proxy/veo/generate",
                method=HttpMethod.POST,
                request_model=Veo2GenVidRequest,
                response_model=Veo2GenVidResponse
            ),
            request=Veo2GenVidRequest(
                instances=instances,
                parameters=parameters
            ),
            auth_token=auth_token
        )

        initial_response = initial_operation.execute()
        operation_name = initial_response.name

        logging.info(f"Veo 動画生成を開始しました（操作名: {operation_name}）")

        # ステータス抽出関数を定義
        def status_extractor(response):
            # 完了状態（成功・失敗を問わず）のみ「completed」を返す
            # エラー検出はポーリング完了後に実施
            return "completed" if response.done else "pending"

        # 進行度抽出関数を定義
        def progress_extractor(response):
            # API が進行度情報を提供する場合、拡張可能
            return None

        # ポーリング操作を定義
        poll_operation = PollingOperation(
            poll_endpoint=ApiEndpoint(
                path="/proxy/veo/poll",
                method=HttpMethod.POST,
                request_model=Veo2GenVidPollRequest,
                response_model=Veo2GenVidPollResponse
            ),
            completed_statuses=["completed"],
            failed_statuses=[],  # 失敗ステータスは空にして、ポーリング完了後にエラー処理
            status_extractor=status_extractor,
            progress_extractor=progress_extractor,
            request=Veo2GenVidPollRequest(
                operationName=operation_name
            ),
            auth_token=auth_token,
            poll_interval=5.0
        )

        # ポーリング操作を実行
        poll_response = poll_operation.execute()

        # 最終レスポンス内のエラーを確認
        # ポーリングレスポンス内のエラーを確認
        if hasattr(poll_response, 'error') and poll_response.error:
            error_message = f"Veo API エラー: {poll_response.error.message}（コード: {poll_response.error.code}）"
            logging.error(error_message)
            raise Exception(error_message)

        # RAI（Responsible AI）によるコンテンツフィルタリングを確認
        if (hasattr(poll_response.response, 'raiMediaFilteredCount') and
            poll_response.response.raiMediaFilteredCount > 0):

            # 理由メッセージが存在するか確認
            if (hasattr(poll_response.response, 'raiMediaFilteredReasons') and
                poll_response.response.raiMediaFilteredReasons):
                reason = poll_response.response.raiMediaFilteredReasons[0]
                error_message = f"Google の Responsible AI 方針によりコンテンツがフィルタリングされました: {reason}（{poll_response.response.raiMediaFilteredCount} 件の動画がフィルタリング済み）"
            else:
                error_message = f"Google の Responsible AI 方針によりコンテンツがフィルタリングされました（{poll_response.response.raiMediaFilteredCount} 件の動画がフィルタリング済み）"

            logging.error(error_message)
            raise Exception(error_message)

        # 動画データを抽出
        video_data = None
        if poll_response.response and hasattr(poll_response.response, 'videos') and poll_response.response.videos and len(poll_response.response.videos) > 0:
            video = poll_response.response.videos[0]

            # 動画が base64 形式か URL 形式かを確認
            if hasattr(video, 'bytesBase64Encoded') and video.bytesBase64Encoded:
                # base64 文字列をバイト列にデコード
                video_data = base64.b64decode(video.bytesBase64Encoded)
            elif hasattr(video, 'gcsUri') and video.gcsUri:
                # URL からダウンロード
                video_url = video.gcsUri
                video_response = requests.get(video_url)
                video_data = video_response.content
            else:
                raise Exception("動画は返却されましたが、データも URL も提供されていません")
        else:
            raise Exception("動画生成は完了しましたが、動画が返却されていません")

        if not video_data:
            raise Exception("動画データが返却されませんでした")

        logging.info("動画生成が正常に完了しました")

        # 動画データを BytesIO オブジェクトに変換
        video_io = io.BytesIO(video_data)

        # VideoFromFile オブジェクトを返却
        return (VideoFromFile(video_io),)