GCP环境下如何在Airflow中无监督自动使用用户账号发送邮件

针对你的需求，以下是两种可行的无监督实现方案，适配你现有GCP+Airflow的环境：

方案一：服务账号模拟用户账号（域范围委派） #

这是GCP环境下最合规的无监督方式，利用Workspace域范围委派让服务账号模拟指定用户发送邮件，无需用户手动授权。

步骤1：配置服务账号的域范围委派 #

1. 登录GCP控制台，进入IAM & Admin > 服务账号，找到你用于Forms/Drive操作的服务账号，点击编辑，勾选「启用G Suite域范围委派」。
2. 复制服务账号的客户端ID（在服务账号详情页的「高级设置」里）。
3. 登录Google Workspace Admin控制台，进入安全 > API控制 > 域范围委派，点击「添加新」：
   • 粘贴刚才的客户端ID
   • 添加授权范围：https://www.googleapis.com/auth/gmail.send（仅发送权限，遵循最小权限原则）
   • 保存配置。

步骤2：在Airflow中配置服务账号连接 #

1. 进入Airflow UI的Admin > Connections，创建或编辑你的Google Cloud连接（比如google_cloud_default）：
   • 连接类型选择「Google Cloud」
   • 在「Extra」字段中填入服务账号的JSON密钥内容（格式：{"keyfile_dict": {"type": "service_account", ...}}）
   • 填写正确的GCP项目ID。

步骤3：编写PythonTask调用Gmail API发送HTML邮件 #

用impersonated_credentials模拟目标用户，调用Gmail API发送带HTML内容的邮件：

import json
import base64
from email.mime.text import MIMEText
from airflow.decorators import task
from airflow.models import Connection
from google.oauth2 import service_account, impersonated_credentials
from googleapiclient.discovery import build

@task
def send_form_email(form_link, recipient_email):
    # 从Airflow连接获取服务账号密钥
    conn = Connection.get_connection_from_secrets("google_cloud_default")
    service_account_info = json.loads(conn.extra_dejson["keyfile_dict"])
    
    # 生成模拟用户的凭据
    source_creds = service_account.Credentials.from_service_account_info(
        service_account_info,
        scopes=["https://www.googleapis.com/auth/gmail.send"]
    )
    target_creds = impersonated_credentials.Credentials(
        source_credentials=source_creds,
        target_principal="your-user@your-domain.com",  # 要模拟的用户邮箱
        target_scopes=["https://www.googleapis.com/auth/gmail.send"],
        lifetime=3600  # 凭据有效期，最长3600秒
    )
    
    # 构建Gmail服务
    gmail_service = build('gmail', 'v1', credentials=target_creds)
    
    # 构造HTML邮件内容（替换成你的格式化内容）
    html_content = f"""
    <html>
      <body>
        <h2>请完成表单填写</h2>
        <p>点击以下链接访问表单：<a href="{form_link}">{form_link}</a></p>
        <p>截止日期：XX年XX月XX日</p>
      </body>
    </html>
    """
    # 封装邮件
    msg = MIMEText(html_content, 'html')
    msg['To'] = recipient_email
    msg['Subject'] = "表单填写通知"
    raw_msg = base64.urlsafe_b64encode(msg.as_bytes()).decode()
    
    # 发送邮件
    gmail_service.users().messages().send(
        userId="me",
        body={"raw": raw_msg}
    ).execute()

方案二：SMTP方式（用用户App密码） #

如果你的用户账号开启了两步验证，可通过Gmail的App密码配置SMTP连接，直接用EmailOperator发送HTML邮件。

步骤1：生成用户的Gmail App密码 #

1. 登录目标用户的Google账号，开启两步验证（必须开启才能生成App密码）。
2. 进入安全 > App密码，选择「邮件」作为应用，「其他（自定义名称）」作为设备，生成App密码（复制保存，只显示一次）。

步骤2：在Airflow中配置SMTP连接 #

1. 进入Airflow UI的Admin > Connections，创建新连接：
   • 连接类型选择「SMTP」
   • 连接ID设为gmail_smtp（自定义）
   • 主机：smtp.gmail.com
   • 端口：587
   • 登录：目标用户的邮箱地址
   • 密码：刚才生成的App密码
   • 加密：STARTTLS
   • 保存配置。

步骤3：用EmailOperator发送HTML邮件 #

直接使用Airflow的EmailOperator，传入HTML内容和表单链接：

from airflow.operators.email import EmailOperator

send_email_task = EmailOperator(
    task_id="send_form_notification",
    to="recipient@example.com",
    subject="请填写表单",
    html_content="""
    <html>
      <body>
        <h3>您好，需要您填写以下表单</h3>
        <p>表单链接：<a href="{{ ti.xcom_pull(task_ids='copy_form') }}">点击访问</a></p>
        <p>如有疑问请联系管理员</p>
      </body>
    </html>
    """,
    smtp_conn_id="gmail_smtp"  # 对应你配置的SMTP连接ID
)

常见问题解决 #

• 服务账号无权限：检查域范围委派是否添加了正确的gmail.send scope，且服务账号已被Workspace管理员授权。
• SMTP配置失败：确认用户账号开启了两步验证，使用的是App密码而非普通密码，且Airflow Worker能访问smtp.gmail.com:587。
• ADC 403错误：本地测试时需确保服务账号已模拟目标用户，且scope包含gmail.send，同时目标用户属于同一个Workspace域。

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