522 lines
21 KiB
Python
522 lines
21 KiB
Python
import asyncio
|
||
import logging
|
||
import os
|
||
import random
|
||
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_payload import format_reply_display, serialize_reply_log
|
||
|
||
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_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
|
||
self.follow_tick = follow_tick
|
||
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._cooldown_override = (
|
||
max(0, int(reply_cooldown_seconds)) if reply_cooldown_seconds is not None else None
|
||
)
|
||
self._cooldown_resolver = cooldown_resolver
|
||
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()
|
||
self._last_reply_at: dict[str, float] = {}
|
||
self._conv_previews: dict[str, str] = {}
|
||
self._conv_names: dict[str, str] = {}
|
||
self._conv_meta: dict[str, dict] = {}
|
||
self._ws_client: Optional[DouyinImWsClient] = None
|
||
self.last_error: str = ""
|
||
self.last_send_risk_notice: str = ""
|
||
|
||
def _reply_key(self, sender: str, content: str) -> str:
|
||
return f"{sender}::{content}"
|
||
|
||
def _cooldown_seconds_sync(self) -> int:
|
||
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:
|
||
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 _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 _handle_incoming(self, msg: dict):
|
||
if self._is_self_message(msg):
|
||
return
|
||
|
||
sender = self._resolve_sender_name(msg)
|
||
content = (msg.get("content") or "").strip()
|
||
unread = int(msg.get("unread_count") or 0)
|
||
conv_id = msg.get("conversation_id") or ""
|
||
sender_uid = str(msg.get("sender_uid") or "")
|
||
sender_avatar = str(msg.get("sender_avatar") or "").strip()
|
||
# 每条 WS 消息带唯一 server_message_id：用它去重，避免“同一用户重复发送
|
||
# 相同文字（如多次‘你好’）被按内容去重而整条丢弃”，这是“有时收不到”的根因。
|
||
# HTTP 轮询的会话预览没有该 ID，则退回按 内容 去重（避免对同一未读重复回复）。
|
||
server_message_id = str(msg.get("server_message_id") or "")
|
||
|
||
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": sender_uid,
|
||
}
|
||
if sender and sender_uid:
|
||
self._conv_names[sender_uid] = sender
|
||
|
||
if not content and unread <= 0:
|
||
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:
|
||
log_key = self._reply_key(sender, content or "[未读]")
|
||
key = self._reply_key(sender, content)
|
||
|
||
log_kwargs = {
|
||
"sender_name": sender,
|
||
"sender_id": conv_id or sender_uid or None,
|
||
"sender_avatar": sender_avatar or self._conv_meta.get(conv_id, {}).get("sender_avatar"),
|
||
"message": content,
|
||
}
|
||
|
||
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",
|
||
)
|
||
|
||
if key in self._replied_keys:
|
||
return
|
||
|
||
prev = self._conv_previews.get(sender)
|
||
should_reply = unread > 0 or (prev and content != prev) or (content and content != "[未读消息]")
|
||
if not should_reply:
|
||
if content:
|
||
self._conv_previews[sender] = content
|
||
return
|
||
|
||
from .hosted_registry import is_hosted
|
||
|
||
if sender_uid and is_hosted(sender_uid):
|
||
# 对方也是本系统托管的账号：若双方都自动回复会形成无限回环，
|
||
# 高频来回发送极易触发抖音风控（7911）/业务拒绝（8004），故直接跳过。
|
||
await self.log_fn(
|
||
**log_kwargs,
|
||
reply=None,
|
||
status="ignored",
|
||
error="对方是本系统托管的另一账号，已跳过自动回复以避免互相回复触发风控",
|
||
)
|
||
self._replied_keys.add(key)
|
||
if content:
|
||
self._conv_previews[sender] = content
|
||
logger.info(
|
||
f"Skip auto-reply to hosted account {sender_uid} (avoid reply loop)"
|
||
)
|
||
return
|
||
|
||
from .protocol import should_skip_auto_reply
|
||
|
||
skip, skip_reason = should_skip_auto_reply(content)
|
||
if skip:
|
||
await self.log_fn(
|
||
**log_kwargs,
|
||
reply=None,
|
||
status="ignored",
|
||
error=skip_reason,
|
||
)
|
||
self._replied_keys.add(key)
|
||
if content:
|
||
self._conv_previews[sender] = content
|
||
logger.info(f"Skip auto-reply to {sender}: {content!r} ({skip_reason})")
|
||
return
|
||
|
||
replies = await self.match_reply(content if content != "[未读消息]" else "")
|
||
if not replies:
|
||
await self.log_fn(
|
||
**log_kwargs,
|
||
reply=None,
|
||
status="ignored",
|
||
error="未配置任何自动回复规则",
|
||
)
|
||
self._replied_keys.add(key)
|
||
if content:
|
||
self._conv_previews[sender] = content
|
||
return
|
||
|
||
peer_key = (sender_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")
|
||
self._replied_keys.add(key)
|
||
if content:
|
||
self._conv_previews[sender] = content
|
||
return
|
||
if peer_key and cooldown > 0:
|
||
self._last_reply_at[peer_key] = time.monotonic()
|
||
|
||
if self.reply_delay_seconds > 0:
|
||
logger.info(f"Delaying reply to {sender} for {self.reply_delay_seconds}s")
|
||
await asyncio.sleep(self.reply_delay_seconds)
|
||
if not self._running:
|
||
return
|
||
|
||
reply_displays = []
|
||
sent_any = False
|
||
send_error = ""
|
||
failed_parts: list[str] = []
|
||
meta = self._conv_meta.get(conv_id, {})
|
||
for index, reply in enumerate(replies):
|
||
if index > 0:
|
||
await asyncio.sleep(0.6)
|
||
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:
|
||
part_err = self.last_error or "IM API 发送失败"
|
||
send_error = part_err
|
||
failed_parts.append(f"{reply_display}: {part_err}")
|
||
else:
|
||
send_error = "缺少会话 ID，无法发送自动回复"
|
||
failed_parts.append(f"{reply_display}: {send_error}")
|
||
if sent:
|
||
sent_any = True
|
||
|
||
combined_display = " | ".join(reply_displays)
|
||
reply_log_content = serialize_reply_log(replies)
|
||
partial = sent_any and bool(failed_parts)
|
||
if partial:
|
||
send_error = "；".join(failed_parts)
|
||
if not sent_any:
|
||
logger.warning(
|
||
f"IM API send failed for [{sender}]: {send_error}"
|
||
)
|
||
if peer_key and cooldown > 0:
|
||
self._last_reply_at.pop(peer_key, None)
|
||
|
||
self._replied_keys.add(key)
|
||
if content:
|
||
self._conv_previews[sender] = content
|
||
|
||
await self.log_fn(
|
||
**log_kwargs,
|
||
reply=reply_log_content,
|
||
status="replied" if sent_any and not partial else ("partial" if partial else "failed"),
|
||
error=None if sent_any and not partial else (send_error or "IM API 发送失败"),
|
||
)
|
||
if sent_any and not partial:
|
||
system_logger.record(
|
||
"自动回复成功",
|
||
detail=f"已回复 {sender}：{combined_display}",
|
||
level="success",
|
||
category="send",
|
||
account_id=self.account_id,
|
||
)
|
||
elif partial:
|
||
system_logger.record(
|
||
"自动回复部分失败",
|
||
detail=f"回复 {sender} 部分成功：{combined_display}；失败：{send_error}",
|
||
level="warning",
|
||
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}, partial={partial}, msgs={len(replies)})"
|
||
)
|
||
|
||
def _index_conversations(self, conversations: list[dict]):
|
||
for conv in conversations:
|
||
conv_id = str(conv.get("conversation_id") or "")
|
||
name = (conv.get("sender_name") or "").strip()
|
||
if conv_id:
|
||
self._conv_meta[conv_id] = conv
|
||
if name:
|
||
self._conv_names[conv_id] = name
|
||
peer_uid = str(conv.get("peer_uid") or "")
|
||
if peer_uid and name:
|
||
self._conv_names[peer_uid] = name
|
||
|
||
async def _poll_conversations(self):
|
||
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()
|
||
self._index_conversations(conversations)
|
||
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 run(self):
|
||
"""主循环：WebSocket + HTTP 轮询"""
|
||
self._running = True
|
||
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,
|
||
)
|
||
|
||
self._ws_client = DouyinImWsClient(
|
||
self.session, self._handle_incoming, account_id=self.account_id
|
||
)
|
||
await self._ws_client.start()
|
||
|
||
# 轮询错峰：多账号同时托管时，若所有账号按同一节奏轮询，请求会在同一
|
||
# 时刻叠峰。这里给每个账号随机相位偏移 + 每轮 ±20% 抖动，把请求摊平。
|
||
# WS 可用时轮询只是兜底，可以适当放缓（KEFU_IM_POLL_INTERVAL_SECONDS 可调）。
|
||
try:
|
||
poll_interval = float(os.getenv("KEFU_IM_POLL_INTERVAL_SECONDS", "") or 15)
|
||
except ValueError:
|
||
poll_interval = 15.0
|
||
poll_interval = max(5.0, poll_interval)
|
||
if has_ws:
|
||
poll_interval = max(poll_interval, 30.0)
|
||
|
||
# 首轮轮询前的随机延迟（相位偏移），批量启动时错开各账号的首波请求
|
||
await asyncio.sleep(random.uniform(0.5, min(10.0, poll_interval)))
|
||
if not self._running:
|
||
return
|
||
|
||
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
|
||
next_poll = time.monotonic() + poll_interval * random.uniform(0.8, 1.2)
|
||
while self._running:
|
||
loop_count += 1
|
||
try:
|
||
if time.monotonic() >= next_poll:
|
||
next_poll = time.monotonic() + poll_interval * random.uniform(0.8, 1.2)
|
||
await self._poll_conversations()
|
||
if loop_count % 6 == 1:
|
||
logger.info(f"IM direct tick #{loop_count} account={self.account_id}")
|
||
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,
|
||
)
|
||
await asyncio.sleep(5)
|
||
|
||
async def stop(self):
|
||
self._running = False
|
||
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,
|
||
"content": str(meta.get("content") or ""),
|
||
"unread_count": int(meta.get("unread_count") or 0),
|
||
})
|
||
return results
|
||
|
||
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。
|
||
"""
|
||
self.last_send_risk_notice = ""
|
||
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
|
||
self.last_send_risk_notice = http.last_send_risk_notice
|
||
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)
|
||
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
|
||
return False, None
|
||
|
||
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,
|
||
)
|
||
my_uid = 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
|