456 lines
15 KiB
Python
456 lines
15 KiB
Python
"""
|
||
微盛群消息一轮：按「询价机器人」席位拉新话，写入 inbox。
|
||
|
||
本文件职责：换票查询、归一、入 inbox、推进测服 last_seq。
|
||
Worker 独立线程轮询，不占 HTTP 槽、不堵回调。入队后由 inbox 线程分发。
|
||
禁止：堵回调；禁止推现网官方 SDK 游标；禁止把存档席当发言人。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import time
|
||
from typing import Any, Callable, Optional, Sequence
|
||
|
||
from agent.channel.queue import get_message_store
|
||
from agent.channel.wshoto.client import WshotoError, WshotoOpenClient, wshoto_ready
|
||
from agent.channel.wshoto.parse import archive_record_to_inbound
|
||
from agent.channel.wshoto.rooms import (
|
||
advance_poll_cursor,
|
||
is_room_hot,
|
||
list_watched_rooms,
|
||
load_last_seq,
|
||
mark_room_hot,
|
||
plan_rooms_for_poll,
|
||
save_last_seq,
|
||
)
|
||
from agent.config import Settings, get_settings
|
||
from agent.redis_coord.client import RedisClient
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 新群还没有游标时，回看窗口必须盖过「存档入库延迟 + 冷群轮转」。
|
||
# 以前 180 秒：存档晚几分钟到、或本轮预算扫不到该群时，人话永久漏掉，last_seq 也永远建不起来。
|
||
_NEW_ROOM_LOOKBACK_SEC = 3600
|
||
# 微盛「请求频率过高」：本轮立刻停，下一轮再扫，避免把活跃群也拖死。
|
||
_RATE_LIMIT_CODE = 10061
|
||
|
||
|
||
def archive_seat_userids(settings: Optional[Settings] = None) -> list[str]:
|
||
"""
|
||
会话存档账号 userid。
|
||
|
||
测服群里的「询价机器人」就是这一席；默认读 WECOM_ARCHIVE_SEAT_USER_IDS（AI001）。
|
||
"""
|
||
cfg = settings or get_settings()
|
||
raw = str(getattr(cfg, "wecom_archive_seat_user_ids", "") or "").strip()
|
||
ids: list[str] = []
|
||
for part in raw.replace(";", ",").split(","):
|
||
uid = part.strip()
|
||
if uid and uid not in ids:
|
||
ids.append(uid)
|
||
return ids
|
||
|
||
|
||
def _record_seq(record: dict[str, Any]) -> int:
|
||
try:
|
||
return int(record.get("seq") or 0)
|
||
except (TypeError, ValueError):
|
||
return 0
|
||
|
||
|
||
def ingest_records(
|
||
records: Sequence[dict[str, Any]],
|
||
*,
|
||
seat_userids: Sequence[str],
|
||
min_seq: int = 0,
|
||
store: Any = None,
|
||
) -> tuple[int, int]:
|
||
"""
|
||
把一页管家记录写入 inbox。
|
||
|
||
返回：(入队成功数, 跳过数)。无 sender / 存档席 / 重复 message_id 都算跳过。
|
||
"""
|
||
book = store if store is not None else get_message_store()
|
||
enqueued = 0
|
||
skipped = 0
|
||
for rec in records:
|
||
if _record_seq(rec) <= min_seq:
|
||
skipped += 1
|
||
continue
|
||
msg = archive_record_to_inbound(rec, seat_userids=seat_userids)
|
||
if msg is None:
|
||
skipped += 1
|
||
continue
|
||
ok, reason = book.enqueue_inbound(msg)
|
||
if ok:
|
||
enqueued += 1
|
||
mark_room_hot(msg.chat_id)
|
||
logger.info(
|
||
"存档入队 sender=%s chat=%s type=%s reason=%s",
|
||
msg.sender_id,
|
||
msg.chat_id,
|
||
msg.msg_type,
|
||
reason,
|
||
)
|
||
else:
|
||
skipped += 1
|
||
return enqueued, skipped
|
||
|
||
|
||
def _needs_target(exc: WshotoError) -> bool:
|
||
text = (exc.msg or "").lower()
|
||
return "target_id" in text or "targetid" in text or "房间" in (exc.msg or "")
|
||
|
||
|
||
def poll_once(
|
||
*,
|
||
client: Optional[WshotoOpenClient] = None,
|
||
settings: Optional[Settings] = None,
|
||
redis: Optional[RedisClient] = None,
|
||
store: Any = None,
|
||
now: Optional[Callable[[], float]] = None,
|
||
) -> dict[str, Any]:
|
||
"""
|
||
拉一页并存入 inbox。
|
||
|
||
已有 last_seq：按 seq 增量查。没有：只看最近若干小时，避免从 0 翻全公司历史。
|
||
已登记协同群按房间号拉（不要带存档席 userid，应用群否则给空页）。
|
||
还没有群时才按席位试拉。
|
||
"""
|
||
cfg = settings or get_settings()
|
||
if not wshoto_ready(cfg):
|
||
return {"ok": False, "reason": "not_configured"}
|
||
seats = archive_seat_userids(cfg)
|
||
if not seats:
|
||
logger.warning("微盛轮询：未配置 WECOM_ARCHIVE_SEAT_USER_IDS，只能靠显示名丢掉询价机器人自己的话")
|
||
api = client or WshotoOpenClient(cfg, redis=redis)
|
||
lookback = max(60, int(getattr(cfg, "wshoto_bootstrap_lookback_sec", 10800) or 10800))
|
||
page_size = int(getattr(cfg, "wshoto_page_size", 50) or 50)
|
||
clock = now or time.time
|
||
rooms = list_watched_rooms(redis=redis)
|
||
scan_ids = [
|
||
x.strip()
|
||
for x in str(getattr(cfg, "wshoto_scan_user_ids", "") or "").replace(";", ",").split(",")
|
||
if x.strip()
|
||
]
|
||
extra: list[dict[str, Any]] = []
|
||
if scan_ids:
|
||
extra = _poll_by_scan_users(
|
||
api,
|
||
userids=scan_ids,
|
||
redis=redis,
|
||
page_size=page_size,
|
||
lookback=lookback,
|
||
clock=clock,
|
||
)
|
||
if rooms:
|
||
# 应用群必须带房间号、不要带存档席 userid，否则管家返回空页。
|
||
records, mode, last = _poll_by_rooms(
|
||
api,
|
||
redis=redis,
|
||
page_size=page_size,
|
||
lookback=lookback,
|
||
clock=clock,
|
||
settings=cfg,
|
||
)
|
||
if extra:
|
||
records = list(records) + extra
|
||
else:
|
||
seat = seats[0] if seats else ""
|
||
last = load_last_seq(f"seat:{seat or 'default'}", redis=redis)
|
||
time_start = 0
|
||
time_end = 0
|
||
if last <= 0:
|
||
time_end = int(clock())
|
||
time_start = time_end - lookback
|
||
try:
|
||
records = api.list_group_messages(
|
||
userid=seat,
|
||
last_seq=last,
|
||
page_size=page_size,
|
||
msg_time_start=time_start,
|
||
msg_time_end=time_end,
|
||
)
|
||
mode = "seat"
|
||
except WshotoError as exc:
|
||
if not _needs_target(exc):
|
||
logger.warning("微盛按席拉取失败 code=%s msg=%s", exc.code, exc.msg)
|
||
return {"ok": False, "reason": "list_failed", "code": exc.code}
|
||
records, mode, last = _poll_by_rooms(
|
||
api,
|
||
redis=redis,
|
||
page_size=page_size,
|
||
lookback=lookback,
|
||
clock=clock,
|
||
settings=cfg,
|
||
)
|
||
if extra:
|
||
records = list(records) + extra
|
||
mode = "rooms"
|
||
# 按房间拉时各群自己的 last_seq 已经过滤过，这里不要再用席游标误丢。
|
||
ingest_min = 0 if mode == "rooms" else last
|
||
enqueued, skipped = ingest_records(
|
||
records,
|
||
seat_userids=seats,
|
||
min_seq=ingest_min,
|
||
store=store,
|
||
)
|
||
max_seq = last
|
||
for rec in records:
|
||
seq = _record_seq(rec)
|
||
if seq > max_seq:
|
||
max_seq = seq
|
||
if max_seq > last and not rooms:
|
||
save_last_seq(f"seat:{(seats[0] if seats else 'default')}", max_seq, redis=redis)
|
||
logger.info(
|
||
"微盛轮询 mode=%s fetched=%s enqueued=%s skipped=%s last_seq=%s→%s",
|
||
mode,
|
||
len(records),
|
||
enqueued,
|
||
skipped,
|
||
last,
|
||
max_seq,
|
||
)
|
||
return {
|
||
"ok": True,
|
||
"mode": mode,
|
||
"fetched": len(records),
|
||
"enqueued": enqueued,
|
||
"skipped": skipped,
|
||
"last_seq": max_seq,
|
||
}
|
||
|
||
|
||
def _poll_by_scan_users(
|
||
api: WshotoOpenClient,
|
||
*,
|
||
userids: Sequence[str],
|
||
redis: Optional[RedisClient],
|
||
page_size: int,
|
||
lookback: int,
|
||
clock: Callable[[], float],
|
||
) -> list[dict[str, Any]]:
|
||
"""
|
||
按人扫新群。空运群是人自建的，存档席没进群时房间名单里没有它。
|
||
|
||
副作用：看到带 @机器人显示名（含旧名）的新房间会登记监听。
|
||
"""
|
||
from agent.channel.wshoto.rooms import watch_collab_room
|
||
from agent.policy.bot_name import text_mentions_bot
|
||
|
||
merged: list[dict[str, Any]] = []
|
||
now = int(clock())
|
||
for uid in userids:
|
||
scope = f"scan:{uid}"
|
||
last = load_last_seq(scope, redis=redis)
|
||
time_start = 0
|
||
time_end = 0
|
||
if last <= 0:
|
||
time_end = now
|
||
time_start = time_end - min(lookback, 8 * 3600)
|
||
try:
|
||
rows = api.list_group_messages(
|
||
userid=uid,
|
||
last_seq=last,
|
||
page_size=page_size,
|
||
msg_time_start=time_start,
|
||
msg_time_end=time_end,
|
||
)
|
||
except WshotoError as exc:
|
||
logger.warning("微盛按人扫群失败 userid=%s code=%s", uid, exc.code)
|
||
continue
|
||
max_seq = last
|
||
for rec in rows:
|
||
seq = _record_seq(rec)
|
||
if seq > max_seq:
|
||
max_seq = seq
|
||
room = str(rec.get("room_id") or rec.get("roomid") or "").strip()
|
||
blob = str(rec.get("content") or "")
|
||
if room and text_mentions_bot(blob):
|
||
try:
|
||
watch_collab_room(room, redis=redis)
|
||
except Exception:
|
||
logger.exception("扫到空运群后登记监听失败 chat=%s", room)
|
||
if max_seq > last:
|
||
save_last_seq(scope, max_seq, redis=redis)
|
||
merged.extend(rows)
|
||
return merged
|
||
|
||
|
||
def _merge_records_by_seq(
|
||
*batches: Sequence[dict[str, Any]],
|
||
min_seq: int = 0,
|
||
) -> list[dict[str, Any]]:
|
||
"""按 seq 去重合并多页；同 seq 后写覆盖。只保留 seq > min_seq。"""
|
||
by_seq: dict[int, dict[str, Any]] = {}
|
||
for batch in batches:
|
||
for rec in batch or ():
|
||
seq = _record_seq(rec)
|
||
if seq <= min_seq:
|
||
continue
|
||
by_seq[seq] = dict(rec)
|
||
return [by_seq[s] for s in sorted(by_seq)]
|
||
|
||
|
||
def _backfill_room_seq_gap(
|
||
api: WshotoOpenClient,
|
||
*,
|
||
room: str,
|
||
last_seq: int,
|
||
rows: list[dict[str, Any]],
|
||
page_size: int,
|
||
lookback: int,
|
||
clock: Callable[[], float],
|
||
) -> list[dict[str, Any]]:
|
||
"""
|
||
增量结果跳号时，用时间窗补中间漏掉的消息。
|
||
|
||
现网：游标 9892，管家增量只回撤回 9899，把「解散群」9898 永久跨过。
|
||
若本页最小 seq > last_seq+1，再打一枪时间窗，按 seq 合并后返回。
|
||
补洞失败则原样返回（调用方仍按本页 max 推进，故补洞要尽量成功）。
|
||
"""
|
||
if last_seq <= 0 or not rows:
|
||
return rows
|
||
page_seqs = [_record_seq(r) for r in rows if _record_seq(r) > 0]
|
||
if not page_seqs:
|
||
return rows
|
||
min_page = min(page_seqs)
|
||
if min_page <= last_seq + 1:
|
||
return rows
|
||
now_i = int(clock())
|
||
window = min(max(60, int(lookback or 0)), _NEW_ROOM_LOOKBACK_SEC)
|
||
logger.warning(
|
||
"微盛房间增量跳号 chat=%s last=%s min_page=%s，时间窗补洞 window=%ss",
|
||
room,
|
||
last_seq,
|
||
min_page,
|
||
window,
|
||
)
|
||
try:
|
||
fill = api.list_group_messages(
|
||
userid="",
|
||
target_id=room,
|
||
last_seq=0,
|
||
page_size=page_size,
|
||
msg_time_start=now_i - window,
|
||
msg_time_end=now_i,
|
||
)
|
||
except WshotoError as exc:
|
||
logger.warning(
|
||
"微盛房间补洞失败 chat=%s code=%s msg=%s",
|
||
room,
|
||
exc.code,
|
||
exc.msg,
|
||
)
|
||
return rows
|
||
merged = _merge_records_by_seq(rows, fill or [], min_seq=last_seq)
|
||
if len(merged) > len(rows):
|
||
logger.info(
|
||
"微盛房间补洞成功 chat=%s added=%s total=%s",
|
||
room,
|
||
len(merged) - len(rows),
|
||
len(merged),
|
||
)
|
||
return merged
|
||
|
||
|
||
def _poll_by_rooms(
|
||
api: WshotoOpenClient,
|
||
*,
|
||
redis: Optional[RedisClient],
|
||
page_size: int,
|
||
lookback: int,
|
||
clock: Callable[[], float],
|
||
settings: Optional[Settings] = None,
|
||
) -> tuple[list[dict[str, Any]], str, int]:
|
||
"""
|
||
只拉已登记协同群。应用群不要带存档席 userid，否则管家给空页。
|
||
|
||
热群每轮必扫；冷群按游标轮转。遇微盛限频或时间预算用尽则本轮停，下一轮续扫。
|
||
禁止把每轮冷群压得太少：销售在冷群发「解散群」时，要等很多轮才会被听见。
|
||
"""
|
||
cfg = settings or get_settings()
|
||
all_rooms = list_watched_rooms(redis=redis)
|
||
if not all_rooms:
|
||
logger.warning("微盛按房间拉取：还没有登记过的协同群")
|
||
return [], "rooms", 0
|
||
max_cold = int(getattr(cfg, "wshoto_rooms_per_poll", 120) or 120)
|
||
max_cold = min(200, max(1, max_cold))
|
||
gap_ms = int(getattr(cfg, "wshoto_room_gap_ms", 100) or 100)
|
||
gap_ms = min(1000, max(0, gap_ms))
|
||
budget_sec = float(getattr(cfg, "wshoto_poll_budget_sec", 20.0) or 20.0)
|
||
budget_sec = min(60.0, max(5.0, budget_sec))
|
||
rooms = plan_rooms_for_poll(all_rooms, max_cold=max_cold, redis=redis)
|
||
merged: list[dict[str, Any]] = []
|
||
floor = 0
|
||
rate_limited = False
|
||
budget_hit = False
|
||
polled = 0
|
||
cold_polled = 0
|
||
t0 = time.time()
|
||
for room in rooms:
|
||
if (time.time() - t0) >= budget_sec:
|
||
budget_hit = True
|
||
break
|
||
scope = f"room:{room}"
|
||
last = load_last_seq(scope, redis=redis)
|
||
time_start = 0
|
||
time_end = 0
|
||
if last <= 0:
|
||
time_end = int(clock())
|
||
time_start = time_end - min(lookback, _NEW_ROOM_LOOKBACK_SEC)
|
||
try:
|
||
rows = api.list_group_messages(
|
||
userid="",
|
||
target_id=room,
|
||
last_seq=last,
|
||
page_size=page_size,
|
||
msg_time_start=time_start,
|
||
msg_time_end=time_end,
|
||
)
|
||
except WshotoError as exc:
|
||
logger.warning("微盛房间拉取失败 chat=%s code=%s msg=%s", room, exc.code, exc.msg)
|
||
if int(exc.code) == _RATE_LIMIT_CODE:
|
||
rate_limited = True
|
||
break
|
||
continue
|
||
polled += 1
|
||
if not is_room_hot(room):
|
||
cold_polled += 1
|
||
# 增量若跳号：管家有时先吐出撤回等高 seq，游标一步跨过中间正文
|
||
# （现网：解散群 seq=9898 被撤回 9899 推过去后永远听不见）。跳号则时间窗补洞。
|
||
rows = _backfill_room_seq_gap(
|
||
api,
|
||
room=room,
|
||
last_seq=last,
|
||
rows=list(rows or []),
|
||
page_size=page_size,
|
||
lookback=lookback,
|
||
clock=clock,
|
||
)
|
||
max_seq = last
|
||
for rec in rows:
|
||
seq = _record_seq(rec)
|
||
if seq > max_seq:
|
||
max_seq = seq
|
||
if max_seq > last:
|
||
save_last_seq(scope, max_seq, redis=redis)
|
||
if last > floor:
|
||
floor = last
|
||
merged.extend(rows)
|
||
if gap_ms > 0:
|
||
time.sleep(gap_ms / 1000.0)
|
||
# 只推进实际扫过的冷群，限频/超时停住的下轮接着扫。
|
||
advance_poll_cursor(cold_polled, redis=redis)
|
||
if rate_limited or budget_hit:
|
||
logger.warning(
|
||
"微盛房间拉取提前结束 rate=%s budget=%s watched=%s planned=%s polled=%s",
|
||
rate_limited,
|
||
budget_hit,
|
||
len(all_rooms),
|
||
len(rooms),
|
||
polled,
|
||
)
|
||
return merged, "rooms", floor
|