请求GCP第二代Cloud Function实现GCS到SFTP的文件传输代码

main.py #

import os
import tempfile
from google.cloud import storage
import paramiko

def gcs_to_sftp(event, context):
    # 替换为你的SFTP配置参数
    SFTP_HOST = "你的SFTP主机名"
    SFTP_USER = "uid"
    SFTP_PASSWORD = "mypassword"
    SFTP_PORT = 22
    SFTP_TARGET_FILE = "test.csv"  # SFTP端保存的文件名，可按需修改

    # 从触发事件中提取GCS存储桶和文件信息
    bucket_name = event['bucket']
    file_name = event['name']

    # 仅处理指定的test.csv文件
    if file_name != "test.csv":
        print(f"跳过非目标文件: {file_name}")
        return

    # 初始化GCS客户端并下载文件到临时目录
    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(file_name)

    with tempfile.NamedTemporaryFile(mode='wb', delete=False) as temp_file:
        blob.download_to_file(temp_file)
        temp_file_path = temp_file.name

    try:
        # 建立SFTP连接并上传文件
        transport = paramiko.Transport((SFTP_HOST, SFTP_PORT))
        transport.connect(username=SFTP_USER, password=SFTP_PASSWORD)
        sftp = paramiko.SFTPClient.from_transport(transport)

        sftp.put(temp_file_path, SFTP_TARGET_FILE)
        print(f"文件 {file_name} 已成功上传至SFTP服务器")

        # 关闭SFTP连接
        sftp.close()
        transport.close()
    except Exception as e:
        print(f"传输失败: {str(e)}")
        raise
    finally:
        # 清理临时文件
        os.unlink(temp_file_path)

requirements.txt #

paramiko>=2.12.0
google-cloud-storage>=2.10.0

注意事项 #

• 替换代码中SFTP_HOST为你的实际SFTP主机地址
• 禁止硬编码密码，建议使用GCP Secret Manager存储敏感信息，通过函数调用动态获取，降低安全风险
• 确保Cloud Function拥有GCS存储桶的storage.objects.get权限，以及出站访问SFTP端口22的网络权限（若使用VPC需配置对应防火墙规则）
• 代码适配Python 3.8环境，符合第二代Cloud Function运行要求

内容的提问来源于stack exchange，提问作者Rv02