forked from erp-dev/erp
82 lines
2.4 KiB
Python
82 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
from dataclasses import dataclass
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.request import Request, urlopen
|
|
|
|
|
|
JPUSH_PUSH_URL = "https://api.jpush.cn/v3/push"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class JPushResponse:
|
|
status_code: int
|
|
raw: dict
|
|
|
|
@property
|
|
def ok(self) -> bool:
|
|
return 200 <= int(self.status_code) < 300 and "error" not in self.raw
|
|
|
|
|
|
def _build_basic_auth_header(*, app_key: str, master_secret: str) -> str:
|
|
token = f"{app_key}:{master_secret}".encode("utf-8")
|
|
return "Basic " + base64.b64encode(token).decode("ascii")
|
|
|
|
|
|
def send_jpush_payload(
|
|
*,
|
|
app_key: str,
|
|
master_secret: str,
|
|
payload: dict,
|
|
timeout_seconds: float = 10.0,
|
|
) -> JPushResponse:
|
|
app_key = str(app_key or "").strip()
|
|
master_secret = str(master_secret or "").strip()
|
|
if not app_key:
|
|
raise ValueError("JPush app_key 未配置")
|
|
if not master_secret:
|
|
raise ValueError("JPush master_secret 未配置")
|
|
if not isinstance(payload, dict) or not payload:
|
|
raise ValueError("JPush payload 不能为空")
|
|
|
|
data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
|
req = Request(
|
|
url=JPUSH_PUSH_URL,
|
|
data=data,
|
|
method="POST",
|
|
headers={
|
|
"Authorization": _build_basic_auth_header(
|
|
app_key=app_key,
|
|
master_secret=master_secret,
|
|
),
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json",
|
|
},
|
|
)
|
|
try:
|
|
with urlopen(req, timeout=float(timeout_seconds)) as resp:
|
|
body = resp.read().decode("utf-8", errors="replace")
|
|
status_code = int(getattr(resp, "status", 200) or 200)
|
|
except HTTPError as exc:
|
|
body = ""
|
|
try:
|
|
body = exc.read().decode("utf-8", errors="replace")
|
|
except Exception:
|
|
pass
|
|
try:
|
|
raw = json.loads(body) if body else {}
|
|
except Exception:
|
|
raw = {"error": {"message": body}}
|
|
raise RuntimeError(f"JPush HTTPError: status={exc.code}, body={raw}") from exc
|
|
except URLError as exc:
|
|
raise RuntimeError(f"JPush URLError: {exc}") from exc
|
|
|
|
try:
|
|
raw = json.loads(body) if body else {}
|
|
except Exception as exc:
|
|
raise RuntimeError(f"JPush 响应不是合法 JSON: {body}") from exc
|
|
|
|
return JPushResponse(status_code=status_code, raw=raw)
|