study-tool
112 lines
3.9 KiB
Python
112 lines
3.9 KiB
Python
"""呆呆面板推送：换 token（缓存到过期）+ 发通知。
|
||
|
||
接口
|
||
POST {base_url}/api/open-api/token {app_key, app_secret} -> token
|
||
POST {base_url}/api/notifications/send Authorization: Bearer <token>
|
||
|
||
配置全部来自系统设置 `notify`（base_url / app_key / app_secret / channel_name），
|
||
本文件不写死任何地址或密钥。换 token 只在内存里缓存，重启后自动重取。
|
||
"""
|
||
|
||
import threading
|
||
import time
|
||
|
||
import requests
|
||
|
||
from study_tool import notify
|
||
from study_tool.config.settings import load_settings
|
||
|
||
TOKEN_PATH = "/api/open-api/token"
|
||
SEND_PATH = "/api/notifications/send"
|
||
TOKEN_TTL = 3600 # 接口没给有效期时的兜底
|
||
|
||
_CACHE = {} # (base_url, app_key) -> (token, 过期时间)
|
||
_LOCK = threading.Lock()
|
||
|
||
|
||
def _base(cfg):
|
||
return (cfg.get("base_url") or "https://corn.hanhan.ltd").rstrip("/")
|
||
|
||
|
||
def _find(obj, names, depth=0):
|
||
"""在返回体里找字段：顶层找不到就往 data / result 里找一层。"""
|
||
if not isinstance(obj, dict) or depth > 3:
|
||
return ""
|
||
for n in names:
|
||
v = obj.get(n)
|
||
if isinstance(v, (str, int, float)) and str(v).strip():
|
||
return v
|
||
for n in ("data", "result", "content", "body"):
|
||
v = _find(obj.get(n), names, depth + 1)
|
||
if v != "":
|
||
return v
|
||
return ""
|
||
|
||
|
||
def _json(r, what):
|
||
try:
|
||
return r.json()
|
||
except Exception:
|
||
raise RuntimeError("%s 返回非 JSON（HTTP %s）：%s" % (what, r.status_code, (r.text or "")[:120]))
|
||
|
||
|
||
class DaidaiPush(notify.Channel):
|
||
id = "daidai_push"
|
||
label = "呆呆面板推送"
|
||
|
||
def _token(self, cfg, force=False):
|
||
ck = (_base(cfg), cfg.get("app_key"))
|
||
with _LOCK:
|
||
hit = _CACHE.get(ck)
|
||
if hit and not force and hit[1] > time.time() + 60:
|
||
return hit[0]
|
||
r = requests.post(
|
||
ck[0] + TOKEN_PATH,
|
||
json={"app_key": cfg["app_key"], "app_secret": cfg["app_secret"]},
|
||
timeout=20,
|
||
)
|
||
j = _json(r, "换 token")
|
||
token = str(_find(j, ("token", "access_token", "accessToken")) or "")
|
||
if not token:
|
||
raise RuntimeError("换 token 失败：%s" % str(j)[:200])
|
||
try:
|
||
ttl = int(float(_find(j, ("expires_in", "expire_in", "expiresIn")) or 0))
|
||
except (TypeError, ValueError):
|
||
ttl = 0
|
||
with _LOCK:
|
||
_CACHE[ck] = (token, time.time() + (ttl or TOKEN_TTL))
|
||
return token
|
||
|
||
def send(self, title, content, context=None):
|
||
cfg = load_settings()["notify"]
|
||
if not cfg.get("app_key") or not cfg.get("app_secret"):
|
||
raise RuntimeError("推送未配置：系统设置 → 推送通知 里填 app_key / app_secret")
|
||
base = _base(cfg)
|
||
payload = {
|
||
"title": title,
|
||
"content": content,
|
||
"content_type": "html",
|
||
"context": {str(k): v for k, v in (context or {}).items()},
|
||
}
|
||
if cfg.get("channel_name"):
|
||
payload["channel_name"] = cfg["channel_name"]
|
||
for attempt in (0, 1):
|
||
token = self._token(cfg, force=attempt == 1)
|
||
r = requests.post(
|
||
base + SEND_PATH,
|
||
headers={"Authorization": "Bearer " + token},
|
||
json=payload,
|
||
timeout=20,
|
||
)
|
||
if r.status_code in (401, 403) and attempt == 0:
|
||
continue # token 过期：丢掉缓存换一张再发一次
|
||
j = _json(r, "推送")
|
||
ok = str(_find(j, ("code",)) or 200) in ("0", "200") or _find(j, ("success", "ok")) in (True, "true")
|
||
if r.status_code >= 400 or not ok:
|
||
raise RuntimeError("推送失败：HTTP %s %s" % (r.status_code, str(j)[:200]))
|
||
return j
|
||
raise RuntimeError("推送失败：token 被拒（换新 token 后仍然 401/403）")
|
||
|
||
|
||
NOTIFY = notify.register(DaidaiPush())
|