""" 空运群同一句去重:BOT 长连接和会话存档补听不要各发一遍。 BOT 与存档的 msgid 经常对不上,按「群 + 人 + 去掉@后的原文」认同一句。 HTTP 和 Worker 是两个进程:Redis 记一份,发布目录再落一份文件,避免 Memory Redis 互看不见。 短协调,不当六态、不当 checkpoint。 """ from __future__ import annotations import logging import time from pathlib import Path from typing import Any, Optional from agent.redis_coord.keys import AIR_INBOUND_TEXT from agent.redis_coord.runtime import get_redis_runtime logger = logging.getLogger(__name__) _TTL_SEC = 3 * 60 def _norm_text(text: str) -> str: body = (text or "").replace("@询价小助手", "").replace("询价小助手", "") return "".join(body.split()) def text_fingerprint(chat_id: str, sender_id: str, text: str) -> str: """同一群同一人同一句(去掉 @询价小助手)。""" return f"{(chat_id or '').strip()}|{(sender_id or '').strip()}|{_norm_text(text)}" def _file_path(fp: str) -> Path: root = Path(__file__).resolve().parents[2] / "var" / "air_inbound" root.mkdir(parents=True, exist_ok=True) safe = "".join(ch if ch.isalnum() else "_" for ch in fp)[:160] return root / f"{safe}.claim" def _redis_get(fp: str) -> bool: try: client = get_redis_runtime().client raw = client.raw.get(client.key(AIR_INBOUND_TEXT, fp)) return bool(raw) except Exception: logger.exception("空运群入站去重读 Redis 失败") return False def _redis_mark(fp: str) -> None: try: client = get_redis_runtime().client client.raw.set(client.key(AIR_INBOUND_TEXT, fp), "1", ex=_TTL_SEC) except Exception: logger.exception("空运群入站去重写 Redis 失败") def _file_fresh(fp: str) -> bool: path = _file_path(fp) try: if not path.exists(): return False return (time.time() - path.stat().st_mtime) < _TTL_SEC except OSError: return False def _file_mark(fp: str) -> None: path = _file_path(fp) try: path.write_text("1", encoding="ascii") except OSError: logger.exception("空运群入站去重写文件失败") def already_handled_air_text(*, chat_id: str, sender_id: str, text: str) -> bool: """这句是否刚被 BOT 办过。给存档补听用。""" fp = text_fingerprint(chat_id, sender_id, text) if not _norm_text(text): return False return _redis_get(fp) or _file_fresh(fp) def mark_air_text(*, chat_id: str, sender_id: str, text: str) -> None: """BOT 或首次存档办完后打标,后到的那路跳过。""" fp = text_fingerprint(chat_id, sender_id, text) if not _norm_text(text): return _redis_mark(fp) _file_mark(fp) def is_archive_source(raw: Optional[dict[str, Any]]) -> bool: src = str((raw or {}).get("source") or "").strip().lower() return src in {"wshoto_archive", "wshoto", "archive"} def reset_air_inbound_for_tests() -> None: """单测清本机标记,避免上一例的同一句挡住这一例。生产不要调。""" root = Path(__file__).resolve().parents[2] / "var" / "air_inbound" if not root.exists(): return for path in root.glob("*.claim"): try: path.unlink() except OSError: logger.exception("空运群入站去重清测试文件失败") def should_skip_air_text( *, chat_id: str, sender_id: str, text: str, raw: Optional[dict[str, Any]] = None, ) -> bool: """ 存档补听到的 @ 句,若 BOT 已经回过就跳过。 BOT 自己再 @ 同一工单号仍要再发一版摘要,不在这里拦。 """ if not is_archive_source(raw): return False if already_handled_air_text(chat_id=chat_id, sender_id=sender_id, text=text): logger.info("空运群存档补听跳过,BOT 已办 chat=%s", chat_id) return True return False