import asyncio import inspect import logging import os import time import weakref from collections import deque from dataclasses import dataclass, field 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]] ReadyFn = Callable[[], Optional[Awaitable[None]]] # These sets live on the FastAPI event-loop thread. They let every account # derive a polling interval from the current population instead of assuming it # is the only hosted account on the server. _active_service_ids: set[int] = set() _ws_service_ids: set[int] = set() def _env_poll_seconds(name: str, default: float, minimum: float = 5.0) -> float: try: return max(minimum, float(os.getenv(name, str(default)))) except (TypeError, ValueError): return default def _conversation_poll_timing( account_id: int, has_ws: bool, population: Optional[int] = None, ) -> tuple[float, float]: """Return the reconciliation interval and a stable per-account stagger. WebSocket is the real-time receive path. HTTP polling is only a safety reconciliation when that path exists, so running it every 15 seconds for hundreds of accounts wastes bandwidth and eventually starves new starts. Accounts without WebSocket keep the original fast polling cadence. """ base_interval = _env_poll_seconds( "KEFU_WS_RECONCILE_INTERVAL_SECONDS" if has_ws else "KEFU_HTTP_POLL_INTERVAL_SECONDS", 120.0 if has_ws else 15.0, ) if population is None: if has_ws: population = len(_ws_service_ids) else: population = len(_active_service_ids - _ws_service_ids) population = max(1, int(population or 0)) # Reserve a bounded request budget for each class. With 500 connected # accounts and the default 1 req/s budget, reconciliation automatically # stretches to 500 seconds instead of permanently saturating the two # shared background-network slots. Disconnected accounts use a separate # budget so they cannot create a 15-second retry storm after an outage. budget_name = "KEFU_WS_POLL_BUDGET_RPS" if has_ws else "KEFU_HTTP_POLL_BUDGET_RPS" try: budget_rps = max(0.05, float(os.getenv(budget_name, "1.0"))) except (TypeError, ValueError): budget_rps = 1.0 interval = max(base_interval, population / budget_rps) spread_ms = max(1, int(interval * 1000)) stagger = ((int(account_id or 0) * 2654435761) % spread_ms) / 1000.0 return interval, stagger def _initial_unread_concurrency() -> int: try: return max( 1, min(8, int(os.getenv("KEFU_INITIAL_UNREAD_CONCURRENCY", "2"))), ) except (TypeError, ValueError): return 2 @dataclass(eq=False) class _InitialUnreadJob: service: "DouyinImService" messages: deque[dict] = field(default_factory=deque) cancelled: bool = False active_task: Optional[asyncio.Task] = None class _InitialUnreadDispatcher: """Process startup unread snapshots with a fixed process-wide worker set. One queued object represents one account and is requeued after each message. This keeps account-local FIFO while preventing a single account from monopolizing both consumers. Only active consumers create handler tasks, so a 500-account start does not create 500 waiting tasks. """ def __init__(self, concurrency: Optional[int] = None) -> None: self.concurrency = max( 1, int(concurrency or _initial_unread_concurrency()), ) self._queue: asyncio.Queue[_InitialUnreadJob] = asyncio.Queue() self._jobs: dict[int, _InitialUnreadJob] = {} self._workers: list[asyncio.Task] = [] self._lock = asyncio.Lock() self._stopping = False async def _ensure_workers(self) -> None: async with self._lock: self._workers = [task for task in self._workers if not task.done()] if self._workers or self._stopping: return for index in range(self.concurrency): self._workers.append( asyncio.create_task( self._worker(index + 1), name=f"initial-unread-worker-{index + 1}", ) ) async def submit( self, service: "DouyinImService", messages: list[dict], ) -> None: pending = [message for message in messages if isinstance(message, dict)] if not pending or not service._running: return await self._ensure_workers() async with self._lock: if self._stopping or not service._running: return service_key = id(service) existing = self._jobs.get(service_key) if existing and not existing.cancelled: existing.messages.extend(pending) return job = _InitialUnreadJob( service=service, messages=deque(pending), ) self._jobs[service_key] = job self._queue.put_nowait(job) logger.debug( "Initial unread queued account=%s count=%s", service.account_id, len(pending), ) async def cancel(self, service: "DouyinImService") -> None: operation: Optional[asyncio.Task] = None current_task = asyncio.current_task() async with self._lock: job = self._jobs.pop(id(service), None) if job is None: return job.cancelled = True job.messages.clear() operation = job.active_task if ( operation and operation is not current_task and not operation.done() ): operation.cancel() if ( operation and operation is not current_task and not operation.done() ): await asyncio.gather(operation, return_exceptions=True) async def _worker(self, worker_number: int) -> None: while True: job = await self._queue.get() operation: Optional[asyncio.Task] = None try: async with self._lock: if ( job.cancelled or not job.service._running or not job.messages ): self._jobs.pop(id(job.service), None) else: message = job.messages.popleft() operation = asyncio.create_task( job.service._handle_incoming(message), name=( "initial-unread-account-" f"{job.service.account_id}" ), ) job.active_task = operation if operation is not None: try: await operation except asyncio.CancelledError: # Cancelling one account cancels only its bounded child # operation; cancelling this worker still shuts it down. if asyncio.current_task().cancelling(): raise except Exception as exc: logger.error( "Initial unread handling failed account=%s " "worker=%s: %s", job.service.account_id, worker_number, exc, ) async with self._lock: if job.active_task is operation: job.active_task = None if ( job.cancelled or not job.service._running or not job.messages ): self._jobs.pop(id(job.service), None) else: # Round-robin across accounts while retaining FIFO # within this account's initial unread snapshot. self._queue.put_nowait(job) except asyncio.CancelledError: if operation and not operation.done(): operation.cancel() await asyncio.gather(operation, return_exceptions=True) raise finally: self._queue.task_done() async def join(self) -> None: await self._queue.join() async def stop(self) -> None: async with self._lock: if self._stopping: workers = list(self._workers) operations = [] else: self._stopping = True workers = list(self._workers) operations = [ job.active_task for job in self._jobs.values() if job.active_task and not job.active_task.done() ] for job in self._jobs.values(): job.cancelled = True job.messages.clear() self._jobs.clear() for operation in operations: operation.cancel() for worker in workers: worker.cancel() if operations: await asyncio.gather(*operations, return_exceptions=True) if workers: await asyncio.gather(*workers, return_exceptions=True) while True: try: self._queue.get_nowait() except asyncio.QueueEmpty: break else: self._queue.task_done() self._workers.clear() _INITIAL_UNREAD_DISPATCHERS = weakref.WeakKeyDictionary() def _get_initial_unread_dispatcher() -> _InitialUnreadDispatcher: loop = asyncio.get_running_loop() dispatcher = _INITIAL_UNREAD_DISPATCHERS.get(loop) if dispatcher is None: dispatcher = _InitialUnreadDispatcher() _INITIAL_UNREAD_DISPATCHERS[loop] = dispatcher return dispatcher async def _shutdown_initial_unread_dispatcher() -> None: loop = asyncio.get_running_loop() dispatcher = _INITIAL_UNREAD_DISPATCHERS.pop(loop, None) if dispatcher is not None: await dispatcher.stop() 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, on_ready: Optional[ReadyFn] = 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._on_ready = on_ready self._ready_notified = False 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 = "" async def _notify_ready(self) -> None: if self._ready_notified: return self._ready_notified = True if not self._on_ready: return result = self._on_ready() if inspect.isawaitable(result): await result 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, immediate_if_idle=True, ) scheduled_wait = 0 if position == 1 else delay_seconds logger.info( "Queued reply to %s for account %s: position=%s wait=%ss interval=%ss", sender, self.account_id, position, scheduled_wait, delay_seconds, ) if position == 1: queue_detail = ( f"{sender} 是当前账号队列的首条任务，等待时间为 0 秒；" f"后续任务仍按 {delay_seconds} 秒间隔排队。" ) else: queue_detail = ( f"{sender} 当前排在第 {position} 位；账号生效间隔为 {delay_seconds} 秒，" "后续任务继续依次排队。" ) system_logger.record( "自动回复已进入账号队列", detail=( f"{queue_detail} 账号内计时与排位独立；" "发送时仍进入全局带宽队列逐条投递。" ), 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], *, enrich_profiles: bool = True, ): 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 ( enrich_profiles and 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, *, initial: bool = False, defer_handlers: bool = False, ) -> list[dict]: controller = get_traffic_controller() async with controller.background_slot( self.account_id, "conversation poll", startup=initial, ): # Do not keep one idle connection pool per hosted account. At 500 # accounts that would retain hundreds of sockets between sparse # reconciliations. Capacity-aware scheduling makes construction # infrequent, while the context manager releases the socket as # soon as this account's poll finishes. async with DouyinImHttpClient( self.session, account_id=self.account_id, ) as http: conversations = await http.get_conversations(enrich_profiles=False) # Capture the previous preview before _index_conversations overwrites # _conv_meta. A conversation-list preview is not inherently a new # message: after startup we only act when it is unread or has actually # changed since the last cache snapshot. This prevents the first # scheduled reconciliation from replying to every historical preview. previous_previews: list[tuple[bool, str]] = [] previous_by_peer = { str(meta.get("peer_uid") or ""): str(meta.get("content") or "") for meta in self._conv_meta.values() if str(meta.get("peer_uid") or "") } for conv in conversations: conv_id = str(conv.get("conversation_id") or "").strip() peer_uid = str( conv.get("peer_uid") or conv.get("sender_id") or "" ).strip() sender_name = str(conv.get("sender_name") or "").strip() prior: Optional[str] = None known = False if conv_id and conv_id in self._conv_meta: known = True prior = str(self._conv_meta[conv_id].get("content") or "") elif sender_name and sender_name in self._conv_previews: known = True prior = str(self._conv_previews.get(sender_name) or "") elif peer_uid and peer_uid in previous_by_peer: known = True prior = previous_by_peer[peer_uid] previous_previews.append((known, prior or "")) # Never enrich every row in a reconciliation snapshot. At 500 accounts # that could turn one list request into tens of thousands of profile # calls. _handle_incoming resolves the peer lazily only for an unread # or genuinely changed conversation after this lightweight cache pass. await self._index_conversations( conversations, enrich_profiles=False, ) unread_total = sum( max(0, int(item.get("unread_count") or 0)) for item in conversations ) if unread_total: logger.info(f"IM unread total: {unread_total}") # Message handling may wait in the global send lane. Do not keep one # of the scarce background HTTP slots occupied while that happens. deferred: list[dict] = [] for conv, (preview_known, previous_preview) in zip( conversations, previous_previews, ): unread = int(conv.get("unread_count") or 0) current_preview = str(conv.get("content") or "") preview_changed = bool( preview_known and current_preview and current_preview != previous_preview ) # Existing previews are useful cache data, but on the authoritative # startup snapshot they are not proof of a newly received message. # The same is true on later polls unless the cached preview changed. if unread > 0 or (not initial and preview_changed): if defer_handlers: deferred.append(conv) else: await self._handle_incoming(conv) return deferred 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", startup=True, ): 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 _active_service_ids.add(self.account_id) 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", startup=True, ): 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", startup=True, ): 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() initial_poll_succeeded = False initial_unread: list[dict] = [] try: initial_unread = await self._poll_conversations( initial=True, defer_handlers=True, ) initial_poll_succeeded = True 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, ) # Batch readiness covers transport discovery plus the authoritative # list/cache snapshot, not potentially slow reply generation, logging, # or outbound sends for pre-existing unread conversations. await self._notify_ready() if initial_unread: await _get_initial_unread_dispatcher().submit( self, initial_unread, ) # A discovered WS URL is enough to start on the low-frequency # reconciliation schedule. If the socket does not actually open, the # first health tick switches the account to fallback mode. This avoids # a second fast HTTP poll just because the handshake needed a moment. ws_connected = bool(has_ws) if ws_connected: _ws_service_ids.add(self.account_id) else: _ws_service_ids.discard(self.account_id) poll_interval, poll_stagger = _conversation_poll_timing( self.account_id, ws_connected, ) loop = asyncio.get_running_loop() if initial_poll_succeeded: initial_retry_interval = poll_interval initial_retry_stagger = poll_stagger else: # If the authoritative first poll failed, retry on the fast HTTP # cadence even when WebSocket connected in the meantime. initial_retry_interval, initial_retry_stagger = ( _conversation_poll_timing(self.account_id, False) ) next_conversation_poll_at = ( loop.time() + initial_retry_interval + initial_retry_stagger ) logger.info( "Conversation reconciliation account=%s interval=%.1fs stagger=%.1fs ws=%s", self.account_id, poll_interval, poll_stagger, "yes" if ws_connected else "no", ) poll_failures = 0 follow_interval = _env_poll_seconds( "KEFU_FOLLOW_POLL_INTERVAL_SECONDS", 60.0, minimum=30.0, ) follow_stagger = ( (int(self.account_id or 0) * 2654435761) % max(1, int(follow_interval * 1000)) ) / 1000.0 next_follow_tick_at = loop.time() + follow_interval + follow_stagger 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 try: current_ws_connected = bool( self._ws_client and getattr(self._ws_client, "connected", False) ) if current_ws_connected != ws_connected: ws_connected = current_ws_connected if ws_connected: _ws_service_ids.add(self.account_id) else: _ws_service_ids.discard(self.account_id) poll_interval, poll_stagger = _conversation_poll_timing( self.account_id, ws_connected, ) candidate_poll_at = loop.time() + poll_interval + poll_stagger # Never postpone an already scheduled reconciliation. # In particular, reconnecting must preserve the earlier # fallback poll that covers messages missed while offline. next_conversation_poll_at = min( next_conversation_poll_at, candidate_poll_at, ) logger.info( "Conversation reconciliation rescheduled account=%s " "interval=%.1fs ws=%s", self.account_id, poll_interval, "yes" if ws_connected else "no", ) else: # Large batches change the active population while older # services are already running. Future polls adopt the # widened capacity interval without moving a poll that is # already scheduled. capacity_interval, _ = _conversation_poll_timing( self.account_id, ws_connected, ) if abs(capacity_interval - poll_interval) >= 1.0: poll_interval = capacity_interval if loop.time() >= next_conversation_poll_at: try: max_poll_seconds = _env_poll_seconds( "KEFU_CONVERSATION_POLL_DEADLINE_SECONDS", 30.0, ) await asyncio.wait_for( self._poll_conversations(), timeout=max_poll_seconds, ) poll_failures = 0 except asyncio.TimeoutError: poll_failures = min(4, poll_failures + 1) logger.debug( "Dropping stale conversation poll account=%s after %.1fs", self.account_id, max_poll_seconds, ) except Exception: poll_failures = min(4, poll_failures + 1) raise finally: # Advance on both success and failure. Otherwise a # past deadline retries every five-second loop tick # during an outage and amplifies traffic. next_conversation_poll_at = ( loop.time() + poll_interval * (2 ** poll_failures) ) # 关注欢迎语：约每 60s 检测一次新粉丝（独立于私信轮询，失败不影响主循环） if self.follow_tick and loop.time() >= next_follow_tick_at: try: await self.follow_tick() except Exception as e: logger.error(f"follow welcome tick error: {e}") finally: next_follow_tick_at = loop.time() + follow_interval 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 _active_service_ids.discard(self.account_id) _ws_service_ids.discard(self.account_id) dispatcher = _INITIAL_UNREAD_DISPATCHERS.get(asyncio.get_running_loop()) if dispatcher is not None: await dispatcher.cancel(self) 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