qinglong/notice.py

import requests
from pydantic import BaseModel, Field
from enum import Enum


class MessageType(Enum):
    DEFAULT = 'DEFAULT',
    TEXT = 'TEXT',
    MARKDOWN = 'MARKDOWN',
    EMAIL = 'EMAIL'


class Email(BaseModel):
    sender: str = Field(..., description='发件人')
    recipient: str = Field(..., description='收件人')
    subject: str = Field(..., description='主题')
    sendTime: str = Field(..., description='发送时间')
    content: str = Field(..., description='内容')


class PushMessage(BaseModel):
    title: str = Field(..., description='标题')
    content: str = Field(..., description='通知内容')
    messageType: MessageType = Field(..., description='消息类型')
    text: str = Field(None, description='文本内容')
    email: Email = Field(None, description='邮件内容')

    def model_dump(self):
        # 重写model_dump方法以正确处理枚举类型
        data = super().model_dump()
        data['messageType'] = self.messageType.value
        return data


def from_email(email: dict):
    sender_display = email["sender"].replace(' ', '\n').replace('"', '')
    recipient_display = email["recipient"].replace(' ', '\n')

    return PushMessage(
        title="新邮件通知",
        content=f'您有一封来自[{email["sender_name"] if email["sender_name"] else email["sender_email"]}]的邮件',
        messageType=MessageType.EMAIL,
        email=Email(
            sender=sender_display,
            recipient=recipient_display,
            subject=email['subject'],
            sendTime=email['date'],
            content=email['content'],
        )
    )


class Push:
    def __init__(self, config: dict):
        self.config = config

    def send_notification(self, data: PushMessage):
        pass


class AiYuFeiFei(Push):
    def __init__(self, config: dict):
        super().__init__(config)
        self.url = f'https://iyuu.cn/{config["token"]}.send'

    def send_notification(self, data: PushMessage):
        data = {
            'text': data.title,
            'desp': data.content
        }
        response = requests.post(self.url, json=data)
        print("爱语飞飞提醒发送结果:", response.json())


class Yo(Push):

    def __init__(self, config: dict):
        super().__init__(config)

    def send_notification(self, data: PushMessage):
        headers = {
            'x-api-key': self.config['token']
        }

        response = requests.post(self.config['url'], json=data.model_dump(), headers=headers)
        print("Yo 提醒发送结果:", response.json())


class WeComWebhook(Push):
    """
    企业微信 Webhook 推送
    """
    def __init__(self, config: dict):
        super().__init__(config)
        self.webhook_url = config.get('webhook_url', '')

    def send_notification(self, data: PushMessage):
        """
        发送企业微信 webhook 通知
        支持发送者、接收者、发送时间、主题等信息
        """
        if data.messageType == MessageType.EMAIL and data.email:
            # 使用文本格式发送，兼容性更好
            content = f"{data.title}\n\n" \
                      f"发件人：{data.email.sender}\n" \
                      f"收件人：{data.email.recipient}\n" \
                      f"发送时间：{data.email.sendTime}\n" \
                      f"主题：{data.email.subject}"
        else:
            content = f"{data.title}\n\n{data.content}"

        payload = {
            "msgtype": "text",
            "text": {
                "content": content
            }
        }

        try:
            response = requests.post(self.webhook_url, json=payload)
            result = response.json()
            print(f"企业微信 Webhook 推送结果：{result}")
            return result
        except Exception as e:
            print(f"企业微信 Webhook 推送失败：{e}")
            raise


class PushFactory:
    @staticmethod
    def create_push(config: dict) -> Push:
        push_type = config.get('type', '')
        if push_type == 'WeComWebhook':
            return WeComWebhook(config)
        elif push_type == 'Yo':
            return Yo(config)
        elif push_type == 'AiYuFeiFei':
            return AiYuFeiFei(config)
        else:
            raise ValueError(f'不支持的推送类型：{push_type}')


if __name__ == '__main__':
    # 测试企业微信 Webhook 推送
    webhook_config = {
        'type': 'WeComWebhook',
        'webhook_url': 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=d5b3282e-7531-462f-9b48-e117fb2797f6'  # 替换为你的企业微信 webhook 地址
    }
    
    # 创建测试邮件消息
    test_email = PushMessage(
        title="新邮件通知",
        content="您有一封来自 [test@example.com] 的邮件",
        messageType=MessageType.EMAIL,
        email=Email(
            sender="张三 <zhangsan@example.com>",
            recipient="李四 <lisi@example.com>",
            subject="测试邮件主题",
            sendTime="2026-03-25 10:30:00",
            content="这是一封测试邮件的内容...\n\n请查收！"
        )
    )
    
    # 测试推送
    try:
        push = PushFactory.create_push(webhook_config)
        print("开始测试企业微信 Webhook 推送...")
        result = push.send_notification(test_email)
        print(f"测试完成！返回结果：{result}")
    except Exception as e:
        print(f"测试失败：{e}")