import json
import requests
import hashlib
import base64
import hmac
import time
import re
from typing import Dict
import smtplib
import uuid
import volcenginesdkiam
import volcenginesdkcore

from volcengine.sms.SmsService import SmsService
from volcengine.ApiInfo import ApiInfo
from volcengine.Credentials import Credentials
from volcengine.base.Service import Service
from volcengine.ServiceInfo import ServiceInfo

# --------------------------- 用户需要自行修改的部分 ---------------------------
AK = "xxxx"
SK = "xxxx"
Region = "cn-beijing"
SMSPhone = "xxxx"
SMSTemplateID = "xxxx"
SMSAccount = "xxxx"
SMSSign = "火山引擎"
# --------------------------- 用户需要自行修改的部分 ---------------------------

sms_service = SmsService()
sms_service.set_ak(AK)
sms_service.set_sk(SK)

configuration = volcenginesdkcore.Configuration()
configuration.ak = AK
configuration.sk = SK
configuration.region = Region
configuration.client_side_validation = True
# set default configuration
volcenginesdkcore.Configuration.set_default(configuration)

def send_sms(
    template_params: dict,
):
    template_params_str = json.dumps(template_params)
    body = {
        "SmsAccount": SMSAccount,
        "Sign": SMSSign,
        "TemplateID": SMSTemplateID,
        "TemplateParam": template_params_str,
        "PhoneNumbers": SMSPhone
    }
    bodyStr = json.dumps(body)
    resp = sms_service.send_sms(bodyStr)
    print(resp)

def handler(event, context):
    # Log incoming request.
    #
    # NOTE: the log here is only for debug purpose. It's recommended to delete those debug/info
    # logs and only print log where error occurs after your business logic is verified and ready to
    # go production, nor the log amount may be too huge.
    print(f"received new request, event content: {event}")

    records = event.get('data', {}).get('Records', [])
    for record in records:
        print(f"get record: {record}")
        message = json.loads(record.get('Sns', '{}').get('Message', '{}'))
        print(f"get message: {message}")
        if message.get("EventResourceType") == "DevInstance":
            event_resource_id = message.get("EventResourceID")
            message_event = message.get("EventType", "")
            event_time = message.get("EventTime")
            devinstance_detail = get_devinstance_detail(event_resource_id)
            creator_trn = devinstance_detail.creator_trn
   
            params = {
                "UserName": creator_trn,
                "AccountID": creator_trn,
                "CustomTaskName": devinstance_detail.name,
                "CustomTaskId": devinstance_detail.id,
                "MessageEvent": message_event,
            }
            send_sms(params)
        else:
            continue
    result = {
        'statusCode': 200,
        'headers': {
            'Content-Type': 'application/json'
        }
    }
    return result
# ==================== MLP Client ====================

SERVICE_VERSION_V2 = "2024-07-01"

SERVICE_NAME = "ml_platform_stg"

service_info = ServiceInfo(
    "open.volcengineapi.com",
    {"Accept": "application/json"},
    Credentials(AK, SK, SERVICE_NAME, Region),
    60 * 5,
    60 * 5,
    "https",
)

api_info_v2 = {
    "GetDevInstance": ApiInfo(
        "POST",
        "/",
        {"Action": "GetDevInstance", "Version": SERVICE_VERSION_V2},
        {},
        {},
    ),
    "GetResourceQueue": ApiInfo(
        "POST",
        "/",
        {"Action": "GetResourceQueue", "Version": SERVICE_VERSION_V2},
        {},
        {},
    ),
}

class MLPClient(Service):
    def __init__(self) -> None:
        super(MLPClient, self).__init__(
            MLPClient.get_service_info(), MLPClient.get_api_info()
        )

    @staticmethod
    def get_service_info() -> ServiceInfo:
        return service_info

    @staticmethod
    def get_api_info() -> Dict[str, ApiInfo]:
        return api_info_v2

    def get_devinstance(self, instance_id: str) -> dict:
        res_str = self.json("GetDevInstance", [], json.dumps({"Id": instance_id}))
        res = json.loads(res_str)
        if not res.get("Result"):
            raise Exception(f"GetDevInstance failed, res: {res}")
        return res.get("Result")

    def get_resource_queue(self, resource_queue_id: str) -> dict:
        res_str = self.json(
            "GetResourceQueue", [], json.dumps({"Id": resource_queue_id})
        )
        res = json.loads(res_str)
        if not res.get("Result"):
            raise Exception(f"GetResourceQueue failed, res: {res}")
        return res.get("Result")

class DevInstanceDetail:
    id: str
    name: str
    resource_group_id: str
    resource_queue_id: str
    creator_trn: str

def get_devinstance_detail(eventResourceId: str) -> DevInstanceDetail:
    return get_devinstance_detail_by_devinstance_id(eventResourceId)

def get_devinstance_detail_by_devinstance_id(devinstance_id: str) -> DevInstanceDetail:
    client = MLPClient()
    res = client.get_devinstance(devinstance_id)
    detail = DevInstanceDetail()
    detail.id = devinstance_id
    detail.name = res.get("Name")
    detail.resource_queue_id = res.get("ResourceQueueId")
    detail.resource_group_id = client.get_resource_queue(detail.resource_queue_id).get(
        "ResourceGroupId"
    )
    detail.creator_trn = res.get("CreatorTrn")
    return detail

效果预览：

需接入火山引擎语音服务产品

import json
import requests
import hashlib
import base64
import hmac
import time
import re
import smtplib
import uuid
from typing import Dict
from volcengine.vms.VmsService import VmsService
from volcengine.ApiInfo import ApiInfo
from volcengine.Credentials import Credentials
from volcengine.base.Service import Service
from volcengine.ServiceInfo import ServiceInfo

# --------------------------- 用户需要自行修改的部分 ---------------------------
AK = "xxxx"
SK = "xxxx"
Region = "cn-beijing"
VMSPhone = "xxxx"
VMSResource = "xxx"
VMSNumberPoolNo = "xxx"
# --------------------------- 用户需要自行修改的部分 ---------------------------

vms_service = VmsService()
vms_service.set_ak(AK)
vms_service.set_sk(SK)

def send_vms(
    phoneParam: dict,
):
    uuid_obj = uuid.uuid4()
    uuid_bytes = uuid_obj.bytes
    b64_str = base64.urlsafe_b64encode(uuid_bytes).rstrip(b'=').decode('utf-8')

    single_openid = b64_str
    single_append_body = {
        "List": [
            {
                "Phone": "18271891346",
                "Type": 1,
                "Resource": VMSResource,
                "NumberPoolNo": VMSNumberPoolNo,
                "SingleOpenId": single_openid,
                "PhoneParam": phoneParam
            },
        ]
    }
    result = vms_service.single_batch_append(single_append_body)
    print(f"received result from vms: {result}")


def handler(event, context):
    # Log incoming request.
    #
    # NOTE: the log here is only for debug purpose. It's recommended to delete those debug/info
    # logs and only print log where error occurs after your business logic is verified and ready to
    # go production, nor the log amount may be too huge.
    print(f"received new request, event content: {event}")

    records = event.get('data', {}).get('Records', [])
    for record in records:
        print(f"get record: {record}")
        message = json.loads(record.get('Sns', '{}').get('Message', '{}'))
        print(f"get message: {message}")
        if message.get("EventResourceType") == "DevInstance":
            event_resource_id = message.get("EventResourceID")
            message_event = message.get("EventType", "")
            event_time = message.get("EventTime")
            devinstance_detail = get_devinstance_detail(event_resource_id)
            params = {
                "CustomTaskName": devinstance_detail.name,
                "CustomTaskID": devinstance_detail.id,
                "MessageEvent": message_event
            }
            send_vms(params)
        else:
            continue


    result = {
        'statusCode': 200,
        'headers': {
            'Content-Type': 'application/json'
        }
    }
    return result
# ==================== MLP Client ====================

SERVICE_VERSION_V2 = "2024-07-01"

SERVICE_NAME = "ml_platform_stg"

service_info = ServiceInfo(
    "open.volcengineapi.com",
    {"Accept": "application/json"},
    Credentials(AK, SK, SERVICE_NAME, Region),
    60 * 5,
    60 * 5,
    "https",
)

api_info_v2 = {
    "GetDevInstance": ApiInfo(
        "POST",
        "/",
        {"Action": "GetDevInstance", "Version": SERVICE_VERSION_V2},
        {},
        {},
    ),
    "GetResourceQueue": ApiInfo(
        "POST",
        "/",
        {"Action": "GetResourceQueue", "Version": SERVICE_VERSION_V2},
        {},
        {},
    ),
}

class MLPClient(Service):
    def __init__(self) -> None:
        super(MLPClient, self).__init__(
            MLPClient.get_service_info(), MLPClient.get_api_info()
        )

    @staticmethod
    def get_service_info() -> ServiceInfo:
        return service_info

    @staticmethod
    def get_api_info() -> Dict[str, ApiInfo]:
        return api_info_v2

    def get_devinstance(self, instance_id: str) -> dict:
        res_str = self.json("GetDevInstance", [], json.dumps({"Id": instance_id}))
        res = json.loads(res_str)
        if not res.get("Result"):
            raise Exception(f"GetDevInstance failed, res: {res}")
        return res.get("Result")

    def get_resource_queue(self, resource_queue_id: str) -> dict:
        res_str = self.json(
            "GetResourceQueue", [], json.dumps({"Id": resource_queue_id})
        )
        res = json.loads(res_str)
        if not res.get("Result"):
            raise Exception(f"GetResourceQueue failed, res: {res}")
        return res.get("Result")

class DevInstanceDetail:
    id: str
    name: str
    resource_group_id: str
    resource_queue_id: str
    creator_trn: str

def get_devinstance_detail(eventResourceId: str) -> DevInstanceDetail:
    return get_devinstance_detail_by_devinstance_id(eventResourceId)

def get_devinstance_detail_by_devinstance_id(devinstance_id: str) -> DevInstanceDetail:
    client = MLPClient()
    res = client.get_devinstance(devinstance_id)
    detail = DevInstanceDetail()
    detail.id = devinstance_id
    detail.name = res.get("Name")
    detail.resource_queue_id = res.get("ResourceQueueId")
    detail.resource_group_id = client.get_resource_queue(detail.resource_queue_id).get(
        "ResourceGroupId"
    )
    detail.creator_trn = res.get("CreatorTrn")
    return detail

import json
import requests
import hashlib
import base64
import hmac
import time
import re
from typing import Dict
import smtplib
import volcenginesdkiam
import volcenginesdkcore

from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.application import MIMEApplication

from volcengine.ApiInfo import ApiInfo
from volcengine.Credentials import Credentials
from volcengine.base.Service import Service
from volcengine.ServiceInfo import ServiceInfo

# --------------------------- 用户需要自行修改的部分 ---------------------------
AK = "xxxx"
SK = "xxxx"
Region = "cn-beijing"
EmailSender = "xxxx@qq.com"
EmailAuthCode = "xxxx"
EmailSubject = "veMLP 事件通知邮件"
# --------------------------- 用户需要自行修改的部分 ---------------------------

configuration = volcenginesdkcore.Configuration()
configuration.ak = AK
configuration.sk = SK
configuration.region = Region
configuration.client_side_validation = True
# set default configuration
volcenginesdkcore.Configuration.set_default(configuration)

# use global default configuration
api_instance = volcenginesdkiam.IAMApi()

