73 lines
2.6 KiB
Python
73 lines
2.6 KiB
Python
"""通知服务：Telegram / 邮件多渠道发送
|
|
|
|
根据 .env 配置的渠道发送通知。未配置的渠道自动跳过。
|
|
"""
|
|
|
|
import smtplib
|
|
from email.mime.text import MIMEText
|
|
|
|
import httpx
|
|
|
|
from app.core.config import settings
|
|
|
|
|
|
def send_telegram(message: str) -> dict:
|
|
"""通过 Telegram Bot 发送消息"""
|
|
if not settings.TELEGRAM_BOT_TOKEN or not settings.TELEGRAM_CHAT_ID:
|
|
return {"ok": False, "channel": "telegram", "message": "未配置 Telegram"}
|
|
url = f"https://api.telegram.org/bot{settings.TELEGRAM_BOT_TOKEN}/sendMessage"
|
|
try:
|
|
resp = httpx.post(
|
|
url,
|
|
json={
|
|
"chat_id": settings.TELEGRAM_CHAT_ID,
|
|
"text": message,
|
|
"parse_mode": "HTML",
|
|
"disable_web_page_preview": True,
|
|
},
|
|
timeout=15,
|
|
)
|
|
resp.raise_for_status()
|
|
return {"ok": True, "channel": "telegram", "message": "发送成功"}
|
|
except Exception as e: # noqa: BLE001
|
|
return {"ok": False, "channel": "telegram", "message": str(e)}
|
|
|
|
|
|
def send_email(message: str, subject: str = "资产续费提醒") -> dict:
|
|
"""通过 SMTP 发送邮件"""
|
|
if not settings.SMTP_HOST or not settings.SMTP_USER:
|
|
return {"ok": False, "channel": "email", "message": "未配置邮件"}
|
|
try:
|
|
msg = MIMEText(message, "plain", "utf-8")
|
|
msg["Subject"] = subject
|
|
msg["From"] = settings.SMTP_FROM or settings.SMTP_USER
|
|
msg["To"] = settings.SMTP_TO or settings.SMTP_USER
|
|
with smtplib.SMTP_SSL(settings.SMTP_HOST, settings.SMTP_PORT, timeout=15) as server:
|
|
server.login(settings.SMTP_USER, settings.SMTP_PASSWORD)
|
|
server.send_message(msg)
|
|
return {"ok": True, "channel": "email", "message": "发送成功"}
|
|
except Exception as e: # noqa: BLE001
|
|
return {"ok": False, "channel": "email", "message": str(e)}
|
|
|
|
|
|
def notify(message: str, subject: str = "资产续费提醒") -> list:
|
|
"""发送到所有已配置的渠道，返回各渠道结果"""
|
|
results = []
|
|
if settings.TELEGRAM_BOT_TOKEN:
|
|
results.append(send_telegram(message))
|
|
if settings.SMTP_HOST:
|
|
results.append(send_email(message, subject))
|
|
if not results:
|
|
results.append({"ok": False, "channel": "none", "message": "未配置任何通知渠道"})
|
|
return results
|
|
|
|
|
|
def configured_channels() -> list:
|
|
"""返回已配置的通知渠道列表"""
|
|
channels = []
|
|
if settings.TELEGRAM_BOT_TOKEN and settings.TELEGRAM_CHAT_ID:
|
|
channels.append("telegram")
|
|
if settings.SMTP_HOST and settings.SMTP_USER:
|
|
channels.append("email")
|
|
return channels
|