import asyncio import logging import time from typing import Awaitable, Callable, Optional from utils import system_logger from .frontier import ensure_frontier_ws from .http_client import DouyinImHttpClient, format_session_credential_summary from .session import DouyinImSession from .ws_client import DouyinImWsClient from .reply_queue import AccountReplyQueue from .traffic_control import get_traffic_controller from .reply_payload import format_reply_display, serialize_reply_log from .conv_util import resolve_peer_uid from .peer_profile import ( enrich_conversation_item, fetch_peer_profile, is_generic_peer_name, ) logger = logging.getLogger("douyin_im.service") MatchReplyFn = Callable[[str], Awaitable[Optional[list[str]]]] LogFn = Callable[..., Awaitable[None]] ReceivedLogFn = Callable[..., Awaitable[None]] class DouyinImService: """抖音 IM 直连服务：WebSocket 实时监听 + HTTP 轮询 + 自动回复""" def __init__( self, session: DouyinImSession, match_reply: MatchReplyFn, log_fn: LogFn, account_id: int, received_log_fn: Optional[ReceivedLogFn] = None, reply_delay_seconds: int = 0, reply_delay_resolver: Optional[Callable[[], Awaitable[int]]] = None, reply_cooldown_seconds: Optional[int] = None, cooldown_resolver: Optional[Callable[[], Awaitable[int]]] = None, refresh_credentials: Optional[Callable[[], Awaitable[bool]]] = None, follow_tick: Optional[Callable[[], Awaitable[None]]] = None, on_session_invalid: Optional[Callable[[str], Awaitable[None]]] = None, ): self.session = session self.match_reply = match_reply self.log_fn = log_fn self.received_log_fn = received_log_fn self.account_id = account_id # 由 worker 注入：周期性检测新粉丝并发送关注欢迎语（约每 60s 触发一次） self.follow_tick = follow_tick # 由 worker 注入：检测到 IM 登录失效（INVALID_REQUEST）时回调，用于自动下线 self.on_session_invalid = on_session_invalid self._session_invalid_strikes = 0 self._session_invalid_fired = False self.reply_delay_seconds = max(0, int(reply_delay_seconds or 0)) # 实时解析账号排队间隔：账号专属优先，否则使用系统默认值。 self._reply_delay_resolver = reply_delay_resolver self._reply_queue = AccountReplyQueue( account_id=self.account_id, on_error=self._on_reply_queue_error, ) # WS 帧与 HTTP 轮询会并发进入；按到达顺序串行完成预处理/入队，确保 FIFO。 self._incoming_lock = asyncio.Lock() # 该账号专属冷却秒数；None 表示继承全局系统设置（仅作为无 resolver 时的兜底） self._cooldown_override = ( max(0, int(reply_cooldown_seconds)) if reply_cooldown_seconds is not None else None ) # 实时解析冷却秒数的回调（账号专属优先，否则全局）；优先于 _cooldown_override self._cooldown_resolver = cooldown_resolver # 由 worker 注入：触发后台重新采集 web_protect/keys（刷新 ts_sign），返回是否刷新成功 self.refresh_credentials = refresh_credentials self._running = False self._replied_keys: set[str] = set() self._logged_keys: set[str] = set() self._received_logged_keys: set[str] = set() # 每个对话/用户最近一次自动回复的时间戳（monotonic 秒），用于冷却窗口去重 self._last_reply_at: dict[str, float] = {} self._conv_previews: dict[str, str] = {} self._conv_names: dict[str, str] = {} # uid/conv_id -> nickname self._conv_meta: dict[str, dict] = {} # conversation_id -> meta self._ws_client: Optional[DouyinImWsClient] = None self.last_error: str = "" def _reply_key(self, conversation_key: str, content: str) -> str: return f"{conversation_key}::{content}" @staticmethod def _reply_queue_merge_keys( conversation_id: str, peer_uid: str, ) -> tuple[str, ...]: """Return every stable identifier currently known for one conversation.""" conversation_id = str(conversation_id or "").strip() peer_uid = str(peer_uid or "").strip() aliases: list[str] = [] if conversation_id: aliases.append(f"conv:{conversation_id}") if peer_uid: aliases.append(f"peer:{peer_uid}") return tuple(aliases) @staticmethod def _merge_reply_queue_details( existing: dict, *, incoming_content: str, sender_name: str, sender_id: str, sender_avatar: Optional[str], conversation_id: str, ) -> dict: """Append one received message while preserving the task's one reply.""" merged = dict(existing or {}) contents = merged.get("incoming_contents") if isinstance(contents, list): contents = list(contents) else: contents = [] if not contents and "incoming_content" in merged: contents.append(str(merged.get("incoming_content") or "")) latest_content = str(incoming_content or "") contents.append(latest_content) merged["incoming_content"] = latest_content merged["incoming_contents"] = contents merged["message_count"] = len(contents) if sender_name: merged["sender_name"] = sender_name if sender_id: merged["sender_id"] = sender_id if sender_avatar: merged["sender_avatar"] = sender_avatar if conversation_id: merged["conversation_id"] = conversation_id return merged def _cooldown_seconds_sync(self) -> int: """无 resolver 时的兜底：账号专属优先，否则取全局设置；0 表示关闭。""" if self._cooldown_override is not None: return self._cooldown_override try: from auth.system_settings import get_cached_settings return max(0, int(get_cached_settings().auto_reply_cooldown_seconds or 0)) except Exception: return 0 async def _resolve_cooldown_seconds(self) -> int: """实时解析冷却秒数：优先用 worker 注入的 resolver（账号优先、否则全局），否则兜底。""" if self._cooldown_resolver is not None: try: return max(0, int(await self._cooldown_resolver() or 0)) except Exception as e: logger.debug(f"cooldown resolver failed: {e}") return self._cooldown_seconds_sync() def _reply_delay_seconds_sync(self) -> int: """无 resolver 时解析排队间隔；0 表示不启用排队规则。""" if self.reply_delay_seconds > 0: return self.reply_delay_seconds try: from auth.system_settings import get_cached_settings return max(0, int(get_cached_settings().auto_reply_delay_seconds or 0)) except Exception: return 0 async def _resolve_reply_delay_seconds(self) -> int: """实时解析账号生效的回复排队间隔。""" if self._reply_delay_resolver is not None: try: return max(0, int(await self._reply_delay_resolver() or 0)) except Exception as exc: logger.debug(f"reply delay resolver failed: {exc}") return self._reply_delay_seconds_sync() def _on_reply_queue_error(self, description: str, exc: BaseException) -> None: system_logger.record( "账号回复队列执行失败", detail=f"{description or '自动回复任务'}：{exc}", level="error", category="send", account_id=self.account_id, ) def _peer_in_cooldown(self, peer_key: str, cooldown: int) -> bool: if cooldown <= 0 or not peer_key: return False last = self._last_reply_at.get(peer_key) if last is None: return False return (time.monotonic() - last) < cooldown def _resolve_sender_name(self, msg: dict) -> str: sender_uid = str(msg.get("sender_uid") or msg.get("sender_name") or "").strip() conv_id = str(msg.get("conversation_id") or "") name = (msg.get("sender_name") or "").strip() if name and not name.isdigit(): return name if sender_uid and self._conv_names.get(sender_uid): return self._conv_names[sender_uid] if conv_id and self._conv_names.get(conv_id): return self._conv_names[conv_id] if sender_uid: return f"用户{sender_uid[-6:]}" if len(sender_uid) > 6 else f"用户{sender_uid}" return "未知用户" def _is_self_message(self, msg: dict) -> bool: sender_uid = str(msg.get("sender_uid") or "").strip() if not sender_uid or not self.session.my_uid: return False try: return int(sender_uid) == int(self.session.my_uid) except (TypeError, ValueError): return False async def _resolve_peer_profile( self, conv_id: str, sender_uid: str, sender: str, sender_avatar: str, ) -> tuple[str, str, str]: my_uid = int(self.session.my_uid or 0) peer_uid = str(sender_uid or "").strip() if (not peer_uid or not peer_uid.isdigit()) and conv_id and my_uid: resolved = resolve_peer_uid(conv_id, my_uid) if resolved: peer_uid = str(resolved) meta = self._conv_meta.get(conv_id, {}) if conv_id else {} name = (sender or meta.get("sender_name") or "").strip() avatar = (sender_avatar or meta.get("sender_avatar") or "").strip() if peer_uid and self._conv_names.get(peer_uid): cached_name = self._conv_names[peer_uid] if is_generic_peer_name(name, peer_uid): name = cached_name if conv_id and self._conv_names.get(conv_id) and is_generic_peer_name(name, peer_uid): name = self._conv_names[conv_id] if peer_uid and (is_generic_peer_name(name, peer_uid) or not avatar): profile = await fetch_peer_profile(self.session, peer_uid, self.account_id) if profile.get("nickname"): name = profile["nickname"] self._conv_names[peer_uid] = name if profile.get("avatar_url"): avatar = profile["avatar_url"] if profile.get("uid"): peer_uid = str(profile["uid"]) if not name: name = self._resolve_sender_name( {"sender_uid": peer_uid, "conversation_id": conv_id, "sender_name": sender} ) return name, avatar, peer_uid async def _fetch_message_by_id(self, conv_id: str, server_message_id: str) -> dict | None: """按 server_message_id 调 get_by_conversation 拉取该条消息的完整数据 （含真实 content / message_type / URL）。命中返回原始消息 dict，否则 None。""" if not conv_id or not server_message_id: return None try: from .auth import DouyinAuth controller = get_traffic_controller() async with controller.background_slot(self.account_id, "message detail fetch"): meta = self._conv_meta.get(conv_id, {}) short_id = str(meta.get("conversation_short_id") or "") auth = DouyinAuth.from_im_session(self.session) my_uid = int(self.session.my_uid or 0) async with DouyinImHttpClient(self.session, account_id=self.account_id) as http: if not short_id: peer_uid = resolve_peer_uid(conv_id, my_uid) if peer_uid: _, short_id, _ = await http.get_conversation_info( auth, int(peer_uid), my_uid, conv_id, 0 ) if short_id: self._conv_meta[conv_id] = { **self._conv_meta.get(conv_id, {}), "conversation_short_id": short_id, } messages = await http.get_conversation_messages( auth, conv_id, int(short_id or 0), limit=20 ) for m in messages: if str(m.get("server_message_id") or "") == server_message_id: return m except Exception as e: logger.debug(f"_fetch_message_by_id failed: {e}") return None async def _enrich_media_content(self, conv_id: str, server_message_id: str, content: str) -> str: """媒体消息（相册图片/语音/视频）WS 推送 content 为空时，按 server_message_id 调 get_by_conversation 拉取真实内容并补全 URL。命中失败则原样返回。""" if not conv_id or not server_message_id or not content: return content try: from .message_content import parse_stored_content, format_im_message, serialize_message_content parsed = parse_stored_content(content) mtype = parsed.get("type") if mtype not in ("image", "voice", "video"): return content if parsed.get("url"): return content # 已有 URL（如商店表情/带 url 的图） m = await self._fetch_message_by_id(conv_id, server_message_id) if m: real = format_im_message(m.get("content") or "", int(m.get("message_type") or 0)) if real.get("url"): enriched = serialize_message_content(real) logger.info( "Enriched media via get_by_conversation: smid=%s type=%s", server_message_id, real.get("type"), ) return enriched except Exception as e: logger.debug(f"_enrich_media_content failed: {e}") return content async def _handle_incoming(self, msg: dict): # asyncio.Lock 按等待顺序唤醒。锁只覆盖解析、去重、规则匹配与入队； # 未启用排队时，真正的网络发送仍在锁外执行，保持原有并发行为。 async with self._incoming_lock: immediate_reply = await self._prepare_incoming(msg) if immediate_reply is not None and self._running: await immediate_reply() async def _prepare_incoming( self, msg: dict, ) -> Optional[Callable[[], Awaitable[None]]]: if self._is_self_message(msg): return conv_id = msg.get("conversation_id") or "" sender_uid = str(msg.get("sender_uid") or "") sender = self._resolve_sender_name(msg) sender_avatar = str(msg.get("sender_avatar") or "").strip() sender, sender_avatar, peer_uid = await self._resolve_peer_profile( conv_id, sender_uid, sender, sender_avatar ) content = (msg.get("content") or "").strip() has_raw_ws = "raw_content" in msg raw_incoming = msg.get("raw_content") if has_raw_ws else None ws_message_type = msg.get("message_type") # 每条 WS 消息带唯一 server_message_id：用它去重，避免“同一用户重复发送 # 相同文字（如多次‘你好’）被按内容去重而整条丢弃”，这是“有时收不到”的根因。 # HTTP 轮询的会话预览没有该 ID，则退回按 内容 去重（避免对同一未读重复回复）。 server_message_id = str(msg.get("server_message_id") or "") # WS 仅推送瘦消息（如 type=26）content 为空：按 server_message_id 回 HTTP 拉取 # 完整消息，补全 content / message_type，确保「接收到的全部信息」都被记录。 if not content and server_message_id and not (raw_incoming or "").strip(): real = await self._fetch_message_by_id(conv_id, server_message_id) if real: real_content = (real.get("content") or "").strip() if real_content: raw_incoming = real.get("content") has_raw_ws = True real_type = real.get("message_type") if real_type is not None: ws_message_type = real_type try: from .message_content import format_im_message, serialize_message_content parsed = format_im_message(real.get("content") or "", int(real_type or 0)) content = serialize_message_content(parsed) if parsed else real_content except Exception: content = real_content logger.info( "Enriched empty WS push via get_by_conversation: smid=%s type=%s", server_message_id, real_type, ) # 相册图片/语音等 WS 推送 content 为空，按 server_message_id 拉取真实内容补 URL content = await self._enrich_media_content(conv_id, server_message_id, content) unread = int(msg.get("unread_count") or 0) if conv_id: self._conv_meta[conv_id] = { **self._conv_meta.get(conv_id, {}), "conversation_id": conv_id, "sender_name": sender, "sender_avatar": sender_avatar or self._conv_meta.get(conv_id, {}).get("sender_avatar"), "content": content or self._conv_meta.get(conv_id, {}).get("content", ""), "unread_count": unread, "peer_uid": peer_uid, } if sender and peer_uid: self._conv_names[peer_uid] = sender if not content and unread <= 0 and raw_incoming is None and not server_message_id: return if content == "[未读消息]" and sender in self._conv_previews: content = self._conv_previews.get(sender, content) if server_message_id: log_key = f"mid:{server_message_id}" key = f"mid:{server_message_id}" else: # HTTP 会话预览通常没有 message_id；必须带 conversation_id/peer_uid， # 否则两个同名用户发送相同内容会被误判成同一条消息。 conversation_key = str(conv_id or peer_uid or sender or "unknown") log_key = self._reply_key(conversation_key, content or "[未读]") key = self._reply_key(conversation_key, content) log_kwargs = { "sender_name": sender, "sender_id": peer_uid or conv_id or None, "sender_avatar": sender_avatar or self._conv_meta.get(conv_id, {}).get("sender_avatar"), "message": content or (raw_incoming if raw_incoming is not None else ""), } # 接收消息原始日志：WS content 原样落库（瘦推送已回 HTTP 补全为真实 content） if self.received_log_fn and has_raw_ws: recv_key = f"recv:mid:{server_message_id}" if server_message_id else f"recv:{log_key}" if recv_key not in self._received_logged_keys: self._received_logged_keys.add(recv_key) message_type = ws_message_type try: message_type = int(message_type) if message_type is not None else None except (TypeError, ValueError): message_type = None await self.received_log_fn( sender_name=sender, sender_id=peer_uid or conv_id or None, sender_avatar=log_kwargs.get("sender_avatar"), raw_content="" if raw_incoming is None else raw_incoming, conversation_id=conv_id or None, message_type=message_type, server_message_id=server_message_id or None, ) if log_key not in self._logged_keys and content: self._logged_keys.add(log_key) await self.log_fn( **log_kwargs, reply=None, status="received", ) try: from .message_content import format_system_log_message, parse_stored_content parsed = parse_stored_content(content) msg_type = parsed.get("type") or "text" detail = format_system_log_message(content) if server_message_id: detail = f"{detail} | mid={server_message_id}" system_logger.record( f"收到{'' if msg_type == 'text' else '['+msg_type+']'}消息：{sender}", detail=detail, level="info", category="recv", account_id=self.account_id, ) except Exception as exc: logger.debug(f"record recv system log failed: {exc}") if key in self._replied_keys: return # WS 与 HTTP 轮询可能同时发现同一条消息。检查后立即占位（中间不 await）， # 防止延迟排队期间被重复加入发送队列。 self._replied_keys.add(key) # 同账号、同会话只保留一个尚未发送的回复任务。后续来信只追加到 # 原任务详情，不改变它的发送时间、位置或已经匹配好的回复。 queue_merge_keys = self._reply_queue_merge_keys(conv_id, peer_uid) if queue_merge_keys and self._running: merge_result = await self._reply_queue.merge_pending( queue_merge_keys, lambda existing: self._merge_reply_queue_details( existing, incoming_content=content or "", sender_name=sender, sender_id=peer_uid or conv_id or "", sender_avatar=log_kwargs.get("sender_avatar"), conversation_id=conv_id, ), ) if merge_result.get("status") == "merged": if content: self._conv_previews[sender] = content message_count = int(merge_result.get("message_count") or 1) logger.info( "Merged message into queued reply for %s on account %s: " "job=%s messages=%s position=%s", sender, self.account_id, merge_result.get("job_id"), message_count, merge_result.get("position"), ) system_logger.record( "同一会话消息已合并到回复队列", detail=( f"{sender} 的新消息已并入原任务；当前共 {message_count} 条消息，" "发送时间和队列位置保持不变。" ), level="info", category="send", account_id=self.account_id, ) return # 收到新消息即尝试自动回复，不按消息类型/托管关系/内容形态过滤 replies = await self.match_reply(content if content != "[未读消息]" else "") if not replies: replies = await self.match_reply("") if not replies: await self.log_fn( **log_kwargs, reply=None, status="ignored", error="未配置任何自动回复规则，请在「自动回复规则」中添加至少一条启用规则", ) if content: self._conv_previews[sender] = content return # 冷却窗口：同一用户在设定时间内，无论发多少条消息，只自动回复一次（账号设置优先，否则全局） peer_key = (peer_uid or conv_id or sender or "").strip() cooldown = await self._resolve_cooldown_seconds() if self._peer_in_cooldown(peer_key, cooldown): logger.info( f"Auto-reply to {sender} skipped: within {cooldown}s cooldown window" ) if content: self._conv_previews[sender] = content system_logger.record( "自动回复已跳过（冷却中）", detail=f"{sender} 在 {cooldown} 秒冷却窗口内重复发送，未重复回复", level="info", category="send", account_id=self.account_id, ) return # 提前标记回复时间，确保冷却窗口内（含延迟期间）的后续消息都被抑制 if peer_key and cooldown > 0: self._last_reply_at[peer_key] = time.monotonic() if content: self._conv_previews[sender] = content delay_seconds = await self._resolve_reply_delay_seconds() if not self._running: return async def send_reply() -> None: await self._send_auto_reply( sender=sender, content=content, conv_id=conv_id, replies=replies, peer_key=peer_key, cooldown=cooldown, log_kwargs=log_kwargs, ) if delay_seconds > 0: position = await self._reply_queue.enqueue( delay_seconds, send_reply, description=f"回复 {sender}", details={ "sender_name": sender, "sender_id": peer_uid or conv_id or None, "sender_avatar": log_kwargs.get("sender_avatar"), "conversation_id": conv_id or None, "incoming_content": content or "", "incoming_contents": [content or ""], "message_count": 1, "replies": list(replies), }, merge_keys=queue_merge_keys, ) logger.info( "Queued reply to %s for account %s: position=%s interval=%ss", sender, self.account_id, position, delay_seconds, ) system_logger.record( "自动回复已进入账号队列", detail=( f"{sender} 当前排在第 {position} 位；账号生效间隔为 {delay_seconds} 秒，" "账号内计时与排位独立；到点后再进入全局带宽队列逐条投递。" ), level="info", category="send", account_id=self.account_id, ) return # 账号与系统均未配置排队间隔：跳过排队规则，保持原来的立即回复。 return send_reply async def _send_auto_reply( self, *, sender: str, content: str, conv_id: str, replies: list[str], peer_key: str, cooldown: int, log_kwargs: dict, ) -> None: """发送一项已匹配的自动回复任务，并记录原有消息/系统日志。""" if not self._running: return reply_displays: list[str] = [] sent_any = False send_error = "" meta = self._conv_meta.get(conv_id, {}) for index, reply in enumerate(replies): if not self._running: send_error = self.last_error or "托管已停止，后续回复已取消" break if index > 0: await asyncio.sleep(0.6) if not self._running: send_error = self.last_error or "托管已停止，后续回复已取消" break reply_display = format_reply_display(reply) reply_displays.append(reply_display) sent = False if conv_id: sent, resolved = await self._send_text( conv_id, reply, conversation_short_id=str(meta.get("conversation_short_id") or ""), ) if sent: if resolved: meta = {**meta, **resolved, "conversation_id": conv_id} self._conv_meta[conv_id] = meta else: send_error = self.last_error or "IM API 发送失败" else: send_error = "缺少会话 ID，无法发送自动回复" if sent: sent_any = True combined_display = " | ".join(reply_displays) # 日志里存结构化内容（单条直接存 payload，多条用 {"messages":[...]} 包裹）， # 这样图片/表情等媒体回复会被前端渲染为真实媒体，而不是被压成 "图片" 占位文字。 reply_log_content = serialize_reply_log(replies) if not sent_any: logger.warning( f"IM API send failed for [{sender}]: {send_error}; reply saved to log only" ) # 发送彻底失败：清除冷却时间戳，避免把没收到回复的用户锁在冷却窗口内 if peer_key and cooldown > 0: self._last_reply_at.pop(peer_key, None) await self.log_fn( **log_kwargs, reply=reply_log_content, status="replied" if sent_any else "failed", error=None if sent_any else (send_error or "IM API 发送失败"), ) if sent_any: system_logger.record( "自动回复成功", detail=f"已回复 {sender}：{combined_display}", level="success", category="send", account_id=self.account_id, ) else: system_logger.record( "自动回复失败", detail=f"回复 {sender} 失败：{send_error}（收到：{content}）", level="error", category="send", account_id=self.account_id, ) logger.info( f"Auto-reply to {sender}: {content!r} -> {combined_display!r} " f"(sent={sent_any}, count={len(replies)})" ) async def _index_conversations(self, conversations: list[dict]): my_uid = int(self.session.my_uid or 0) for raw in conversations: conv = enrich_conversation_item(raw, my_uid) conv_id = str(conv.get("conversation_id") or "") name = (conv.get("sender_name") or "").strip() avatar = str(conv.get("sender_avatar") or "").strip() peer_uid = str(conv.get("peer_uid") or "") if peer_uid and (is_generic_peer_name(name, peer_uid) or not avatar): profile = await fetch_peer_profile(self.session, peer_uid, self.account_id) if profile.get("nickname"): name = profile["nickname"] conv["sender_name"] = name if profile.get("avatar_url"): avatar = profile["avatar_url"] conv["sender_avatar"] = avatar if conv_id: self._conv_meta[conv_id] = { **conv, "sender_name": name, "sender_avatar": avatar or None, "peer_uid": peer_uid, } if name: self._conv_names[conv_id] = name if peer_uid and name: self._conv_names[peer_uid] = name async def _poll_conversations(self): controller = get_traffic_controller() async with controller.background_slot(self.account_id, "conversation poll"): async with DouyinImHttpClient(self.session, account_id=self.account_id) as http: unread_total = await http.get_unread_count() if unread_total: logger.info(f"IM unread total: {unread_total}") conversations = await http.get_conversations() await self._index_conversations(conversations) # Message handling may wait in the global send lane. Do not keep one # of the scarce background HTTP slots occupied while that happens. for conv in conversations: unread = int(conv.get("unread_count") or 0) if unread > 0 or conv.get("content"): await self._handle_incoming(conv) async def _verify_account_uid(self): """启动时用 query/user 接口核验账号真实 UID，修正采集端可能取错的 my_uid/device_id。 采集端从 tea_cache 推断的 my_uid 可能是访客/对方 id，会导致会话列表为 0、 创建会话 INVALID_REQUEST。这里在建连前先校正，保证后续所有请求身份正确。 """ if getattr(self.session, "uid_verified", False) and self.session.my_uid: return try: from .auth import DouyinAuth auth = DouyinAuth.from_im_session(self.session) controller = get_traffic_controller() async with controller.background_slot(self.account_id, "account UID verify"): async with DouyinImHttpClient(self.session, account_id=self.account_id) as http: old = int(self.session.my_uid or 0) resolved = await asyncio.to_thread(http._resolve_authoritative_uid, auth) if resolved and old and int(resolved) != old: system_logger.record( "已自动校正账号 UID", detail=f"采集端识别 UID={old}，接口核验真实 UID={resolved}，已修正后再建立私信连接。", level="info", category="system", account_id=self.account_id, ) except Exception as e: logger.warning(f"启动核验账号 UID 失败（沿用采集值）：{e}") async def run(self): """主循环：WebSocket + HTTP 轮询""" self._running = True await self._reply_queue.start() await self._verify_account_uid() # ensure_frontier_ws 可能触发签名/HTTP（阻塞），放线程池避免多账号启动时卡死事件循环 controller = get_traffic_controller() async with controller.background_slot(self.account_id, "frontier discovery"): await asyncio.to_thread(ensure_frontier_ws, self.session) has_ws = bool(self.session.frontier_ws_url()) cred_summary = format_session_credential_summary(self.session) logger.info(cred_summary) logger.info( f"Starting IM direct service for account {self.account_id} " f"(ws={'yes' if has_ws else 'no'})" ) system_logger.record( "私信托管已启动", detail=f"实时接收通道：{'已就绪' if has_ws else '不可用（仅 HTTP 轮询）'}\n{cred_summary}", level="success" if has_ws else "warning", category="system", account_id=self.account_id, ) try: from .emoji_pack import ensure_emoji_map, is_fresh if not is_fresh(): async with controller.background_slot(self.account_id, "emoji preload"): await asyncio.to_thread(ensure_emoji_map, self.session) except Exception as e: logger.debug(f"emoji map preload failed: {e}") self._ws_client = DouyinImWsClient( self.session, self._handle_incoming, account_id=self.account_id ) await self._ws_client.start() try: await self._poll_conversations() except Exception as e: logger.warning(f"Initial conversation poll failed: {e}") system_logger.record( "首次会话轮询失败", detail=f"{e}", level="warning", category="poll", account_id=self.account_id, ) loop_count = 0 while self._running: # The initial poll above is authoritative. Sleep before the next # recurring tick so startup cannot issue two back-to-back polls. await asyncio.sleep(5) if not self._running: break loop_count += 1 try: if loop_count % 3 == 0: await self._poll_conversations() if loop_count % 6 == 0: logger.info(f"IM direct tick #{loop_count} account={self.account_id}") # 关注欢迎语：约每 60s 检测一次新粉丝（独立于私信轮询，失败不影响主循环） if self.follow_tick and loop_count % 12 == 0: try: await self.follow_tick() except Exception as e: logger.error(f"follow welcome tick error: {e}") except Exception as e: logger.error(f"IM poll error: {e}") system_logger.record( "会话轮询出错", detail=f"拉取会话/未读时出错：{e}", level="error", category="poll", account_id=self.account_id, ) async def stop(self): self._running = False await get_traffic_controller().send_queue.cancel_account(self.account_id) await self._reply_queue.stop() if self._ws_client: await self._ws_client.stop() def get_cached_conversations(self) -> list[dict]: """返回运行中缓存的会话（来自 WS / 轮询）。""" results = [] seen = set() for conv_id, meta in self._conv_meta.items(): name = (meta.get("sender_name") or "").strip() key = conv_id or name if not key or key in seen: continue seen.add(key) results.append({ "conversation_id": conv_id, "sender_name": name or f"会话{conv_id[-8:]}" if conv_id else "未知用户", "sender_avatar": meta.get("sender_avatar") or None, "sender_id": str(meta.get("peer_uid") or meta.get("sender_id") or conv_id or ""), "peer_uid": str(meta.get("peer_uid") or ""), "content": str(meta.get("content") or ""), "unread_count": int(meta.get("unread_count") or 0), }) return results async def get_reply_queue_snapshot(self) -> list[dict]: """返回当前账号自动回复队列的可管理快照。""" return await self._reply_queue.snapshot() async def send_queued_reply_now(self, job_id: str) -> dict: """把指定自动回复任务移入账号紧急队列；实际发送仍由单消费者串行执行。""" return await self._reply_queue.send_now(job_id) async def _send_text( self, conversation_id: str, content: str, conversation_short_id: str = "", ) -> tuple[bool, Optional[dict]]: """发送一条私信；若因签名凭证失效(7911)失败，刷新 web_protect 后自动重试一次。 返回 (是否成功, 解析到的会话 meta)。失败原因写入 self.last_error。 """ for attempt in range(2): async with DouyinImHttpClient(self.session, account_id=self.account_id) as http: sent = await http.send_text_message( conversation_id, content, conversation_short_id=conversation_short_id, ) self.last_error = http.last_error needs_refresh = http.last_send_needs_refresh if sent: resolved = http.last_send_meta.get(conversation_id) self.session.conv_meta.update(http.session.conv_meta) self._session_invalid_strikes = 0 # 发送成功 → 登录有效 return True, resolved # 仅在“签名凭证失效”时刷新并重试一次 if attempt == 0 and needs_refresh and self.refresh_credentials: logger.warning( f"Send hit credential-expiry(7911) for {conversation_id}; " "refreshing web_protect and retrying once..." ) try: refreshed = await self.refresh_credentials() except Exception as e: logger.warning(f"refresh_credentials raised: {e}") refreshed = False if refreshed: continue break await self._note_session_invalid(self.last_error) return False, None async def _note_session_invalid(self, error: str) -> None: """根据发送失败原因判断 IM 是否已退出登录；连续 INVALID_REQUEST 即触发自动下线。 INVALID_REQUEST 来自 create_conversation/发送：会话/签名被抖音判为无效，强相关于「登录失效」。 而 8xxx/7xxx 等业务错误（关系/频控/内容）说明请求已到达抖音、登录仍有效，重置计数。 """ err = error or "" if "INVALID_REQUEST" not in err: self._session_invalid_strikes = 0 return self._session_invalid_strikes += 1 if self._session_invalid_strikes < 2 or self._session_invalid_fired: return self._session_invalid_fired = True reason = "IM 会话失效(INVALID_REQUEST)，登录可能已退出" logger.warning( f"Account {self.account_id} {reason}；连续 {self._session_invalid_strikes} 次 -> 自动下线" ) system_logger.record( "IM 登录失效，自动下线", detail=f"{reason}（连续 {self._session_invalid_strikes} 次发送返回 INVALID_REQUEST）。" "请停止托管后用浏览器模式重新登录并打开私信页，再重新启动托管。", level="error", category="auth", account_id=self.account_id, ) self._running = False # 让主循环尽快退出 if self.on_session_invalid: try: await self.on_session_invalid(reason) except Exception as e: logger.error(f"on_session_invalid handler error: {e}") async def send_message(self, conversation_id: str, content: str) -> bool: """手动发送私信""" from .conv_util import normalize_conversation_id from .auth import DouyinAuth auth = DouyinAuth() auth.perepare_auth( self.session.cookie_header(), self.session.web_protect_str, self.session.keys_str, ) if getattr(self.session, "uid_verified", False) and self.session.my_uid: my_uid = self.session.my_uid else: my_uid = await asyncio.to_thread(lambda: auth.get_uid()) or self.session.my_uid if my_uid: conversation_id = normalize_conversation_id(conversation_id, my_uid) meta = self._conv_meta.get(conversation_id, {}) sent, resolved = await self._send_text( conversation_id, content, conversation_short_id=str(meta.get("conversation_short_id") or ""), ) if sent and resolved: self._conv_meta[conversation_id] = { **meta, **resolved, "conversation_id": conversation_id, } return sent