如何使用Google Cloud Functions从WinSCP导入文件并配置每周执行触发器

一、前置说明 #

你提到的WinSCP是SFTP/FTP客户端工具，实际需求为通过云函数访问WinSCP对应的目标SFTP服务器，拉取指定文件到GCP环境，以下是完整操作步骤：

二、开发实现云函数SFTP文件导入功能 #

2.1 前置准备 #

• 启用GCP项目的Cloud Functions、Secret Manager、Cloud Storage（如果需要将导入文件存到GCS）API
• 提前在Secret Manager中存储SFTP连接敏感信息：主机地址、端口、用户名、密码/SSH私钥，避免硬编码到代码中
• 若目标SFTP服务器部署在内网环境，需提前配置云函数的VPC接入权限，确保云函数出口可以访问SFTP服务的22端口（默认SFTP端口）

2.2 云函数基础配置 #

• 运行环境选择Python 3.9及以上版本
• 内存配置根据拉取的文件大小调整，建议最小配置256M
• 超时时间根据文件拉取耗时调整，最大可设为9分钟
• 在requirements.txt中添加依赖项：

paramiko==3.4.0
google-cloud-secret-manager==2.16.3

2.3 核心功能代码示例 #

import paramiko
from google.cloud import secretmanager
import os

# 初始化Secret Manager客户端
secret_client = secretmanager.SecretManagerServiceClient()
PROJECT_ID = os.getenv("GCP_PROJECT")

def get_secret(secret_name):
    """从Secret Manager读取敏感信息"""
    name = f"projects/{PROJECT_ID}/secrets/{secret_name}/versions/latest"
    response = secret_client.access_secret_version(request={"name": name})
    return response.payload.data.decode("UTF-8")

def sftp_file_import(event, context):
    # 读取SFTP连接信息
    sftp_host = get_secret("sftp_host")
    sftp_port = int(get_secret("sftp_port"))
    sftp_user = get_secret("sftp_user")
    sftp_pwd = get_secret("sftp_password")
    # 待拉取的SFTP文件路径
    remote_file_path = "/remote/path/your_target_file.csv"
    # 本地临时存储路径（云函数临时目录内存放，拉取后可上传到GCS等存储）
    local_temp_path = "/tmp/your_target_file.csv"

    # 建立SFTP连接
    transport = paramiko.Transport((sftp_host, sftp_port))
    transport.connect(username=sftp_user, password=sftp_pwd)
    sftp = paramiko.SFTPClient.from_transport(transport)

    try:
        # 拉取文件到云函数临时目录
        sftp.get(remote_file_path, local_temp_path)
        # 此处可添加后续逻辑：比如将文件上传到Cloud Storage、写入BigQuery等
        print(f"文件{remote_file_path}导入成功")
    except Exception as e:
        print(f"文件导入失败：{str(e)}")
        raise
    finally:
        # 关闭连接
        sftp.close()
        transport.close()

• 云函数入口函数设置为sftp_file_import
• 代码部署完成后可手动触发测试，验证SFTP连接、文件拉取逻辑正常

三、配置周级自动执行触发器 #

该功能通过Cloud Scheduler实现定时触发云函数，步骤如下：

• 启用Cloud Scheduler API
• 新建定时作业：
  • 频率配置：使用cron表达式设置执行周期，例如0 2 * * 1代表每周一凌晨2点执行，时区选择你所在的对应时区（如Asia/Shanghai）
  • 目标类型选择「Cloud Functions」，在下拉列表中选择上一步创建的云函数
  • 服务账号配置：为Cloud Scheduler的默认服务账号授予Cloud Functions Invoker权限，确保调度服务有权限调用云函数
  • 可选配置：如果需要动态传入拉取文件路径等参数，可在请求体中添加自定义JSON参数，云函数中直接读取event参数即可获取
• 作业创建完成后可手动触发测试，验证调度链路正常

四、常见问题排查 #

• 若云函数连接SFTP超时，优先检查VPC配置、防火墙规则是否放开SFTP端口访问权限
• 若文件拉取失败，检查SFTP账号的目录访问权限、目标文件路径是否正确

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