diff --git a/admin/dashboard/blueprints/contacts.py b/admin/dashboard/blueprints/contacts.py index 27a11c5..0cc71e8 100644 --- a/admin/dashboard/blueprints/contacts.py +++ b/admin/dashboard/blueprints/contacts.py @@ -203,6 +203,73 @@ def _extract_emoji_meta(attachment_url: str, image_path: str): return md5, total_length +def _parse_positive_int(value): + """将任意输入尽量解析为正整数，失败时返回 0。 + + 说明： + 1. 前端可能传 total_length / totalLength / len，类型也可能是字符串； + 2. 统一在这里收口，避免每个分支都重复写 try/except。 + """ + try: + parsed = int(value) + except Exception: + return 0 + return parsed if parsed > 0 else 0 + + +def _get_emoji_asset_by_md5(message_storage, md5: str): + """从消息存储中按 md5 反查表情原始记录。 + + 说明： + 1. 优先走 message_storage 自身方法，兼容未来把查询逻辑上移； + 2. 若当前实例没有该方法，则回退到底层 message_db； + 3. 查不到时返回 None，让上层决定是否报错。 + """ + if not message_storage or not md5: + return None + + if hasattr(message_storage, "get_emoji_asset_by_md5"): + return message_storage.get_emoji_asset_by_md5(md5) + + message_db = getattr(message_storage, "message_db", None) + if message_db and hasattr(message_db, "get_emoji_asset_by_md5"): + return message_db.get_emoji_asset_by_md5(md5) + + return None + + +def _resolve_emoji_send_meta(message_storage, md5: str, total_length: int): + """补全发送表情所需的 md5 与 total_length。 + + 说明： + 1. wechat_ipad 的 SendEmoji 接口并不是只要 md5，还必须带 TotalLen； + 2. 当前端只传了 md5 或长度为空时，这里尝试从历史消息里反查原始 XML； + 3. 返回值始终是“规整后的 md5 + total_length”，方便发送分支直接使用。 + """ + normalized_md5 = _safe_text(md5).strip().lower() + normalized_total_length = _parse_positive_int(total_length) + + if not re.fullmatch(r"[0-9a-f]{16,64}", normalized_md5): + return "", 0 + + if normalized_total_length > 0: + return normalized_md5, normalized_total_length + + asset = _get_emoji_asset_by_md5(message_storage, normalized_md5) + if not asset: + return normalized_md5, 0 + + resolved_md5, resolved_total_length = _extract_emoji_meta( + _safe_text(asset.get("attachment_url")), + _safe_text(asset.get("image_path")) + ) + if resolved_md5 and resolved_md5 != normalized_md5: + # 历史数据如果出现大小写或异常值，以前端传入的 md5 为准，避免串表情。 + logger.warning(f"表情参数回填命中 md5 不一致，request_md5={normalized_md5}, record_md5={resolved_md5}") + + return normalized_md5, _parse_positive_int(resolved_total_length) + + def _normalize_recent_message(server, raw_message: dict, chat_type: str, target_wxid: str): sender = _safe_text(raw_message.get("sender")).strip() message_type = str(raw_message.get("message_type", "")) @@ -830,20 +897,30 @@ def api_send_message(): elif msg_type == 'emoji': if not isinstance(content, dict): return jsonify({'success': False, 'message': '表情参数格式错误'}) - md5 = _safe_text(content.get('md5')).strip().lower() - total_length = int(content.get('total_length') or 0) - if not re.fullmatch(r"[0-9a-f]{16,64}", md5) or total_length <= 0: - return jsonify({'success': False, 'message': '缺少表情 md5 或长度'}) - try: - loop = get_or_create_loop() - future = asyncio.run_coroutine_threadsafe(server.client.send_emoji_message(wxid, md5, total_length), loop) - future.result(timeout=20) - except Exception as e: - logger.error(f"发送表情失败 md5={md5} len={total_length} wxid={wxid}: {e}") - return jsonify({'success': False, 'message': f'表情发送失败: {str(e)}'}), 500 + # 表情发送必须同时具备 md5 和 total_length。 + # 当前前端有时只拿得到 md5，因此这里优先使用请求体里的长度， + # 拿不到时再去历史消息表里反查，避免“参数明明看起来对，但接口还是发不出去”。 + md5, total_length = _resolve_emoji_send_meta( + getattr(server, "message_storage", None), + content.get('md5'), + content.get('total_length') or content.get('totalLength') or content.get('len') + ) + if not md5: + return jsonify({'success': False, 'message': '表情 md5 格式不正确'}) + if total_length <= 0: + return jsonify({'success': False, 'message': '该表情缺少 total_length，无法仅凭 md5 发送'}) + + # 表情发送改为和文本/图片一致的异步提交通道，避免 HTTP 请求线程 + # 同步等待队列结果导致“高概率卡住”的体验问题。 + logger.info(f"提交表情发送任务 wxid={wxid} md5={md5} total_length={total_length}") + send_message_in_thread(server.client.send_emoji_message, wxid, md5, total_length) return jsonify({ 'success': True, - 'message': '表情发送成功' + 'message': '表情消息已提交到 iPad 通道', + 'data': { + 'md5': md5, + 'total_length': total_length + } }) else: diff --git a/db/message_storage.py b/db/message_storage.py index 79dcaf2..9bbf852 100644 --- a/db/message_storage.py +++ b/db/message_storage.py @@ -645,6 +645,26 @@ class MessageStorageDB(BaseDBOperator): """ return self.execute_query(sql, (limit,)) or [] + def get_emoji_asset_by_md5(self, md5: str) -> Optional[Dict]: + """根据表情 md5 精确查找最近一条表情消息。 + + 说明： + 1. 后台聊天发送表情时，前端偶尔只能拿到 md5，拿不到 total_length； + 2. wechat_ipad 的 SendEmoji 接口要求同时提供 Md5 和 TotalLen； + 3. 这里直接从历史消息表里按 md5 反查最近一条原始记录，给发送接口补全长度。 + """ + sql = """ + SELECT message_id, group_id, sender, timestamp, message_type, attachment_url, image_path + FROM messages + WHERE message_type IN ('47', '1048625', '1090519089') + AND attachment_url IS NOT NULL + AND attachment_url <> '' + AND attachment_url LIKE %s + ORDER BY timestamp DESC + LIMIT 1 + """ + return self.execute_query(sql, (f'%md5="{md5}"%',), fetch_one=True) + def get_messages_by_date_range(self, group_id: str, start_date: str, end_date: str = None, min_content_length: int = 6, max_results: int = 5000) -> List[Dict]: """按日期范围获取消息（支持按天总结） diff --git a/wechat_ipad/client/message.py b/wechat_ipad/client/message.py index 859091b..6fd981c 100644 --- a/wechat_ipad/client/message.py +++ b/wechat_ipad/client/message.py @@ -506,15 +506,26 @@ class MessageMixin(WechatAPIClientBase): if not self.wxid: raise UserLoggedOut("请先登录") - async with aiohttp.ClientSession() as session: + # 表情发送接口历史上最容易出现“接口长时间不返回，导致整个消息队列被拖住”的问题， + # 因此这里单独加总超时和更细的日志，方便区分“参数错误”和“接口无响应”两类故障。 + timeout = aiohttp.ClientTimeout(total=20) + async with aiohttp.ClientSession(timeout=timeout) as session: json_param = {"Wxid": self.wxid, "ToWxid": wxid, "Md5": md5, "TotalLen": total_length} - response = await session.post(f'http://{self.ip}:{self.port}/api/Msg/SendEmoji', json=json_param) - json_resp = await response.json() + try: + self.logging.info("开始发送表情消息: 对方wxid:{} md5:{} 总长度:{}", wxid, md5, total_length) + response = await session.post(f'http://{self.ip}:{self.port}/api/Msg/SendEmoji', json=json_param) + json_resp = await response.json(content_type=None) + except asyncio.TimeoutError as exc: + self.logging.error("发送表情消息超时: 对方wxid:{} md5:{} 总长度:{}", wxid, md5, total_length) + raise TimeoutError("SendEmoji 接口调用超时") from exc if json_resp.get("Success"): - self.logging.info("发送表情消息: 对方wxid:{} md5:{} 总长度:{}", wxid, md5, total_length) - return json_resp.get("Data").get("emojiItem") + data = json_resp.get("Data") or {} + self.logging.info("发送表情消息成功: 对方wxid:{} md5:{} 总长度:{}", wxid, md5, total_length) + return data.get("emojiItem") or data.get("EmojiItem") or data else: + self.logging.error("发送表情消息失败: 对方wxid:{} md5:{} 总长度:{} resp:{}", + wxid, md5, total_length, json_resp) self.error_handler(json_resp) async def send_card_message(self, wxid: str, card_wxid: str, card_nickname: str, card_alias: str = "") -> tuple[