notify.py

#!/usr/bin/env python3
# _*_ coding:utf-8 _*_
import json
import os
import threading
import smtplib
from email.mime.text import MIMEText
from email.header import Header
from email.utils import formataddr
from dotenv import load_dotenv
import requests

load_dotenv()
# 原先的 print 函数和主线程的锁
_print = print
mutex = threading.Lock()


# 定义新的 print 函数
def print(text, *args, **kw):
    """
    使输出有序进行，不出现多线程同一时间输出导致错乱的问题。
    """
    with mutex:
        _print(text, *args, **kw)


# 通知服务
# fmt: off
push_config = {
    'HITOKOTO': True,                  # 启用一言（随机句子）
    
    'SMTP_SERVER': '',                  # SMTP 发送邮件服务器，形如 smtp.exmail.qq.com:465
    'SMTP_SSL': 'false',                # SMTP 发送邮件服务器是否使用 SSL，填写 true 或 false
    'SMTP_EMAIL': '',                   # SMTP 收发件邮箱，通知将会由自己发给自己
    'SMTP_PASSWORD': '',                # SMTP 登录密码，也可能为特殊口令，视具体邮件服务商说明而定
    'SMTP_NAME': '',                    # SMTP 收发件人姓名，可随意填写
    
    'SENDKEY': '',                      # Server酱的 SENDKEY
    
    'WX_TOKEN': '',                     # wxpusher的 Token
    'UIDS':''
}
notify_function = []
# fmt: on

# 首先读取 面板变量 或者 github action 运行变量
for k in push_config:
    if os.getenv(k):
        v = os.getenv(k)
        push_config[k] = v


def smtp(title: str, content: str) -> None:
    """
    使用 SMTP 邮件 推送消息。
    """
    if (
        not push_config.get("SMTP_SERVER")
        or not push_config.get("SMTP_SSL")
        or not push_config.get("SMTP_EMAIL")
        or not push_config.get("SMTP_PASSWORD")
        or not push_config.get("SMTP_NAME")
    ):
        print(
            "SMTP 邮件 的 SMTP_SERVER 或者 SMTP_SSL 或者 SMTP_EMAIL 或者 SMTP_PASSWORD 或者 SMTP_NAME 未设置!!\n取消推送"
        )
        return
    print("SMTP 邮件 服务启动")

    message = MIMEText(content, "plain", "utf-8")
    message["From"] = formataddr(
        (
            Header(push_config.get("SMTP_NAME"), "utf-8").encode(),
            push_config.get("SMTP_EMAIL"),
        )
    )
    message["To"] = formataddr(
        (
            Header(push_config.get("SMTP_NAME"), "utf-8").encode(),
            push_config.get("SMTP_EMAIL"),
        )
    )
    message["Subject"] = Header(title, "utf-8")

    try:
        smtp_server = (
            smtplib.SMTP_SSL(push_config.get("SMTP_SERVER"))
            if push_config.get("SMTP_SSL") == "true"
            else smtplib.SMTP(push_config.get("SMTP_SERVER"))
        )
        smtp_server.login(
            push_config.get("SMTP_EMAIL"), push_config.get("SMTP_PASSWORD")
        )
        smtp_server.sendmail(
            push_config.get("SMTP_EMAIL"),
            push_config.get("SMTP_EMAIL"),
            message.as_bytes(),
        )
        smtp_server.close()
        print("SMTP 邮件 推送成功！")
    except Exception as e:
        print(f"SMTP 邮件 推送失败！{e}")


def server_chan(text, desp=''):
    """
    Server酱推送
    Args:
        text (_type_): _通知标题_
        desp (_type_): _通知内容_
    """
    if not push_config.get("SENDKEY"):
        print("Server酱 推送失败！未配置 Server酱 SENDKEY")
        return
    print("Server酱 推送中...(*≧︶≦))(￣▽￣* )ゞ")

    url = f'https://sctapi.ftqq.com/{push_config.get("SENDKEY")}.send'
    data = {'text': text, 'desp': desp, 'channel':'9|18'}
    response = requests.post(url, data=data)
    print(response.text)
    return response.text


def wx_pusher(title, content):
    """
    wx_pusher推送
    Args:
        wxpuser_token (_type_): _token_
        uids (_type_): _description_
        msg (_type_): _description_
    """
    if not push_config.get("WX_TOKEN") or not push_config.get("UIDS"):
        print("WX_TOKEN 或者 UIDS 未配置,请检查")
        return
    uids:list = str(push_config.get("UIDS")).split(';')
    url = "https://wxpusher.zjiecode.com/api/send/message"
    payload = json.dumps(
        {
            "appToken": f'{push_config.get("WX_TOKEN")}',
            "content": f"## 吾爱破解Task任务执行日志 \n{title}",  # Title
            "summary": f"{content}",  #
            "contentType": 3,
            "uids": uids,
            "verifyPay": False,
        }
    )
    headers = {
        "Content-Type": "application/json",
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/116.0.0.0 Safari/537.36",
    }
    while True:
        try:
            r = requests.request("POST", url, headers=headers, data=payload).json()
            # print(json.dumps(r))
            msg = r["msg"]
            status = r["data"][0]["status"]
            print(f"🎁 WxPusher {status} - {msg}")
            break
        except requests.exceptions.ConnectionError as e:
            print(f"ConnectionError:{e}")


def one() -> str:
    """
    获取一条一言。
    :return:
    """
    url = "https://v1.hitokoto.cn/"
    res = requests.get(url).json()
    return res["hitokoto"] + "    ----" + res["from"]


if (
    push_config.get("SMTP_SERVER")
    and push_config.get("SMTP_SSL")
    and push_config.get("SMTP_EMAIL")
    and push_config.get("SMTP_PASSWORD")
    and push_config.get("SMTP_NAME")
):
    notify_function.append(smtp)
if push_config.get("SENDKEY"):
    notify_function.append(server_chan)
if push_config.get("WX_TOKEN") and push_config.get("UIDS"):
    notify_function.append(wx_pusher)


def send(title: str, content: str) -> None:
    if not content:
        print(f"{title} 推送内容为空！")
        return

    hitokoto = bool(push_config.get("HITOKOTO"))

    text = one() if hitokoto else ""
    title += "\n" + text
    print(title)
    ts = [
        threading.Thread(target=mode, args=(title, content), name=mode.__name__)
        for mode in notify_function
    ]
    [t.start() for t in ts]
    [t.join() for t in ts]


def main():
    send("吾爱Task签到反馈", "吾爱Task签到反馈-BYS")


if __name__ == "__main__":
    main()
    # print(one())
    # print()