27、扣子-COZE智能体-批量EXCEL内容收发邮件

import smtplibimport json# ======= 发件人配置（建议后续改为环境变量）=======FROM_ADDR = "xxx@aliyun.com"  # 替换为你的发件邮箱PASSWORD = "xxx"              # 邮箱授权码，正式环境请勿写死SMTP_SERVER = "smtp.aliyun.com"  # 对应邮箱的SMTP服务器地址SMTP_PORT = 465                # 阿里云SSL端口，其他邮箱需对应修改def parse_addrs(value):    """    将收件人/抄送地址字段转换为列表，支持多种格式：    - 单个邮箱："a@b.com"    - 分号分隔："a@b.com;b@c.com"    - 逗号分隔："a@b.com,b@c.com"    - 直接传入列表    """    if not value:        return []    if isinstance(value, list):        return [x.strip() for x in value if x.strip()]    if isinstance(value, str):        parts = value.split(";") if ";" in value else value.split(",")        return [p.strip() for p in parts if p.strip()]    return []def send_mail_from_row(row: dict):    """    根据表格单行数据发送邮件，row示例：    {        "姓名": "张三",        "发送的邮箱": "8xxxxxxxx@qq.com",        "抄送的地址": "hi123@sohu.com",        "标题": "测试邮件标题",        "邮箱的内容": "测试邮件正文"    }    """    name = row.get("姓名", "")    to_field = row.get("发送的邮箱", "")    cc_field = row.get("抄送的地址", "")    subject = row.get("标题", "自动邮件通知")    body = row.get("邮箱的内容", "")    to_addrs = parse_addrs(to_field)    cc_addrs = parse_addrs(cc_field)    all_receivers = to_addrs + cc_addrs    if not all_receivers:        raise ValueError("收件人地址为空，请检查发送的邮箱/抄送的地址字段")    if not body:        body = f"您好 {name}，这是一封自动发送的测试邮件。"    # 手动拼接邮件内容，兼容性更强    header_lines = [        f"From: {FROM_ADDR}",        f"To: {', '.join(to_addrs)}" if to_addrs else "",        f"Cc: {', '.join(cc_addrs)}" if cc_addrs else "",        f"Subject: {subject}"    ]    message = "\r\n".join([line for line in header_lines if line]) + "\r\n\r\n" + body    # 发送邮件    with smtplib.SMTP_SSL(SMTP_SERVER, SMTP_PORT) as server:        server.login(FROM_ADDR, PASSWORD)        server.sendmail(FROM_ADDR, all_receivers, message.encode("utf-8"))    return {        "name": name,        "to_addrs": to_addrs,        "cc_addrs": cc_addrs,        "subject": subject    }# ============ Coze 节点入口 ============async def main(args: Args) -> Output:    params = args.params    input_obj = params.get("input", {})    fields = input_obj.get("fields")    if fields is None:        return {            "status": "error",            "error": "未找到'fields'字段，请检查数据格式",            "params": params        }    # 解析表格单行数据    if isinstance(fields, str):        try:            row = json.loads(fields)        except Exception as e:            return {                "status": "error",                "error": f"解析数据失败: {str(e)}",                "raw_fields": fields            }    elif isinstance(fields, dict):        row = fields    else:        return {            "status": "error",            "error": "数据格式错误，fields需为字符串或对象",            "type": str(type(fields))        }    # 执行发件    try:        info = send_mail_from_row(row)        return {            "status": "success",            "detail": info,            "row": row        }    except Exception as e:        return {            "status": "error",            "error": str(e),            "row": row        }