From e284d43cc75fb5a8390274698d79383362420675 Mon Sep 17 00:00:00 2001 From: qihu-bot Date: Sat, 15 Aug 2026 18:13:47 +0800 Subject: [PATCH] =?UTF-8?q?#970=20=E5=BE=AE=E4=BF=A1Bot=20P1=E5=A2=9E?= =?UTF-8?q?=E5=BC=BA=20-=20sendtyping=20+=20=E8=AF=AD=E9=9F=B3ASR=20+=20Se?= =?UTF-8?q?ssion=20Guard?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1.语音消息ASR: 提取voice_item.text(微信自带转文字) 2.sendtyping正在输入状态: getconfig拿typing_ticket缓存24h + sendtyping发送 3.Session Guard增强: 会话过期记录告警日志供前台提示重扫 --- backend/python/app/api/im_bot_user.py | 68 ++++++++++++++++++++++++++- 1 file changed, 66 insertions(+), 2 deletions(-) diff --git a/backend/python/app/api/im_bot_user.py b/backend/python/app/api/im_bot_user.py index a246682..d319d4e 100644 --- a/backend/python/app/api/im_bot_user.py +++ b/backend/python/app/api/im_bot_user.py @@ -54,6 +54,9 @@ WECHAT_TEXT_MAX_LEN = 1500 # 每个已连接账号一个后台接收线程，key = 账号表主键 id _wechat_receiver_threads: dict[int, threading.Thread] = {} +# typing_ticket 缓存：{ilink_user_id: (ticket, 写入时间戳)}，24h 有效 +_typing_ticket_cache: dict[str, tuple[str, float]] = {} + # ============================================================ # 通用 DB 工具 @@ -307,6 +310,45 @@ def _wechat_send_message(baseurl: str, bot_token: str, to_user_id: str, context_ return all_ok, last_ret, {} +def _wechat_get_typing_ticket(baseurl: str, bot_token: str, ilink_user_id: str, context_token: str) -> str: + """getconfig 获取 typing_ticket（按 ilink_user_id 缓存 24h）。""" + global _typing_ticket_cache + _now = time.time() + _hit = _typing_ticket_cache.get(ilink_user_id) + if _hit and (_now - _hit[1]) < 86400: + return _hit[0] + try: + resp = requests.post( + baseurl.rstrip("/") + "/ilink/bot/getconfig", + headers=_wechat_headers(bot_token), + json={"ilink_user_id": ilink_user_id, "context_token": context_token}, + timeout=10, + ) + body = resp.json() if resp.content else {} + ticket = body.get("typing_ticket", "") if isinstance(body, dict) else "" + if ticket: + _typing_ticket_cache[ilink_user_id] = (ticket, _now) + return ticket + except Exception: + return "" + + +def _wechat_send_typing(baseurl: str, bot_token: str, ilink_user_id: str, typing_ticket: str, status: int = 1) -> bool: + """发送/取消"正在输入"状态。status=1 typing，2 cancel。失败静默（不阻塞主流程）。""" + if not ilink_user_id or not typing_ticket: + return False + try: + resp = requests.post( + baseurl.rstrip("/") + "/ilink/bot/sendtyping", + headers=_wechat_headers(bot_token), + json={"ilink_user_id": ilink_user_id, "typing_ticket": typing_ticket, "status": status}, + timeout=10, + ) + return resp.ok + except Exception: + return False + + def _platform_relay_url(raw_url: str) -> str: """把 AGNES 等境外媒体 URL 改写为平台中转地址，使微信/iLink 可内联渲染。 @@ -367,8 +409,14 @@ def _wechat_send_media(baseurl: str, bot_token: str, to_user_id: str, context_to def _wechat_extract_text(msg: dict) -> str: parts = [] for it in (msg.get("item_list") or []): - if it.get("type") == 1 and it.get("text_item", {}).get("text"): + _t = it.get("type") + if _t == 1 and it.get("text_item", {}).get("text"): parts.append(it["text_item"]["text"]) + elif _t == 3: + # 语音消息：微信已自带 ASR 转文字（voice_item.text），直接复用 + voice = it.get("voice_item") or {} + if voice.get("text"): + parts.append("[语音] " + str(voice["text"])) if not parts: for k in ("text", "content", "message", "msg"): v = msg.get(k) @@ -480,7 +528,7 @@ def _mirror_to_ai_chat(user_id: str, user_content: str, assistant_content: str): # 后台接收线程（每用户独立） # ============================================================ def _mark_wechat_expired(account_id: int): - """发送会话失效（ret=-14/-2）：标记账号需重扫，并清除失效 token。""" + """发送会话失效（ret=-14/-2）：标记账号需重扫，清除失效 token，并记录告警日志。""" db = None try: db = SessionLocal() @@ -488,6 +536,14 @@ def _mark_wechat_expired(account_id: int): "UPDATE im_bot_wechat_account SET status='expired', bot_token=NULL, " "updated_at=:ts WHERE id=:aid" ), {"aid": account_id, "ts": _beijing_now()}) + # Session Guard：记录账号信息，供前台提示用户重扫 + row = _execute_raw(db, text("SELECT user_id, username FROM im_bot_wechat_account WHERE id=:aid"), {"aid": account_id}) + _uid = row[0][0] if row else "?" + try: + with open("/tmp/wechat_session_expired.log", "a", encoding="utf-8") as _f: + _f.write(_beijing_now().isoformat() + " SESSION_EXPIRED account_id=%s user_id=%s\n" % (account_id, _uid)) + except Exception: + pass except Exception: pass finally: @@ -530,6 +586,14 @@ def _wechat_receiver_loop(account_id: int): + " ctx=" + repr(ctx) + " raw=" + json.dumps(m, ensure_ascii=False) + "\n") except Exception: pass + # 发送"正在输入"状态（尽力而为，失败静默） + try: + _bot_id = m.get("to_user_id") or d.get("ilink_bot_id") or "" + _ticket = _wechat_get_typing_ticket(baseurl, bot_token, _bot_id, ctx) + if _bot_id and _ticket: + _wechat_send_typing(baseurl, bot_token, _bot_id, _ticket, 1) + except Exception: + pass # 先生成回复（此时历史尚未包含本条，避免重复） import logging as _lrx, inspect as _insp _lrx.warning(f'[receiver] calling _qihua_reply({repr(content)}, {repr(owner_uid)})')