/<body>`` 发送，适合快速测试。地址与密钥默认从环境变量读取，也可在构造时显式传入： - ``BARK_BASE_URL``：服务器地址，默认 ``http://47.108.226.213:10086``。 - ``BARK_DEVICE_KEY``：设备密钥。 """ def __init__( self, base_url: Optional[str] = None, device_key: Optional[str] = None, timeout: float = 10.0, ) -> None: self.base_url = (base_url or DEFAULT_BASE_URL).rstrip("/") self.device_key = device_key or DEFAULT_DEVICE_KEY self.timeout = timeout def push(self, notification: Notification) -> Dict[str, Any]: """通过 ``POST /push`` 推送一条完整通知，返回服务器 JSON 响应。""" payload = self._payload(notification) payload["device_key"] = self.device_key data = json.dumps(payload, ensure_ascii=False).encode("utf-8") return self._request("/push", data=data) def notify(self, title: str, body: str = "", **kwargs: Any) -> Dict[str, Any]: """GET 快捷推送：``/<key>/<title>/<body>?<params>``。 ``kwargs`` 中除 ``title/body`` 外的键会转成查询参数，布尔值转 ``true/false``，例如 ``level="critical"``、``group="x"``。 """ encoded_title = urllib.parse.quote(title) encoded_body = urllib.parse.quote(body) path = f"/{self.device_key}/{encoded_title}/{encoded_body}" params = _normalize_query(kwargs) params.pop("title", None) params.pop("body", None) if params: path += "?" + urllib.parse.urlencode(params) return self._request(path) # ---------- 内部工具 ---------- @staticmethod def _payload(notification: Notification) -> Dict[str, Any]: """把 Notification 转成 POST /push 的 JSON body（camelCase 键）。""" payload: Dict[str, Any] = { "title": notification.title, "body": notification.body, "level": _value_of(notification.level) or Level.DEFAULT.value, } optional: Dict[str, Any] = { "subtitle": notification.subtitle, "sound": _value_of(notification.sound), "badge": notification.badge, "group": notification.group, "icon": notification.icon, "image": notification.image, "copy": notification.copy, "url": notification.url, "isArchive": notification.is_archive, "call": notification.call, "ttl": notification.ttl, "volume": notification.volume, } for key, value in optional.items(): if value is not None: payload[key] = value return payload def _request(self, path: str, data: Optional[bytes] = None) -> Dict[str, Any]: url = self.base_url + path req = urllib.request.Request(url, data=data) if data is not None: req.add_header("Content-Type", "application/json") try: with urllib.request.urlopen(req, timeout=self.timeout) as resp: raw = resp.read().decode("utf-8") except urllib.error.HTTPError as exc: raw = exc.read().decode("utf-8", errors="replace") except urllib.error.URLError as exc: raise BarkError(f"网络错误: {exc}") from exc try: result = json.loads(raw) except json.JSONDecodeError: result = {"code": -1, "message": raw.strip()} if result.get("code") != 200: raise BarkError(f"Bark 返回异常: code={result.get('code')} message={result.get('message')}") return result

"""Bark 推送客户端：基于标准 Bark 服务器 HTTP API，零第三方依赖。""" from __future__ import annotations import json import os import urllib.error import urllib.parse import urllib.request from typing import Any, Dict, Optional, Union from .types import Level, Notification, Sound DEFAULT_BASE_URL = os.environ.get("BARK_BASE_URL", "http://47.108.226.213:10086") DEFAULT_DEVICE_KEY = os.environ.get("BARK_DEVICE_KEY", "pvd3kWwHNFjKXNjiC2sMe7") class BarkError(RuntimeError): """Bark 服务器返回失败或网络异常。""" def _value_of(value: Union[Level, Sound, str, None]) -> Optional[str]: """枚举成员取其值，普通字符串原样返回。""" if isinstance(value, (Level, Sound)): return value.value return value def _normalize_query(params: Dict[str, Any]) -> Dict[str, Any]: """把 Python 值转成 GET 查询参数可接受的类型。""" out: Dict[str, Any] = {} for key, value in params.items(): if value is None: continue if isinstance(value, bool): out[key] = "true" if value else "false" else: out[key] = _value_of(value) return out class BarkClient: """Bark 通知客户端。支持两种推送方式： - :meth:`push`：通过 ``POST /push`` 发送，支持全部参数（推荐）。 - :meth:`notify`：通过 ``GET ///<body>`` 发送，适合快速测试。地址与密钥默认从环境变量读取，也可在构造时显式传入： - ``BARK_BASE_URL``：服务器地址，默认 ``http://47.108.226.213:10086``。 - ``BARK_DEVICE_KEY``：设备密钥。 """ def __init__( self, base_url: Optional[str] = None, device_key: Optional[str] = None, timeout: float = 10.0, ) -> None: self.base_url = (base_url or DEFAULT_BASE_URL).rstrip("/") self.device_key = device_key or DEFAULT_DEVICE_KEY self.timeout = timeout def push(self, notification: Notification) -> Dict[str, Any]: """通过 ``POST /push`` 推送一条完整通知，返回服务器 JSON 响应。""" payload = self._payload(notification) payload["device_key"] = self.device_key data = json.dumps(payload, ensure_ascii=False).encode("utf-8") return self._request("/push", data=data) def notify(self, title: str, body: str = "", **kwargs: Any) -> Dict[str, Any]: """GET 快捷推送：``/<key>/<title>/<body>?<params>``。 ``kwargs`` 中除 ``title/body`` 外的键会转成查询参数，布尔值转 ``true/false``，例如 ``level="critical"``、``group="x"``。 """ encoded_title = urllib.parse.quote(title) encoded_body = urllib.parse.quote(body) path = f"/{self.device_key}/{encoded_title}/{encoded_body}" params = _normalize_query(kwargs) params.pop("title", None) params.pop("body", None) if params: path += "?" + urllib.parse.urlencode(params) return self._request(path) # ---------- 内部工具 ---------- @staticmethod def _payload(notification: Notification) -> Dict[str, Any]: """把 Notification 转成 POST /push 的 JSON body（camelCase 键）。""" payload: Dict[str, Any] = { "title": notification.title, "body": notification.body, "level": _value_of(notification.level) or Level.DEFAULT.value, } optional: Dict[str, Any] = { "subtitle": notification.subtitle, "sound": _value_of(notification.sound), "badge": notification.badge, "group": notification.group, "icon": notification.icon, "image": notification.image, "copy": notification.copy, "url": notification.url, "isArchive": notification.is_archive, "call": notification.call, "ttl": notification.ttl, "volume": notification.volume, } for key, value in optional.items(): if value is not None: payload[key] = value return payload def _request(self, path: str, data: Optional[bytes] = None) -> Dict[str, Any]: url = self.base_url + path req = urllib.request.Request(url, data=data) if data is not None: req.add_header("Content-Type", "application/json") try: with urllib.request.urlopen(req, timeout=self.timeout) as resp: raw = resp.read().decode("utf-8") except urllib.error.HTTPError as exc: raw = exc.read().decode("utf-8", errors="replace") except urllib.error.URLError as exc: raise BarkError(f"网络错误: {exc}") from exc try: result = json.loads(raw) except json.JSONDecodeError: result = {"code": -1, "message": raw.strip()} if result.get("code") != 200: raise BarkError(f"Bark 返回异常: code={result.get('code')} message={result.get('message')}") return result