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),)
