Pyhton检测邮箱是否可用

🏦 依赖

• Python3

pip install httpx smtplib datetime

✅ 计划任务

*/5 * * * * cd PY脚本目录 && PY解释器路径 check_smtp_active.py

✅ 日志输出

• 自动生成在 PY脚本目录 logs 下

# ========= 文件信息 =========
# Author: 000
# Email: [e-mail]
# Date: 2025-04-19 16:00:00
# LastEditTime: 2025-04-19 16:00:00
# Description: 检测邮箱发送是否正常
# =========================


# Python3: pip install httpx smtplib datetime


import os,logging,asyncio,httpx,ssl,smtplib
from email.message import EmailMessage
from typing import List, Tuple, Union
from logging.handlers import RotatingFileHandler
from datetime import datetime

# 日志设置
LOG_DIR = "logs"
SCREENSHOT_DIR = os.path.join(LOG_DIR, "screenshots")
REPORT_DIR = os.path.join(LOG_DIR, "reports")
os.makedirs(LOG_DIR, exist_ok=True)
os.makedirs(SCREENSHOT_DIR, exist_ok=True)
os.makedirs(REPORT_DIR, exist_ok=True)

log_filename = os.path.join(LOG_DIR, f"check_stmp_active_{datetime.now().strftime('%Y-%m-%d')}.log")
log_handler = RotatingFileHandler(log_filename, maxBytes=10*1024*1024, backupCount=5, encoding="utf-8")
log_formatter = logging.Formatter("[%(asctime)s] [%(levelname)s] %(message)s", datefmt='%Y-%m-%d %H:%M:%S')
log_handler.setFormatter(log_formatter)

logger = logging.getLogger()
logger.setLevel(logging.INFO)
logger.addHandler(log_handler)

console_handler = logging.StreamHandler()
console_handler.setFormatter(log_formatter)
logger.addHandler(console_handler)




class SMTPMonitor:
    def __init__(self, name, smtp_server, smtp_port, username, password, sender, recipient):
        self.name = name
        self.smtp_server = smtp_server
        self.smtp_port = smtp_port
        self.username = username
        self.password = password
        self.sender = sender
        self.recipient = recipient

    def send_test_email(self) -> Union[bool, Tuple[bool, str]]:
        msg = EmailMessage()
        msg.set_content(f"This is a test email from {self.name}.")
        msg["Subject"] = f"SMTP Monitor Test from {self.name}"
        msg["From"] = self.sender
        msg["To"] = self.recipient

        try:
            context = ssl.create_default_context()
            with smtplib.SMTP_SSL(self.smtp_server, self.smtp_port, context=context) as server:
                server.login(self.username, self.password)
                server.send_message(msg)
            logger.info(f"✅ [{self.name}] SMTP test email sent successfully.")
            return True
        except Exception as e:
            logger.error(f"❌ [{self.name}] Failed to send SMTP email: {e}")
            return False, str(e)

    def get_context_info(self) -> str:
        """返回错误上下文信息，用于 Telegram 报警"""
        return (
            f"📌 名称: {self.name}\n"
            f"📮 SMTP: {self.smtp_server}:{self.smtp_port}\n"
            f"📤 发件人: {self.sender}\n"
            f"📥 收件人: {self.recipient}"
        )


class TelegramNotifier:
    def __init__(self, bot_token: str, chat_id: str):
        self.bot_token = bot_token
        self.chat_id = chat_id
        self.api_url = f"https://api.telegram.org/bot{bot_token}/sendMessage"

    async def send_alert(self, context_info: str, error_msg: str):
        text = (
            "🚨 *SMTP 邮局发送失败*\n"
            f"{context_info}\n"
            f"❌ 错误: `{error_msg}`"
        )
        async with httpx.AsyncClient() as client:
            try:
                response = await client.post(self.api_url, json={
                    "chat_id": self.chat_id,
                    "text": text,
                    "parse_mode": "Markdown"
                })
                logger.info(f"📨 Telegram alert sent: {response.status_code}")
            except Exception as e:
                logger.error(f"❌ Failed to send Telegram alert: {e}")


class MonitorRunner:
    def __init__(self, smtp_monitors: List[SMTPMonitor], notifier: TelegramNotifier):
        self.smtp_monitors = smtp_monitors
        self.notifier = notifier

    async def run_all(self):
        tasks = []
        for monitor in self.smtp_monitors:
            result = monitor.send_test_email()
            if result is not True:
                _, error_msg = result if isinstance(result, tuple) else (False, "Unknown error")
                context = monitor.get_context_info()
                tasks.append(self.notifier.send_alert(context, error_msg))
        if tasks:
            await asyncio.gather(*tasks)

if __name__ == "__main__":
    # === 多个邮局配置 ===
    smtp_monitors = [
        SMTPMonitor(
            name="项目-AAAA",
            smtp_server="mail.aaaa.com",
            smtp_port=465,
            username="[e-mail]",
            password="123456",
            sender="[e-mail]",
            recipient="[e-mail]"
        ),
        SMTPMonitor(
            name="项目-BBBB",
            smtp_server="mail.bbbb.vip",
            smtp_port=465,
            username="[e-mail]",
            password="123456",
            sender="[e-mail]",
            recipient="[e-mail]"
        ),
    ]

    # === Telegram 配置 ===
    telegram_notifier = TelegramNotifier(
        bot_token="000000000000:000000000000000",
        chat_id="-00000000000"
    )

    # === 执行所有检查 ===
    runner = MonitorRunner(smtp_monitors, telegram_notifier)
    asyncio.run(runner.run_all())