diff --git a/pyproject.toml b/pyproject.toml index 5b3c31d..9d03b6b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -15,6 +15,7 @@ dependencies = [ "pyside6-fluent-widgets==1.11.1", "pyyaml==6.0.2", "requests==2.33.0", + "websocket-client==1.9.2", "pyuac==0.0.3", "ruff>=0.15.5", ] diff --git a/src/one_dragon/base/operation/notify_pool.py b/src/one_dragon/base/operation/notify_pool.py index d8db80d..e58fff3 100644 --- a/src/one_dragon/base/operation/notify_pool.py +++ b/src/one_dragon/base/operation/notify_pool.py @@ -21,10 +21,13 @@ class NotifyPool: last_image 属性从 items 尾部遍历获取，用于 APP 级别结束通知附带截图。 """ + DEFAULT_MAX_ITEMS: int = 200 + DEFAULT_MAX_IMAGES: int = 10 + def __init__(self) -> None: self.items: list[NotifyPoolItem] = [] - self.max_items: int = 200 - self.max_images: int = 10 + self.max_items: int = self.DEFAULT_MAX_ITEMS + self.max_images: int = self.DEFAULT_MAX_IMAGES self._image_count: int = 0 def add(self, content: str, image: MatLike | None = None) -> None: @@ -62,4 +65,6 @@ def last_image(self) -> MatLike | None: def clear(self) -> None: self.items.clear() + self.max_items = self.DEFAULT_MAX_ITEMS + self.max_images = self.DEFAULT_MAX_IMAGES self._image_count = 0 diff --git a/src/one_dragon/base/push/channel/discord.py b/src/one_dragon/base/push/channel/discord.py index 58cb6ec..669f318 100644 --- a/src/one_dragon/base/push/channel/discord.py +++ b/src/one_dragon/base/push/channel/discord.py @@ -114,22 +114,22 @@ def push( # 发送消息 message_url = f"{api_host}/channels/{channel_id}/messages" - message_payload = {"content": f"{title}\n{content}"} + message_payload = self._build_message_payload(title, content, image is not None) files = None if image is not None: image_data = self.image_to_bytes(image) if image_data is not None: image_data.seek(0) - files = {'file': ('image.png', image_data, 'image/png')} - data = {'payload_json': json.dumps(message_payload)} + files = {'files[0]': ('screenshot.jpg', image_data, 'image/jpeg')} + data = {'payload_json': json.dumps(message_payload, ensure_ascii=False)} if "Content-Type" in headers: del headers["Content-Type"] else: - data = json.dumps(message_payload) + data = json.dumps(self._build_message_payload(title, content), ensure_ascii=False) headers["Content-Type"] = "application/json" else: - data = json.dumps(message_payload) + data = json.dumps(message_payload, ensure_ascii=False) headers["Content-Type"] = "application/json" response = requests.post(message_url, headers=headers, data=data, files=files, timeout=30) @@ -141,6 +141,21 @@ def push( log.error("Discord 推送异常", exc_info=True) return False, f"Discord 推送异常: {str(e)}" + @staticmethod + def _build_message_payload(title: str, content: str, with_thumbnail: bool = False) -> dict: + if not with_thumbnail: + return {"content": f"{title}\n{content}"} + + return { + "embeds": [ + { + "title": title, + "description": content, + "thumbnail": {"url": "attachment://screenshot.jpg"}, + } + ], + } + def validate_config(self, config: dict[str, str]) -> tuple[bool, str]: """ 验证 Discord 配置 @@ -160,4 +175,4 @@ def validate_config(self, config: dict[str, str]) -> tuple[bool, str]: if len(user_id) == 0: return False, "用户 ID 不能为空" - return True, "配置验证通过" \ No newline at end of file + return True, "配置验证通过" diff --git a/src/one_dragon/base/push/channel/one_bot.py b/src/one_dragon/base/push/channel/one_bot.py index 2bcf927..742d170 100644 --- a/src/one_dragon/base/push/channel/one_bot.py +++ b/src/one_dragon/base/push/channel/one_bot.py @@ -91,6 +91,7 @@ def push( url += "" if url.endswith("/send_msg") else "/send_msg" headers = {'Content-Type': "application/json"} + proxies = self.get_proxy(proxy_url) message = [{"type": "text", "data": {"text": f"{title}\n{content}"}}] if image is not None: @@ -112,7 +113,13 @@ def push( data_private["message_type"] = "private" data_private["user_id"] = user_id try: - response_private = requests.post(url, data=json.dumps(data_private), headers=headers, timeout=15) + response_private = requests.post( + url, + data=json.dumps(data_private), + headers=headers, + proxies=proxies, + timeout=15, + ) response_private.raise_for_status() result_private = response_private.json() @@ -133,7 +140,13 @@ def push( data_group["message_type"] = "group" data_group["group_id"] = group_id try: - response_group = requests.post(url, data=json.dumps(data_group), headers=headers, timeout=15) + response_group = requests.post( + url, + data=json.dumps(data_group), + headers=headers, + proxies=proxies, + timeout=15, + ) response_group.raise_for_status() result_group = response_group.json() @@ -212,11 +225,18 @@ def _send_forward( url: str, data: dict, headers: dict[str, str], + proxies: dict[str, str | None], label: str, ) -> tuple[bool, str]: """发送单批合并转发请求""" try: - resp = requests.post(url, data=json.dumps(data), headers=headers, timeout=30) + resp = requests.post( + url, + data=json.dumps(data), + headers=headers, + proxies=proxies, + timeout=30, + ) resp.raise_for_status() result = resp.json() if result.get('status') == 'ok': @@ -255,6 +275,7 @@ def push_merged( token = config.get('TOKEN', '') headers = {'Content-Type': 'application/json'} + proxies = self.get_proxy(proxy_url) if token and len(token) > 0: headers['Authorization'] = f'Bearer {token}' @@ -293,6 +314,7 @@ def push_merged( base_url + '/send_private_forward_msg', {'user_id': user_id, 'messages': batch}, headers, + proxies, f'私聊合并转发{suffix}', ) if ok: @@ -305,6 +327,7 @@ def push_merged( base_url + '/send_group_forward_msg', {'group_id': group_id, 'messages': batch}, headers, + proxies, f'群聊合并转发{suffix}', ) if ok: diff --git a/src/one_dragon/base/push/channel/qq_bot.py b/src/one_dragon/base/push/channel/qq_bot.py new file mode 100644 index 0000000..54ca8f7 --- /dev/null +++ b/src/one_dragon/base/push/channel/qq_bot.py @@ -0,0 +1,771 @@ +""" +QQ 官方机器人推送渠道 + +通过 QQ 开放平台的机器人接口发送单聊（C2C）与群聊消息，不需要部署 OneBot / NapCat 等协议端。 +收件方用 openid / group_openid 标识，两者都没有查询接口，只能从机器人网关的事件里取， +界面上点“绑定”按钮走本文件的绑定流程自动获取。 +""" + +from __future__ import annotations + +import contextlib +import hashlib +import json +import random +import threading +import time +from collections.abc import Callable, Iterator + +import requests +import websocket +from cv2.typing import MatLike + +from one_dragon.base.push.push_channel import PushChannel +from one_dragon.base.push.push_channel_config import ( + FieldTypeEnum, + PushChannelConfigField, +) +from one_dragon.utils.log_utils import log + +_GATEWAY_URL: str = 'https://api.sgroup.qq.com/gateway' +_INTENTS: int = 33554432 # 1 << 25，群聊与单聊事件 +_BIND_TIMEOUT_SECONDS: int = 60 # 绑定整体超时 +_RECV_TIMEOUT_SECONDS: float = 1.0 # 单次收消息超时，用来定期检查取消并发送心跳 + +# 平台侧常见拒绝原因，拼在错误信息后面，省得用户拿错误码去查文档 +_ERROR_HINTS: dict[int, str] = { + 40034105: '主动消息无权限：群聊需群主在手机 QQ 的群机器人设置里打开“机器人在群内主动发言”', + 40034100: '主动消息发送太频繁：同一接收方每分钟最多 20 条、每天最多 1000 条', + 40054004: '机器人还没添加对方为好友，单聊需先把机器人加为好友', + 40054013: '接收方拒收了消息：需要接收方在手机 QQ 打开该机器人资料页，开启“接收机器人推送”', + 40054007: '消息长度超出上限', + 40034011: 'markdown 内容无效', +} + + +class QQBot(PushChannel): + """QQ 官方机器人推送渠道实现类""" + + # 换取 access_token 的地址、机器人消息接口的基础地址 + _TOKEN_URL: str = 'https://bots.qq.com/app/getAppAccessToken' + _API_BASE_URL: str = 'https://api.sgroup.qq.com' + + _MAX_RETRY: int = 3 # 幂等请求的重试次数，首次之外最多再试 3 次 + _RETRY_DELAY: float = 1.5 # 重试基础间隔（秒），按次数翻倍 + _FILE_TYPE_IMAGE: int = 1 # 富媒体上传的文件类型：图片 + _MD5_10M_SIZE: int = 10002432 # 秒传校验用：文件前 10002432 字节 + + # access_token 缓存，避免每次推送都换 token 触发限频；换了 AppID 就重新取 + _last_access_token_info: tuple[str, str, float] = ('', '', 0.0) + _access_token_lock: threading.Lock = threading.Lock() + + def __init__(self) -> None: + """初始化 QQ 官方机器人推送渠道配置""" + config_schema = [ + PushChannelConfigField( + var_suffix="APP_ID", + title="机器人 AppID", + icon="APPLICATION", + field_type=FieldTypeEnum.TEXT, + placeholder="QQ 开放平台 -> 机器人 -> 开发设置 -> AppID", + required=True + ), + PushChannelConfigField( + var_suffix="CLIENT_SECRET", + title="机器人 AppSecret", + icon="VPN", + field_type=FieldTypeEnum.TEXT, + placeholder="QQ 开放平台 -> 机器人 -> 开发设置 -> AppSecret", + required=True + ), + PushChannelConfigField( + var_suffix="OPENID", + title="单聊 OpenID", + icon="PEOPLE", + field_type=FieldTypeEnum.TEXT, + placeholder="点上方“绑定单聊”自动填入" + ), + PushChannelConfigField( + var_suffix="GROUP_OPENID", + title="群聊 GroupOpenID", + icon="PEOPLE", + field_type=FieldTypeEnum.TEXT, + placeholder="点上方“绑定群聊”自动填入" + ) + ] + + PushChannel.__init__( + self, + channel_id='QQ_BOT', + channel_name='QQ 官方机器人', + config_schema=config_schema + ) + + @staticmethod + def _to_int(value: object, default: int) -> int: + """把接口返回值转成 int（可能是字符串），转换失败时用默认值""" + try: + return int(str(value)) + except (TypeError, ValueError): + return default + + @staticmethod + def _error_hint(code: object) -> str: + """把平台错误码翻译成一句人话，没有对应说明时返回空串""" + hint = _ERROR_HINTS.get(QQBot._to_int(code, 0)) + return f'（{hint}）' if hint else '' + + def push( + self, + config: dict[str, str], + title: str, + content: str, + image: MatLike | None = None, + proxy_url: str | None = None, + ) -> tuple[bool, str]: + """ + 推送消息到 QQ 官方机器人 + + Args: + config: 配置字典，包含 APP_ID、CLIENT_SECRET、OPENID、GROUP_OPENID + title: 消息标题 + content: 消息内容 + image: 图片数据（可选，走富媒体分片上传） + proxy_url: 代理地址 + + Returns: + tuple[bool, str]: 是否成功、错误信息 + """ + try: + return self._send(config, title, content, image, proxy_url) + except Exception as e: + log.error('QQ 官方机器人推送异常', exc_info=True) + return False, f'推送异常：{e}' + + def validate_config(self, config: dict[str, str]) -> tuple[bool, str]: + """ + 验证 QQ 官方机器人配置 + + Args: + config: 配置字典 + + Returns: + tuple[bool, str]: 验证是否通过、错误信息 + """ + app_id = config.get('APP_ID', '').strip() + client_secret = config.get('CLIENT_SECRET', '').strip() + openid = config.get('OPENID', '').strip() + group_openid = config.get('GROUP_OPENID', '').strip() + + if len(app_id) == 0: + return False, "机器人 AppID 不能为空" + + if len(client_secret) == 0: + return False, "机器人 AppSecret 不能为空" + + if len(openid) == 0 and len(group_openid) == 0: + return False, "请先点“绑定单聊”或“绑定群聊”获取接收方 OpenID" + + return True, "配置验证通过" + + def _send( + self, + config: dict[str, str], + title: str, + content: str, + image: MatLike | None, + proxy_url: str | None, + ) -> tuple[bool, str]: + """逐个接收目标发送消息，最后附图。单个目标失败不影响其他目标，但有接收方没收到就按失败返回，图片失败只降级为纯文本。""" + ok, msg = self.validate_config(config) + if not ok: + log.error(f"QQ 官方机器人校验失败：{msg}") + return False, msg + + proxies = self.get_proxy(proxy_url) + access_token, err = self._get_access_token( + config.get('APP_ID', '').strip(), + config.get('CLIENT_SECRET', '').strip(), + proxies, + ) + if len(err) > 0: + log.error(f"QQ 官方机器人{err}") + return False, err + + targets = self._targets(config) + ok_count = 0 + errors: list[str] = [] + payload: dict[str, object] = { + 'msg_type': 2, + 'markdown': {'content': self._to_markdown(title, content)}, + } + for target_name, base_path in targets: + ok, err = self._send_message(base_path, payload, access_token, proxies) + if not ok: + errors.append(f"{target_name}消息发送失败：{err}") + log.warning(f"QQ 官方机器人{target_name}消息发送失败：{err}") + continue + + ok_count += 1 + + if image is None: + continue + ok, err = self._send_image(base_path, image, access_token, proxies) + if not ok: + # 文本已经送达，图片失败只记警告，不把整次推送判为失败 + log.warning(f"QQ 官方机器人{target_name}图片发送失败，已降级为纯文本：{err}") + + if len(errors) > 0: + # 有接收方没收到就按失败返回：只返回成功的话调用方只会提示成功，错误被吞掉 + prefix = f"{ok_count}/{len(targets)} 个接收方推送成功；" if ok_count > 0 else '' + return False, f"{prefix}{'；'.join(errors)}" + + log.info(f"QQ 官方机器人推送成功，目标数：{ok_count}") + return True, "推送成功" + + def _targets(self, config: dict[str, str]) -> list[tuple[str, str]]: + """构造接收目标列表，元素为 (目标名称, 接口路径前缀)。""" + targets: list[tuple[str, str]] = [] + openid = config.get('OPENID', '').strip() + group_openid = config.get('GROUP_OPENID', '').strip() + + if len(openid) > 0: + targets.append(('单聊', f'/v2/users/{openid}')) + if len(group_openid) > 0: + targets.append(('群聊', f'/v2/groups/{group_openid}')) + return targets + + @staticmethod + def _to_markdown(title: str, content: str) -> str: + """标题渲染成二级标题、正文另起一段；没有标题时只发正文""" + if len(title.strip()) == 0: + return content + return f'## {title}\n\n{content}' + + def _get_access_token( + self, + app_id: str, + client_secret: str, + proxies: dict[str, str | None], + ) -> tuple[str, str]: + """换取 access_token，按 AppID 缓存，提前 60 秒视为过期""" + with self._access_token_lock: + cached_app_id, cached_token, expires_at = self._last_access_token_info + if cached_app_id == app_id and len(cached_token) > 0 and expires_at > time.time(): + return cached_token, '' + + data, err = self._post( + self._TOKEN_URL, + {'appId': app_id, 'clientSecret': client_secret}, + proxies, + retries=self._MAX_RETRY, + ) + if data is None: + return '', f"获取 access_token 失败：{err}" + + access_token = str(data.get('access_token') or '') + if len(access_token) == 0: + return '', f"获取 access_token 失败：{data}" + + # 解析失败按 60 秒处理，避免把异常响应缓存两小时 + expires_in = self._to_int(data.get('expires_in'), 60) + self._last_access_token_info = (app_id, access_token, time.time() + max(expires_in - 60, 30)) + return access_token, '' + + def _send_message( + self, + base_path: str, + payload: dict[str, object], + access_token: str, + proxies: dict[str, str | None], + ) -> tuple[bool, str]: + """发消息到目标的 /messages 接口。发消息非幂等，重试会让用户收到重复通知，只尝试一次。""" + data, err = self._post( + f"{self._API_BASE_URL}{base_path}/messages", + payload, + proxies, + access_token=access_token, + timeout=20, + ) + return data is not None, err + + def _send_image( + self, + base_path: str, + image: MatLike, + access_token: str, + proxies: dict[str, str | None], + ) -> tuple[bool, str]: + """发送图片消息（msg_type=7）：先分片上传换 file_info，再发富媒体消息""" + image_io = self.image_to_bytes(image) + if image_io is None: + return False, '图片转换失败' + + file_info, err = self._upload_image(base_path, image_io.getvalue(), access_token, proxies) + if len(err) > 0: + return False, err + + ok, err = self._send_message( + base_path, + {'msg_type': 7, 'media': {'file_info': file_info}}, + access_token, + proxies, + ) + return ok, err + + def _upload_image( + self, + base_path: str, + image_bytes: bytes, + access_token: str, + proxies: dict[str, str | None], + ) -> tuple[str, str]: + """ + 分片上传图片换 file_info：upload_prepare 拿分片地址 -> 逐片 PUT -> upload_part_finish -> files 合并。 + 单聊与群聊的上传接口不互通，必须走对应目标的路径。上传幂等，各步都可重试。 + """ + file_name = 'screenshot.jpg' + prepared, err = self._post( + f"{self._API_BASE_URL}{base_path}/upload_prepare", + { + 'file_type': self._FILE_TYPE_IMAGE, + 'file_size': str(len(image_bytes)), + 'file_name': file_name, + 'md5': hashlib.md5(image_bytes).hexdigest(), + 'sha1': hashlib.sha1(image_bytes).hexdigest(), + 'md5_10m': hashlib.md5(image_bytes[: self._MD5_10M_SIZE]).hexdigest(), + }, + proxies, + access_token=access_token, + retries=self._MAX_RETRY, + timeout=20, + ) + if prepared is None: + return '', f'上传准备失败：{err}' + + upload_id = str(prepared.get('upload_id') or '') + parts = prepared.get('parts') + if len(upload_id) == 0 or not isinstance(parts, list): + return '', f'上传准备返回不完整：{prepared}' + + default_block_size = self._to_int(prepared.get('block_size'), 0) + offset = 0 + for part in parts: + if not isinstance(part, dict): + continue + + block_size = self._to_int(part.get('block_size'), default_block_size) + chunk = image_bytes[offset: offset + block_size] if block_size > 0 else image_bytes[offset:] + if len(chunk) == 0: + break + + presigned_url = str(part.get('presigned_url') or '') + if len(presigned_url) == 0: + return '', '上传准备未返回分片地址' + + ok, err = self._put_chunk(presigned_url, chunk, proxies) + if not ok: + return '', err + + _, err = self._post( + f"{self._API_BASE_URL}{base_path}/upload_part_finish", + { + 'upload_id': upload_id, + 'part_index': part.get('index'), + 'block_size': str(len(chunk)), + 'md5': hashlib.md5(chunk).hexdigest(), + }, + proxies, + access_token=access_token, + retries=self._MAX_RETRY, + timeout=20, + ) + if len(err) > 0: + return '', f'分片确认失败：{err}' + offset += len(chunk) + + # 合并分片，取回 file_info + merged, err = self._post( + f"{self._API_BASE_URL}{base_path}/files", + { + 'file_type': self._FILE_TYPE_IMAGE, + 'upload_id': upload_id, + }, + proxies, + access_token=access_token, + retries=self._MAX_RETRY, + timeout=20, + ) + if merged is None: + return '', f'分片合并失败：{err}' + + file_info = str(merged.get('file_info') or '') + if len(file_info) == 0: + return '', f'分片合并未返回 file_info：{merged}' + return file_info, '' + + def _put_chunk( + self, + presigned_url: str, + chunk: bytes, + proxies: dict[str, str | None], + ) -> tuple[bool, str]: + """通过预签名地址 PUT 上传单个分片，网络异常与 400/429/5xx 重试""" + err = '' + for attempt in range(self._MAX_RETRY + 1): + try: + response = requests.put( + presigned_url, + data=chunk, + headers={'Content-Type': 'application/octet-stream'}, + proxies=proxies, + timeout=60, + ) + except requests.RequestException as e: + err = f'分片上传请求异常：{e}' + else: + if response.status_code < 300: + return True, '' + detail = (response.text or '').strip() or f"HTTP {response.status_code}" + err = f'分片上传失败（HTTP {response.status_code}）：{detail}' + if response.status_code < 500 and response.status_code not in (400, 429): + return False, err + + if attempt < self._MAX_RETRY: + delay = self._RETRY_DELAY * (2 ** attempt) + log.warning(f'QQ 官方机器人{err}，{delay:.1f} 秒后重试') + time.sleep(delay) + + return False, err + + def _post( + self, + url: str, + payload: dict[str, object], + proxies: dict[str, str | None], + access_token: str | None = None, + retries: int = 0, + timeout: int = 15, + ) -> tuple[dict[str, object] | None, str]: + """ + POST JSON，返回 (响应体, 错误信息)。 + + retries > 0 只给幂等请求（换 token、分片上传）用；发消息非幂等，保持默认的不重试。 + 只有没拿到响应（网络异常）和服务端临时故障才重试，业务错误重试也没用。 + """ + headers = {'Content-Type': 'application/json'} + if access_token is not None: + headers['Authorization'] = f'QQBot {access_token}' + + err = '' + for attempt in range(retries + 1): + try: + response = requests.post(url, json=payload, headers=headers, proxies=proxies, timeout=timeout) + except requests.RequestException as e: + err = f'请求异常：{e}' + else: + detail = (response.text or '').strip() or f"HTTP {response.status_code}" + if response.status_code < 400: + try: + data = json.loads(detail) + except ValueError: + return None, f'响应不是 JSON 对象：{detail}' + if not isinstance(data, dict): + return None, f'响应不是 JSON 对象：{detail}' + # 业务错误也会随 HTTP 200 返回 + if not data.get('code'): + return data, '' + return None, f"接口返回错误码 {data.get('code')}：{data.get('message') or detail}{self._error_hint(data.get('code'))}" + + err = f'请求失败（HTTP {response.status_code}）：{detail}' + with contextlib.suppress(ValueError, AttributeError): + err += self._error_hint(json.loads(detail).get('code')) + # 400 可能携带可重试的业务码（如分片未就绪），按可重试处理 + if response.status_code < 500 and response.status_code not in (400, 429): + return None, err + + if attempt < retries: + delay = self._RETRY_DELAY * (2 ** attempt) + log.warning(f'QQ 官方机器人{err}，{delay:.1f} 秒后重试') + time.sleep(delay) + + return None, err + + +def bind_user_openid( + app_id: str, + client_secret: str, + on_verify_code: Callable[[str], None], + cancel_event: threading.Event, +) -> tuple[str, str]: + """ + 等用户在 QQ 里私聊机器人发送验证码，获取单聊 openid + + Args: + app_id: 机器人 AppID + client_secret: 机器人 AppSecret + on_verify_code: 拿到验证码后的回调，界面用它提示用户发什么 + cancel_event: 用户取消绑定时置位 + + Returns: + tuple[str, str]: 单聊 openid、错误信息 + """ + session, err = _open_session( + app_id, + client_secret, + cancel_event, + time.time() + _BIND_TIMEOUT_SECONDS, + auth_message='QQ 机器人鉴权失败，请检查 AppID/AppSecret，并确认已开通单聊事件权限', + timeout_message='绑定超时，请在 60 秒内用手机 QQ 私聊机器人发送验证码', + ) + if session is None: + return '', err + + try: + # Identify 发出后即可显示验证码，READY 事件在接收循环里被忽略 + verify_code = _generate_verify_code() + on_verify_code(verify_code) + + for event_type, data in session.events(): + if event_type == 'C2C_MESSAGE_CREATE': + author = data.get('author') + openid = str(author.get('user_openid') or '') if isinstance(author, dict) else '' + if len(openid) > 0 and verify_code in str(data.get('content') or ''): + return openid, '' + elif event_type == 'FRIEND_ADD': + openid = str(data.get('openid') or '') + if len(openid) > 0: + return openid, '' + return '', session.error + finally: + session.close() + + +def bind_group_openid( + app_id: str, + client_secret: str, + on_verify_code: Callable[[str], None], + on_status: Callable[[str], None], + cancel_event: threading.Event, +) -> tuple[str, str]: + """ + 等机器人被拉进群，或用户在群里 @机器人 发送验证码，获取群 group_openid + + Args: + app_id: 机器人 AppID + client_secret: 机器人 AppSecret + on_verify_code: 拿到验证码后的回调，界面用它提示用户发什么 + on_status: 收到群消息但不是有效验证码时的回调 + cancel_event: 用户取消绑定时置位 + + Returns: + tuple[str, str]: 群 openid、错误信息 + """ + session, err = _open_session( + app_id, + client_secret, + cancel_event, + time.time() + _BIND_TIMEOUT_SECONDS, + auth_message='QQ 机器人鉴权失败，请检查 AppID/AppSecret，并确认已开通群聊事件权限', + timeout_message='绑定超时，请在 60 秒内把机器人拉进群，或在群里 @机器人 发送验证码', + ) + if session is None: + return '', err + + try: + # 群聊要等 READY 再提示验证码，否则用户在订阅生效前把机器人拉进群，入群事件会丢 + if not session.wait_ready(): + return '', session.error + + verify_code = _generate_verify_code() + on_verify_code(verify_code) + + for event_type, data in session.events(): + group_openid = str(data.get('group_openid') or '') + if event_type in ('GROUP_AT_MESSAGE_CREATE', 'GROUP_MESSAGE_CREATE'): + matched = verify_code in str(data.get('content') or '') + if len(group_openid) > 0 and matched: + return group_openid, '' + on_status('已收到群内 @机器人消息，但未获取到群 OpenID，请重试' if matched + else '已收到群内 @机器人消息，但验证码不匹配，请发送当前显示的验证码') + elif event_type == 'GROUP_ADD_ROBOT': + if len(group_openid) > 0: + return group_openid, '' + on_status('已收到机器人入群事件，但未获取到群 OpenID，请重试') + return '', session.error + finally: + session.close() + + +class _GatewaySession: + """已鉴权的网关连接，负责心跳、事件分发、超时与取消""" + + def __init__( + self, + ws: websocket.WebSocket, + heartbeat_ms: int, + deadline: float, + cancel_event: threading.Event, + auth_message: str, + timeout_message: str, + ) -> None: + self._ws: websocket.WebSocket = ws + self._heartbeat_seconds: float = max(heartbeat_ms / 1000, 5) + self._deadline: float = deadline + self._cancel_event: threading.Event = cancel_event + self._auth_message: str = auth_message + self._timeout_message: str = timeout_message + self._seq: int | None = None + self._next_heartbeat: float = time.time() + self._heartbeat_seconds + self.error: str = '' + + def wait_ready(self) -> bool: + """等到 READY，确认鉴权与事件订阅已生效""" + return any(event_type == 'READY' for event_type, _ in self._receive_events()) + + def events(self) -> Iterator[tuple[str, dict]]: + """持续产出网关推送的事件，连接中断时停下，原因见 error""" + yield from self._receive_events() + + def close(self) -> None: + _close_quietly(self._ws) + + def _receive_events(self) -> Iterator[tuple[str, dict]]: + while True: + if self._cancel_event.is_set(): + self.error = '已取消绑定' + return + if time.time() > self._deadline: + self.error = self._timeout_message + return + + try: + raw = self._ws.recv() + except websocket.WebSocketTimeoutException: + raw = '' + except websocket.WebSocketConnectionClosedException: + self.error = 'QQ 网关连接已断开，请重试' + return + + self._heartbeat() + + if len(raw) == 0: + continue + try: + payload = json.loads(raw) + except ValueError: + continue + if not isinstance(payload, dict): + continue + + if payload.get('s') is not None: + self._seq = int(payload['s']) + + op = payload.get('op') + if op == 9: + self.error = self._auth_message + return + if op == 0 and len(str(payload.get('t') or '')) > 0: + data = payload.get('d') + yield str(payload['t']), data if isinstance(data, dict) else {} + + def _heartbeat(self) -> None: + now = time.time() + if now < self._next_heartbeat: + return + self._ws.send(json.dumps({'op': 1, 'd': self._seq})) + self._next_heartbeat = now + self._heartbeat_seconds + + +def _open_session( + app_id: str, + client_secret: str, + cancel_event: threading.Event, + deadline: float, + auth_message: str, + timeout_message: str, +) -> tuple[_GatewaySession | None, str]: + """ + 换取 token、连上网关并完成鉴权握手 + + Returns: + tuple[_GatewaySession | None, str]: 网关连接、错误信息 + """ + if len(app_id) == 0 or len(client_secret) == 0: + return None, '请先填写机器人 AppID 与 AppSecret' + + # 复用渠道的取 token 逻辑，绑定与推送走同一套地址和重试；直连，不读系统代理 + proxies: dict[str, str | None] = {'http': None, 'https': None, 'all': None} + access_token, err = QQBot()._get_access_token(app_id, client_secret, proxies) + if len(err) > 0: + return None, err + + gateway_url, err = _get_gateway_url(access_token, proxies) + if len(err) > 0: + return None, err + + try: + ws = websocket.create_connection(gateway_url, timeout=10) + except websocket.WebSocketException as e: + return None, f'连接 QQ 网关失败：{e}' + + try: + hello = json.loads(ws.recv()) + except (websocket.WebSocketException, ValueError) as e: + _close_quietly(ws) + return None, f'连接 QQ 网关失败：{e}' + + # Hello（op=10）里带心跳间隔，别的 opcode 说明握手异常 + if hello.get('op') != 10: + _close_quietly(ws) + return None, f"网关返回异常 opcode {hello.get('op')}，期望 Hello (10)" + + hello_data = hello.get('d') if isinstance(hello.get('d'), dict) else {} + try: + heartbeat_ms = int(hello_data.get('heartbeat_interval') or 45000) + except (TypeError, ValueError): + heartbeat_ms = 45000 + + try: + ws.send(json.dumps({ + 'op': 2, + 'd': {'token': f'QQBot {access_token}', 'intents': _INTENTS, 'shard': [0, 1]}, + })) + except websocket.WebSocketException as e: + _close_quietly(ws) + return None, f'连接 QQ 网关失败：{e}' + + # 握手用较长超时，之后靠短超时的收消息循环定期检查取消与发送心跳 + ws.settimeout(_RECV_TIMEOUT_SECONDS) + log.info('QQ 官方机器人已连接网关，等待消息事件') + return _GatewaySession(ws, heartbeat_ms, deadline, cancel_event, auth_message, timeout_message), '' + + +def _close_quietly(ws: websocket.WebSocket) -> None: + """关闭连接，失败不影响绑定结果""" + with contextlib.suppress(Exception): + ws.close() + + +def _get_gateway_url(access_token: str, proxies: dict[str, str | None]) -> tuple[str, str]: + """获取 WebSocket 网关地址""" + try: + response = requests.get( + _GATEWAY_URL, + headers={'Authorization': f'QQBot {access_token}'}, + proxies=proxies, + timeout=15, + ) + data = response.json() + except (requests.RequestException, ValueError) as e: + return '', f'获取网关地址失败：{e}' + + url = str(data.get('url') or '') if isinstance(data, dict) else '' + if len(url) == 0: + return '', f'获取网关地址失败：{data}' + return url, '' + + +def _generate_verify_code() -> str: + """生成 4 位数字验证码，用户发回这条消息即确认是本人""" + return ''.join(random.choice('0123456789') for _ in range(4)) diff --git a/src/one_dragon/base/push/push_channel.py b/src/one_dragon/base/push/push_channel.py index 2d22054..933b925 100644 --- a/src/one_dragon/base/push/push_channel.py +++ b/src/one_dragon/base/push/push_channel.py @@ -166,7 +166,7 @@ def image_to_base64(self, image: MatLike, max_bytes: int | None = None) -> str | image_bytes.seek(0) return base64.b64encode(image_bytes.getvalue()).decode('utf-8') - def get_proxy(self, proxy_url: str) -> dict | None: + def get_proxy(self, proxy_url: str | None) -> dict[str, str | None]: """ 获取代理配置 @@ -174,9 +174,10 @@ def get_proxy(self, proxy_url: str) -> dict | None: proxy_url: 代理地址 Returns: - dict | None: 代理配置 + dict[str, str | None]: 代理配置 """ if proxy_url is not None and proxy_url != "": return {"http": proxy_url, "https": proxy_url} - else: - return None + + # 显式传入空代理，避免 requests 自动读取 ALL_PROXY 等环境变量。 + return {"http": None, "https": None, "all": None} diff --git a/src/one_dragon/base/push/push_service.py b/src/one_dragon/base/push/push_service.py index e18f820..4618a4e 100644 --- a/src/one_dragon/base/push/push_service.py +++ b/src/one_dragon/base/push/push_service.py @@ -24,6 +24,7 @@ from one_dragon.base.push.channel.push_me import PushMe from one_dragon.base.push.channel.push_plus import PushPlus from one_dragon.base.push.channel.q_msg import QMsg +from one_dragon.base.push.channel.qq_bot import QQBot from one_dragon.base.push.channel.server_chan import ServerChan from one_dragon.base.push.channel.smtp import Smtp from one_dragon.base.push.channel.synology_chat import SynologyChat @@ -73,6 +74,7 @@ def init_push_channels(self) -> None: self._add_channel(Webhook()) self._add_channel(DingDingBot()) self._add_channel(FeiShu()) + self._add_channel(QQBot()) self._add_channel(WorkWeixinBot()) self._add_channel(WorkWeixinApp()) self._add_channel(OneBot()) @@ -215,7 +217,8 @@ def get_channel_config(self, channel_id: str) -> dict[str, str]: field_name=field.var_suffix, default_value=field.default, ) - config[field.var_suffix] = value + # 未配置的渠道字段可能来自 YAML/env 的 None；统一成字符串交给渠道校验。 + config[field.var_suffix] = '' if value is None else str(value) return config @@ -326,7 +329,7 @@ def push_merged_async( future = self._executor.submit( self.push_merged, title, - items, + items[:], channel_id, ) future.add_done_callback(thread_utils.handle_future_result) diff --git a/src/one_dragon_qt/view/setting/setting_push_interface.py b/src/one_dragon_qt/view/setting/setting_push_interface.py index c5d8f46..c6ad9d2 100644 --- a/src/one_dragon_qt/view/setting/setting_push_interface.py +++ b/src/one_dragon_qt/view/setting/setting_push_interface.py @@ -1,7 +1,13 @@ +import contextlib import json +from collections.abc import Callable +import cv2 +import numpy as np +from cv2.typing import MatLike +from PySide6.QtCore import QThread, Signal from PySide6.QtWidgets import QWidget -from qfluentwidgets import FluentIcon, InfoBar, InfoBarPosition, PushButton +from qfluentwidgets import FluentIcon, InfoBar, InfoBarPosition, PushButton, SettingCard from one_dragon.base.config.config_item import ConfigItem from one_dragon.base.controller.pc_clipboard import PcClipboard @@ -27,17 +33,39 @@ from one_dragon_qt.widgets.setting_card.expand_setting_card_group import ( ExpandSettingCardGroup, ) +from one_dragon_qt.widgets.setting_card.help_card import HelpCard from one_dragon_qt.widgets.setting_card.key_value_setting_card import ( KeyValueSettingCard, ) from one_dragon_qt.widgets.setting_card.multi_push_setting_card import ( MultiPushSettingCard, ) +from one_dragon_qt.widgets.setting_card.qq_bot_bind_card import QQBotBindCard from one_dragon_qt.widgets.setting_card.switch_setting_card import SwitchSettingCard from one_dragon_qt.widgets.setting_card.text_setting_card import TextSettingCard +from one_dragon_qt.widgets.setting_card.yaml_config_adapter import YamlConfigAdapter from one_dragon_qt.widgets.vertical_scroll_interface import VerticalScrollInterface +class PushTestRunner(QThread): + """后台发送测试推送 避免网络请求卡住界面线程。""" + + # 不能用 finished 命名 会遮蔽 QThread 内置的线程结束信号 + push_done = Signal(bool, str) + + def __init__(self, push_func: Callable[[], tuple[bool, str]]) -> None: + super().__init__() + self._push_func: Callable[[], tuple[bool, str]] = push_func + + def run(self) -> None: + try: + ok, msg = self._push_func() + except Exception as e: + ok, msg = False, f'测试推送失败: {e}' + + self.push_done.emit(ok, msg) + + class SettingPushInterface(VerticalScrollInterface): def __init__(self, ctx, parent=None): @@ -50,10 +78,19 @@ def __init__(self, ctx, parent=None): nav_icon=FluentIcon.MESSAGE ) self.ctx = ctx + self._error_bar: InfoBar | None = None + self._info_bar: InfoBar | None = None + self._push_test_runner: PushTestRunner | None = None def get_content_widget(self) -> QWidget: content_widget = Column() + self.help_opt = HelpCard( + title='设置说明', + content='配置运行通知的推送方式或渠道，所有已配置的渠道都会启用', + ) + content_widget.add_widget(self.help_opt) + self.title_opt = TextSettingCard( icon=FluentIcon.MESSAGE, title='自定义通知标题', @@ -102,12 +139,9 @@ def get_content_widget(self) -> QWidget: ) self.notification_method_opt.value_changed.connect(self._update_notification_ui) - channel_group = ExpandSettingCardGroup( - icon=FluentIcon.MESSAGE, - title='通知方式', - initial_expand=True, - ) + channel_group = ExpandSettingCardGroup(icon=FluentIcon.MESSAGE, title='通知方式') channel_group.addHeaderWidget(self.notification_method_opt.combo_box) + channel_group.setExpand(True) content_widget.add_widget(channel_group) # 预创建特殊卡片（稍后按渠道分配） @@ -131,6 +165,12 @@ def get_content_widget(self) -> QWidget: self.email_service_opt.combo_box.setCurrentIndex(-1) self.email_service_opt.setVisible(False) + # QQ 官方机器人的接收方 openid 只能从机器人网关收到的事件里取，用绑定按钮自动回填 + self.qq_bot_bind_card = QQBotBindCard(get_app_credentials=self._get_qq_bot_credentials) + self.qq_bot_bind_card.openid_bound.connect(self._on_qq_bot_openid_bound) + self.qq_bot_bind_card.bind_failed.connect(self._show_error_message) + self.qq_bot_bind_card.setVisible(False) + # 按渠道组织卡片 self.push_channel_cards: dict[str, list] = {} for channel in self.ctx.push_service.channels: @@ -144,6 +184,11 @@ def get_content_widget(self) -> QWidget: channel_group.addSettingCard(self.curl_btn) for field in channel.config_schema: + # QQ 官方机器人的绑定按钮放在接收方字段上方 + if channel.channel_id == 'QQ_BOT' and field.var_suffix == 'OPENID': + channel_cards.append(self.qq_bot_bind_card) + channel_group.addSettingCard(self.qq_bot_bind_card) + card = self._create_card(channel.channel_id, field) channel_cards.append(card) channel_group.addSettingCard(card) @@ -235,37 +280,95 @@ def _send_test_message(self): self._show_error_message(str(e)) return - try: - ok, msg = self.ctx.push_service.push( - title=gt('测试推送通知'), - content=gt('这是一条测试消息'), - channel_id=test_method, - ) - if not ok: - self._show_error_message(msg) - else: - self._show_success_message("已向当前通知方式发送测试消息") - except ValueError as e: - self._show_error_message(str(e)) - except Exception as e: - self._show_error_message(f"测试推送失败: {str(e)}") + self._start_test_push(test_method, "已向当前通知方式发送测试消息") def _send_test_all_message(self): """发送测试消息到所有已配置的通知方式""" - try: - self._show_success_message("正在向所有已配置的通知方式发送测试消息...") + self._start_test_push(None, "已向所有已配置的通知方式发送测试消息") + + def _start_test_push(self, channel_id: str | None, success_msg: str) -> None: + """ + 后台发送测试推送 避免网络请求卡住界面线程 + + 推送期间禁用按钮 既提示正在发送 也避免连点堆积多次推送。 + + Args: + channel_id: 推送渠道ID 为空时使用所有通过配置校验的渠道 + success_msg: 推送成功时的提示文案 + """ + # 上一次未结束时直接返回 覆盖字段会让仍在运行的线程被回收导致崩溃 + if self._push_test_runner is not None and self._push_test_runner.isRunning(): + return + + # 先生成截图再禁用按钮 生成失败时按钮仍可点击 不会卡在禁用状态 + image = self._get_test_screenshot() + + self.test_current_btn.setDisabled(True) + self.test_all_btn.setDisabled(True) + self._clear_error_bar() + self._info_bar = InfoBar.info( + title='提示', + content="正在发送测试消息...", + orient=InfoBarPosition.TOP, + isClosable=True, + duration=-1, + parent=self + ) + + def push_test() -> tuple[bool, str]: ok, msg = self.ctx.push_service.push( title=gt('测试推送通知'), content=gt('这是一条测试消息'), + image=image, + channel_id=channel_id, ) - if not ok: - self._show_error_message(msg) - else: - self._show_success_message("已向所有已配置的通知方式发送测试消息") - except ValueError as e: - self._show_error_message(str(e)) - except Exception as e: - self._show_error_message(f"测试推送失败: {str(e)}") + # 单渠道成功时渠道也会回「推送成功」这类文案 只有全渠道的空 msg 才代表全部成功 + return ok, ('' if ok and channel_id is not None else msg) + + self._push_test_runner = PushTestRunner(push_func=push_test) + self._push_test_runner.push_done.connect( + lambda ok, msg: self._on_test_push_finished(ok, msg, success_msg) + ) + self._push_test_runner.start() + + def _on_test_push_finished(self, ok: bool, msg: str, success_msg: str) -> None: + """测试推送结束后的界面反馈""" + self.test_current_btn.setDisabled(False) + self.test_all_btn.setDisabled(False) + + if self._info_bar is not None: + with contextlib.suppress(RuntimeError): + self._info_bar.close() + self._info_bar = None + + # 部分渠道失败时 ok 仍为 True 错误信息通过 msg 带回来 + if not ok or len(msg) > 0: + self._show_error_message(msg if msg else "测试推送失败") + else: + self._show_success_message(success_msg) + + def _get_test_screenshot(self) -> MatLike | None: + """生成通知测试用标准色相图。""" + if not self.ctx.push_service.push_config.send_image: + return None + + size = 600 + center = size // 2 + y, x = np.ogrid[:size, :size] + dx = x - center + dy = center - y + radius = np.sqrt(dx * dx + dy * dy) + angle = (np.arctan2(dy, dx) + 2 * np.pi) % (2 * np.pi) + + hsv = np.zeros((size, size, 3), dtype=np.uint8) + hsv[:, :, 0] = (angle * 180 / np.pi / 2).astype(np.uint8) + hsv[:, :, 1] = np.clip(radius / (size / 2) * 255, 0, 255).astype(np.uint8) + hsv[:, :, 2] = 255 + + image = cv2.cvtColor(hsv, cv2.COLOR_HSV2BGR) + image[radius > center] = (45, 45, 45) + cv2.circle(image, (center, center), center - 2, (255, 255, 255), 2) + return image def _on_email_service_selected(self, text): config = PushEmailServices.get_configs(str(text)) @@ -275,17 +378,17 @@ def _on_email_service_selected(self, text): smtp_port = config.get("port", 465) smtp_ssl = str(config.get("secure", True)).lower() if "secure" in config else "true" # 找到对应的TextSettingCard并赋值 - server_card = getattr(self, "smtp_server_push_card", None) + server_card: SettingCard = getattr(self, "smtp_server_push_card", None) if server_card is not None: host = f"{smtp_server}:{smtp_port}" server_card.setValue(host) - adapter = getattr(server_card, "adapter", None) + adapter: YamlConfigAdapter = getattr(server_card, "adapter", None) if adapter is not None: adapter.set_value(host) - ssl_card = getattr(self, "smtp_ssl_push_card", None) + ssl_card: SettingCard = getattr(self, "smtp_ssl_push_card", None) if ssl_card is not None: ssl_card.setValue(smtp_ssl) - adapter = getattr(ssl_card, "adapter", None) + adapter: YamlConfigAdapter = getattr(ssl_card, "adapter", None) if adapter is not None: adapter.set_value(smtp_ssl) @@ -302,6 +405,24 @@ def _set_proxy_input_visibility(self): """设置代理输入框的可见性""" self.proxy_input_opt.setVisible(self.proxy_opt.getValue() == PushProxy.PERSONAL.value.value) + def _get_qq_bot_credentials(self) -> tuple[str, str]: + """取 QQ 机器人配置里的 AppID 与 AppSecret""" + channel_config = self.ctx.push_service.get_channel_config('QQ_BOT') + return channel_config.get('APP_ID', ''), channel_config.get('CLIENT_SECRET', '') + + def _on_qq_bot_openid_bound(self, is_group: bool, openid: str) -> None: + """绑定成功，写进配置并回填输入框""" + field_name = 'GROUP_OPENID' if is_group else 'OPENID' + self.ctx.push_service.push_config.update_channel_config_value('QQ_BOT', field_name, openid) + + card: TextSettingCard | None = getattr(self, f'qq_bot_{field_name.lower()}_push_card', None) + if card is not None: + card.setValue(openid) + if card.adapter is not None: + card.adapter.set_value(openid) + + self._show_success_message('绑定成功') + def on_interface_shown(self) -> None: VerticalScrollInterface.on_interface_shown(self) @@ -384,8 +505,19 @@ def _validate_json_format(self, json_str: str) -> bool: except (json.JSONDecodeError, TypeError): return False + def _clear_error_bar(self) -> None: + """关闭常驻的错误提示""" + if self._error_bar is None: + return + + # 用户手动点掉后底层对象可能已被删除 + with contextlib.suppress(RuntimeError): + self._error_bar.close() + self._error_bar = None + def _show_success_message(self, message: str): """显示成功消息提示""" + self._clear_error_bar() InfoBar.success( title='成功', content=message, @@ -396,12 +528,13 @@ def _show_success_message(self, message: str): ) def _show_error_message(self, message: str): - """显示错误消息提示""" - InfoBar.error( + """显示错误消息提示，不自动消失，新的错误替换旧的""" + self._clear_error_bar() + self._error_bar = InfoBar.error( title='错误', content=message, orient=InfoBarPosition.TOP, isClosable=True, - duration=5000, + duration=-1, parent=self ) diff --git a/src/one_dragon_qt/widgets/setting_card/qq_bot_bind_card.py b/src/one_dragon_qt/widgets/setting_card/qq_bot_bind_card.py new file mode 100644 index 0000000..0e9e110 --- /dev/null +++ b/src/one_dragon_qt/widgets/setting_card/qq_bot_bind_card.py @@ -0,0 +1,224 @@ +import contextlib +import threading +from collections.abc import Callable + +from PySide6.QtCore import QThread, Signal +from PySide6.QtWidgets import QApplication, QWidget +from qfluentwidgets import FluentIcon, InfoBar, PushButton + +from one_dragon.base.push.channel.qq_bot import ( + bind_group_openid, + bind_user_openid, +) +from one_dragon.utils.log_utils import log +from one_dragon_qt.widgets.setting_card.multi_push_setting_card import ( + MultiPushSettingCard, +) + + +class QQBotBindRunner(QThread): + """后台跑 QQ 机器人绑定，等验证码期间不阻塞界面""" + + status_signal = Signal(str) + verify_signal = Signal(str) # 验证码文案，卡片另外弹一条常驻提示 + result_signal = Signal(bool, str, str) # 是否群聊、openid、错误信息 + + def __init__( + self, + app_id: str, + client_secret: str, + is_group: bool, + parent: QWidget | None = None, + ) -> None: + """ + 初始化绑定任务 + + Args: + app_id: 机器人 AppID + client_secret: 机器人 AppSecret + is_group: 绑定群聊还是单聊 + parent: 父组件 + """ + QThread.__init__(self, parent) + self._app_id: str = app_id + self._client_secret: str = client_secret + self._is_group: bool = is_group + self._cancel_event: threading.Event = threading.Event() + + def cancel(self) -> None: + """用户点取消时置位，绑定流程会在下一次收消息超时时退出""" + self._cancel_event.set() + + def run(self) -> None: + """跑绑定流程，结果通过 result_signal 发回界面线程""" + try: + if self._is_group: + openid, err = bind_group_openid( + self._app_id, + self._client_secret, + self._on_group_verify_code, + self.status_signal.emit, + self._cancel_event, + ) + else: + openid, err = bind_user_openid( + self._app_id, + self._client_secret, + self._on_user_verify_code, + self._cancel_event, + ) + self.result_signal.emit(self._is_group, openid, err) + except Exception as e: + log.error('QQ 官方机器人绑定异常', exc_info=True) + self.result_signal.emit(self._is_group, '', f'绑定失败：{e}') + + def _on_user_verify_code(self, code: str) -> None: + """单聊的验证码提示：让用户私聊机器人发出去""" + msg = f'验证码 {code}，请用手机 QQ 私聊机器人发送' + self.status_signal.emit(msg) + self.verify_signal.emit(msg) + + def _on_group_verify_code(self, code: str) -> None: + """群聊的验证码提示：拉进群或在群里 @机器人 发出去""" + msg = f'验证码 {code}，请把机器人拉进群或在群里 @机器人 发送' + self.status_signal.emit(msg) + self.verify_signal.emit(msg) + + +class QQBotBindCard(MultiPushSettingCard): + """QQ 官方机器人接收方绑定卡片 + + 单聊和群聊的 openid 只能从机器人网关收到的事件里取，卡片提供两个绑定按钮， + 绑定跑在后台线程，成功后通过 openid_bound 把结果交给界面写入配置。 + """ + + openid_bound = Signal(bool, str) # 是否群聊、openid + bind_failed = Signal(str) # 错误信息 + + def __init__( + self, + get_app_credentials: Callable[[], tuple[str, str]], + parent: QWidget | None = None, + ) -> None: + """ + 初始化绑定卡片 + + Args: + get_app_credentials: 取当前 AppID 与 AppSecret 的回调，点绑定时才读最新值 + parent: 父组件 + """ + self._get_app_credentials: Callable[[], tuple[str, str]] = get_app_credentials + self._runner: QQBotBindRunner | None = None + self._verify_bar: InfoBar | None = None + + self.bind_user_btn = PushButton(text='绑定单聊', icon=FluentIcon.PEOPLE) + self.bind_user_btn.clicked.connect(lambda: self._start_bind(False)) + self.bind_group_btn = PushButton(text='绑定群聊', icon=FluentIcon.PEOPLE) + self.bind_group_btn.clicked.connect(lambda: self._start_bind(True)) + self.cancel_btn = PushButton(text='取消', icon=FluentIcon.CANCEL) + self.cancel_btn.clicked.connect(self._cancel_bind) + self.cancel_btn.setVisible(False) + + MultiPushSettingCard.__init__( + self, + icon=FluentIcon.SYNC, + title="自动获取接收方 *", + content='点绑定后按提示用 QQ 给机器人发验证码，OpenID 自动填入下方', + btn_list=[self.bind_user_btn, self.bind_group_btn, self.cancel_btn], + parent=parent, + ) + + # 绑定最长要等 60 秒，退出时先取消再等线程收尾，否则线程还在跑就被父对象析构，Qt 会直接终止进程 + QApplication.instance().aboutToQuit.connect(self._stop_runner) + + def _start_bind(self, is_group: bool) -> None: + """ + 开始绑定接收方 + + Args: + is_group: 绑定群聊还是单聊 + """ + if self._runner is not None: + return + + app_id, client_secret = self._get_app_credentials() + if len(app_id) == 0 or len(client_secret) == 0: + self.bind_failed.emit('请先填写 AppID 与 AppSecret，再点绑定') + return + + self._runner = QQBotBindRunner(app_id, client_secret, is_group, self) + self._runner.status_signal.connect(self.setContent) + self._runner.verify_signal.connect(self._show_verify_code) + self._runner.result_signal.connect(self._on_bind_result) + self._runner.finished.connect(self._on_bind_finished) + self._set_binding(True) + self.setContent('正在连接 QQ 网关…') + self._runner.start() + + def _cancel_bind(self) -> None: + """取消绑定，断开网关连接""" + if self._runner is not None: + self._runner.cancel() + + def _stop_runner(self) -> None: + """程序退出前停掉绑定线程，避免线程还在跑就被父对象析构""" + runner = self._runner + if runner is None: + return + + # 退出路径上不需要再弹“已取消绑定”的提示 + runner.result_signal.disconnect(self._on_bind_result) + runner.cancel() + if not runner.wait(3000): + log.warning('QQ 机器人绑定线程未在退出前结束') + + def _show_verify_code(self, msg: str) -> None: + """验证码另弹一条不自动消失的提示，卡片里那行灰色小字容易被忽略""" + self._close_verify_bar() + self._verify_bar = InfoBar.warning( + title='绑定 QQ 机器人', + content=msg, + isClosable=True, + duration=-1, + parent=self.window(), + ) + + def _on_bind_result(self, is_group: bool, openid: str, error: str) -> None: + """绑定结束，成功则把 openid 交给界面写进配置""" + if len(error) > 0: + self.setContent(error) + self.bind_failed.emit(error) + return + + target = '群聊' if is_group else '单聊' + self.setContent(f'绑定成功，已自动填入{target} OpenID') + self.openid_bound.emit(is_group, openid) + + def _on_bind_finished(self) -> None: + """绑定结束，收起验证码提示并复位按钮""" + self._close_verify_bar() + runner = self._runner + self._runner = None + self._set_binding(False) + if runner is not None: + runner.deleteLater() + + def _set_binding(self, binding: bool) -> None: + """绑定期间禁用绑定按钮，只留取消""" + self.cancel_btn.setVisible(binding) + self.bind_user_btn.setEnabled(not binding) + self.bind_group_btn.setEnabled(not binding) + # 禁用按钮会让焦点落到第一个输入框并被全选，这里把焦点留在按钮上 + if binding: + self.cancel_btn.setFocus() + else: + self.bind_user_btn.setFocus() + + def _close_verify_bar(self) -> None: + """关闭常驻的验证码提示，用户已手动点掉时底层对象可能已被删除""" + if self._verify_bar is None: + return + + with contextlib.suppress(RuntimeError): + self._verify_bar.close() + self._verify_bar = None diff --git a/uv.lock b/uv.lock index 56e76e2..5ca54b8 100644 --- a/uv.lock +++ b/uv.lock @@ -196,6 +196,7 @@ dependencies = [ { name = "pyyaml" }, { name = "requests" }, { name = "ruff" }, + { name = "websocket-client" }, ] [package.metadata] @@ -212,6 +213,7 @@ requires-dist = [ { name = "pyyaml", specifier = "==6.0.2" }, { name = "requests", specifier = "==2.33.0" }, { name = "ruff", specifier = ">=0.15.5" }, + { name = "websocket-client", specifier = "==1.9.2" }, ] [[package]] @@ -3361,3 +3363,12 @@ sdist = { url = "https://files.pythonhosted.org/packages/53/0c/06f8b233b8fd13b9e wheels = [ { url = "https://files.pythonhosted.org/packages/7f/3e/5db95bcf282c52709639744ca2a8b149baccf648e39c8cc87553df9eae0c/urllib3-2.7.0-py3-none-any.whl", hash = "sha256:9fb4c81ebbb1ce9531cce37674bbc6f1360472bc18ca9a553ede278ef7276897", size = 131087, upload-time = "2026-05-07T16:13:17.151Z" }, ] + +[[package]] +name = "websocket-client" +version = "1.9.2" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/d8/cb/a5abcc2891249f393827c650c6296660ce40374ac22d99ab9aea41f9d2a2/websocket_client-1.9.2.tar.gz", hash = "sha256:0fcb57545848be86992e128218fd96dd87a6769ffdb1a968dff79632b85604d0", size = 84110, upload-time = "2026-08-31T14:08:40.964Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/d5/d2/cc4dc1271e464942db7ee278baae2daa99ee77cb2af744025c04da585a3e/websocket_client-1.9.2-py3-none-any.whl", hash = "sha256:e1a673830a9c7bfa47b1cd3d5e4178f4c9651d80a4eab02c9c23a1c3ec6250ce", size = 95786, upload-time = "2026-08-31T14:08:39.899Z" }, +]