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')

    print("邮件内容:", email)

    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 PushFactory:
    @staticmethod
    def create_push(config: dict) -> Push:
        if config['type'] == 'Yo':
            return Yo(config)
        elif config['type'] == 'AiYuFeiFei':
            return AiYuFeiFei(config)
        else:
            raise ValueError(f'不支持的推送类型: {config.type}')