Pub/Sub触发的Cloud Function无法保存邮件附件至Cloud Storage的问题排查求助

我正在搭建一个带Pub/Sub触发器的Google Cloud Function，用来处理邮件附件，目的是把允许的邮箱发来的附件存到指定的Cloud Storage桶里。

问题概述 #

Pub/Sub触发器能正常工作，日志也显示函数被调用了，但附件就是存不到桶里。有时候明明测试邮件里有有效附件，日志却显示No valid attachments found in the email，偶尔还会出现邮件无法投递的情况，日志里也没更多错误细节。

环境配置细节 #

Cloud Function #

• 触发器：Pub/Sub Topic
• 运行时：Python 3.9
• 权限：相关服务账号已经被授予了Cloud Functions Invoker、Pub/Sub Publisher、Storage Admin这些必要角色

Pub/Sub Topic #

用默认设置创建的：添加了订阅，拉取投递类型，消息保留7天

预期流程 #

1. 邮件附件通过Pub/Sub消息传递
2. 函数处理Pub/Sub消息，验证发件人和主题，把附件存到指定Cloud Storage桶里

当前观察到的现象 #

• Pub/Sub消息能成功发布，函数日志显示已触发调用
• 但没有任何附件被保存到存储桶
• 部分情况下，明明邮件带了有效附件，日志却提示No valid attachments found in the email
• 偶尔会出现邮件无法投递的情况，日志里没有进一步的错误信息

参考代码片段 #

我把函数代码贴在这里，希望大家能帮忙找找问题：

@functions_framework.cloud_event
def handle_email_attachment(cloud_event):
    """Handles email attachments sent via Pub/Sub."""
    try:
        pubsub_message = base64.b64decode(cloud_event.data["message"]["data"]).decode("utf-8")
        email_message = message_from_bytes(base64.b64decode(pubsub_message))

        # Extract sender, recipient, and subject
        sender_email = email_message.get("from", "").strip()
        recipient_email = email_message.get("to", "").strip()
        subject = email_message.get("subject", "").strip()

        # Validate sender email
        if sender_email not in ALLOWED_EMAILS:
            print(f"Email from {sender_email} is not allowed. Ignoring.")
            return

        # Process attachments
        for part in email_message.walk():
            if part.get_content_maintype() == 'multipart':
                continue
            if part.get('Content-Disposition') is None:
                continue

            file_name = part.get_filename()
            file_data = part.get_payload(decode=True)
            if file_name:
                save_to_storage(sender_email, file_name, file_data)

    except Exception as e:
        print(f"Error processing email attachment: {e}")

def save_to_storage(sender_email, file_name, file_data):
    """Save the attachment to Cloud Storage."""
    folder_name = sender_email.replace("@", "at").replace(".", "_")
    bucket = storage_client.bucket(ATTACHMENT_BUCKET)
    blob = bucket.blob(f"{folder_name}/{file_name}")
    blob.upload_from_string(file_data)
    print(f"File {file_name} uploaded to folder {folder_name} in bucket {ATTACHMENT_BUCKET}.")

几个疑问点 #

• Pub/Sub消息结构：处理邮件附件是不是需要特定的Pub/Sub消息结构？
• 邮件字段验证：Gmail或者其他域名的邮件，它们的“from”或“to”字段格式会不会在Pub/Sub消息里有差异？
• 附件处理：是不是有些特定的内容类型（比如text/csv、application/vnd.ms-excel）需要在函数里显式处理？
• 测试环境调试：针对处理附件的Pub/Sub触发Cloud Function，有没有额外的调试步骤可以做？

备注：内容来源于stack exchange，提问作者Bhushan tarale