Lee
19afaffbb7
All checks were successful
Java Maven 3.9.9 & JDK 26 CI/CD Pipeline / build-and-deploy (push) Successful in 3m22s
38 lines
1.6 KiB
Python
38 lines
1.6 KiB
Python
"""QQ notifier (reserved). Transport: NapCat OneBot HTTP API."""
|
|
import os
|
|
|
|
import httpx
|
|
|
|
from .base import ImageAttachment, Notifier
|
|
|
|
|
|
class QQNotifier(Notifier):
|
|
name = "qq"
|
|
|
|
def __init__(self, onebot_http: str = '', access_token: str = '',
|
|
targets: list[dict] | None = None) -> None:
|
|
self.onebot_http = (onebot_http or os.getenv("QQ_ONEBOT_HTTP", "")).rstrip("/")
|
|
self.access_token = access_token or os.getenv("ONEBOT_ACCESS_TOKEN", "")
|
|
self.targets = targets if targets is not None else []
|
|
if not self.onebot_http:
|
|
raise RuntimeError("QQ OneBot HTTP endpoint not configured (reserved)")
|
|
|
|
async def send(self, title: str, text: str,
|
|
images: list[ImageAttachment] | None = None) -> None:
|
|
if not self.targets:
|
|
raise RuntimeError("QQ targets not configured (reserved)")
|
|
headers = {}
|
|
if self.access_token:
|
|
headers["Authorization"] = f"Bearer {self.access_token}"
|
|
body = f"{title}\n{text}" if title else text
|
|
async with httpx.AsyncClient(timeout=20, trust_env=False) as client:
|
|
for t in self.targets:
|
|
if t.get("group_id"):
|
|
await client.post(f"{self.onebot_http}/send_group_msg",
|
|
headers=headers,
|
|
json={"group_id": t["group_id"], "message": body})
|
|
elif t.get("user_id"):
|
|
await client.post(f"{self.onebot_http}/send_private_msg",
|
|
headers=headers,
|
|
json={"user_id": t["user_id"], "message": body})
|