def send_email_via_qq_smtp(
    sender: str,
    auth_code: str,
    receiver: str,
    subject: str,
    content: str,
    attachment_path: str = None,
    is_html: bool = False
):
    """
    使用 QQ 邮箱 SMTP 发送邮件
    :param sender: 发件人邮箱（需已开启 SMTP）
    :param auth_code: SMTP 授权码（非邮箱登录密码）
    :param receiver: 收件人邮箱（多个用逗号分隔，如 "a@qq.com,b@qq.com"）
    :param subject: 邮件主题
    :param content: 邮件正文（纯文本或 HTML）
    :param attachment_path: 附件文件路径（可选）
    :param is_html: 是否以 HTML 格式发送正文
    """
    # 创建邮件对象
    msg = MIMEMultipart()
    msg["From"] = sender
    msg["To"] = receiver
    msg["Subject"] = subject

    # 添加正文（纯文本或 HTML）
    content_type = "html" if is_html else "plain"
    msg.attach(MIMEText(content, content_type, "utf-8"))

    # 添加附件（如果存在）
    if attachment_path:
        with open(attachment_path, "rb") as f:
            attachment = MIMEApplication(f.read())
            attachment.add_header(
                "Content-Disposition",
                "attachment",
                filename=attachment_path.split("/")[-1]  # 提取文件名
            )
            msg.attach(attachment)

    try:
        # 连接 SMTP 服务器（SSL 加密）
        with smtplib.SMTP_SSL("smtp.qq.com", 465) as server:
            server.login(sender, auth_code)
            server.sendmail(sender, receiver.split(","), msg.as_string())
            server.quit()
            print("邮件发送成功！")
    except Exception as e:
        print(f"邮件发送失败: {str(e)}")

def get_email_content(
    id,
    name,
    resource_group_id,
    resource_queue_id,
    creator_trn,
    event_time,
    message_event
):
    return f"""
    <div>
        <p>您好，</p>
        <p><b>ID</b>：{id}</p>
        <p><b>名称</b>：{name}</p>
        <p><b>资源组 ID</b>：{resource_group_id}</p>
        <p><b>队列 ID</b>：{resource_queue_id}</p>
        <p><b>创建人</b>：{creator_trn}</p>
        <p><b>事件时间</b>：{event_time}</p>
        <p><b>事件类型</b>：{message_event}</p>
        <p>感谢您的使用。</p>
    </div>
    """

def handler(event, context):
    print(f"received new request, event content: {event}")

    records = event.get('data', {}).get('Records', [])
    for record in records:
        message = json.loads(record.get('Sns', '{}').get('Message', '{}'))
        if message.get("EventResourceType") == "DevInstance":
            event_resource_id = message.get("EventResourceID")
            message_event = message.get("EventType", "")
            event_time = message.get("EventTime")
            devinstance_detail = get_devinstance_detail(event_resource_id)
            creator_trn = devinstance_detail.creator_trn

            # 确保有有效的收件人邮箱
            recipients = ["2512790809@qq.com"]

            # 生成邮件内容
            try:
                email_content = get_email_content(
                    devinstance_detail.id,
                    devinstance_detail.name,
                    devinstance_detail.resource_group_id,
                    devinstance_detail.resource_queue_id,
                    devinstance_detail.creator_trn,
                    event_time,
                    message_event,
                )
            except Exception as e:
                print(f"Error generating email content: {e}")
                continue

            # 发送邮件
            for recipient in recipients:
                try:
                    email_config = {
                        "sender": EmailSender,
                        "auth_code": EmailAuthCode,
                        "receiver": recipient,
                        "subject": EmailSubject,
                        "content": email_content,
                        "is_html": True
                    }
                    send_email_via_qq_smtp(**email_config)
                    print(f"Email sent to: {recipient}")
                except Exception as e:
                    print(f"Failed to send email to {recipient}: {e}")
        else:
            print(f"Skipping non-CustomTask event: {message.get('EventResourceType')}")

    return {
        'statusCode': 200,
        'headers': {
            'Content-Type': 'application/json'
        }
    }
# ==================== MLP Client ====================

SERVICE_VERSION_V2 = "2024-07-01"

SERVICE_NAME = "ml_platform_stg"

service_info = ServiceInfo(
    "open.volcengineapi.com",
    {"Accept": "application/json"},
    Credentials(AK, SK, SERVICE_NAME, Region),
    60 * 5,
    60 * 5,
    "https",
)

api_info_v2 = {
    "GetDevInstance": ApiInfo(
        "POST",
        "/",
        {"Action": "GetDevInstance", "Version": SERVICE_VERSION_V2},
        {},
        {},
    ),
    "GetResourceQueue": ApiInfo(
        "POST",
        "/",
        {"Action": "GetResourceQueue", "Version": SERVICE_VERSION_V2},
        {},
        {},
    ),
}

class MLPClient(Service):
    def __init__(self) -> None:
        super(MLPClient, self).__init__(
            MLPClient.get_service_info(), MLPClient.get_api_info()
        )

    @staticmethod
    def get_service_info() -> ServiceInfo:
        return service_info

    @staticmethod
    def get_api_info() -> Dict[str, ApiInfo]:
        return api_info_v2

    def get_devinstance(self, instance_id: str) -> dict:
        res_str = self.json("GetDevInstance", [], json.dumps({"Id": instance_id}))
        res = json.loads(res_str)
        if not res.get("Result"):
            raise Exception(f"GetDevInstance failed, res: {res}")
        return res.get("Result")

    def get_resource_queue(self, resource_queue_id: str) -> dict:
        res_str = self.json(
            "GetResourceQueue", [], json.dumps({"Id": resource_queue_id})
        )
        res = json.loads(res_str)
        if not res.get("Result"):
            raise Exception(f"GetResourceQueue failed, res: {res}")
        return res.get("Result")

class DevInstanceDetail:
    id: str
    name: str
    resource_group_id: str
    resource_queue_id: str
    creator_trn: str

def get_devinstance_detail(eventResourceId: str) -> DevInstanceDetail:
    return get_devinstance_detail_by_devinstance_id(eventResourceId)

def get_devinstance_detail_by_devinstance_id(devinstance_id: str) -> DevInstanceDetail:
    client = MLPClient()
    res = client.get_devinstance(devinstance_id)
    detail = DevInstanceDetail()
    detail.id = devinstance_id
    detail.name = res.get("Name")
    detail.resource_queue_id = res.get("ResourceQueueId")
    detail.resource_group_id = client.get_resource_queue(detail.resource_queue_id).get(
        "ResourceGroupId"
    )
    detail.creator_trn = res.get("CreatorTrn")
    return detail

预览效果

import json
import requests
import hashlib
import base64
import hmac
import time
import re
import smtplib
import uuid
import volcenginesdkiam
import volcenginesdkcore

from volcengine.sms.SmsService import SmsService
from volcenginesdkiam import GetUserResponse
from volcenginesdkcore.rest import ApiException

# --------------------------- 用户需要自行修改的部分 ---------------------------
AK = "xxxx"
SK = "xxxx"
Region = "cn-beijing"
SMSPhone = "xxxx"
SMSTemplateID = "xxxx"
SMSAccount = "xxxx"
SMSSign = "火山引擎"
# --------------------------- 用户需要自行修改的部分 ---------------------------

def _get_user_email(resp: GetUserResponse):
    if is_email(resp.user.user_name):
        return resp.user.user_name

    if is_email(resp.user.user_email):
        return resp.user.user_email

    return None

sms_service = SmsService()
sms_service.set_ak(AK)
sms_service.set_sk(SK)

configuration = volcenginesdkcore.Configuration()
configuration.ak = AK
configuration.sk = SK
configuration.region = Region
configuration.client_side_validation = True
# set default configuration
volcenginesdkcore.Configuration.set_default(configuration)

# use global default configuration
api_instance = volcenginesdkiam.IAMApi()


def is_email(email):
    # 定义邮箱的正则表达式
    pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+.[a-zA-Z]{2,}$'
    # 使用正则表达式匹配
    if re.match(pattern, email):
        return True
    else:
        return False


