qinglong/check_new_email.py

import json

from mysql_db import DBManager
from email_client import EmailClient
from notice import PushFactory, PushMessage, MessageType, Email

EMAIL_CONFIG_SQL = '''
            select email_address,    -- 邮件地址
                email_password,         -- 邮件密码
                connection_type,        -- 连接类型
                receive_server_address, -- 接收邮件服务器地址
                receive_server_port,    -- 接收邮件服务器端口号
                push_config,             -- 爱语飞飞token
                last_check_uid          -- 上次检查的UID
            from check_email_config;
        '''

EMAIL_UPDATE_LAST_CHECK_UID_SQL = '''
            update check_email_config
                set last_check_uid = %s
            where email_address = %s;
        '''


def fetch_all_data(db_manager):
    """
    获取所有数据
    :param db_manager: 数据库管理器
    :return:
    """
    try:
        with db_manager.managed_cursor() as cursor:
            cursor.execute(EMAIL_CONFIG_SQL)
            records = cursor.fetchall()
            if not records:
                return []
            return [
                {
                    "email_account": record[0],
                    "email_password": record[1],
                    "connection_type": record[2],
                    "receive_server_address": record[3],
                    "receive_server_port": record[4],
                    "push_config": record[5],
                    "last_check_uid": record[6]
                }
                for record in records
            ]

    except Exception as e:
        print(f"查询邮箱配置失败: {e}")
        return []


def update_last_check_uid(db_manager, email_account, last_check_uid):
    """
    更新检查邮件的UID
    :param db_manager: 数据库管理器
    :param email_account: 邮箱账号
    :param last_check_uid: 最后检查的UID
    :return:
    """
    try:
        with db_manager.managed_cursor() as cursor:
            cursor.execute(EMAIL_UPDATE_LAST_CHECK_UID_SQL, (last_check_uid, email_account))
            db_manager.connection.commit()
            print(f"更新 {email_account} 的本次检测UID为: {last_check_uid}")
    except Exception as e:
        print(f"更新UID失败: {e}")
        db_manager.connection.rollback()


def check_new_email(email_config: dict):
    """
    检查新邮件
    :param email_config: 邮箱配置
    :return:
    """
    if 'push_config' not in email_config:
        print("没有配置推送服务")
        return

    push_config = json.loads(email_config['push_config'])

    email_account = email_config['email_account']
    print(f"正在检查 {email_account} 的邮件...")

    # 检查新邮件
    client = EmailClient(email_config)
    new_emails = client.check_new_email(email_config['last_check_uid'])
    print(f"新邮件数量: {len(new_emails)}")

    # 发送通知
    current_email = None
    try:
        for email in new_emails:
            current_email = email
            message = to_message(email)
            PushFactory.create_push(push_config).send_notification(message)

    except Exception as e:
        print(f"邮箱账户{email_account} 发送邮件[{current_email['subject']}]通知失败: {e}")

    # 更新最大UID
    if current_email and current_email['uid'] > email_config['last_check_uid']:
        update_last_check_uid(db_manager, email_account, current_email['uid'])

def to_message(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'],
        )
    )


if __name__ == '__main__':
    db_manager = DBManager()
    records = fetch_all_data(db_manager)
    if not records:
        print("没有查询到邮箱配置")
    else:
        for row in records:
            check_new_email(row)