Create gcs_uploader.py

gcs_uploader.py

@@ -0,0 +1,64 @@
+# gcs_uploader.py - Google Cloud Storage 上传工具
+import os
+from google.cloud import storage
+from datetime import timedelta
+
+def upload_video_to_gcs(
+    video_file_path: str,
+    bucket_name: str = "n8n-33-arvin",
+    destination_prefix: str = "video-demo/",
+    credentials_path: str = None,
+    expires: int = 3600
+):
+    """
+    上传视频到 Google Cloud Storage，返回 (local_path, signed_url)
+    
+    Args:
+        video_file_path (str): 本地视频文件路径
+        bucket_name (str): GCS 存储桶名称
+        destination_prefix (str): GCS 对象前缀（如 "video-demo/"）
+        credentials_path (str): 服务账号 JSON 路径（可选）
+        expires (int): 签名 URL 有效期（秒），默认 1 小时
+    
+    Returns:
+        tuple: (local_path, signed_url) 或 (None, error_message)
+    """
+    if not video_file_path or not os.path.isfile(video_file_path):
+        return None, "❌ 视频文件无效或不存在"
+
+    try:
+        # 设置凭据（优先使用传入路径，其次环境变量）
+        if credentials_path:
+            os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = credentials_path
+        elif "GOOGLE_APPLICATION_CREDENTIALS" not in os.environ:
+            # 尝试从 HF Secrets 获取 base64 编码的凭据（可选扩展）
+            pass
+
+        # 初始化客户端
+        storage_client = storage.Client()
+        bucket = storage_client.bucket(bucket_name)
+
+        # 构造目标对象名
+        original_name = os.path.basename(video_file_path)
+        destination_blob_name = f"{destination_prefix.rstrip('/')}/{original_name}"
+
+        blob = bucket.blob(destination_blob_name)
+        
+        # 上传文件
+        blob.upload_from_filename(video_file_path)
+        
+        # 生成签名 URL
+        signed_url = blob.generate_signed_url(
+            version="v4",
+            expiration=timedelta(seconds=expires),
+            method="GET"
+        )
+
+        print(f"[GCS] ✅ 上传成功: {destination_blob_name}")
+        print(f"[GCS] 🔗 签名链接: {signed_url}")
+        return video_file_path, signed_url
+
+    except Exception as e:
+        error_msg = f"❌ GCS 上传失败: {type(e).__name__} - {str(e)}"
+        print(f"[GCS] {error_msg}")
+        return None, error_msg