def get_user_email(user_id: str):
    if user_id is None:
        return user_id
    elif user_id == 0:
        return "主账号"
    try:
        resp = api_instance.get_user(
            volcenginesdkiam.GetUserRequest(
                id=user_id,
            ))
        email = _get_user_email(resp)
        return email if email else user_id
    except ApiException as e:
        print("Exception when calling IAMApi->get_user: %s
" % e)
        return user_id

def send_sms(
    template_params: dict,
):
    template_params_str = json.dumps(template_params)
    body = {
        "SmsAccount": SMSAccount,
        "Sign": SMSSign,
        "TemplateID": SMSTemplateID,
        "TemplateParam": template_params_str,
        "PhoneNumbers": SMSPhone
    }
    bodyStr = json.dumps(body)
    resp = sms_service.send_sms(bodyStr)
    print(resp)


def handler(event, context):
    # Log incoming request.
    #
    # NOTE: the log here is only for debug purpose. It's recommended to delete those debug/info
    # logs and only print log where error occurs after your business logic is verified and ready to
    # go production, nor the log amount may be too huge.
    print(f"received new request, event content: {event}")

    records = event.get('data', {}).get('Records', [])
    for record in records:
        print(f"get record: {record}")
        message = json.loads(record.get('Sns', '{}').get('Message', '{}'))
        print(f"get message: {message}")
        custom_task_id = ""
        if message.get('EventResourceType') == 'CustomTask':
            custom_task_id = message.get('EventResourceID')
            message_event = message.get('EventType', '')[10:]
            custom_task = message.get('Detail', {}).get('New') or message.get('Detail', {}).get('Old')
            custom_task_name = custom_task.get('Name')
            creator_email = get_user_email(custom_task.get('CreatorUserId'))
            accountId = message.get('AccountID')
            # if custom_task.get('CreatorUserId') != 6138818:
            #    continue
        else:
            continue
        params = {
            "UserName": creator_email,
            "AccountID": accountId,
            "CustomTaskName": custom_task_name,
            "CustomTaskId": custom_task_id,
            "MessageEvent": message_event,
        }
        send_sms(params)

    result = {
        'statusCode': 200,
        'headers': {
            'Content-Type': 'application/json'
        }
    }
    return result

效果预览：

需接入火山引擎语音服务产品

import json
import requests
import hashlib
import base64
import hmac
import time
import re
import smtplib
import uuid
from volcengine.vms.VmsService import VmsService

# --------------------------- 用户需要自行修改的部分 ---------------------------
AK = "xxxx"
SK = "xxxx"
Region = "cn-beijing"
VMSPhone = "xxxx"
VMSResource = "xxx"
VMSNumberPoolNo = "xxx"
# --------------------------- 用户需要自行修改的部分 ---------------------------

vms_service = VmsService()
vms_service.set_ak(AK)
vms_service.set_sk(SK)

def send_vms(
    phoneParam: dict,
):
    uuid_obj = uuid.uuid4()
    uuid_bytes = uuid_obj.bytes
    b64_str = base64.urlsafe_b64encode(uuid_bytes).rstrip(b'=').decode('utf-8')

    single_openid = b64_str
    single_append_body = {
        "List": [
            {
                "Phone": "xxxx",
                "Type": 1,
                "Resource": VMSResource,
                "NumberPoolNo": VMSNumberPoolNo,
                "SingleOpenId": single_openid,
                "PhoneParam": phoneParam
            },
        ]
    }
    result = vms_service.single_batch_append(single_append_body)
    print(f"received result from vms: {result}")


def handler(event, context):
    # Log incoming request.
    #
    # NOTE: the log here is only for debug purpose. It's recommended to delete those debug/info
    # logs and only print log where error occurs after your business logic is verified and ready to
    # go production, nor the log amount may be too huge.
    print(f"received new request, event content: {event}")

    records = event.get('data', {}).get('Records', [])
    for record in records:
        print(f"get record: {record}")
        message = json.loads(record.get('Sns', '{}').get('Message', '{}'))
        print(f"get message: {message}")
        if message.get('EventResourceType') == 'CustomTask':
            custom_task_id = message.get('EventResourceID')
            message_event = message.get('EventType', '')[10:]
            custom_task = message.get('Detail', {}).get('New') or message.get('Detail', {}).get('Old')
            custom_task_name = custom_task.get('Name')
            # if custom_task.get('CreatorUserId') != xxxx:
            #    continue
        else:
            continue

        params = {
            "CustomTaskName": custom_task_name,
            "CustomTaskID": custom_task_id,
            "MessageEvent": message_event
        }
        send_vms(params)

    result = {
        'statusCode': 200,
        'headers': {
            'Content-Type': 'application/json'
        }
    }
    return result

import json
import requests
import hashlib
import base64
import hmac
import time
import re
import smtplib
import volcenginesdkiam
import volcenginesdkcore

from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.application import MIMEApplication

from volcenginesdkiam import GetUserResponse
from volcenginesdkcore.rest import ApiException

# --------------------------- 用户需要自行修改的部分 ---------------------------
AK = "xxxx"
SK = "xxxx"
Region = "cn-beijing"
EmailSender = "xxxx@qq.com"
EmailAuthCode = "xxxx"
EmailSubject = "veMLP 事件通知邮件"


def _get_user_email(resp: GetUserResponse):
    if is_email(resp.user.user_name):
        return resp.user.user_name

    if is_email(resp.user.user_email):
        return resp.user.user_email

    return None
# --------------------------- 用户需要自行修改的部分 ---------------------------


configuration = volcenginesdkcore.Configuration()
configuration.ak = AK
configuration.sk = SK
configuration.region = Region
configuration.client_side_validation = True
# set default configuration
volcenginesdkcore.Configuration.set_default(configuration)

# use global default configuration
api_instance = volcenginesdkiam.IAMApi()


def is_email(email):
    # 定义邮箱的正则表达式
    pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+.[a-zA-Z]{2,}$'
    # 使用正则表达式匹配
    if re.match(pattern, email):
        return True
    else:
        return False


def get_user_email(user_id: str):
    if user_id is None:
        return user_id
    elif user_id == 0:
        return "主账号"
    try:
        resp = api_instance.get_user(
            volcenginesdkiam.GetUserRequest(
                id=user_id,
            ))
        email = _get_user_email(resp)
        return email if email else user_id
    except ApiException as e:
        print("Exception when calling IAMApi->get_user: %s
" % e)
        return user_id

def send_email_via_qq_smtp(
    sender: str,
    auth_code: str,
    receiver: str,
    subject: str,
    content: str,
    attachment_path: str = None,
    is_html: bool = False
):
    """
    使用 QQ 邮箱 SMTP 发送邮件
    :param sender: 发件人邮箱（需已开启 SMTP）
    :param auth_code: SMTP 授权码（非邮箱登录密码）
    :param receiver: 收件人邮箱（多个用逗号分隔，如 "a@qq.com,b@qq.com"）
    :param subject: 邮件主题
    :param content: 邮件正文（纯文本或 HTML）
    :param attachment_path: 附件文件路径（可选）
    :param is_html: 是否以 HTML 格式发送正文
    """
    # 创建邮件对象
    msg = MIMEMultipart()
    msg["From"] = sender
    msg["To"] = receiver
    msg["Subject"] = subject

    # 添加正文（纯文本或 HTML）
    content_type = "html" if is_html else "plain"
    msg.attach(MIMEText(content, content_type, "utf-8"))

    # 添加附件（如果存在）
    if attachment_path:
        with open(attachment_path, "rb") as f:
            attachment = MIMEApplication(f.read())
            attachment.add_header(
                "Content-Disposition",
                "attachment",
                filename=attachment_path.split("/")[-1]  # 提取文件名
            )
            msg.attach(attachment)

    try:
        # 连接 SMTP 服务器（SSL 加密）
        with smtplib.SMTP_SSL("smtp.qq.com", 465) as server:
            server.login(sender, auth_code)
            server.sendmail(sender, receiver.split(","), msg.as_string())
            server.quit()
            print("邮件发送成功！")
    except Exception as e:
        print(f"邮件发送失败: {str(e)}")

def get_email_content(
    custom_task_id,
    custom_task_name,
    resource_group_id,
    resource_queue_id,
    creator,
    operator,
    event_time,
    message_event
):
    return f"""
    <div>
        <p>您好，</p>
        <p>您有一个新的自定义任务：</p>
        <p><b>自定义任务 ID</b>：{custom_task_id}</p>
        <p><b>自定义任务名称</b>：{custom_task_name}</p>
        <p><b>资源组 ID</b>：{resource_group_id}</p>
        <p><b>队列 ID</b>：{resource_queue_id}</p>
        <p><b>创建人</b>：{creator}</p>
        <p><b>操作人</b>：{operator}</p>
        <p><b>事件时间</b>：{event_time}</p>
        <p><b>事件类型</b>：{message_event}</p>
        <p>感谢您的使用。</p>
    </div>
    """

def handler(event, context):
    # Log incoming request.
    #
    # NOTE: the log here is only for debug purpose. It's recommended to delete those debug/info
    # logs and only print log where error occurs after your business logic is verified and ready to
    # go production, nor the log amount may be too huge.
    print(f"received new request, event content: {event}")

    records = event.get('data', {}).get('Records', [])
    for record in records:
        print(f"get record: {record}")
        message = json.loads(record.get('Sns', '{}').get('Message', '{}'))
        print(f"get message: {message}")
        if message.get('EventResourceType') == 'CustomTask':
            custom_task_id = message.get('EventResourceID')
            message_event = message.get('EventType', '')[10:]
            event_time = message.get('EventTime')
            custom_task = message.get('Detail', {}).get('New') or message.get('Detail', {}).get('Old')
            custom_task_name = custom_task.get('Name')
            resource_group_id = custom_task.get('ResourceGroupId')
            resource_queue_id = custom_task.get('ResourceQueueId')
            # <at email="{operator}"></at>
            creator_email = get_user_email(custom_task.get('CreatorUserId'))
            creator = f'<at email="{creator_email}"></at>' if is_email(creator_email) else creator_email
            if message.get('EventTigerByUser'):
                operator_email = get_user_email(message.get('UserID'))
                operator = f'<at email="{operator_email}"></at>' if is_email(operator_email) else operator_email
            else:
                operator = "平台"
        else:
            continue

        email_content = get_email_content(
            custom_task_id,
            custom_task_name,
            resource_group_id,
            resource_queue_id,
            creator_email,
            operator_email,
            event_time,
            message_event,
        )
        print(f"get email_content: {email_content}")
        email_config = {
            "sender": EmailSender,
            "auth_code": EmailAuthCode,
            "receiver": operator_email,
            "subject": EmailSubject,
            "content": email_content,
            "is_html": True
        }
        send_email_via_qq_smtp(**email_config)

        email_config = {
            "sender": EmailSender,
            "auth_code": EmailAuthCode,
            "receiver": creator_email,
            "subject": EmailSubject,
            "content": email_content,
            "is_html": True
        }
        send_email_via_qq_smtp(**email_config)

    result = {
        'statusCode': 200,
        'headers': {
            'Content-Type': 'application/json'
        }
    }
    return result

预览效果

import json
import requests
import hashlib
import base64
import hmac
import time
import re
import re
from typing import Dict
import smtplib
import uuid
import volcenginesdkcore

from volcengine.sms.SmsService import SmsService
from volcengine.ApiInfo import ApiInfo
from volcengine.Credentials import Credentials
from volcengine.base.Service import Service
from volcengine.ServiceInfo import ServiceInfo

# --------------------------- 用户需要自行修改的部分 ---------------------------
AK = "xxxx"
SK = "xxxx"
Region = "cn-beijing"
SMSPhone = "xxxx"
SMSTemplateID = "xxxx"
SMSAccount = "xxxx"
SMSSign = "火山引擎"
UseMultiRole = True
# --------------------------- 用户需要自行修改的部分 ---------------------------

sms_service = SmsService()
sms_service.set_ak(AK)
sms_service.set_sk(SK)

configuration = volcenginesdkcore.Configuration()
configuration.ak = AK
configuration.sk = SK
configuration.region = Region
configuration.client_side_validation = True
# set default configuration
volcenginesdkcore.Configuration.set_default(configuration)

def send_sms(
    template_params: dict,
):
    template_params_str = json.dumps(template_params)
    body = {
        "SmsAccount": SMSAccount,
        "Sign": SMSSign,
        "TemplateID": SMSTemplateID,
        "TemplateParam": template_params_str,
        "PhoneNumbers": SMSPhone
    }
    bodyStr = json.dumps(body)
    resp = sms_service.send_sms(bodyStr)
    print(resp)

def handler(event, context):
    # Log incoming request.
    #
    # NOTE: the log here is only for debug purpose. It's recommended to delete those debug/info
    # logs and only print log where error occurs after your business logic is verified and ready to
    # go production, nor the log amount may be too huge.
    print(f"received new request, event content: {event}")

    records = event.get('data', {}).get('Records', [])
    for record in records:
        print(f"get record: {record}")
        message = json.loads(record.get('Sns', '{}').get('Message', '{}'))
        print(f"get message: {message}")
        if message.get('EventResourceType') == 'ServiceEvent':
            event_resource_id = message.get('EventResourceID')
            message_event = message.get('EventType', '')
            event_time = message.get('EventTime')
            inferenceDetail = get_inference_detail(message_event, event_resource_id)
            params = {
                "CustomTaskName": inferenceDetail.service_name,
                "CustomTaskId":  inferenceDetail.service_id,
                "MessageEvent": message_event,
            }
            send_sms(params)
        else:
            continue
    result = {
        'statusCode': 200,
        'headers': {
            'Content-Type': 'application/json'
        }
    }
    return result

SERVICE_VERSION_V1 = '2021-10-01'
SERVICE_VERSION_V2 = '2024-07-01'

service_info = ServiceInfo(
    'open.volcengineapi.com',
    {'Accept': 'application/json'},
    Credentials(AK, SK, 'ml_platform_stg', Region),
    60 * 5, 
    60 * 5,
    "https"
)

api_info_v1 = {
    'GetDeployment': ApiInfo('POST', '/',{
        "Action": "GetDeployment", "Version": SERVICE_VERSION_V1}, {}, {}),
    'GetService': ApiInfo('POST', '/',{
        "Action": "GetService", "Version": SERVICE_VERSION_V1}, {}, {}),
}

api_info_v2 = {
    'GetDeployment': ApiInfo('POST', '/',{
        "Action": "GetDeployment", "Version": SERVICE_VERSION_V2}, {}, {}),
    'GetService': ApiInfo('POST', '/',{
        "Action": "GetService", "Version": SERVICE_VERSION_V2}, {}, {}),
    'GetResourceQueue': ApiInfo('POST', '/',{
        "Action": "GetResourceQueue", "Version": SERVICE_VERSION_V2}, {}, {}),
}

class MLPClientV1(Service):
    def __init__(self) -> None:
        super(MLPClientV1, self).__init__(MLPClientV1.get_service_info(), MLPClientV1.get_api_info())
    
    @staticmethod
    def get_service_info() -> ServiceInfo:
        return service_info

    @staticmethod
    def get_api_info() -> Dict[str, ApiInfo]:
        return api_info_v1
    
    def get_deployment(self, service_id: str, deployment_id: str) -> dict:
        res_str = self.json('GetDeployment', [],json.dumps({
            "DeploymentId": deployment_id,
            "ServiceId": service_id
        }))
        res = json.loads(res_str)
        if not res.get('Result'):
            raise Exception(f'GetDeployment failed, res: {res}')
        return res.get('Result')
    
    def get_service(self, service_id: str) -> dict:
        res_str = self.json('GetService', [],json.dumps({
            "ServiceID": service_id
        }))
        res = json.loads(res_str)
        if not res.get('Result'):
            raise Exception(f'GetService failed, res: {res}')
        return res.get('Result')
    

class MLPClientV2(Service):
    def __init__(self) -> None:
        super(MLPClientV2, self).__init__(MLPClientV2.get_service_info(), MLPClientV2.get_api_info())
    
    @staticmethod
    def get_service_info() -> ServiceInfo:
        return service_info

    @staticmethod
    def get_api_info() -> Dict[str, ApiInfo]:
        return api_info_v2
    
    def get_deployment(self, deployment_id: str) -> dict:
        res_str = self.json('GetDeployment', [],json.dumps({
            "Id": deployment_id
        }))
        res = json.loads(res_str)
        if not res.get('Result'):
            raise Exception(f'GetDeployment failed, res: {res}')
        return res.get('Result')
    
    def get_service(self, service_id: str) -> dict:
        res_str = self.json('GetService',[], json.dumps({
            "Id": service_id
        }))
        res = json.loads(res_str)

        if not res.get('Result'):
            raise Exception(f'GetService failed, res: {res}')
        return res.get('Result')
    
    def get_resource_queue(self, resource_queue_id: str) -> dict:
        res_str = self.json('GetResourceQueue', [], json.dumps({
            "Id": resource_queue_id
        }))
        res = json.loads(res_str)
        if not res.get('Result'):
            raise Exception(f'GetResourceQueue failed, res: {res}')
        return res.get('Result')

class InferenceDetail():
    id: str
    name: str
    resource_group_id: str
    resource_queue_id: str
    service_id: str
    service_name: str

def get_inference_detail(eventType: str, eventResourceId: str) -> InferenceDetail:
    if eventType.startswith('Deployment'):
        return get_inference_detail_by_deployment_id(eventResourceId)
    elif eventType.startswith('Service'):
        return get_inference_detail_by_service_id(eventResourceId)
    elif eventType.startswith('Instance'):
        return get_inference_detail_by_instance_id(eventResourceId)
    else:
        return None
    
def get_inference_detail_by_service_id(service_id: str) -> InferenceDetail:
    detail = InferenceDetail()
    detail.id = service_id
    if UseMultiRole:
        client = MLPClientV2()
        res = client.get_service(service_id)
        detail.name = res.get('Name')
        detail.service_id = service_id
        detail.service_name = res.get('Name')
    else:
        client = MLPClientV1()
        res = client.get_service(service_id)
        detail.name = res.get('ServiceName')
        detail.service_id = service_id
        detail.service_name = res.get('Name')
    return detail

def get_inference_detail_by_deployment_id(deployment_id: str) -> InferenceDetail:
    detail = InferenceDetail()
    detail.id = deployment_id
    detail.resource_type = 'Deployment'
    if UseMultiRole:
        clientv2 = MLPClientV2()
        res = clientv2.get_deployment(deployment_id)
        detail.name = res.get('Name')
        detail.resource_queue_id = res.get('ResourceQueueId')
        resource_group_id = clientv2.get_resource_queue(detail.resource_queue_id).get('ResourceGroupId')
        detail.resource_group_id = resource_group_id
        service_id = deployment_id[:22]
        service = clientv2.get_service(service_id)
        detail.service_id = service_id
        detail.service_name = service.get('Name')
    else:
        clientv1 = MLPClientV1()
        clientv2 = MLPClientV2()
        service_id = deployment_id[:22]
        res = clientv1.get_deployment(service_id, deployment_id)
        detail.name = res.get('Name')
        detail.resource_queue_id = res.get('ResourceQueueId')
        resource_group_id = clientv2.get_resource_queue(detail.resource_queue_id).get('ResourceGroupId')
        detail.resource_group_id = resource_group_id
        detail.service_id = service_id
        service_info = clientv1.get_service(service_id)
        detail.service_name = service_info.get('ServiceName')
    return detail

def get_inference_detail_by_instance_id(instance_id: str) -> InferenceDetail:
    detail = InferenceDetail()
    detail.id = instance_id
    detail.resource_type = 'Instance'
    if UseMultiRole:
        clientv2 = MLPClientV2()
        deployment_id = instance_id[:31]
        res = clientv2.get_deployment(deployment_id)
        detail.resource_queue_id = res.get('ResourceQueueId')
        resource_group_id = clientv2.get_resource_queue(detail.resource_queue_id).get('ResourceGroupId')
        detail.resource_group_id = resource_group_id
        service_id = deployment_id[:22]
        service = clientv2.get_service(service_id)
        detail.service_id = service_id
        detail.service_name = service.get('Name')
    else:
        clientv1 = MLPClientV1()
        clientv2 = MLPClientV2()
        service_id = instance_id[:22]
        deployment_id = instance_id[:28]
        res = clientv1.get_deployment(service_id, deployment_id)
        detail.resource_queue_id = res.get('ResourceQueueId')
        resource_group_id = clientv2.get_resource_queue(detail.resource_queue_id).get('ResourceGroupId')
        detail.resource_group_id = resource_group_id
        detail.service_id = service_id
        service_info = clientv1.get_service(service_id)
        detail.service_name = service_info.get('ServiceName')
    return detail

效果预览：

需接入火山引擎语音服务产品

import json
import requests
import hashlib
import base64
import hmac
import time
import re
from typing import Dict
import smtplib
import uuid
import volcenginesdkcore
from volcengine.vms.VmsService import VmsService
from volcengine.ApiInfo import ApiInfo
from volcengine.Credentials import Credentials
from volcengine.base.Service import Service
from volcengine.ServiceInfo import ServiceInfo

# --------------------------- 用户需要自行修改的部分 ---------------------------
AK = "xxxx"
SK = "xxxx"
Region = "cn-beijing"
VMSPhone = "xxxx"
VMSResource = "xxx"
VMSNumberPoolNo = "xxx"
UseMultiRole = True
# --------------------------- 用户需要自行修改的部分 ---------------------------

vms_service = VmsService()
vms_service.set_ak(AK)
vms_service.set_sk(SK)

def send_vms(
    phoneParam: dict,
):
    uuid_obj = uuid.uuid4()
    uuid_bytes = uuid_obj.bytes
    b64_str = base64.urlsafe_b64encode(uuid_bytes).rstrip(b'=').decode('utf-8')

    single_openid = b64_str
    single_append_body = {
        "List": [
            {
                "Phone": "18271891346",
                "Type": 1,
                "Resource": VMSResource,
                "NumberPoolNo": VMSNumberPoolNo,
                "SingleOpenId": single_openid,
                "PhoneParam": phoneParam
            },
        ]
    }
    result = vms_service.single_batch_append(single_append_body)
    print(f"received result from vms: {result}")

def handler(event, context):
    # Log incoming request.
    #
    # NOTE: the log here is only for debug purpose. It's recommended to delete those debug/info
    # logs and only print log where error occurs after your business logic is verified and ready to
    # go production, nor the log amount may be too huge.
    print(f"received new request, event content: {event}")

    records = event.get('data', {}).get('Records', [])
    for record in records:
        print(f"get record: {record}")
        message = json.loads(record.get('Sns', '{}').get('Message', '{}'))
        print(f"get message: {message}")
        if message.get('EventResourceType') == 'ServiceEvent':
            event_resource_id = message.get('EventResourceID')
            message_event = message.get('EventType', '')
            event_time = message.get('EventTime')
            inferenceDetail = get_inference_detail(message_event, event_resource_id)
            params = {
                "CustomTaskName": inferenceDetail.service_name,
                "CustomTaskID":  inferenceDetail.service_id,
                "MessageEvent": message_event
            }
            send_vms(params)
        else:
            continue

    result = {
        'statusCode': 200,
        'headers': {
            'Content-Type': 'application/json'
        }
    }
    return result

SERVICE_VERSION_V1 = '2021-10-01'
SERVICE_VERSION_V2 = '2024-07-01'

service_info = ServiceInfo(
    'open.volcengineapi.com',
    {'Accept': 'application/json'},
    Credentials(AK, SK, 'ml_platform_stg', Region),
    60 * 5, 
    60 * 5,
    "https"
)

api_info_v1 = {
    'GetDeployment': ApiInfo('POST', '/',{
        "Action": "GetDeployment", "Version": SERVICE_VERSION_V1}, {}, {}),
    'GetService': ApiInfo('POST', '/',{
        "Action": "GetService", "Version": SERVICE_VERSION_V1}, {}, {}),
}

api_info_v2 = {
    'GetDeployment': ApiInfo('POST', '/',{
        "Action": "GetDeployment", "Version": SERVICE_VERSION_V2}, {}, {}),
    'GetService': ApiInfo('POST', '/',{
        "Action": "GetService", "Version": SERVICE_VERSION_V2}, {}, {}),
    'GetResourceQueue': ApiInfo('POST', '/',{
        "Action": "GetResourceQueue", "Version": SERVICE_VERSION_V2}, {}, {}),
}

class MLPClientV1(Service):
    def __init__(self) -> None:
        super(MLPClientV1, self).__init__(MLPClientV1.get_service_info(), MLPClientV1.get_api_info())
    
    @staticmethod
    def get_service_info() -> ServiceInfo:
        return service_info

    @staticmethod
    def get_api_info() -> Dict[str, ApiInfo]:
        return api_info_v1
    
    def get_deployment(self, service_id: str, deployment_id: str) -> dict:
        res_str = self.json('GetDeployment', [],json.dumps({
            "DeploymentId": deployment_id,
            "ServiceId": service_id
        }))
        res = json.loads(res_str)
        if not res.get('Result'):
            raise Exception(f'GetDeployment failed, res: {res}')
        return res.get('Result')
    
    def get_service(self, service_id: str) -> dict:
        res_str = self.json('GetService', [],json.dumps({
            "ServiceID": service_id
        }))
        res = json.loads(res_str)
        if not res.get('Result'):
            raise Exception(f'GetService failed, res: {res}')
        return res.get('Result')
    

class MLPClientV2(Service):
    def __init__(self) -> None:
        super(MLPClientV2, self).__init__(MLPClientV2.get_service_info(), MLPClientV2.get_api_info())
    
    @staticmethod
    def get_service_info() -> ServiceInfo:
        return service_info

    @staticmethod
    def get_api_info() -> Dict[str, ApiInfo]:
        return api_info_v2
    
    def get_deployment(self, deployment_id: str) -> dict:
        res_str = self.json('GetDeployment', [],json.dumps({
            "Id": deployment_id
        }))
        res = json.loads(res_str)
        if not res.get('Result'):
            raise Exception(f'GetDeployment failed, res: {res}')
        return res.get('Result')
    
    def get_service(self, service_id: str) -> dict:
        res_str = self.json('GetService',[], json.dumps({
            "Id": service_id
        }))
        res = json.loads(res_str)

        if not res.get('Result'):
            raise Exception(f'GetService failed, res: {res}')
        return res.get('Result')
    
    def get_resource_queue(self, resource_queue_id: str) -> dict:
        res_str = self.json('GetResourceQueue', [], json.dumps({
            "Id": resource_queue_id
        }))
        res = json.loads(res_str)
        if not res.get('Result'):
            raise Exception(f'GetResourceQueue failed, res: {res}')
        return res.get('Result')

class InferenceDetail():
    id: str
    name: str
    resource_group_id: str
    resource_queue_id: str
    service_id: str
    service_name: str

def get_inference_detail(eventType: str, eventResourceId: str) -> InferenceDetail:
    if eventType.startswith('Deployment'):
        return get_inference_detail_by_deployment_id(eventResourceId)
    elif eventType.startswith('Service'):
        return get_inference_detail_by_service_id(eventResourceId)
    elif eventType.startswith('Instance'):
        return get_inference_detail_by_instance_id(eventResourceId)
    else:
        return None
    
def get_inference_detail_by_service_id(service_id: str) -> InferenceDetail:
    detail = InferenceDetail()
    detail.id = service_id
    if UseMultiRole:
        client = MLPClientV2()
        res = client.get_service(service_id)
        detail.name = res.get('Name')
        detail.service_id = service_id
        detail.service_name = res.get('Name')
    else:
        client = MLPClientV1()
        res = client.get_service(service_id)
        detail.name = res.get('ServiceName')
        detail.service_id = service_id
        detail.service_name = res.get('Name')
    return detail

def get_inference_detail_by_deployment_id(deployment_id: str) -> InferenceDetail:
    detail = InferenceDetail()
    detail.id = deployment_id
    detail.resource_type = 'Deployment'
    if UseMultiRole:
        clientv2 = MLPClientV2()
        res = clientv2.get_deployment(deployment_id)
        detail.name = res.get('Name')
        detail.resource_queue_id = res.get('ResourceQueueId')
        resource_group_id = clientv2.get_resource_queue(detail.resource_queue_id).get('ResourceGroupId')
        detail.resource_group_id = resource_group_id
        service_id = deployment_id[:22]
        service = clientv2.get_service(service_id)
        detail.service_id = service_id
        detail.service_name = service.get('Name')
    else:
        clientv1 = MLPClientV1()
        clientv2 = MLPClientV2()
        service_id = deployment_id[:22]
        res = clientv1.get_deployment(service_id, deployment_id)
        detail.name = res.get('Name')
        detail.resource_queue_id = res.get('ResourceQueueId')
        resource_group_id = clientv2.get_resource_queue(detail.resource_queue_id).get('ResourceGroupId')
        detail.resource_group_id = resource_group_id
        detail.service_id = service_id
        service_info = clientv1.get_service(service_id)
        detail.service_name = service_info.get('ServiceName')
    return detail

def get_inference_detail_by_instance_id(instance_id: str) -> InferenceDetail:
    detail = InferenceDetail()
    detail.id = instance_id
    detail.resource_type = 'Instance'
    if UseMultiRole:
        clientv2 = MLPClientV2()
        deployment_id = instance_id[:31]
        res = clientv2.get_deployment(deployment_id)
        detail.resource_queue_id = res.get('ResourceQueueId')
        resource_group_id = clientv2.get_resource_queue(detail.resource_queue_id).get('ResourceGroupId')
        detail.resource_group_id = resource_group_id
        service_id = deployment_id[:22]
        service = clientv2.get_service(service_id)
        detail.service_id = service_id
        detail.service_name = service.get('Name')
    else:
        clientv1 = MLPClientV1()
        clientv2 = MLPClientV2()
        service_id = instance_id[:22]
        deployment_id = instance_id[:28]
        res = clientv1.get_deployment(service_id, deployment_id)
        detail.resource_queue_id = res.get('ResourceQueueId')
        resource_group_id = clientv2.get_resource_queue(detail.resource_queue_id).get('ResourceGroupId')
        detail.resource_group_id = resource_group_id
        detail.service_id = service_id
        service_info = clientv1.get_service(service_id)
        detail.service_name = service_info.get('ServiceName')
    return detail

import json
import requests
import hashlib
import base64
import hmac
import time
import re
import smtplib
from typing import Dict
import volcenginesdkiam
import volcenginesdkcore

from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.application import MIMEApplication

from volcengine.ApiInfo import ApiInfo
from volcengine.Credentials import Credentials
from volcengine.base.Service import Service
from volcengine.ServiceInfo import ServiceInfo

# --------------------------- 用户需要自行修改的部分 ---------------------------
AK = "xxxx"
SK = "xxxx"
Region = "cn-beijing"
EmailSender = "xxxx@qq.com"
EmailAuthCode = "xxxx"
EmailSubject = "veMLP 事件通知邮件"
UseMultiRole = True
# --------------------------- 用户需要自行修改的部分 ---------------------------

configuration = volcenginesdkcore.Configuration()
configuration.ak = AK
configuration.sk = SK
configuration.region = Region
configuration.client_side_validation = True
# set default configuration
volcenginesdkcore.Configuration.set_default(configuration)

def send_email_via_qq_smtp(
    sender: str,
    auth_code: str,
    receiver: str,
    subject: str,
    content: str,
    attachment_path: str = None,
    is_html: bool = False
):
    """
    使用 QQ 邮箱 SMTP 发送邮件
    :param sender: 发件人邮箱（需已开启 SMTP）
    :param auth_code: SMTP 授权码（非邮箱登录密码）
    :param receiver: 收件人邮箱（多个用逗号分隔，如 "a@qq.com,b@qq.com"）
    :param subject: 邮件主题
    :param content: 邮件正文（纯文本或 HTML）
    :param attachment_path: 附件文件路径（可选）
    :param is_html: 是否以 HTML 格式发送正文
    """
    # 创建邮件对象
    msg = MIMEMultipart()
    msg["From"] = sender
    msg["To"] = receiver
    msg["Subject"] = subject

    # 添加正文（纯文本或 HTML）
    content_type = "html" if is_html else "plain"
    msg.attach(MIMEText(content, content_type, "utf-8"))

    # 添加附件（如果存在）
    if attachment_path:
        with open(attachment_path, "rb") as f:
            attachment = MIMEApplication(f.read())
            attachment.add_header(
                "Content-Disposition",
                "attachment",
                filename=attachment_path.split("/")[-1]  # 提取文件名
            )
            msg.attach(attachment)

    try:
        # 连接 SMTP 服务器（SSL 加密）
        with smtplib.SMTP_SSL("smtp.qq.com", 465) as server:
            server.login(sender, auth_code)
            server.sendmail(sender, receiver.split(","), msg.as_string())
            server.quit()
            print("邮件发送成功！")
    except Exception as e:
        print(f"邮件发送失败: {str(e)}")

def get_email_content(
    id,
    service_id,
    service_name,
    resource_group_id,
    resource_queue_id,
    event_time,
    message_event
):
    return f"""
    <div>
        <p>您好，</p>
        <p>您有一个新的在线服务事件：</p>
        <p><b>ID</b>：{id}</p>
        <p><b>服务ID</b>：{service_id}</p>
        <p><b>服务名称</b>：{service_name}</p>
        <p><b>资源组 ID</b>：{resource_group_id}</p>
        <p><b>队列 ID</b>：{resource_queue_id}</p>
        <p><b>事件时间</b>：{event_time}</p>
        <p><b>事件类型</b>：{message_event}</p>
        <p>感谢您的使用。</p>
    </div>
    """

def handler(event, context):
    # Log incoming request.
    #
    # NOTE: the log here is only for debug purpose. It's recommended to delete those debug/info
    # logs and only print log where error occurs after your business logic is verified and ready to
    # go production, nor the log amount may be too huge.
    print(f"received new request, event content: {event}")

    records = event.get('data', {}).get('Records', [])
    for record in records:
        print(f"get record: {record}")
        message = json.loads(record.get('Sns', '{}').get('Message', '{}'))
        print(f"get message: {message}")
        if message.get('EventResourceType') == 'ServiceEvent':
            event_resource_id = message.get('EventResourceID')
            message_event = message.get('EventType', '')
            event_time = message.get('EventTime')
            inferenceDetail = get_inference_detail(message_event, event_resource_id)

            # 确保有有效的收件人邮箱
            recipients = ["zhucaili@bytedance.com"]
            if not recipients:
                print("No valid email recipients found.")
                continue

            # 生成邮件内容
            try:
                email_content = get_email_content(
                    inferenceDetail.id,
                    inferenceDetail.service_id,
                    inferenceDetail.service_name,
                    inferenceDetail.resource_group_id,
                    inferenceDetail.resource_queue_id,
                    event_time,
                    message_event,
                )
            except Exception as e:
                print(f"Error generating email content: {e}")
                continue

            # 发送邮件
            for recipient in recipients:
                try:
                    email_config = {
                        "sender": EmailSender,
                        "auth_code": EmailAuthCode,
                        "receiver": recipient,
                        "subject": EmailSubject,
                        "content": email_content,
                        "is_html": True
                    }
                    send_email_via_qq_smtp(**email_config)
                    print(f"Email sent to: {recipient}")
                except Exception as e:
                    print(f"Failed to send email to {recipient}: {e}")
        else:
            print(f"Skipping non-CustomTask event: {message.get('EventResourceType')}")

    return {
        'statusCode': 200,
        'headers': {
            'Content-Type': 'application/json'
        }
    }

SERVICE_VERSION_V1 = '2021-10-01'
SERVICE_VERSION_V2 = '2024-07-01'

service_info = ServiceInfo(
    'open.volcengineapi.com',
    {'Accept': 'application/json'},
    Credentials(AK, SK, 'ml_platform_stg', Region),
    60 * 5, 
    60 * 5,
    "https"
)

api_info_v1 = {
    'GetDeployment': ApiInfo('POST', '/',{
        "Action": "GetDeployment", "Version": SERVICE_VERSION_V1}, {}, {}),
    'GetService': ApiInfo('POST', '/',{
        "Action": "GetService", "Version": SERVICE_VERSION_V1}, {}, {}),
}

api_info_v2 = {
    'GetDeployment': ApiInfo('POST', '/',{
        "Action": "GetDeployment", "Version": SERVICE_VERSION_V2}, {}, {}),
    'GetService': ApiInfo('POST', '/',{
        "Action": "GetService", "Version": SERVICE_VERSION_V2}, {}, {}),
    'GetResourceQueue': ApiInfo('POST', '/',{
        "Action": "GetResourceQueue", "Version": SERVICE_VERSION_V2}, {}, {}),
}

class MLPClientV1(Service):
    def __init__(self) -> None:
        super(MLPClientV1, self).__init__(MLPClientV1.get_service_info(), MLPClientV1.get_api_info())
    
    @staticmethod
    def get_service_info() -> ServiceInfo:
        return service_info

    @staticmethod
    def get_api_info() -> Dict[str, ApiInfo]:
        return api_info_v1
    
    def get_deployment(self, service_id: str, deployment_id: str) -> dict:
        res_str = self.json('GetDeployment', [],json.dumps({
            "DeploymentId": deployment_id,
            "ServiceId": service_id
        }))
        res = json.loads(res_str)
        if not res.get('Result'):
            raise Exception(f'GetDeployment failed, res: {res}')
        return res.get('Result')
    
    def get_service(self, service_id: str) -> dict:
        res_str = self.json('GetService', [],json.dumps({
            "ServiceID": service_id
        }))
        res = json.loads(res_str)
        if not res.get('Result'):
            raise Exception(f'GetService failed, res: {res}')
        return res.get('Result')
    

class MLPClientV2(Service):
    def __init__(self) -> None:
        super(MLPClientV2, self).__init__(MLPClientV2.get_service_info(), MLPClientV2.get_api_info())
    
    @staticmethod
    def get_service_info() -> ServiceInfo:
        return service_info

    @staticmethod
    def get_api_info() -> Dict[str, ApiInfo]:
        return api_info_v2
    
    def get_deployment(self, deployment_id: str) -> dict:
        res_str = self.json('GetDeployment', [],json.dumps({
            "Id": deployment_id
        }))
        res = json.loads(res_str)
        if not res.get('Result'):
            raise Exception(f'GetDeployment failed, res: {res}')
        return res.get('Result')
    
    def get_service(self, service_id: str) -> dict:
        res_str = self.json('GetService',[], json.dumps({
            "Id": service_id
        }))
        res = json.loads(res_str)

        if not res.get('Result'):
            raise Exception(f'GetService failed, res: {res}')
        return res.get('Result')
    
    def get_resource_queue(self, resource_queue_id: str) -> dict:
        res_str = self.json('GetResourceQueue', [], json.dumps({
            "Id": resource_queue_id
        }))
        res = json.loads(res_str)
        if not res.get('Result'):
            raise Exception(f'GetResourceQueue failed, res: {res}')
        return res.get('Result')

class InferenceDetail():
    id: str
    name: str
    resource_group_id: str
    resource_queue_id: str
    creator: str
    service_id: str
    service_name: str

def get_inference_detail(eventType: str, eventResourceId: str) -> InferenceDetail:
    if eventType.startswith('Deployment'):
        return get_inference_detail_by_deployment_id(eventResourceId)
    elif eventType.startswith('Service'):
        return get_inference_detail_by_service_id(eventResourceId)
    elif eventType.startswith('Instance'):
        return get_inference_detail_by_instance_id(eventResourceId)
    else:
        return None
    
def get_inference_detail_by_service_id(service_id: str) -> InferenceDetail:
    detail = InferenceDetail()
    detail.id = service_id
    if UseMultiRole:
        client = MLPClientV2()
        res = client.get_service(service_id)
        detail.name = res.get('Name')
        detail.creator = res.get('CreatedBy')
        detail.service_id = service_id
        detail.service_name = res.get('Name')
    else:
        client = MLPClientV1()
        res = client.get_service(service_id)
        detail.name = res.get('ServiceName')
        detail.creator = res.get('CreatorUserId')
        detail.service_id = service_id
        detail.service_name = res.get('Name')
    return detail

def get_inference_detail_by_deployment_id(deployment_id: str) -> InferenceDetail:
    detail = InferenceDetail()
    detail.id = deployment_id
    detail.resource_type = 'Deployment'
    if UseMultiRole:
        clientv2 = MLPClientV2()
        res = clientv2.get_deployment(deployment_id)
        detail.name = res.get('Name')
        detail.creator = res.get('CreatedBy')
        detail.resource_queue_id = res.get('ResourceQueueId')
        resource_group_id = clientv2.get_resource_queue(detail.resource_queue_id).get('ResourceGroupId')
        detail.resource_group_id = resource_group_id
        service_id = deployment_id[:22]
        service = clientv2.get_service(service_id)
        detail.service_id = service_id
        detail.service_name = service.get('Name')
    else:
        clientv1 = MLPClientV1()
        clientv2 = MLPClientV2()
        service_id = deployment_id[:22]
        res = clientv1.get_deployment(service_id, deployment_id)
        detail.name = res.get('Name')
        detail.creator = res.get('UserId')
        detail.resource_queue_id = res.get('ResourceQueueId')
        resource_group_id = clientv2.get_resource_queue(detail.resource_queue_id).get('ResourceGroupId')
        detail.resource_group_id = resource_group_id
        detail.service_id = service_id
        service_info = clientv1.get_service(service_id)
        detail.service_name = service_info.get('ServiceName')
    return detail

def get_inference_detail_by_instance_id(instance_id: str) -> InferenceDetail:
    detail = InferenceDetail()
    detail.id = instance_id
    detail.resource_type = 'Instance'
    if UseMultiRole:
        clientv2 = MLPClientV2()
        deployment_id = instance_id[:31]
        res = clientv2.get_deployment(deployment_id)
        detail.creator = res.get('CreatedBy')
        detail.resource_queue_id = res.get('ResourceQueueId')
        resource_group_id = clientv2.get_resource_queue(detail.resource_queue_id).get('ResourceGroupId')
        detail.resource_group_id = resource_group_id
        service_id = deployment_id[:22]
        service = clientv2.get_service(service_id)
        detail.service_id = service_id
        detail.service_name = service.get('Name')
    else:
        clientv1 = MLPClientV1()
        clientv2 = MLPClientV2()
        service_id = instance_id[:22]
        deployment_id = instance_id[:28]
        res = clientv1.get_deployment(service_id, deployment_id)
        detail.creator = res.get('UserId')
        detail.resource_queue_id = res.get('ResourceQueueId')
        resource_group_id = clientv2.get_resource_queue(detail.resource_queue_id).get('ResourceGroupId')
        detail.resource_group_id = resource_group_id
        detail.service_id = service_id
        service_info = clientv1.get_service(service_id)
        detail.service_name = service_info.get('ServiceName')
    return detail

预览效果

import os
import re
import json
import smtplib
import requests
import volcenginesdkiam
import volcenginesdkcore

from volcenginesdkiam import GetUserResponse
from volcenginesdkcore.rest import ApiException
from volcengine.sms.SmsService import SmsService

from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.application import MIMEApplication

# --------------------------- 用户需要自行修改的部分 ---------------------------
AK = "xxxx"
SK = "xxxx"
Region = "cn-beijing"
SMSPhone = "xxxx"
SMSTemplateID = "xxxx"
SMSAccount = "xxxx"
SMSSign = "火山引擎"
# --------------------------- 用户需要自行修改的部分 ---------------------------


type_color = {
    "CreatePipelineEvent": "green",
    "UpdatePipelineEvent": "green",
    "DeletePipelineEvent": "yellow",
    "PipelineInstanceInitializedEvent": "blue",
    "PipelineInstanceRunningEvent": "green",
    "PipelineInstanceSuccessEvent": "green",
    "PipelineInstanceFailedEvent": "red",
    "PipelineInstanceKillingEvent": "green",
    "PipelineInstanceKilledEvent": "yellow",
}

status_hint= {
    "CreatePipelineEvent": "创建成功",
    "UpdatePipelineEvent": "更新成功",
    "DeletePipelineEvent": "删除成功",
    "PipelineInstanceInitializedEvent": "初始化中",
    "PipelineInstanceRunningEvent": "运行中",
    "PipelineInstanceSuccessEvent": "运行成功",
    "PipelineInstanceFailedEvent": "运行失败",
    "PipelineInstanceKillingEvent": "停止中",
    "PipelineInstanceKilledEvent": "已停止"
}

def is_email(email):
    # 定义邮箱的正则表达式
    pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+.[a-zA-Z]{2,}$'
    # 使用正则表达式匹配
    if re.match(pattern, email):
        return True
    else:
        return False

def _get_user_email(resp: GetUserResponse):
    if is_email(resp.user.user_name):
        return resp.user.user_name

    if is_email(resp.user.user_email):
        return resp.user.user_email
    return None

def get_pipeline_card(
    region,
    pipeline_id,
    pipeline_name,
    creator,
    operator,
    event_time,
    event_type,
    event_service,
    event_template
):
    return {
        "config": {
            "update_multi": True
        },
        "i18n_elements": {
            "zh_cn": [
                {
                    "tag": "column_set",
                    "horizontal_spacing": "8px",
                    "horizontal_align": "left",
                    "columns": [
                        {
                            "tag": "column",
                            "width": "weighted",
                            "elements": [
                                {
                                    "tag": "markdown",
                                    "content": f"**工作流**: {pipeline_name}(ID: {pipeline_id})",
                                    "text_align": "left",
                                    "text_size": "normal",
                                    "icon": {
                                        "tag": "standard_icon",
                                        "token": "card_outlined",
                                        "color": "grey"
                                    }
                                },
                                {
                                    "tag": "markdown",
                                    "content": f"**创建人：**{creator}",
                                    "text_align": "left",
                                    "text_size": "normal",
                                    "icon": {
                                        "tag": "standard_icon",
                                        "token": "member_outlined",
                                        "color": "grey"
                                    }
                                },
                                {
                                    "tag": "markdown",
                                    "content": f"**操作人：**{operator}",
                                    "text_align": "left",
                                    "text_size": "normal",
                                    "icon": {
                                        "tag": "standard_icon",
                                        "token": "member_outlined",
                                        "color": "grey"
                                    }
                                },
                                {
                                    "tag": "markdown",
                                    "content": f"**事件时间：** {event_time}",
                                    "text_align": "left",
                                    "text_size": "normal",
                                    "icon": {
                                        "tag": "standard_icon",
                                        "token": "time_outlined",
                                        "color": "grey"
                                    }
                                }
                            ],
                            "vertical_align": "top",
                            "vertical_spacing": "8px",
                            "weight": 2
                        }
                    ]
                },
                {
                    "tag": "hr"
                },
                {
                    "tag": "action",
                    "actions": [
                        {
                            "tag": "button",
                            "text": {
                                "tag": "plain_text",
                                "content": "查看详情"
                            },
                            "url": f"https://console.volcengine.com/{event_service.replace('_', '-')}/region:{event_service.replace('_', '-')}+{region}/pipeline/detail/{pipeline_id}",
                            "type": "default",
                            "width": "default",
                            "size": "medium",
                            "icon": {
                                "tag": "standard_icon",
                                "token": "details_outlined"
                            }
                        }
                    ]

                },
                {
                    "tag": "hr"
                }
            ]
        },
        "i18n_header": {
            "zh_cn": {
                "title": {
                    "tag": "plain_text",
                    "content": f"【火山引擎-机器学习平台】【{region}】 工作流 {pipeline_id} {status_hint.get(event_type)}"
                },
                "subtitle": {
                    "tag": "plain_text",
                    "content": ""
                },
                "template": f"{event_template}"
            }
        }
    }

def get_pipeline_instance_card(
        region,
        pipeline_instance_id,
        pipeline_instance_name,
        pipeline_id,
        parent_pipeline_instance_id,
        creator,
        operator,
        event_time,
        event_type,
        event_service,
        event_template
):
    return {
        "config": {
            "update_multi": True
        },
        "i18n_elements": {
            "zh_cn": [
                {
                    "tag": "column_set",
                    "horizontal_spacing": "8px",
                    "horizontal_align": "left",
                    "columns": [
                        {
                            "tag": "column",
                            "width": "weighted",
                            "elements": [
                                {
                                    "tag": "markdown",
                                    "content": f"**工作流**: {pipeline_id}",
                                    "text_align": "left",
                                    "text_size": "normal",
                                    "icon": {
                                        "tag": "standard_icon",
                                        "token": "card_outlined",
                                        "color": "grey"
                                    }
                                },
                                {
                                    "tag": "markdown",
                                    "content": f"**工作流实例**: {pipeline_instance_name}(ID: {pipeline_instance_id})",
                                    "text_align": "left",
                                    "text_size": "normal",
                                    "icon": {
                                        "tag": "standard_icon",
                                        "token": "card_outlined",
                                        "color": "grey"
                                    }
                                },
                                {
                                    "tag": "markdown",
                                    "content": f"**是否重试**: {'是' if parent_pipeline_instance_id is not None else '否'}",
                                    "text_align": "left",
                                    "text_size": "normal",
                                    "icon": {
                                        "tag": "standard_icon",
                                        "token": "card_outlined",
                                        "color": "grey"
                                    }
                                },
                                {
                                    "tag": "markdown",
                                    "content": f"**创建人：**{creator}",
                                    "text_align": "left",
                                    "text_size": "normal",
                                    "icon": {
                                        "tag": "standard_icon",
                                        "token": "member_outlined",
                                        "color": "grey"
                                    }
                                },
                                {
                                    "tag": "markdown",
                                    "content": f"**操作人：**{operator}",
                                    "text_align": "left",
                                    "text_size": "normal",
                                    "icon": {
                                        "tag": "standard_icon",
                                        "token": "member_outlined",
                                        "color": "grey"
                                    }
                                },
                                {
                                    "tag": "markdown",
                                    "content": f"**事件时间：** {event_time}",
                                    "text_align": "left",
                                    "text_size": "normal",
                                    "icon": {
                                        "tag": "standard_icon",
                                        "token": "time_outlined",
                                        "color": "grey"
                                    }
                                }
                            ],
                            "vertical_align": "top",
                            "vertical_spacing": "8px",
                            "weight": 2
                        }
                    ]
                },
                {
                    "tag": "hr"
                },
                {
                    "tag": "action",
                    "actions": [
                        {
                            "tag": "button",
                            "text": {
                                "tag": "plain_text",
                                "content": "查看详情"
                            },
                            "url": f"https://console.volcengine.com/{event_service.replace('_', '-')}/region:{event_service.replace('_', '-')}+{region}/pipeline/detail/{pipeline_id}?activeBar=operationRecord&selectedId={pipeline_instance_id}",
                            "type": "default",
                            "width": "default",
                            "size": "medium",
                            "icon": {
                                "tag": "standard_icon",
                                "token": "details_outlined"
                            }
                        }
                    ]

                }
            ]
        },
        "i18n_header": {
            "zh_cn": {
                "title": {
                    "tag": "plain_text",
                    "content": f"【火山引擎-机器学习平台】【{region}】 工作流实例 {pipeline_instance_id} {status_hint.get(event_type)}"
                },
                "subtitle": {
                    "tag": "plain_text",
                    "content": ""
                },
                "template": f"{event_template}"
            }
        }
    }


class PipelineEvent:
    def __init__(self, region, webhook, email_sender, email_auth_code, sms_phone, sms_tpl_id, sms_account, sms_sign):
        self.region = region

        configuration = volcenginesdkcore.Configuration()
        configuration.ak = AK
        configuration.sk = SK
        configuration.region = region
        configuration.client_side_validation = True
        volcenginesdkcore.Configuration.set_default(configuration)
        self.iam_client = volcenginesdkiam.IAMApi()

        self.sms_service = SmsService()
        self.sms_service.set_ak(AK)
        self.sms_service.set_sk(SK)
        self.sms_phone = sms_phone
        self.sms_tpl_id = sms_tpl_id
        self.sms_account = sms_account
        self.sms_sign = sms_sign

    def get_user_email(self, user_id: str):
        if user_id is None:
            return user_id
        elif user_id == 0:
            return "主账号"
        try:
            resp = self.iam_client.get_user(
                volcenginesdkiam.GetUserRequest(
                    id=user_id,
                ))
            email = _get_user_email(resp)
            return email if email else user_id
        except ApiException as e:
            print("Exception when calling IAMApi->get_user: %s
" % e)
            return user_id

    def get_user_name(self, user_id: str):
        if user_id is None:
            return user_id
        elif user_id == 0:
            return "主账号"
        try:
            resp = self.iam_client.get_user(
                volcenginesdkiam.GetUserRequest(
                    id=user_id,
                ))
            return resp.user.user_name
        except ApiException as e:
            print("Exception when calling IAMApi->get_user: %s
" % e)
            return user_id

    def send_sms(self, params: dict):
        template_params = json.dumps(params)
        body = {
            "SmsAccount": self.sms_account,
            "Sign": self.sms_sign,
            "TemplateID": self.sms_tpl_id,
            "TemplateParam": template_params,
            "PhoneNumbers": self.sms_phone
        }
        body_str = json.dumps(body)
        resp = self.sms_service.send_sms(body_str)
        print(resp)

    def handle(self, msg: dict):
        resource_id = msg.get('EventResourceID') # pipeline id or pipeline instance id
        event_type = msg.get('EventType')
        event_time = msg.get('EventTime')
        event_service = msg.get('EventService')
        account_id = msg.get('AccountID')
        identity_type = msg.get('IdentityType')
        user_id = msg.get('UserID')
        resource_type = msg.get('EventResourceType')

        operator = "平台"
        operator_email = None
        user_name = None
        if msg.get('EventTigerByUser'):
            operator_email = self.get_user_email(msg.get('UserID'))
            operator = f'<at email="{operator_email}"></at>' if is_email(operator_email) else operator_email
            user_name = self.get_user_name(msg.get('UserID'))

        extra_info = json.loads(msg.get('Detail').get('Message', {}))
        creator_email = self.get_user_email(extra_info.get('OwnerId'))
        creator = f'<at email="{creator_email}"></at>' if is_email(creator_email) else creator_email
        if user_name is None:
            user_name = self.get_user_name(extra_info.get('OwnerId'))

        recipients = merge_email_recipients(operator_email, creator_email)

        card = {}
        email_config = {}
        params = {}
        if resource_type == 'Pipeline':
            pipeline_name = extra_info.get('PipelineName')

            params = {
                "UserName": user_name,
                "AccountID": account_id,
                "ResourcName": pipeline_name,
                "ResourceId": resource_id,
                "EventName": status_hint[event_type],
            }

        elif resource_type == 'PipelineInstance':
            pipeline_instance_name = extra_info.get('PipelineInstanceName')
            pipeline_id = extra_info.get('PipelineId')
            # trigger = extra_info.get("Trigger") # Manual
            # diag_info = extra_info.get("DiagInfo")
            # reason = extra_info.get("Reason")
            parent_pipeline_instance_id = extra_info.get("ParentPipelineInstanceId", None)

            params = {
                "UserName": user_name,
                "AccountID": account_id,
                "ResourcName": pipeline_instance_name,
                "ResourceId": resource_id,
                "EventName": status_hint[event_type],
            }

        else:
            return

        self.send_sms(params)

pipeline_event = PipelineEvent(
    region=Region,
    sms_phone=SMSPhone,
    sms_tpl_id=SMSTemplateID ,
    sms_account=SMSAccount,
    sms_sign=SMSSign,
)

def handler(event, context):
    print(f"received new request, event content: {event}")

    records = event.get('data', {}).get('Records', [])
    for record in records:
        print(f"get record: {record}")
        message = json.loads(record.get('Sns', '{}').get('Message', '{}'))
        print(f"get message: {message}")
        pipeline_event.handle(message)

    result = {
        'statusCode': 200,
        'headers': {
            'Content-Type': 'application/json'
        }
    }
    return result

import os
import re
import json
import smtplib
import requests
import volcenginesdkiam
import volcenginesdkcore

from volcenginesdkiam import GetUserResponse
from volcenginesdkcore.rest import ApiException
from volcengine.sms.SmsService import SmsService

from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.application import MIMEApplication

# --------------------------- 用户需要自行修改的部分 ---------------------------
AK = "xxxx"
SK = "xxxx"
Region = "cn-beijing"
EmailSender = "xxxx@qq.com"
EmailAuthCode = "xxxx"
# --------------------------- 用户需要自行修改的部分 ---------------------------

type_color = {
    "CreatePipelineEvent": "green",
    "UpdatePipelineEvent": "green",
    "DeletePipelineEvent": "yellow",
    "PipelineInstanceInitializedEvent": "blue",
    "PipelineInstanceRunningEvent": "green",
    "PipelineInstanceSuccessEvent": "green",
    "PipelineInstanceFailedEvent": "red",
    "PipelineInstanceKillingEvent": "green",
    "PipelineInstanceKilledEvent": "yellow",
}

status_hint= {
    "CreatePipelineEvent": "创建成功",
    "UpdatePipelineEvent": "更新成功",
    "DeletePipelineEvent": "删除成功",
    "PipelineInstanceInitializedEvent": "初始化中",
    "PipelineInstanceRunningEvent": "运行中",
    "PipelineInstanceSuccessEvent": "运行成功",
    "PipelineInstanceFailedEvent": "运行失败",
    "PipelineInstanceKillingEvent": "停止中",
    "PipelineInstanceKilledEvent": "已停止"
}

def is_email(email):
    # 定义邮箱的正则表达式
    pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+.[a-zA-Z]{2,}$'
    # 使用正则表达式匹配
    if re.match(pattern, email):
        return True
    else:
        return False

def _get_user_email(resp: GetUserResponse):
    if is_email(resp.user.user_name):
        return resp.user.user_name

    if is_email(resp.user.user_email):
        return resp.user.user_email
    return None

def get_pipeline_card(
    region,
    pipeline_id,
    pipeline_name,
    creator,
    operator,
    event_time,
    event_type,
    event_service,
    event_template
):
    return {
        "config": {
            "update_multi": True
        },
        "i18n_elements": {
            "zh_cn": [
                {
                    "tag": "column_set",
                    "horizontal_spacing": "8px",
                    "horizontal_align": "left",
                    "columns": [
                        {
                            "tag": "column",
                            "width": "weighted",
                            "elements": [
                                {
                                    "tag": "markdown",
                                    "content": f"**工作流**: {pipeline_name}(ID: {pipeline_id})",
                                    "text_align": "left",
                                    "text_size": "normal",
                                    "icon": {
                                        "tag": "standard_icon",
                                        "token": "card_outlined",
                                        "color": "grey"
                                    }
                                },
                                {
                                    "tag": "markdown",
                                    "content": f"**创建人：**{creator}",
                                    "text_align": "left",
                                    "text_size": "normal",
                                    "icon": {
                                        "tag": "standard_icon",
                                        "token": "member_outlined",
                                        "color": "grey"
                                    }
                                },
                                {
                                    "tag": "markdown",
                                    "content": f"**操作人：**{operator}",
                                    "text_align": "left",
                                    "text_size": "normal",
                                    "icon": {
                                        "tag": "standard_icon",
                                        "token": "member_outlined",
                                        "color": "grey"
                                    }
                                },
                                {
                                    "tag": "markdown",
                                    "content": f"**事件时间：** {event_time}",
                                    "text_align": "left",
                                    "text_size": "normal",
                                    "icon": {
                                        "tag": "standard_icon",
                                        "token": "time_outlined",
                                        "color": "grey"
                                    }
                                }
                            ],
                            "vertical_align": "top",
                            "vertical_spacing": "8px",
                            "weight": 2
                        }
                    ]
                },
                {
                    "tag": "hr"
                },
                {
                    "tag": "action",
                    "actions": [
                        {
                            "tag": "button",
                            "text": {
                                "tag": "plain_text",
                                "content": "查看详情"
                            },
                            "url": f"https://console.volcengine.com/{event_service.replace('_', '-')}/region:{event_service.replace('_', '-')}+{region}/pipeline/detail/{pipeline_id}",
                            "type": "default",
                            "width": "default",
                            "size": "medium",
                            "icon": {
                                "tag": "standard_icon",
                                "token": "details_outlined"
                            }
                        }
                    ]

                },
                {
                    "tag": "hr"
                }
            ]
        },
        "i18n_header": {
            "zh_cn": {
                "title": {
                    "tag": "plain_text",
                    "content": f"【火山引擎-机器学习平台】【{region}】 工作流 {pipeline_id} {status_hint.get(event_type)}"
                },
                "subtitle": {
                    "tag": "plain_text",
                    "content": ""
                },
                "template": f"{event_template}"
            }
        }
    }

def get_pipeline_instance_card(
        region,
        pipeline_instance_id,
        pipeline_instance_name,
        pipeline_id,
        parent_pipeline_instance_id,
        creator,
        operator,
        event_time,
        event_type,
        event_service,
        event_template
):
    return {
        "config": {
            "update_multi": True
        },
        "i18n_elements": {
            "zh_cn": [
                {
                    "tag": "column_set",
                    "horizontal_spacing": "8px",
                    "horizontal_align": "left",
                    "columns": [
                        {
                            "tag": "column",
                            "width": "weighted",
                            "elements": [
                                {
                                    "tag": "markdown",
                                    "content": f"**工作流**: {pipeline_id}",
                                    "text_align": "left",
                                    "text_size": "normal",
                                    "icon": {
                                        "tag": "standard_icon",
                                        "token": "card_outlined",
                                        "color": "grey"
                                    }
                                },
                                {
                                    "tag": "markdown",
                                    "content": f"**工作流实例**: {pipeline_instance_name}(ID: {pipeline_instance_id})",
                                    "text_align": "left",
                                    "text_size": "normal",
                                    "icon": {
                                        "tag": "standard_icon",
                                        "token": "card_outlined",
                                        "color": "grey"
                                    }
                                },
                                {
                                    "tag": "markdown",
                                    "content": f"**是否重试**: {'是' if parent_pipeline_instance_id is not None else '否'}",
                                    "text_align": "left",
                                    "text_size": "normal",
                                    "icon": {
                                        "tag": "standard_icon",
                                        "token": "card_outlined",
                                        "color": "grey"
                                    }
                                },
                                {
                                    "tag": "markdown",
                                    "content": f"**创建人：**{creator}",
                                    "text_align": "left",
                                    "text_size": "normal",
                                    "icon": {
                                        "tag": "standard_icon",
                                        "token": "member_outlined",
                                        "color": "grey"
                                    }
                                },
                                {
                                    "tag": "markdown",
                                    "content": f"**操作人：**{operator}",
                                    "text_align": "left",
                                    "text_size": "normal",
                                    "icon": {
                                        "tag": "standard_icon",
                                        "token": "member_outlined",
                                        "color": "grey"
                                    }
                                },
                                {
                                    "tag": "markdown",
                                    "content": f"**事件时间：** {event_time}",
                                    "text_align": "left",
                                    "text_size": "normal",
                                    "icon": {
                                        "tag": "standard_icon",
                                        "token": "time_outlined",
                                        "color": "grey"
                                    }
                                }
                            ],
                            "vertical_align": "top",
                            "vertical_spacing": "8px",
                            "weight": 2
                        }
                    ]
                },
                {
                    "tag": "hr"
                },
                {
                    "tag": "action",
                    "actions": [
                        {
                            "tag": "button",
                            "text": {
                                "tag": "plain_text",
                                "content": "查看详情"
                            },
                            "url": f"https://console.volcengine.com/{event_service.replace('_', '-')}/region:{event_service.replace('_', '-')}+{region}/pipeline/detail/{pipeline_id}?activeBar=operationRecord&selectedId={pipeline_instance_id}",
                            "type": "default",
                            "width": "default",
                            "size": "medium",
                            "icon": {
                                "tag": "standard_icon",
                                "token": "details_outlined"
                            }
                        }
                    ]

                }
            ]
        },
        "i18n_header": {
            "zh_cn": {
                "title": {
                    "tag": "plain_text",
                    "content": f"【火山引擎-机器学习平台】【{region}】 工作流实例 {pipeline_instance_id} {status_hint.get(event_type)}"
                },
                "subtitle": {
                    "tag": "plain_text",
                    "content": ""
                },
                "template": f"{event_template}"
            }
        }
    }

def get_pipeline_email_content(
    region: str,
    user_name: str,
    user_id : str,
    account_id: str,
    event_service: str,
    pipeline_id: str,
    pipeline_name: str,
    message_event: str,
):
    return f"""<html>
<body style="font-family: '微软雅黑', Arial, sans-serif; line-height: 1.6;">
    <p>尊敬的{user_name} {user_id}（主账号ID：{account_id}）：</p>
    <p style="text-indent: 2em;">您好！您的工作流 <a href="https://console.volcengine.com/{event_service.replace('_', '-')}/region:{event_service.replace('_', '-')}+{region}/pipeline/detail/{pipeline_id}" target="_blank">{pipeline_name}(ID:{pipeline_id})</a> {message_event}。您可通过链接进入工作流详情页。</p>
    <p style="text-indent: 2em;">感谢您对火山引擎机器学习平台的支持！</p>
</body>
</html>
"""

def get_pipeline_instance_email_content(
    region: str,
    user_name: str,
    user_id: str,
    account_id: str,
    event_service: str,
    pipeline_id: str,
    pipeline_instance_id: str,
    pipeline_instance_name: str,
    message_event: str,
):
    return f"""<html>
<body style="font-family: '微软雅黑', Arial, sans-serif; line-height: 1.6;">
    <p>尊敬的{user_name} {user_id}（主账号ID：{account_id}）：</p>
    <p style="text-indent: 2em;">您好！您的工作流实例 <a href="https://console.volcengine.com/{event_service.replace('_', '-')}/region:{event_service.replace('_', '-')}+{region}/pipeline/detail/{pipeline_id}?activeBar=operationRecord&selectedId={pipeline_instance_id}" target="_blank">{pipeline_instance_name}(ID:{pipeline_instance_id})</a> {message_event}。您可通过链接进入工作流详情页。</p>
    <p style="text-indent: 2em;">感谢您对火山引擎机器学习平台的支持！</p>
</body>
</html>
"""

def merge_email_recipients(operator_email, creator_email):
    recipients = set()
    if operator_email and is_email(operator_email):
        recipients.add(operator_email)
    if creator_email and is_email(creator_email):
        recipients.add(creator_email)
    return recipients

class PipelineEvent:
    def __init__(self, region, webhook, email_sender, email_auth_code, sms_phone, sms_tpl_id, sms_account, sms_sign):
        self.region = region
        self.email_sender = email_sender
        self.email_auth_code = email_auth_code

        configuration = volcenginesdkcore.Configuration()
        configuration.ak = AK
        configuration.sk = SK
        configuration.region = region
        configuration.client_side_validation = True
        volcenginesdkcore.Configuration.set_default(configuration)
        self.iam_client = volcenginesdkiam.IAMApi()

    def get_user_email(self, user_id: str):
        if user_id is None:
            return user_id
        elif user_id == 0:
            return "主账号"
        try:
            resp = self.iam_client.get_user(
                volcenginesdkiam.GetUserRequest(
                    id=user_id,
                ))
            email = _get_user_email(resp)
            return email if email else user_id
        except ApiException as e:
            print("Exception when calling IAMApi->get_user: %s
" % e)
            return user_id

    def get_user_name(self, user_id: str):
        if user_id is None:
            return user_id
        elif user_id == 0:
            return "主账号"
        try:
            resp = self.iam_client.get_user(
                volcenginesdkiam.GetUserRequest(
                    id=user_id,
                ))
            return resp.user.user_name
        except ApiException as e:
            print("Exception when calling IAMApi->get_user: %s
" % e)
            return user_id

    def send_email_via_qq_smtp(
            self,
            sender: str,
            auth_code: str,
            receivers: str,
            subject: str,
            content: str,
            attachment_path: str = None,
            is_html: bool = False
    ):
        """
        使用 QQ 邮箱 SMTP 发送邮件
        :param sender: 发件人邮箱（需已开启 SMTP）
        :param auth_code: SMTP 授权码（非邮箱登录密码）
        :param receivers: 收件人邮箱（多个用逗号分隔，如 "a@qq.com,b@qq.com"）
        :param subject: 邮件主题
        :param content: 邮件正文（纯文本或 HTML）
        :param attachment_path: 附件文件路径（可选）
        :param is_html: 是否以 HTML 格式发送正文
        """
        if receivers == '':
            print("No valid email recipients found.")
            return

        # 创建邮件对象
        msg = MIMEMultipart()
        msg["From"] = sender
        msg["To"] = receivers
        msg["Subject"] = subject

        # 添加正文（纯文本或 HTML）
        content_type = "html" if is_html else "plain"
        msg.attach(MIMEText(content, content_type, "utf-8"))

        # 添加附件（如果存在）
        if attachment_path:
            with open(attachment_path, "rb") as f:
                attachment = MIMEApplication(f.read())
                attachment.add_header(
                    "Content-Disposition",
                    "attachment",
                    filename=attachment_path.split("/")[-1]  # 提取文件名
                )
                msg.attach(attachment)
        try:
            # 连接 SMTP 服务器（SSL 加密）
            with smtplib.SMTP_SSL("smtp.qq.com", 465) as server:
                server.login(sender, auth_code)
                server.sendmail(sender, receivers.split(","), msg.as_string())
                server.quit()
                print("邮件发送成功！")
        except Exception as e:
            print(f"邮件发送失败: {str(e)}")


    def handle(self, msg: dict):
        resource_id = msg.get('EventResourceID') # pipeline id or pipeline instance id
        event_type = msg.get('EventType')
        event_time = msg.get('EventTime')
        event_service = msg.get('EventService')
        account_id = msg.get('AccountID')
        identity_type = msg.get('IdentityType')
        user_id = msg.get('UserID')
        resource_type = msg.get('EventResourceType')

        operator = "平台"
        operator_email = None
        user_name = None
        if msg.get('EventTigerByUser'):
            operator_email = self.get_user_email(msg.get('UserID'))
            operator = f'<at email="{operator_email}"></at>' if is_email(operator_email) else operator_email
            user_name = self.get_user_name(msg.get('UserID'))

        extra_info = json.loads(msg.get('Detail').get('Message', {}))
        creator_email = self.get_user_email(extra_info.get('OwnerId'))
        creator = f'<at email="{creator_email}"></at>' if is_email(creator_email) else creator_email
        if user_name is None:
            user_name = self.get_user_name(extra_info.get('OwnerId'))

        recipients = merge_email_recipients(operator_email, creator_email)

        card = {}
        email_config = {}
        params = {}
        if resource_type == 'Pipeline':
            pipeline_name = extra_info.get('PipelineName')

            email_config = {
                "sender": self.email_sender,
                "auth_code": self.email_auth_code,
                "receivers": ','.join(recipients),
                "subject": f"【火山引擎-机器学习平台】【{self.region}】【{identity_type}】 工作流 {resource_id} 于{event_time}进入{status_hint.get(event_type)}",
                "content": get_pipeline_email_content(
                    self.region,
                    user_name,
                    user_id,
                    account_id,
                    event_service,
                    resource_id,
                    pipeline_name,
                    status_hint[event_type]),
                "is_html": True
            }

            params = {
                "UserName": user_name,
                "AccountID": account_id,
                "ResourcName": pipeline_name,
                "ResourceId": resource_id,
                "EventName": status_hint[event_type],
            }

        elif resource_type == 'PipelineInstance':
            pipeline_instance_name = extra_info.get('PipelineInstanceName')
            pipeline_id = extra_info.get('PipelineId')
            # trigger = extra_info.get("Trigger") # Manual
            # diag_info = extra_info.get("DiagInfo")
            # reason = extra_info.get("Reason")
            parent_pipeline_instance_id = extra_info.get("ParentPipelineInstanceId", None)

            email_config = {
                "sender": self.email_sender,
                "auth_code": self.email_auth_code,
                "receivers": ','.join(recipients),
                "subject": f"【火山引擎-机器学习平台】【{self.region}】 工作流实例 {resource_id} 于{event_time}进入 {status_hint.get(event_type)}",
                "content": get_pipeline_instance_email_content(
                    self.region,
                    user_name,
                    user_id,
                    account_id,
                    event_service,
                    pipeline_id,
                    resource_id,
                    pipeline_instance_name,
                    status_hint[event_type]),
                "is_html": True
            }

            params = {
                "UserName": user_name,
                "AccountID": account_id,
                "ResourcName": pipeline_instance_name,
                "ResourceId": resource_id,
                "EventName": status_hint[event_type],
            }

        else:
            return

        # self.send_webhook(card)
        self.send_email_via_qq_smtp(**email_config)
        # self.send_sms(params)

pipeline_event = PipelineEvent(
    region=Region,
    email_sender=EmailSender,
    email_auth_code=EmailAuthCode ,
)

def handler(event, context):
    print(f"received new request, event content: {event}")

    records = event.get('data', {}).get('Records', [])
    for record in records:
        print(f"get record: {record}")
        message = json.loads(record.get('Sns', '{}').get('Message', '{}'))
        print(f"get message: {message}")
        pipeline_event.handle(message)

    result = {
        'statusCode': 200,
        'headers': {
            'Content-Type': 'application/json'
        }
    }
    return result