85 lines
2.8 KiB
Python
85 lines
2.8 KiB
Python
"""推送通知：邮件(SMTP) + Server酱 + 企业微信机器人 + PushPlus。
|
||
|
||
任意渠道配置了凭证即启用；notify() 会向所有已配置渠道发送，返回各渠道结果。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import smtplib
|
||
import ssl
|
||
from email.mime.text import MIMEText
|
||
from email.header import Header
|
||
|
||
import requests
|
||
|
||
import config
|
||
|
||
|
||
def channels_status():
|
||
return {
|
||
"email": bool(config.SMTP_HOST and config.SMTP_USER and config.SMTP_TO),
|
||
"serverchan": bool(config.SERVERCHAN_KEY),
|
||
"wecom": bool(config.WECOM_WEBHOOK),
|
||
"pushplus": bool(config.PUSHPLUS_TOKEN),
|
||
}
|
||
|
||
|
||
def any_enabled():
|
||
return any(channels_status().values())
|
||
|
||
|
||
def _send_email(title, content):
|
||
msg = MIMEText(content, "plain", "utf-8")
|
||
msg["Subject"] = Header(title, "utf-8")
|
||
msg["From"] = config.SMTP_USER
|
||
tos = [x.strip() for x in config.SMTP_TO.split(",") if x.strip()]
|
||
msg["To"] = ",".join(tos)
|
||
ctx = ssl.create_default_context()
|
||
if config.SMTP_PORT == 465:
|
||
with smtplib.SMTP_SSL(config.SMTP_HOST, config.SMTP_PORT, context=ctx, timeout=15) as srv:
|
||
srv.login(config.SMTP_USER, config.SMTP_PASSWORD)
|
||
srv.sendmail(config.SMTP_USER, tos, msg.as_string())
|
||
else:
|
||
with smtplib.SMTP(config.SMTP_HOST, config.SMTP_PORT, timeout=15) as srv:
|
||
srv.starttls(context=ctx)
|
||
srv.login(config.SMTP_USER, config.SMTP_PASSWORD)
|
||
srv.sendmail(config.SMTP_USER, tos, msg.as_string())
|
||
return True
|
||
|
||
|
||
def _send_serverchan(title, content):
|
||
url = f"https://sctapi.ftqq.com/{config.SERVERCHAN_KEY}.send"
|
||
r = requests.post(url, data={"title": title, "desp": content}, timeout=12)
|
||
r.raise_for_status()
|
||
return r.json().get("code", 0) == 0
|
||
|
||
|
||
def _send_wecom(title, content):
|
||
r = requests.post(config.WECOM_WEBHOOK,
|
||
json={"msgtype": "text", "text": {"content": f"{title}\n{content}"}}, timeout=12)
|
||
r.raise_for_status()
|
||
return r.json().get("errcode", -1) == 0
|
||
|
||
|
||
def _send_pushplus(title, content):
|
||
r = requests.post("https://www.pushplus.plus/send",
|
||
json={"token": config.PUSHPLUS_TOKEN, "title": title, "content": content}, timeout=12)
|
||
r.raise_for_status()
|
||
return r.json().get("code", 0) == 200
|
||
|
||
|
||
def notify(title, content):
|
||
"""向所有已配置渠道推送，返回 {渠道: 'ok'|错误信息}。"""
|
||
st = channels_status()
|
||
senders = {"email": _send_email, "serverchan": _send_serverchan,
|
||
"wecom": _send_wecom, "pushplus": _send_pushplus}
|
||
result = {}
|
||
for ch, on in st.items():
|
||
if not on:
|
||
continue
|
||
try:
|
||
senders[ch](title, content)
|
||
result[ch] = "ok"
|
||
except Exception as e:
|
||
result[ch] = repr(e)[:120]
|
||
return result
|