diff --git a/digital-avatar-app/backend/main.py b/digital-avatar-app/backend/main.py index a5d1e4d..c34ebac 100644 --- a/digital-avatar-app/backend/main.py +++ b/digital-avatar-app/backend/main.py @@ -6,7 +6,6 @@ import logging from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.triggers.interval import IntervalTrigger -import redis as redis_lib from database import init_db, SessionLocal from models import Avatar, Authorization, Organization, TokenAccount, TokenPlan @@ -24,7 +23,6 @@ from responses import ok logger = logging.getLogger(__name__) takeover_scheduler = None -takeover_db = None app = FastAPI(title="会会数字分身 API", version="1.0.0") @@ -113,7 +111,7 @@ def seed(): @app.on_event("startup") def on_startup(): - global takeover_scheduler, takeover_db + global takeover_scheduler init_db() seed() @@ -123,47 +121,43 @@ def on_startup(): # --- Takeover scheduler --- try: - # Initialize Redis (optional) - redis_client = None - redis_url = os.getenv("REDIS_URL", "") - if redis_url: - try: - redis_client = redis_lib.from_url(redis_url) - redis_client.ping() - except Exception as e: - logger.warning(f"Redis connection failed, delayed takeover will degrade to immediate: {e}") - - # Initialize Box IM client + # BOXIM production endpoints are intentionally separate from the login API. from services.boxim_client import BoxIMClient boxim_config = { - "HUIHUI_IM_BASE_URL": os.getenv("HUIHUI_IM_BASE_URL", "http://192.168.1.200:60040"), + "HUIHUI_PLATFORM_BASE_URL": os.getenv( + "HUIHUI_PLATFORM_BASE_URL", "https://open.99hui.com/api" + ), + "BOXIM_API_BASE_URL": os.getenv( + "BOXIM_API_BASE_URL", "https://im.99hui.com/api" + ), "HUIHUI_APP_ID": os.getenv("HUIHUI_APP_ID", ""), "HUIHUI_ACCESS_ID": os.getenv("HUIHUI_ACCESS_ID", ""), "HUIHUI_ACCESS_SECRET": os.getenv("HUIHUI_ACCESS_SECRET", ""), + "BOXIM_TIMEOUT_SECONDS": os.getenv("BOXIM_TIMEOUT_SECONDS", "20"), } boxim_client = BoxIMClient(boxim_config) - # Initialize takeover service from services.takeover_service import TakeoverService - takeover_db = SessionLocal() - takeover_service = TakeoverService(takeover_db, boxim_client, redis_client) + takeover_service = TakeoverService(SessionLocal, boxim_client) - # AsyncIOScheduler awaits the service coroutine instead of dropping it. + poll_interval = max(0.5, float(os.getenv("BOXIM_POLL_INTERVAL_SECONDS", "1"))) takeover_scheduler = AsyncIOScheduler() takeover_scheduler.add_job( takeover_service.poll_and_process_messages, - trigger=IntervalTrigger(seconds=10), + trigger=IntervalTrigger(seconds=poll_interval), id="takeover_message_poll", + max_instances=1, + coalesce=True, ) takeover_scheduler.start() - logger.info("Takeover message polling scheduler started (interval=10s)") + logger.info("BOXIM takeover scheduler started (interval=%ss)", poll_interval) except Exception as e: stop_takeover_scheduler() logger.warning(f"Failed to initialize takeover scheduler, app will continue without it: {e}") def stop_takeover_scheduler(): - global takeover_scheduler, takeover_db + global takeover_scheduler if takeover_scheduler is not None: try: @@ -174,13 +168,6 @@ def stop_takeover_scheduler(): finally: takeover_scheduler = None - if takeover_db is not None: - try: - takeover_db.close() - finally: - takeover_db = None - - @app.on_event("shutdown") def on_shutdown(): stop_takeover_scheduler() diff --git a/digital-avatar-app/backend/models.py b/digital-avatar-app/backend/models.py index 72e6a44..38b1484 100644 --- a/digital-avatar-app/backend/models.py +++ b/digital-avatar-app/backend/models.py @@ -1,6 +1,17 @@ import uuid -from sqlalchemy import Column, String, Integer, Float, DateTime, Text, JSON, Boolean +from sqlalchemy import ( + Boolean, + Column, + DateTime, + Float, + Index, + Integer, + JSON, + String, + Text, + UniqueConstraint, +) from sqlalchemy.sql import func from database import Base @@ -74,6 +85,76 @@ class Authorization(Base): } +class TakeoverCursor(Base): + """Durable BOXIM polling cursor for one avatar owner.""" + + __tablename__ = "takeover_cursors" + id = Column(String, primary_key=True, default=lambda: uuid.uuid4().hex) + avatar_id = Column(String, nullable=False, unique=True, index=True) + owner_id = Column(String, nullable=False, default="", index=True) + boxim_owner_id = Column(String, default="") + last_message_id = Column(String, default="0") + initialized = Column(Boolean, default=False) + last_polled_at = Column(DateTime) + last_error = Column(Text, default="") + created_at = Column(DateTime, server_default=func.now()) + updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now()) + + +class TakeoverMessage(Base): + """BOXIM message receipt used for audit, deduplication, and chat context.""" + + __tablename__ = "takeover_messages" + __table_args__ = ( + UniqueConstraint("owner_id", "boxim_message_id", name="uq_takeover_message_owner_boxim"), + Index("ix_takeover_message_conversation", "owner_id", "peer_id", "send_time"), + ) + + id = Column(String, primary_key=True, default=lambda: uuid.uuid4().hex) + avatar_id = Column(String, nullable=False, index=True) + owner_id = Column(String, nullable=False, index=True) + boxim_message_id = Column(String, nullable=False) + boxim_local_id = Column(String, nullable=True) + peer_id = Column(String, nullable=False, index=True) + direction = Column(String, nullable=False) # incoming | outgoing + message_type = Column(Integer, default=0) + content = Column(Text, default="") + is_avatar = Column(Boolean, default=False) + send_time = Column(DateTime, nullable=False) + created_at = Column(DateTime, server_default=func.now()) + + +class TakeoverReplyTask(Base): + """Restart-safe three-second BOXIM reply task.""" + + __tablename__ = "takeover_reply_tasks" + __table_args__ = ( + UniqueConstraint("owner_id", "trigger_message_id", name="uq_takeover_task_owner_trigger"), + Index("ix_takeover_task_due", "status", "scheduled_at"), + Index("ix_takeover_task_conversation", "owner_id", "peer_id", "status"), + ) + + id = Column(String, primary_key=True, default=lambda: uuid.uuid4().hex) + avatar_id = Column(String, nullable=False, index=True) + owner_id = Column(String, nullable=False, index=True) + peer_id = Column(String, nullable=False, index=True) + trigger_message_id = Column(String, nullable=False) + source_message_ids = Column(JSON, default=list) + prompt = Column(Text, default="") + response_text = Column(Text, default="") + status = Column(String, default="pending") + scheduled_at = Column(DateTime, nullable=False) + locked_at = Column(DateTime) + sent_at = Column(DateTime) + attempts = Column(Integer, default=0) + last_error = Column(Text, default="") + cancel_reason = Column(String, default="") + boxim_local_id = Column(String, nullable=False) + boxim_sent_message_id = Column(String, default="") + created_at = Column(DateTime, server_default=func.now()) + updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now()) + + class Organization(Base): __tablename__ = "organizations" id = Column(String, primary_key=True, default=lambda: uuid.uuid4().hex) diff --git a/digital-avatar-app/backend/requirements.txt b/digital-avatar-app/backend/requirements.txt index 00d7d76..f744392 100644 --- a/digital-avatar-app/backend/requirements.txt +++ b/digital-avatar-app/backend/requirements.txt @@ -7,5 +7,4 @@ httpx pypdf python-docx openpyxl -redis>=5.0 apscheduler>=3.10 diff --git a/digital-avatar-app/backend/routers/authorizations.py b/digital-avatar-app/backend/routers/authorizations.py index 59031d5..69b8481 100644 --- a/digital-avatar-app/backend/routers/authorizations.py +++ b/digital-avatar-app/backend/routers/authorizations.py @@ -11,7 +11,7 @@ router = APIRouter(tags=["授权"]) TARGET_TYPES = {"user", "organization", "application"} PERMISSION_ORDER = ("friend", "chat", "publish", "browse", "interact", "takeover") ALLOWED_PERMISSIONS = set(PERMISSION_ORDER) -AVATAR_PERMISSION_ORDER = PERMISSION_ORDER[:-1] +AVATAR_PERMISSION_ORDER = PERMISSION_ORDER AVATAR_PERMISSION_KEY = "authorizationPermissions" DEFAULT_AVATAR_PERMISSIONS = ["friend", "chat"] LEGACY_PERMISSION_MAP = { diff --git a/digital-avatar-app/backend/routers/takeover.py b/digital-avatar-app/backend/routers/takeover.py index 9fc3095..a27ff53 100644 --- a/digital-avatar-app/backend/routers/takeover.py +++ b/digital-avatar-app/backend/routers/takeover.py @@ -1,9 +1,12 @@ -"""数字分身单聊接管配置 API。""" +"""数字分身 BOXIM 单聊接管 API。""" + +from datetime import datetime, timedelta from fastapi import APIRouter, Body, Depends, Header from sqlalchemy.orm import Session from database import get_db +from models import TakeoverCursor, TakeoverReplyTask, User from responses import fail, ok from routers.authorizations import _require_authorization from routers.avatars import _require_owned_avatar @@ -11,6 +14,53 @@ from routers.avatars import _require_owned_avatar router = APIRouter(tags=["分身接管"]) +@router.get("/avatar/{avatar_id}/takeover/status") +def get_takeover_status( + avatar_id: str, + authorization: str = Header(None), + db: Session = Depends(get_db), +): + avatar = _require_owned_avatar(db, avatar_id, authorization) + permissions = (avatar.config or {}).get("authorizationPermissions", []) + enabled = isinstance(permissions, list) and "takeover" in permissions + user = db.query(User).filter(User.huihui_user_id == avatar.owner_id).first() + cursor = db.query(TakeoverCursor).filter(TakeoverCursor.avatar_id == avatar.id).first() + pending_count = ( + db.query(TakeoverReplyTask) + .filter( + TakeoverReplyTask.avatar_id == avatar.id, + TakeoverReplyTask.status.in_(("pending", "generating", "ready", "sending")), + ) + .count() + ) + + if not enabled: + status, message = "disabled", "主动接管未开启" + elif not user or not user.huihui_token: + status, message = "needs_login", "请重新登录会会生产账号以连接 BOXIM" + elif cursor and cursor.last_error: + status, message = "error", cursor.last_error + elif ( + cursor + and cursor.initialized + and cursor.last_polled_at + and cursor.last_polled_at >= datetime.utcnow() - timedelta(seconds=15) + ): + status, message = "ready", "BOXIM 已连接，收到私聊消息 3 秒后自动回复" + else: + status, message = "connecting", "正在连接 BOXIM" + + return ok( + { + "enabled": enabled, + "status": status, + "message": message, + "pendingCount": pending_count, + "lastPolledAt": cursor.last_polled_at.isoformat() if cursor and cursor.last_polled_at else None, + } + ) + + def _has(payload: dict, camel_key: str, snake_key: str) -> bool: return camel_key in payload or snake_key in payload diff --git a/digital-avatar-app/backend/services/boxim_client.py b/digital-avatar-app/backend/services/boxim_client.py index 726da2a..1adf9bc 100644 --- a/digital-avatar-app/backend/services/boxim_client.py +++ b/digital-avatar-app/backend/services/boxim_client.py @@ -1,76 +1,178 @@ -"""盒子 IM 客户端 — 封装网易云信 IM 接口调用""" +"""Client for Huihui's self-hosted BOXIM production APIs.""" + import hashlib import random +import secrets import string -from datetime import datetime -from typing import Optional +import time +from datetime import datetime, timedelta, timezone +from typing import Any import httpx +_CN_TZ = timezone(timedelta(hours=8)) + + +class BoxIMError(RuntimeError): + def __init__(self, message: str, *, code: Any = None, auth_error: bool = False): + super().__init__(message) + self.code = code + self.auth_error = auth_error + + class BoxIMClient: - """盒子 IM 客户端，通过会会平台网关调用网易云信 IM""" + """Exchange Huihui credentials and call BOXIM's private-message API.""" def __init__(self, config: dict): - self.base_url = config.get("HUIHUI_IM_BASE_URL", "http://192.168.1.200:60040") + self.platform_base_url = config.get( + "HUIHUI_PLATFORM_BASE_URL", "https://open.99hui.com/api" + ).rstrip("/") + self.im_base_url = config.get( + "BOXIM_API_BASE_URL", "https://im.99hui.com/api" + ).rstrip("/") self.app_id = config.get("HUIHUI_APP_ID", "") self.access_id = config.get("HUIHUI_ACCESS_ID", "") self.access_secret = config.get("HUIHUI_ACCESS_SECRET", "") + self.timeout = float(config.get("BOXIM_TIMEOUT_SECONDS", 20)) - def _build_sign_params(self, extra: dict) -> dict: - """构建带签名的请求参数（复用 news_service 签名模式）""" - nonce = "".join(random.choices(string.ascii_lowercase + string.digits, k=12)) - timestamp = datetime.now().strftime("%Y%m%d%H%M%S") # 24小时制 + def _build_sign_params(self, extra: dict | None = None) -> dict: + """Build the same signed form used by Huihui's current production app.""" params = { "appId": self.app_id, "accessId": self.access_id, - "nonce": nonce, - "timestamp": timestamp, - **extra, + "nonce": "".join(random.choices(string.ascii_lowercase + string.digits, k=12)), + "timestamp": datetime.now(_CN_TZ).strftime("%Y%m%d%H%M%S"), + "signType": "MD5", + "signVersion": "1.0", + **(extra or {}), } - # 计算签名 — 排序 key, 过滤空值, 拼接后加 accessSecret, MD5 大写 - keys = sorted(params.keys()) + params.pop("accessSecret", None) + params.pop("signature", None) sign_parts = [] - for k in keys: - if k in ("signature", "accessSecret"): + for key in sorted(params): + value = params[key] + if value in (None, "", []): continue - v = params.get(k) - if v and v != "" and v != []: - sign_parts.append(f"{k}={v}") - sign_str = "&".join(sign_parts) + f"&accessSecret={self.access_secret}" - signature = hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper() - params["signature"] = signature - params["signType"] = "MD5" - params["signVersion"] = "1.0" + if isinstance(value, list): + continue + sign_parts.append(f"{key}={value}") + sign_source = "&".join(sign_parts) + f"&accessSecret={self.access_secret}" + params["signature"] = hashlib.md5(sign_source.encode("utf-8")).hexdigest().upper() return params - async def get_credentials(self, user_id: str) -> Optional[dict]: - """获取用户的网易云信 IM 凭证 (accid, token)""" - params = self._build_sign_params({"userId": user_id}) - async with httpx.AsyncClient(timeout=10) as client: - r = await client.post( - f"{self.base_url}/box/netease", - params=params, - ) - data = r.json() - if data.get("code") in (0, 200): - return data.get("data", {}) - return None + @staticmethod + def _response_payload(response: httpx.Response) -> dict: + try: + payload = response.json() + except ValueError as exc: + raise BoxIMError("BOXIM 返回了无效响应") from exc + if not isinstance(payload, dict): + raise BoxIMError("BOXIM 返回格式不正确") + return payload - async def send_p2p_message( - self, from_accid: str, to_accid: str, content: str - ) -> bool: - """发送单聊消息（文本）""" - params = self._build_sign_params({ - "from": from_accid, - "to": to_accid, - "msgType": "text", - "content": content, - }) - async with httpx.AsyncClient(timeout=10) as client: - r = await client.post( - f"{self.base_url}/box/message/send/p2p", - params=params, + async def exchange_access_token(self, huihui_token: str) -> dict: + """Exchange a production Huihui token for a BOXIM access token.""" + if not huihui_token: + raise BoxIMError("缺少会会登录凭证", auth_error=True) + if not (self.app_id and self.access_id and self.access_secret): + raise BoxIMError("会会开放平台凭证未配置", auth_error=True) + + headers = { + "Authorization": f"Bearer {huihui_token}", + "appId": self.app_id, + "windowAppId": self.app_id, + } + async with httpx.AsyncClient(timeout=self.timeout, follow_redirects=True) as client: + response = await client.post( + f"{self.platform_base_url}/im/box/netease", + headers=headers, + data=self._build_sign_params(), ) - data = r.json() - return data.get("code") in (0, 200) + payload = self._response_payload(response) + data = payload.get("data") or {} + code = payload.get("code") + if response.status_code >= 400 or code not in (0, 200, "0", "200"): + raise BoxIMError( + payload.get("message") or "BOXIM 授权失败", + code=code or response.status_code, + auth_error=response.status_code in (400, 401, 403) + or code in (400, 401, 40100, 403, "400", "401", "40100", "403"), + ) + if not data.get("accessToken"): + raise BoxIMError("会会未返回 BOXIM 访问凭证", auth_error=True) + return data + + async def _request( + self, + method: str, + path: str, + access_token: str, + *, + params: dict | None = None, + json: dict | None = None, + ) -> Any: + headers = {"accessToken": access_token} + async with httpx.AsyncClient(timeout=self.timeout) as client: + response = await client.request( + method, + f"{self.im_base_url}{path}", + headers=headers, + params=params, + json=json, + ) + payload = self._response_payload(response) + code = payload.get("code") + if response.status_code >= 400 or code not in (200, "200"): + raise BoxIMError( + payload.get("message") or "BOXIM 请求失败", + code=code or response.status_code, + auth_error=response.status_code in (400, 401, 403) + or code in (400, 401, 403, "400", "401", "403"), + ) + return payload.get("data") + + async def get_self(self, access_token: str) -> dict: + data = await self._request("GET", "/user/self", access_token) + if not isinstance(data, dict) or data.get("id") is None: + raise BoxIMError("BOXIM 未返回当前用户信息") + return data + + async def fetch_private_messages(self, access_token: str, min_id: str = "0") -> list[dict]: + data = await self._request( + "GET", + "/message/private/loadOfflineMessage", + access_token, + params={"minId": str(min_id or "0")}, + ) + if data is None: + return [] + if not isinstance(data, list): + raise BoxIMError("BOXIM 私聊消息格式不正确") + return [item for item in data if isinstance(item, dict)] + + async def send_private_message( + self, + access_token: str, + peer_id: str, + content: str, + *, + local_id: int | str | None = None, + ) -> dict: + local_id = int(local_id or (int(time.time() * 1000) * 1000 + secrets.randbelow(1000))) + data = await self._request( + "POST", + "/message/private/send", + access_token, + json={ + "localId": local_id, + "recvId": int(peer_id) if str(peer_id).isdigit() else peer_id, + "content": content, + "type": 0, + "receipt": False, + "atUserIds": [], + }, + ) + if not isinstance(data, dict): + raise BoxIMError("BOXIM 未返回发送结果") + return data diff --git a/digital-avatar-app/backend/services/takeover_service.py b/digital-avatar-app/backend/services/takeover_service.py index c11c6e7..916f675 100644 --- a/digital-avatar-app/backend/services/takeover_service.py +++ b/digital-avatar-app/backend/services/takeover_service.py @@ -1,184 +1,565 @@ -"""Takeover service — message listening, decision, reply execution.""" -import json +"""Restart-safe automatic replies over Huihui's self-hosted BOXIM.""" + +import asyncio +import hashlib import logging -import os -from typing import Optional +import re +import secrets +import time +from datetime import datetime, timedelta +from typing import Callable -import httpx from sqlalchemy.orm import Session -from models import Avatar, Authorization -from services.boxim_client import BoxIMClient +from models import ( + Avatar, + TakeoverCursor, + TakeoverMessage, + TakeoverReplyTask, + User, +) +from services.boxim_client import BoxIMClient, BoxIMError logger = logging.getLogger(__name__) +ACTIVE_TASK_STATUSES = ("pending", "generating", "ready", "sending") +GENERATABLE_TASK_STATUSES = ("pending",) +MAX_PROMPT_LENGTH = 4000 +MAX_STALE_SECONDS = 120 +STUCK_LOCK_SECONDS = 90 +TAKEOVER_PERMISSION = "takeover" + + +def _utcnow() -> datetime: + return datetime.utcnow() + + +def _takeover_enabled(avatar: Avatar | None) -> bool: + if not avatar or avatar.status != "active": + return False + permissions = (avatar.config or {}).get("authorizationPermissions", []) + return isinstance(permissions, list) and TAKEOVER_PERMISSION in permissions + + +def _boxim_time(value, fallback: datetime) -> datetime: + try: + timestamp = float(value) + if timestamp > 10_000_000_000: + timestamp /= 1000 + return datetime.utcfromtimestamp(timestamp) + except (TypeError, ValueError, OSError, OverflowError): + return fallback + + +def _numeric_id(value) -> int: + try: + return int(value) + except (TypeError, ValueError): + return 0 + + +def _plain_text_reply(value: str) -> str: + """BOXIM is plain text, so remove Markdown markers without damaging paragraphs.""" + text = (value or "").replace("\r\n", "\n").replace("\r", "\n") + text = re.sub(r"```(?:\w+)?\n?(.*?)```", r"\1", text, flags=re.S) + text = re.sub(r"\*\*(.*?)\*\*|__(.*?)__", lambda m: m.group(1) or m.group(2), text) + text = re.sub(r"(? Optional[Authorization]: - """Check whether takeover is enabled for the given target user.""" - avatar = ( - self.db.query(Avatar) - .filter(Avatar.owner_id == owner_huihui_id, Avatar.status == "active") - .first() - ) - if not avatar: - return None - - auth = ( - self.db.query(Authorization) - .filter(Authorization.avatar_id == avatar.id) - .filter(Authorization.target_id == from_user_id) - .filter(Authorization.target_type == "user") - .filter(Authorization.status == "active") - .filter(Authorization.takeover_enabled == True) - .first() - ) - permissions = set(auth.permissions or []) if auth else set() - return auth if auth and "takeover" in permissions else None - - async def generate_reply(self, avatar_id: str, message: str) -> str: - """Call the avatar chat endpoint to generate a reply.""" - try: - async with httpx.AsyncClient(timeout=30) as client: - r = await client.post( - f"{self._chat_api_base}/avatar/{avatar_id}/chat", - json={"message": message, "history": []}, - ) - data = r.json() - if data.get("code") in (0, 200): - return data.get("data", {}).get("answer", "") - logger.warning(f"Avatar chat API returned error code: {data}") - return "" - except Exception as e: - logger.error(f"Failed to call avatar chat API: {e}") - return "" - - async def execute_takeover(self, auth: Authorization, message: dict) -> bool: - """Execute takeover: generate a reply and send it as the owner via IM.""" - try: - # Resolve owner through Avatar model - avatar = self.db.query(Avatar).filter(Avatar.id == auth.avatar_id).first() - if not avatar: - logger.warning(f"Avatar not found: {auth.avatar_id}") - return False - - owner_huihui_id = avatar.owner_id - credentials = await self.boxim.get_credentials(owner_huihui_id) - if not credentials: - logger.warning(f"Cannot obtain IM credentials for owner: {owner_huihui_id}") - return False - - reply = await self.generate_reply(auth.avatar_id, message.get("content", "")) - if not reply: - logger.warning("Avatar did not generate a reply") - return False - - success = await self.boxim.send_p2p_message( - from_accid=credentials["accid"], - to_accid=message.get("from_accid", ""), - content=reply, - ) - if success: - logger.info(f"Takeover reply sent successfully: {reply[:50]}...") - return success - except Exception as e: - logger.error(f"Takeover execution failed: {e}") - return False - - def enqueue_delayed_message(self, auth: Authorization, message: dict): - """Write a message into the Redis delayed queue (TTL = delay + 10s buffer).""" - if not self.redis: - logger.warning("Redis not configured, degrading to immediate takeover") - return - - avatar = self.db.query(Avatar).filter(Avatar.id == auth.avatar_id).first() - owner_huihui_id = avatar.owner_id if avatar else "" - key = f"takeover:delayed:{auth.target_id}:{message.get('msg_id', '')}" - value = json.dumps({ - "avatar_id": auth.avatar_id, - "from_accid": message.get("from_accid", ""), - "content": message.get("content", ""), - "owner_huihui_id": owner_huihui_id, - }) - self.redis.setex(key, auth.takeover_delay_seconds + 10, value) - logger.info(f"Message enqueued to delayed queue: {key}") - - async def process_delayed_queue(self): - """Process expired messages from the delayed queue. - - Scans Redis keys matching the takeover:delayed: pattern and dispatches - each to execute_takeover after resolving the Authorization. - """ - if not self.redis: - return - try: - pattern = "takeover:delayed:*" - keys = self.redis.keys(pattern) - for key in keys: - raw = self.redis.get(key) - if not raw: - continue - data = json.loads(raw) - auth = ( - self.db.query(Authorization) - .filter(Authorization.target_id == key.split(":")[2]) - .first() - ) - if auth: - message = { - "msg_id": key.split(":")[-1], - "from_accid": data.get("from_accid", ""), - "content": data.get("content", ""), - } - await self.execute_takeover(auth, message) - self.redis.delete(key) - except Exception as e: - logger.error(f"Failed to process delayed queue: {e}") + self.reply_delay_seconds = reply_delay_seconds + self.now = now + self._sessions: dict[str, dict] = {} + self._run_lock = asyncio.Lock() async def poll_and_process_messages(self): - """Periodic polling job: fetch unread messages and process each.""" + """Run one complete cycle; polling always happens before reply dispatch.""" + if self._run_lock.locked(): + return + async with self._run_lock: + self._recover_stuck_tasks() + avatar_ids = self._enabled_avatar_ids() + self._cancel_disabled_tasks(set(avatar_ids)) + for avatar_id in avatar_ids: + await self._sync_avatar(avatar_id) + + generated = await self._prepare_replies() + if generated: + # Catch a human reply sent while the model was preparing its answer. + for avatar_id in avatar_ids: + await self._sync_avatar(avatar_id) + await self._dispatch_ready_replies() + + def _enabled_avatar_ids(self) -> list[str]: + db = self.session_factory() try: - messages = await self.fetch_unread_messages() - for msg in messages: - await self.process_message(msg) - except Exception as e: - logger.error(f"poll_and_process_messages failed: {e}") + return [ + avatar.id + for avatar in db.query(Avatar).filter(Avatar.status == "active").all() + if _takeover_enabled(avatar) + ] + finally: + db.close() - async def fetch_unread_messages(self) -> list: - """Fetch unread messages from Box IM. Stub — replace with real API call.""" - logger.debug("fetch_unread_messages: no real API wired yet") - return [] + def _cancel_disabled_tasks(self, enabled_avatar_ids: set[str]): + db = self.session_factory() + try: + tasks = ( + db.query(TakeoverReplyTask) + .filter(TakeoverReplyTask.status.in_(ACTIVE_TASK_STATUSES)) + .all() + ) + changed = False + for task in tasks: + if task.avatar_id not in enabled_avatar_ids: + task.status = "cancelled" + task.cancel_reason = "takeover_disabled" + task.locked_at = None + changed = True + if changed: + db.commit() + finally: + db.close() - async def process_message(self, message: dict): - """Process a single message: check takeover, dispatch immediate or delayed.""" - owner_id = message.get("owner_huihui_id", "") - from_id = message.get("from_accid", "") + def _recover_stuck_tasks(self): + db = self.session_factory() + try: + threshold = self.now() - timedelta(seconds=STUCK_LOCK_SECONDS) + tasks = ( + db.query(TakeoverReplyTask) + .filter( + TakeoverReplyTask.status.in_(("generating", "sending")), + TakeoverReplyTask.locked_at.isnot(None), + TakeoverReplyTask.locked_at < threshold, + ) + .all() + ) + for task in tasks: + task.status = "pending" if task.status == "generating" else "ready" + task.locked_at = None + task.last_error = "上次处理意外中断，已自动恢复" + if tasks: + db.commit() + finally: + db.close() - auth = self.check_takeover_enabled(owner_id, from_id) - if not auth: + async def _boxim_session(self, user: User) -> dict: + token_fingerprint = hashlib.sha256((user.huihui_token or "").encode()).hexdigest() + cached = self._sessions.get(user.id) + if ( + cached + and cached["expires_at"] > time.monotonic() + and cached["token_fingerprint"] == token_fingerprint + ): + return cached + + token_data = await self.boxim.exchange_access_token(user.huihui_token) + access_token = token_data["accessToken"] + profile = await self.boxim.get_self(access_token) + try: + expires_in = int(token_data.get("accessTokenExpiresIn") or 3600) + except (TypeError, ValueError): + expires_in = 3600 + if expires_in > 86_400: + expires_in //= 1000 + cache_for = max(60, min(expires_in - 60, 3600)) + cached = { + "access_token": access_token, + "boxim_owner_id": str(profile["id"]), + "expires_at": time.monotonic() + cache_for, + "token_fingerprint": token_fingerprint, + } + self._sessions[user.id] = cached + return cached + + def _forget_boxim_session(self, user_id: str): + self._sessions.pop(user_id, None) + + async def _sync_avatar(self, avatar_id: str) -> bool: + db = self.session_factory() + try: + avatar = db.query(Avatar).filter(Avatar.id == avatar_id).first() + if not _takeover_enabled(avatar): + return False + user = db.query(User).filter(User.huihui_user_id == avatar.owner_id).first() + cursor = db.query(TakeoverCursor).filter(TakeoverCursor.avatar_id == avatar.id).first() + if not cursor: + cursor = TakeoverCursor(avatar_id=avatar.id, owner_id=avatar.owner_id) + db.add(cursor) + db.flush() + if not user or not user.huihui_token: + cursor.last_error = "请重新登录会会生产账号后再开启主动接管" + cursor.last_polled_at = self.now() + db.commit() + return False + + try: + session = await self._boxim_session(user) + owner_boxim_id = session["boxim_owner_id"] + if cursor.boxim_owner_id and cursor.boxim_owner_id != owner_boxim_id: + cursor.initialized = False + cursor.last_message_id = "0" + cursor.boxim_owner_id = owner_boxim_id + messages = await self.boxim.fetch_private_messages( + session["access_token"], cursor.last_message_id or "0" + ) + except Exception as exc: + if isinstance(exc, BoxIMError) and exc.auth_error: + self._forget_boxim_session(user.id) + cursor.last_error = "BOXIM 授权已失效，请重新登录会会生产账号" + else: + cursor.last_error = f"BOXIM 暂时连接失败：{str(exc)[:160]}" + cursor.last_polled_at = self.now() + db.commit() + logger.warning("BOXIM sync failed for avatar %s: %s", avatar.id, exc) + return False + + messages.sort(key=lambda item: (_numeric_id(item.get("id")), item.get("sendTime") or 0)) + priming = not bool(cursor.initialized) + max_message_id = _numeric_id(cursor.last_message_id) + for message in messages: + self._record_message( + db, + avatar, + cursor.boxim_owner_id, + message, + schedule_reply=not priming, + ) + max_message_id = max(max_message_id, _numeric_id(message.get("id"))) + + cursor.last_message_id = str(max_message_id) + cursor.initialized = True + cursor.last_polled_at = self.now() + cursor.last_error = "" + db.commit() + return True + except Exception: + db.rollback() + logger.exception("Failed to persist BOXIM messages for avatar %s", avatar_id) + return False + finally: + db.close() + + def _record_message( + self, + db: Session, + avatar: Avatar, + boxim_owner_id: str, + message: dict, + *, + schedule_reply: bool, + ): + message_id = str(message.get("id") or "").strip() + if not message_id: + return + local_id = str(message.get("localId") or "").strip() or None + if ( + db.query(TakeoverMessage) + .filter( + TakeoverMessage.owner_id == avatar.owner_id, + TakeoverMessage.boxim_message_id == message_id, + ) + .first() + ): return - if auth.takeover_mode == "immediate": - await self.execute_takeover(auth, message) - elif not self.redis: - logger.warning("Redis not configured, executing delayed takeover immediately") - await self.execute_takeover(auth, message) + send_id = str(message.get("sendId") or "") + recv_id = str(message.get("recvId") or "") + if send_id == boxim_owner_id: + direction, peer_id = "outgoing", recv_id + elif recv_id == boxim_owner_id: + direction, peer_id = "incoming", send_id else: - self.enqueue_delayed_message(auth, message) + return + if not peer_id: + return + + now = self.now() + send_time = _boxim_time(message.get("sendTime"), now) + is_avatar = False + if direction == "outgoing" and local_id: + is_avatar = bool( + db.query(TakeoverReplyTask) + .filter( + TakeoverReplyTask.owner_id == avatar.owner_id, + TakeoverReplyTask.boxim_local_id == local_id, + TakeoverReplyTask.status == "sent", + ) + .first() + ) + + event = TakeoverMessage( + avatar_id=avatar.id, + owner_id=avatar.owner_id, + boxim_message_id=message_id, + boxim_local_id=local_id, + peer_id=peer_id, + direction=direction, + message_type=int(message.get("type") or 0), + content=str(message.get("content") or ""), + is_avatar=is_avatar, + send_time=send_time, + ) + db.add(event) + db.flush() + + if direction == "outgoing": + if not is_avatar: + self._cancel_conversation(db, avatar.owner_id, peer_id, "owner_replied") + return + if not schedule_reply or event.message_type != 0 or not event.content.strip(): + return + if (now - send_time).total_seconds() > MAX_STALE_SECONDS: + return + self._schedule_reply(db, avatar, event) + + @staticmethod + def _cancel_conversation(db: Session, owner_id: str, peer_id: str, reason: str): + tasks = ( + db.query(TakeoverReplyTask) + .filter( + TakeoverReplyTask.owner_id == owner_id, + TakeoverReplyTask.peer_id == peer_id, + TakeoverReplyTask.status.in_(ACTIVE_TASK_STATUSES), + ) + .all() + ) + for task in tasks: + task.status = "cancelled" + task.cancel_reason = reason + task.locked_at = None + + def _schedule_reply(self, db: Session, avatar: Avatar, event: TakeoverMessage): + active_tasks = ( + db.query(TakeoverReplyTask) + .filter( + TakeoverReplyTask.owner_id == avatar.owner_id, + TakeoverReplyTask.peer_id == event.peer_id, + TakeoverReplyTask.status.in_(("pending", "generating", "ready")), + ) + .order_by(TakeoverReplyTask.created_at.desc()) + .all() + ) + prompt_parts = [] + source_ids = [] + if active_tasks: + latest = active_tasks[0] + prompt_parts.append(latest.prompt) + source_ids.extend(latest.source_message_ids or []) + for task in active_tasks: + task.status = "cancelled" + task.cancel_reason = "newer_incoming_message" + task.locked_at = None + prompt_parts.append(event.content.strip()) + source_ids.append(event.boxim_message_id) + prompt = "\n".join(part for part in prompt_parts if part).strip()[-MAX_PROMPT_LENGTH:] + due_at = event.send_time + timedelta(seconds=self.reply_delay_seconds) + task_id = secrets.token_hex(16) + local_id = int(time.time() * 1000) * 1000 + secrets.randbelow(1000) + db.add( + TakeoverReplyTask( + id=task_id, + avatar_id=avatar.id, + owner_id=avatar.owner_id, + peer_id=event.peer_id, + trigger_message_id=event.boxim_message_id, + source_message_ids=source_ids, + prompt=prompt, + status="pending", + scheduled_at=due_at, + boxim_local_id=str(local_id), + ) + ) + + async def _prepare_replies(self) -> int: + db = self.session_factory() + try: + task_ids = [ + row[0] + for row in ( + db.query(TakeoverReplyTask.id) + .filter( + TakeoverReplyTask.status.in_(GENERATABLE_TASK_STATUSES), + TakeoverReplyTask.response_text == "", + ) + .order_by(TakeoverReplyTask.created_at.asc()) + .limit(10) + .all() + ) + ] + finally: + db.close() + + generated = 0 + for task_id in task_ids: + if await asyncio.to_thread(self._generate_reply, task_id): + generated += 1 + return generated + + def _generate_reply(self, task_id: str) -> bool: + db = self.session_factory() + try: + task = db.query(TakeoverReplyTask).filter(TakeoverReplyTask.id == task_id).first() + if not task or task.status != "pending": + return False + avatar = db.query(Avatar).filter(Avatar.id == task.avatar_id).first() + if not _takeover_enabled(avatar): + task.status = "cancelled" + task.cancel_reason = "takeover_disabled" + db.commit() + return False + + task.status = "generating" + task.locked_at = self.now() + db.commit() + + excluded_ids = set(task.source_message_ids or []) + events = ( + db.query(TakeoverMessage) + .filter( + TakeoverMessage.owner_id == task.owner_id, + TakeoverMessage.peer_id == task.peer_id, + ) + .order_by(TakeoverMessage.send_time.desc()) + .limit(30) + .all() + ) + history = [] + for event in reversed(events): + if event.boxim_message_id in excluded_ids or not event.content.strip(): + continue + history.append( + { + "role": "user" if event.direction == "incoming" else "assistant", + "content": event.content.strip(), + } + ) + history = history[-10:] + + from routers.chat import _resolve_reply + + result = _resolve_reply(db, avatar, task.prompt, history) + answer = _plain_text_reply(result.get("answer", "")) + db.refresh(task) + if task.status != "generating": + return False + if not answer: + raise RuntimeError("分身没有生成有效回复") + task.response_text = answer + task.status = "ready" + task.locked_at = None + task.last_error = "" + db.commit() + return True + except Exception as exc: + db.rollback() + task = db.query(TakeoverReplyTask).filter(TakeoverReplyTask.id == task_id).first() + if task and task.status in ("pending", "generating"): + task.attempts = (task.attempts or 0) + 1 + task.status = "pending" if task.attempts < 3 else "failed" + task.locked_at = None + task.last_error = str(exc)[:300] + db.commit() + logger.warning("Failed to prepare takeover reply %s: %s", task_id, exc) + return False + finally: + db.close() + + async def _dispatch_ready_replies(self): + db = self.session_factory() + try: + task_ids = [ + row[0] + for row in ( + db.query(TakeoverReplyTask.id) + .filter( + TakeoverReplyTask.status == "ready", + TakeoverReplyTask.scheduled_at <= self.now(), + ) + .order_by(TakeoverReplyTask.scheduled_at.asc()) + .limit(10) + .all() + ) + ] + finally: + db.close() + + for task_id in task_ids: + await self._send_task(task_id) + + async def _send_task(self, task_id: str) -> bool: + db = self.session_factory() + user = None + try: + task = db.query(TakeoverReplyTask).filter(TakeoverReplyTask.id == task_id).first() + if not task or task.status != "ready": + return False + avatar = db.query(Avatar).filter(Avatar.id == task.avatar_id).first() + if not _takeover_enabled(avatar): + task.status = "cancelled" + task.cancel_reason = "takeover_disabled" + db.commit() + return False + if (self.now() - task.scheduled_at).total_seconds() > MAX_STALE_SECONDS: + task.status = "cancelled" + task.cancel_reason = "stale_reply" + db.commit() + return False + user = db.query(User).filter(User.huihui_user_id == task.owner_id).first() + if not user or not user.huihui_token: + raise BoxIMError("缺少会会登录凭证", auth_error=True) + + task.status = "sending" + task.locked_at = self.now() + db.commit() + session = await self._boxim_session(user) + result = await self.boxim.send_private_message( + session["access_token"], + task.peer_id, + task.response_text, + local_id=task.boxim_local_id, + ) + db.refresh(task) + if task.status != "sending": + return False + task.status = "sent" + task.sent_at = self.now() + task.locked_at = None + task.last_error = "" + task.boxim_sent_message_id = str(result.get("id") or "") + db.commit() + logger.info("BOXIM takeover reply sent for task %s", task.id) + return True + except Exception as exc: + db.rollback() + if user and isinstance(exc, BoxIMError) and exc.auth_error: + self._forget_boxim_session(user.id) + task = db.query(TakeoverReplyTask).filter(TakeoverReplyTask.id == task_id).first() + if task and task.status in ("ready", "sending"): + task.attempts = (task.attempts or 0) + 1 + task.status = "ready" if task.attempts < 3 else "failed" + task.locked_at = None + task.last_error = str(exc)[:300] + if task.status == "ready": + task.scheduled_at = self.now() + timedelta(seconds=2 ** task.attempts) + db.commit() + logger.warning("Failed to send takeover reply %s: %s", task_id, exc) + return False + finally: + db.close() diff --git a/digital-avatar-app/backend/tests/conftest.py b/digital-avatar-app/backend/tests/conftest.py index a3bd5a7..9d2068f 100644 --- a/digital-avatar-app/backend/tests/conftest.py +++ b/digital-avatar-app/backend/tests/conftest.py @@ -2,7 +2,14 @@ import uuid import pytest from database import init_db, SessionLocal -from models import Authorization, Avatar, User +from models import ( + Authorization, + Avatar, + TakeoverCursor, + TakeoverMessage, + TakeoverReplyTask, + User, +) @pytest.fixture(scope="session", autouse=True) @@ -84,10 +91,20 @@ def authorization_context(): } finally: db.rollback() - db.query(Authorization).filter( - Authorization.avatar_id.in_([avatar.id, other_avatar.id]) + avatar_ids = [avatar.id, other_avatar.id] + db.query(TakeoverReplyTask).filter( + TakeoverReplyTask.avatar_id.in_(avatar_ids) ).delete(synchronize_session=False) - db.query(Avatar).filter(Avatar.id.in_([avatar.id, other_avatar.id])).delete( + db.query(TakeoverMessage).filter( + TakeoverMessage.avatar_id.in_(avatar_ids) + ).delete(synchronize_session=False) + db.query(TakeoverCursor).filter( + TakeoverCursor.avatar_id.in_(avatar_ids) + ).delete(synchronize_session=False) + db.query(Authorization).filter( + Authorization.avatar_id.in_(avatar_ids) + ).delete(synchronize_session=False) + db.query(Avatar).filter(Avatar.id.in_(avatar_ids)).delete( synchronize_session=False ) db.query(User).filter(User.id.in_([owner.id, other.id])).delete( diff --git a/digital-avatar-app/backend/tests/test_authorizations_api.py b/digital-avatar-app/backend/tests/test_authorizations_api.py index e3b8742..2da8e49 100644 --- a/digital-avatar-app/backend/tests/test_authorizations_api.py +++ b/digital-avatar-app/backend/tests/test_authorizations_api.py @@ -108,13 +108,13 @@ def test_avatar_permission_settings_default_and_persist(authorization_context): updated = client.put( endpoint, headers=context["owner_headers"], - json={"permissions": ["interact", "publish", "friend", "friend"]}, + json={"permissions": ["interact", "takeover", "publish", "friend", "friend"]}, ).json() assert updated["code"] == 200 - assert updated["data"]["permissions"] == ["friend", "publish", "interact"] + assert updated["data"]["permissions"] == ["friend", "publish", "interact", "takeover"] reloaded = client.get(endpoint, headers=context["owner_headers"]).json() - assert reloaded["data"]["permissions"] == ["friend", "publish", "interact"] + assert reloaded["data"]["permissions"] == ["friend", "publish", "interact", "takeover"] def test_avatar_permission_settings_allow_all_disabled(authorization_context): @@ -137,7 +137,7 @@ def test_avatar_permission_settings_validate_owner_and_permissions(authorization invalid = client.put( endpoint, headers=context["owner_headers"], - json={"permissions": ["takeover"]}, + json={"permissions": ["admin"]}, ).json() assert invalid["code"] == 400 diff --git a/digital-avatar-app/backend/tests/test_boxim_client.py b/digital-avatar-app/backend/tests/test_boxim_client.py index 5bd4d0c..6a65b23 100644 --- a/digital-avatar-app/backend/tests/test_boxim_client.py +++ b/digital-avatar-app/backend/tests/test_boxim_client.py @@ -1,134 +1,118 @@ -"""Tests for the Box IM client (Netease Yunxin gateway wrapper).""" -import pytest +"""Contract tests for the self-hosted BOXIM client.""" + from unittest.mock import AsyncMock, MagicMock, patch +import pytest + +from services.boxim_client import BoxIMClient, BoxIMError + @pytest.fixture -def mock_config(): +def config(): return { - "HUIHUI_IM_BASE_URL": "http://192.168.1.200:60040", + "HUIHUI_PLATFORM_BASE_URL": "https://open.example/api", + "BOXIM_API_BASE_URL": "https://im.example/api", "HUIHUI_APP_ID": "test_app", "HUIHUI_ACCESS_ID": "test_access", "HUIHUI_ACCESS_SECRET": "test_secret", } -def _make_mock_response(json_data: dict): - """Create a properly configured mock for httpx.Response.""" - mock_response = MagicMock() - mock_response.json.return_value = json_data - return mock_response +def _response(payload: dict, status_code: int = 200): + response = MagicMock() + response.status_code = status_code + response.json.return_value = payload + return response -def _patch_httpx_client(json_data: dict): - """Patch httpx.AsyncClient so that `async with httpx.AsyncClient() as c: await c.post(...)` returns json_data.""" - mock_client = AsyncMock() - mock_client.post.return_value = _make_mock_response(json_data) - - mock_cm = AsyncMock() - mock_cm.__aenter__.return_value = mock_client - mock_cm.__aexit__.return_value = None - - return patch("httpx.AsyncClient", return_value=mock_cm) +def _client_patch(*, post_payload=None, request_payload=None, status_code=200): + client = AsyncMock() + if post_payload is not None: + client.post.return_value = _response(post_payload, status_code) + if request_payload is not None: + client.request.return_value = _response(request_payload, status_code) + context = AsyncMock() + context.__aenter__.return_value = client + context.__aexit__.return_value = None + return patch("services.boxim_client.httpx.AsyncClient", return_value=context), client @pytest.mark.asyncio -async def test_get_credentials(mock_config): - """get_credentials should return accid and token from the gateway response.""" - with _patch_httpx_client({"code": 200, "data": {"accid": "user123", "token": "tok_xyz"}}): - from services.boxim_client import BoxIMClient +async def test_exchange_access_token_uses_huihui_bearer_and_signed_form(config): + mocked, client = _client_patch( + post_payload={"code": 0, "data": {"accessToken": "box-token", "accessTokenExpiresIn": 3600}} + ) + with mocked: + result = await BoxIMClient(config).exchange_access_token("huihui-token") - client = BoxIMClient(mock_config) - result = await client.get_credentials("user123") - - assert result["accid"] == "user123" - assert result["token"] == "tok_xyz" + assert result["accessToken"] == "box-token" + call = client.post.await_args + assert call.args[0] == "https://open.example/api/im/box/netease" + assert call.kwargs["headers"]["Authorization"] == "Bearer huihui-token" + assert call.kwargs["data"]["appId"] == "test_app" + assert len(call.kwargs["data"]["signature"]) == 32 @pytest.mark.asyncio -async def test_send_p2p_message_success(mock_config): - """send_p2p_message should return True when the gateway responds with code 200.""" - with _patch_httpx_client({"code": 200}): - from services.boxim_client import BoxIMClient +async def test_get_self_and_incremental_private_messages_use_boxim_header(config): + client_instance = BoxIMClient(config) + mocked, client = _client_patch( + request_payload={"code": 200, "data": {"id": 42, "nickName": "Owner"}} + ) + with mocked: + profile = await client_instance.get_self("box-token") + assert profile["id"] == 42 + assert client.request.await_args.kwargs["headers"] == {"accessToken": "box-token"} - client = BoxIMClient(mock_config) - result = await client.send_p2p_message("owner_acc", "target_acc", "Hello") - - assert result is True + mocked, client = _client_patch( + request_payload={"code": 200, "data": [{"id": 101, "sendId": 7, "recvId": 42}]} + ) + with mocked: + messages = await client_instance.fetch_private_messages("box-token", "100") + assert messages[0]["id"] == 101 + assert client.request.await_args.kwargs["params"] == {"minId": "100"} @pytest.mark.asyncio -async def test_send_p2p_message_failure(mock_config): - """send_p2p_message should return False when the gateway responds with a non-200 code.""" - with _patch_httpx_client({"code": 500, "message": "error"}): - from services.boxim_client import BoxIMClient +async def test_send_private_message_matches_boxim_payload(config): + mocked, client = _client_patch( + request_payload={"code": 200, "data": {"id": 88, "localId": 12345}} + ) + with mocked: + result = await BoxIMClient(config).send_private_message( + "box-token", "77", "你好", local_id="12345" + ) - client = BoxIMClient(mock_config) - result = await client.send_p2p_message("owner_acc", "target_acc", "Hello") - - assert result is False + assert result["id"] == 88 + call = client.request.await_args + assert call.args[:2] == ("POST", "https://im.example/api/message/private/send") + assert call.kwargs["json"] == { + "localId": 12345, + "recvId": 77, + "content": "你好", + "type": 0, + "receipt": False, + "atUserIds": [], + } @pytest.mark.asyncio -async def test_get_credentials_returns_none_on_error(mock_config): - """get_credentials should return None when the gateway responds with an error code.""" - with _patch_httpx_client({"code": 500, "message": "user not found"}): - from services.boxim_client import BoxIMClient - - client = BoxIMClient(mock_config) - result = await client.get_credentials("nonexistent") - - assert result is None +async def test_boxim_auth_error_is_explicit(config): + mocked, _ = _client_patch( + request_payload={"code": 400, "message": "未登录"}, status_code=200 + ) + with mocked, pytest.raises(BoxIMError) as exc_info: + await BoxIMClient(config).get_self("expired") + assert exc_info.value.auth_error is True -def test_build_sign_params_contains_required_fields(mock_config): - """_build_sign_params should produce appId, accessId, nonce, timestamp, signature, signType, signVersion.""" - from services.boxim_client import BoxIMClient - - client = BoxIMClient(mock_config) - params = client._build_sign_params({"userId": "u1"}) - - assert "appId" in params - assert "accessId" in params - assert "nonce" in params - assert "timestamp" in params - assert "signature" in params +def test_sign_params_include_production_required_fields(config): + params = BoxIMClient(config)._build_sign_params() + assert params["appId"] == "test_app" + assert params["accessId"] == "test_access" assert params["signType"] == "MD5" assert params["signVersion"] == "1.0" assert len(params["nonce"]) == 12 - - -def test_build_sign_params_excludes_signature_and_accessSecret_from_signing_string(mock_config): - """signature and accessSecret must be excluded from the signing string to match news_service.py.""" - from services.boxim_client import BoxIMClient - - client = BoxIMClient(mock_config) - - # Pass params that already contain a stale "signature" value - params_with_stale_sig = client._build_sign_params({ - "userId": "u1", - "signature": "OLD_STALE_SIG", - }) - - # The returned signature must be freshly computed (32-char MD5 uppercase), - # NOT the stale value we passed in. - assert params_with_stale_sig["signature"] != "OLD_STALE_SIG" - assert len(params_with_stale_sig["signature"]) == 32 - - # Calling with the same extra params but no stale signature should also work. - params_clean = client._build_sign_params({"userId": "u1"}) - assert len(params_clean["signature"]) == 32 - - -def test_build_sign_params_signature_is_deterministic(mock_config): - """Same inputs should produce valid MD5 signatures.""" - from services.boxim_client import BoxIMClient - - client = BoxIMClient(mock_config) - - params1 = client._build_sign_params({"userId": "u1"}) - params2 = client._build_sign_params({"userId": "u1"}) - - assert params1["signature"] is not None - assert params2["signature"] is not None - assert len(params1["signature"]) == 32 # MD5 hex length + assert len(params["timestamp"]) == 14 + assert len(params["signature"]) == 32 + assert "accessSecret" not in params diff --git a/digital-avatar-app/backend/tests/test_takeover_api.py b/digital-avatar-app/backend/tests/test_takeover_api.py index 902a2d5..7d0d527 100644 --- a/digital-avatar-app/backend/tests/test_takeover_api.py +++ b/digital-avatar-app/backend/tests/test_takeover_api.py @@ -1,10 +1,12 @@ -"""Tests for the authorization takeover configuration endpoint.""" +"""Tests for takeover configuration and BOXIM connection status.""" + +from datetime import datetime from fastapi.testclient import TestClient from database import SessionLocal from main import app -from models import Authorization +from models import Authorization, TakeoverCursor, TakeoverReplyTask, User client = TestClient(app) @@ -123,3 +125,81 @@ def test_takeover_is_limited_to_active_user_authorizations(authorization_context ).json() assert response["code"] == 400 assert "单聊接管" in response["message"] + + +def test_takeover_status_reports_disabled_and_requires_owner_login(authorization_context): + context = authorization_context + endpoint = f"/api/avatar/{context['avatar'].id}/takeover/status" + + disabled = client.get(endpoint, headers=context["owner_headers"]) + assert disabled.status_code == 200 + assert disabled.json()["data"]["status"] == "disabled" + + client.put( + f"/api/avatar/{context['avatar'].id}/permission-settings", + headers=context["owner_headers"], + json={"permissions": ["chat", "takeover"]}, + ) + needs_login = client.get(endpoint, headers=context["owner_headers"]).json()["data"] + assert needs_login["enabled"] is True + assert needs_login["status"] == "needs_login" + assert "BOXIM" in needs_login["message"] + + assert client.get(endpoint).status_code == 401 + assert client.get(endpoint, headers=context["other_headers"]).status_code == 403 + + +def test_takeover_status_reports_ready_pending_count_and_errors(authorization_context): + context = authorization_context + avatar_id = context["avatar"].id + endpoint = f"/api/avatar/{avatar_id}/takeover/status" + client.put( + f"/api/avatar/{avatar_id}/permission-settings", + headers=context["owner_headers"], + json={"permissions": ["chat", "takeover"]}, + ) + + db = SessionLocal() + try: + owner = db.query(User).filter(User.id == context["owner"].id).one() + owner.huihui_token = "production-login-token" + cursor = TakeoverCursor( + avatar_id=avatar_id, + owner_id=owner.huihui_user_id, + boxim_owner_id="100", + last_message_id="10", + initialized=True, + last_polled_at=datetime.utcnow(), + ) + task = TakeoverReplyTask( + avatar_id=avatar_id, + owner_id=owner.huihui_user_id, + peer_id="200", + trigger_message_id="11", + source_message_ids=["11"], + prompt="你好", + status="pending", + scheduled_at=datetime.utcnow(), + boxim_local_id="123", + ) + db.add_all([cursor, task]) + db.commit() + finally: + db.close() + + ready = client.get(endpoint, headers=context["owner_headers"]).json()["data"] + assert ready["status"] == "ready" + assert ready["pendingCount"] == 1 + assert ready["lastPolledAt"] + + db = SessionLocal() + try: + cursor = db.query(TakeoverCursor).filter(TakeoverCursor.avatar_id == avatar_id).one() + cursor.last_error = "BOXIM 暂时不可用" + db.commit() + finally: + db.close() + + failed = client.get(endpoint, headers=context["owner_headers"]).json()["data"] + assert failed["status"] == "error" + assert failed["message"] == "BOXIM 暂时不可用" diff --git a/digital-avatar-app/backend/tests/test_takeover_scheduler.py b/digital-avatar-app/backend/tests/test_takeover_scheduler.py index 5d02f3f..111c119 100644 --- a/digital-avatar-app/backend/tests/test_takeover_scheduler.py +++ b/digital-avatar-app/backend/tests/test_takeover_scheduler.py @@ -1,239 +1,83 @@ -"""Tests for the scheduled takeover message polling.""" -import json -import pytest -from unittest.mock import MagicMock, patch, AsyncMock +"""Tests for the BOXIM takeover scheduler lifecycle.""" + +from unittest.mock import AsyncMock, MagicMock, patch -def test_app_has_startup_event(): - """Verify the app has a startup event configured.""" +def test_app_has_startup_and_shutdown_events(): from main import app - startup_handlers = [handler for handler in app.router.on_startup] - assert len(startup_handlers) > 0 - -@patch("services.takeover_service.TakeoverService") -@patch("services.boxim_client.BoxIMClient") -@patch("main.redis_lib.from_url") -@patch("main.AsyncIOScheduler") -def test_scheduler_initialized_with_redis(mock_scheduler_class, mock_redis_from_url, mock_boxim_cls, mock_takeover_cls): - """Verify scheduler is initialized when Redis is available.""" - mock_redis = MagicMock() - mock_redis.ping.return_value = None - mock_redis_from_url.return_value = mock_redis - - mock_boxim = MagicMock() - mock_boxim_cls.return_value = mock_boxim - - mock_takeover = MagicMock() - mock_takeover.poll_and_process_messages = AsyncMock() - mock_takeover_cls.return_value = mock_takeover - - with patch("main.init_db"), patch("main.seed"), patch.dict("os.environ", {"REDIS_URL": "redis://localhost:6379"}): - from main import on_startup - on_startup() - - mock_scheduler_class.return_value.add_job.assert_called_once() - scheduled_callable = mock_scheduler_class.return_value.add_job.call_args.args[0] - call_kwargs = mock_scheduler_class.return_value.add_job.call_args[1] - assert scheduled_callable is mock_takeover.poll_and_process_messages - assert call_kwargs["id"] == "takeover_message_poll" - mock_scheduler_class.return_value.start.assert_called_once_with() + assert app.router.on_startup + assert app.router.on_shutdown @patch("services.takeover_service.TakeoverService") @patch("services.boxim_client.BoxIMClient") @patch("main.AsyncIOScheduler") -def test_scheduler_starts_without_redis(mock_scheduler_class, mock_boxim_cls, mock_takeover_cls): - """App should start even when REDIS_URL is not set.""" - mock_boxim = MagicMock() - mock_boxim_cls.return_value = mock_boxim - - mock_takeover = MagicMock() - mock_takeover.poll_and_process_messages = AsyncMock() - mock_takeover_cls.return_value = mock_takeover - - with patch("main.init_db"), patch("main.seed"), patch.dict("os.environ", {"REDIS_URL": ""}, clear=False): - from main import on_startup - on_startup() - - mock_scheduler_class.return_value.add_job.assert_called_once() - - -@patch("services.takeover_service.TakeoverService") -@patch("services.boxim_client.BoxIMClient") -@patch("main.redis_lib.from_url") -@patch("main.AsyncIOScheduler") -def test_scheduler_starts_when_redis_fails(mock_scheduler_class, mock_redis_from_url, mock_boxim_cls, mock_takeover_cls): - """App should start even when Redis ping fails.""" - mock_redis_from_url.side_effect = ConnectionError("Connection refused") - - mock_boxim = MagicMock() - mock_boxim_cls.return_value = mock_boxim - - mock_takeover = MagicMock() - mock_takeover.poll_and_process_messages = AsyncMock() - mock_takeover_cls.return_value = mock_takeover - - with patch("main.init_db"), patch("main.seed"), patch.dict("os.environ", {"REDIS_URL": "redis://badhost:6379"}): - from main import on_startup - on_startup() - - mock_scheduler_class.return_value.add_job.assert_called_once() - - -@patch("main.AsyncIOScheduler") -def test_scheduler_fails_gracefully(mock_scheduler_class): - """If scheduler init raises, the app should still start (exception caught).""" - mock_scheduler_class.side_effect = RuntimeError("Scheduler crash") - - with patch("main.init_db"), patch("main.seed"): - from main import on_startup - on_startup() - - # No exception should propagate - - -def test_scheduler_shutdown_releases_resources(): - """Shutdown should stop polling and close its dedicated database session.""" +def test_scheduler_uses_boxim_and_restart_safe_service( + mock_scheduler_class, + mock_boxim_class, + mock_takeover_class, +): import main - mock_scheduler = MagicMock() - mock_scheduler.running = True - mock_db = MagicMock() - main.takeover_scheduler = mock_scheduler - main.takeover_db = mock_db + scheduler = MagicMock() + mock_scheduler_class.return_value = scheduler + boxim = MagicMock() + mock_boxim_class.return_value = boxim + takeover = MagicMock() + takeover.poll_and_process_messages = AsyncMock() + mock_takeover_class.return_value = takeover + + environment = { + "HUIHUI_PLATFORM_BASE_URL": "https://open.example/api", + "BOXIM_API_BASE_URL": "https://im.example/api", + "HUIHUI_APP_ID": "app-id", + "HUIHUI_ACCESS_ID": "access-id", + "HUIHUI_ACCESS_SECRET": "secret", + "BOXIM_POLL_INTERVAL_SECONDS": "1", + } + with patch("main.init_db"), patch("main.seed"), patch.dict( + "os.environ", environment, clear=False + ): + main.on_startup() + + config = mock_boxim_class.call_args.args[0] + assert config["HUIHUI_PLATFORM_BASE_URL"] == "https://open.example/api" + assert config["BOXIM_API_BASE_URL"] == "https://im.example/api" + mock_takeover_class.assert_called_once_with(main.SessionLocal, boxim) + + scheduler.add_job.assert_called_once() + scheduled_callable = scheduler.add_job.call_args.args[0] + job_options = scheduler.add_job.call_args.kwargs + assert scheduled_callable is takeover.poll_and_process_messages + assert job_options["id"] == "takeover_message_poll" + assert job_options["trigger"].interval.total_seconds() == 1 + assert job_options["max_instances"] == 1 + assert job_options["coalesce"] is True + scheduler.start.assert_called_once_with() + + main.takeover_scheduler = None + + +@patch("main.AsyncIOScheduler") +def test_scheduler_failure_does_not_stop_the_api(mock_scheduler_class): + import main + + mock_scheduler_class.side_effect = RuntimeError("scheduler crash") + with patch("main.init_db"), patch("main.seed"): + main.on_startup() + + assert main.takeover_scheduler is None + + +def test_shutdown_stops_only_the_scheduler(): + import main + + scheduler = MagicMock() + scheduler.running = True + main.takeover_scheduler = scheduler main.on_shutdown() - mock_scheduler.shutdown.assert_called_once_with(wait=False) - mock_db.close.assert_called_once_with() + scheduler.shutdown.assert_called_once_with(wait=False) assert main.takeover_scheduler is None - assert main.takeover_db is None - - -# --- poll_and_process_messages --- - - -@pytest.fixture -def mock_db(): - return MagicMock() - - -@pytest.fixture -def mock_boxim(): - return AsyncMock() - - -@pytest.mark.asyncio -async def test_poll_and_process_messages_calls_fetch_and_process(mock_db, mock_boxim): - """poll_and_process_messages should fetch messages and process each.""" - from services.takeover_service import TakeoverService - - service = TakeoverService(mock_db, mock_boxim) - service.fetch_unread_messages = AsyncMock(return_value=[ - {"owner_huihui_id": "owner_1", "from_accid": "user_1", "content": "hi"}, - {"owner_huihui_id": "owner_2", "from_accid": "user_2", "content": "hello"}, - ]) - service.process_message = AsyncMock() - - await service.poll_and_process_messages() - - service.fetch_unread_messages.assert_awaited_once() - assert service.process_message.await_count == 2 - - -@pytest.mark.asyncio -async def test_poll_and_process_messages_handles_errors(mock_db, mock_boxim): - """poll_and_process_messages should not crash on fetch failure.""" - from services.takeover_service import TakeoverService - - service = TakeoverService(mock_db, mock_boxim) - service.fetch_unread_messages = AsyncMock(side_effect=ConnectionError("Box IM down")) - - await service.poll_and_process_messages() - # No exception should propagate - - -# --- process_message --- - - -@pytest.fixture -def mock_auth(): - auth = MagicMock() - auth.takeover_enabled = True - auth.takeover_mode = "immediate" - auth.takeover_delay_seconds = 30 - auth.avatar_id = "avatar_123" - auth.target_id = "target_user_123" - return auth - - -@pytest.mark.asyncio -async def test_process_message_immediate_mode(mock_db, mock_boxim, mock_auth): - """When takeover_mode is 'immediate', execute_takeover should be called.""" - from services.takeover_service import TakeoverService - - service = TakeoverService(mock_db, mock_boxim) - service.check_takeover_enabled = MagicMock(return_value=mock_auth) - service.execute_takeover = AsyncMock(return_value=True) - service.enqueue_delayed_message = MagicMock() - - message = {"owner_huihui_id": "owner_1", "from_accid": "user_1", "content": "hi"} - await service.process_message(message) - - service.execute_takeover.assert_awaited_once_with(mock_auth, message) - service.enqueue_delayed_message.assert_not_called() - - -@pytest.mark.asyncio -async def test_process_message_delayed_mode(mock_db, mock_boxim, mock_auth): - """When takeover_mode is not 'immediate', message should be enqueued.""" - from services.takeover_service import TakeoverService - - mock_auth.takeover_mode = "delayed" - - service = TakeoverService(mock_db, mock_boxim, MagicMock()) - service.check_takeover_enabled = MagicMock(return_value=mock_auth) - service.execute_takeover = AsyncMock() - service.enqueue_delayed_message = MagicMock() - - message = {"owner_huihui_id": "owner_1", "from_accid": "user_1", "content": "hi"} - await service.process_message(message) - - service.enqueue_delayed_message.assert_called_once_with(mock_auth, message) - service.execute_takeover.assert_not_awaited() - - -@pytest.mark.asyncio -async def test_process_message_delayed_mode_without_redis_falls_back_immediately(mock_db, mock_boxim, mock_auth): - """A missing Redis connection must not silently drop delayed replies.""" - from services.takeover_service import TakeoverService - - mock_auth.takeover_mode = "delayed" - service = TakeoverService(mock_db, mock_boxim) - service.check_takeover_enabled = MagicMock(return_value=mock_auth) - service.execute_takeover = AsyncMock(return_value=True) - service.enqueue_delayed_message = MagicMock() - - message = {"owner_huihui_id": "owner_1", "from_accid": "user_1", "content": "hi"} - await service.process_message(message) - - service.execute_takeover.assert_awaited_once_with(mock_auth, message) - service.enqueue_delayed_message.assert_not_called() - - -@pytest.mark.asyncio -async def test_process_message_no_takeover(mock_db, mock_boxim): - """When takeover is not enabled, nothing should happen.""" - from services.takeover_service import TakeoverService - - service = TakeoverService(mock_db, mock_boxim) - service.check_takeover_enabled = MagicMock(return_value=None) - service.execute_takeover = AsyncMock() - service.enqueue_delayed_message = MagicMock() - - message = {"owner_huihui_id": "owner_1", "from_accid": "user_1", "content": "hi"} - await service.process_message(message) - - service.execute_takeover.assert_not_awaited() - service.enqueue_delayed_message.assert_not_called() diff --git a/digital-avatar-app/backend/tests/test_takeover_service.py b/digital-avatar-app/backend/tests/test_takeover_service.py index f1fa2b5..0d98fce 100644 --- a/digital-avatar-app/backend/tests/test_takeover_service.py +++ b/digital-avatar-app/backend/tests/test_takeover_service.py @@ -1,322 +1,196 @@ -"""Tests for the TakeoverService — message listening, decision, reply execution.""" +"""End-to-end service tests for BOXIM takeover timing and human priority.""" + +from datetime import datetime, timedelta, timezone +from unittest.mock import patch + import pytest -from unittest.mock import AsyncMock, patch, MagicMock -from services.takeover_service import TakeoverService -from models import Authorization, Avatar +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker +from sqlalchemy.pool import StaticPool + +from database import Base +from models import Avatar, TakeoverCursor, TakeoverMessage, TakeoverReplyTask, User +from services.takeover_service import TakeoverService, _plain_text_reply + + +class Clock: + def __init__(self): + self.value = datetime(2026, 8, 19, 10, 0, 0) + + def now(self): + return self.value + + def advance(self, seconds: int): + self.value += timedelta(seconds=seconds) + + def millis(self): + return int(self.value.replace(tzinfo=timezone.utc).timestamp() * 1000) + + +class FakeBoxIM: + def __init__(self): + self.messages = [] + self.sent = [] + + async def exchange_access_token(self, huihui_token): + assert huihui_token == "prod-huihui-token" + return {"accessToken": "box-token", "accessTokenExpiresIn": 3600} + + async def get_self(self, access_token): + assert access_token == "box-token" + return {"id": 100} + + async def fetch_private_messages(self, access_token, min_id="0"): + assert access_token == "box-token" + return [item.copy() for item in self.messages if int(item["id"]) > int(min_id)] + + async def send_private_message(self, access_token, peer_id, content, *, local_id=None): + self.sent.append({"peerId": str(peer_id), "content": content, "localId": str(local_id)}) + return {"id": 900 + len(self.sent), "localId": int(local_id)} @pytest.fixture -def mock_db(): - db = MagicMock() - return db +def service_context(): + engine = create_engine( + "sqlite://", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + session_factory = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False) + Base.metadata.create_all(engine) + db = session_factory() + user = User( + id="owner-local", + huihui_user_id="owner-huihui", + huihui_token="prod-huihui-token", + app_token="app-token", + ) + avatar = Avatar( + id="avatar-1", + owner_id=user.huihui_user_id, + name="分身", + status="active", + config={"authorizationPermissions": ["chat", "takeover"]}, + ) + db.add_all([user, avatar]) + db.commit() + db.close() - -@pytest.fixture -def mock_boxim(): - client = AsyncMock() - client.get_credentials.return_value = {"accid": "owner_acc", "token": "tok"} - client.send_p2p_message.return_value = True - return client - - -@pytest.fixture -def mock_auth(): - auth = MagicMock(spec=Authorization) - auth.takeover_enabled = True - auth.takeover_mode = "immediate" - auth.takeover_delay_seconds = 30 - auth.avatar_id = "avatar_123" - auth.target_id = "target_user_123" - auth.target_type = "user" - auth.status = "active" - auth.permissions = ["chat", "takeover"] - return auth - - -@pytest.fixture -def mock_avatar(): - avatar = MagicMock(spec=Avatar) - avatar.id = "avatar_123" - avatar.owner_id = "owner_huihui_123" - return avatar - - -# --- check_takeover_enabled --- - - -def test_check_takeover_enabled_returns_auth_when_enabled(mock_db, mock_auth, mock_boxim, mock_avatar): - avatar_filter = MagicMock() - avatar_filter.first.return_value = mock_avatar - avatar_query = MagicMock() - avatar_query.filter.return_value = avatar_filter - - auth_filter = MagicMock() - auth_filter.filter.return_value = auth_filter - auth_filter.first.return_value = mock_auth - - def query_side_effect(model): - if model == Avatar: - return avatar_query - return auth_filter - - mock_db.query.side_effect = query_side_effect - - service = TakeoverService(mock_db, mock_boxim) - result = service.check_takeover_enabled("owner_huihui_123", "target_user_123") - assert result == mock_auth - - -def test_check_takeover_enabled_returns_none_when_no_avatar(mock_db, mock_boxim): - avatar_filter = MagicMock() - avatar_filter.first.return_value = None - avatar_query = MagicMock() - avatar_query.filter.return_value = avatar_filter - mock_db.query.return_value = avatar_query - - service = TakeoverService(mock_db, mock_boxim) - result = service.check_takeover_enabled("owner_123", "target_123") - assert result is None - - -def test_check_takeover_enabled_returns_none_when_disabled(mock_db, mock_boxim, mock_avatar): - avatar_filter = MagicMock() - avatar_filter.first.return_value = mock_avatar - avatar_query = MagicMock() - avatar_query.filter.return_value = avatar_filter - - disabled_auth = MagicMock(spec=Authorization) - disabled_auth.takeover_enabled = False - disabled_auth.permissions = [] - auth_filter = MagicMock() - auth_filter.filter.return_value = auth_filter - auth_filter.first.return_value = disabled_auth - - def query_side_effect(model): - if model == Avatar: - return avatar_query - return auth_filter - - mock_db.query.side_effect = query_side_effect - - service = TakeoverService(mock_db, mock_boxim) - result = service.check_takeover_enabled("owner_123", "target_123") - assert result is None - - -def test_check_takeover_enabled_filters_by_owner_and_target(mock_db, mock_boxim, mock_avatar, mock_auth): - """Verify that queries use the correct filter arguments.""" - avatar_filter = MagicMock() - avatar_filter.first.return_value = mock_avatar - avatar_query = MagicMock() - avatar_query.filter.return_value = avatar_filter - - auth_filter = MagicMock() - auth_filter.filter.return_value = auth_filter - auth_filter.first.return_value = mock_auth - - call_order = [] - - def query_side_effect(model): - if model == Avatar: - call_order.append("Avatar") - return avatar_query - call_order.append("Authorization") - return auth_filter - - mock_db.query.side_effect = query_side_effect - - service = TakeoverService(mock_db, mock_boxim) - service.check_takeover_enabled("owner_huihui_123", "target_user_123") - - assert "Avatar" in call_order - assert "Authorization" in call_order - - -# --- generate_reply --- + clock = Clock() + boxim = FakeBoxIM() + service = TakeoverService(session_factory, boxim, now=clock.now) + return session_factory, service, boxim, clock @pytest.mark.asyncio -async def test_generate_reply_returns_answer(mock_boxim): - mock_db = MagicMock() - with patch("services.takeover_service.httpx.AsyncClient") as mock_client_class: - mock_response = MagicMock() - mock_response.json.return_value = {"code": 200, "data": {"answer": "Hello back"}} - mock_client_class.return_value.__aenter__.return_value.post.return_value = mock_response +async def test_first_sync_primes_cursor_without_replying_to_history(service_context): + session_factory, service, boxim, clock = service_context + boxim.messages = [ + {"id": 10, "localId": 1, "sendId": 200, "recvId": 100, "sendTime": clock.millis(), "type": 0, "content": "旧消息"} + ] - service = TakeoverService(mock_db, mock_boxim) - result = await service.generate_reply("avatar_123", "Hello") - assert result == "Hello back" + with patch("routers.chat._resolve_reply", return_value={"answer": "不应发送"}): + await service.poll_and_process_messages() + + db = session_factory() + try: + cursor = db.query(TakeoverCursor).one() + assert cursor.initialized is True + assert cursor.last_message_id == "10" + assert db.query(TakeoverMessage).count() == 1 + assert db.query(TakeoverReplyTask).count() == 0 + assert boxim.sent == [] + finally: + db.close() @pytest.mark.asyncio -async def test_generate_reply_handles_empty_answer(mock_boxim): - """generate_reply should return empty string when answer is missing.""" - mock_db = MagicMock() - with patch("services.takeover_service.httpx.AsyncClient") as mock_client_class: - mock_response = MagicMock() - mock_response.json.return_value = {"code": 200, "data": {}} - mock_client_class.return_value.__aenter__.return_value.post.return_value = mock_response +async def test_incoming_message_is_prepared_then_sent_at_three_seconds(service_context): + session_factory, service, boxim, clock = service_context + await service.poll_and_process_messages() + boxim.messages.append( + {"id": 11, "localId": 2, "sendId": 200, "recvId": 100, "sendTime": clock.millis(), "type": 0, "content": "你好"} + ) - service = TakeoverService(mock_db, mock_boxim) - result = await service.generate_reply("avatar_123", "Hello") - assert result == "" + with patch("routers.chat._resolve_reply", return_value={"answer": "**你好**\n\n很高兴见到你"}): + await service.poll_and_process_messages() + assert boxim.sent == [] + + clock.advance(2) + await service.poll_and_process_messages() + assert boxim.sent == [] + + clock.advance(1) + await service.poll_and_process_messages() + assert boxim.sent == [{"peerId": "200", "content": "你好\n很高兴见到你", "localId": boxim.sent[0]["localId"]}] + + db = session_factory() + try: + task = db.query(TakeoverReplyTask).one() + assert task.status == "sent" + assert task.sent_at == clock.now() + finally: + db.close() @pytest.mark.asyncio -async def test_generate_reply_handles_error_code(mock_boxim): - """generate_reply should return empty string when API returns error code.""" - mock_db = MagicMock() - with patch("services.takeover_service.httpx.AsyncClient") as mock_client_class: - mock_response = MagicMock() - mock_response.json.return_value = {"code": 500, "message": "Internal error"} - mock_client_class.return_value.__aenter__.return_value.post.return_value = mock_response +async def test_owner_message_cancels_pending_reply(service_context): + session_factory, service, boxim, clock = service_context + await service.poll_and_process_messages() + boxim.messages.append( + {"id": 21, "localId": 3, "sendId": 200, "recvId": 100, "sendTime": clock.millis(), "type": 0, "content": "在吗"} + ) + with patch("routers.chat._resolve_reply", return_value={"answer": "在的"}): + await service.poll_and_process_messages() - service = TakeoverService(mock_db, mock_boxim) - result = await service.generate_reply("avatar_123", "Hello") - assert result == "" + clock.advance(2) + boxim.messages.append( + {"id": 22, "localId": 4, "sendId": 100, "recvId": 200, "sendTime": clock.millis(), "type": 0, "content": "我来回复"} + ) + await service.poll_and_process_messages() + clock.advance(2) + await service.poll_and_process_messages() - -# --- execute_takeover --- + db = session_factory() + try: + task = db.query(TakeoverReplyTask).filter(TakeoverReplyTask.trigger_message_id == "21").one() + assert task.status == "cancelled" + assert task.cancel_reason == "owner_replied" + assert boxim.sent == [] + finally: + db.close() @pytest.mark.asyncio -async def test_execute_takeover_success(mock_db, mock_boxim, mock_auth, mock_avatar): - avatar_filter = MagicMock() - avatar_filter.first.return_value = mock_avatar - avatar_query = MagicMock() - avatar_query.filter.return_value = avatar_filter - mock_db.query.return_value = avatar_query +async def test_quick_successive_messages_are_coalesced_into_one_reply(service_context): + session_factory, service, boxim, clock = service_context + await service.poll_and_process_messages() + boxim.messages.append( + {"id": 31, "localId": 5, "sendId": 200, "recvId": 100, "sendTime": clock.millis(), "type": 0, "content": "第一句"} + ) + with patch("routers.chat._resolve_reply", return_value={"answer": "第一版"}): + await service.poll_and_process_messages() - with patch("services.takeover_service.httpx.AsyncClient") as mock_client_class: - mock_response = MagicMock() - mock_response.json.return_value = {"code": 200, "data": {"answer": "Hello back"}} - mock_client_class.return_value.__aenter__.return_value.post.return_value = mock_response + clock.advance(1) + boxim.messages.append( + {"id": 32, "localId": 6, "sendId": 200, "recvId": 100, "sendTime": clock.millis(), "type": 0, "content": "第二句"} + ) + with patch("routers.chat._resolve_reply", return_value={"answer": "合并回复"}) as resolver: + await service.poll_and_process_messages() + assert resolver.call_args.args[2] == "第一句\n第二句" - service = TakeoverService(mock_db, mock_boxim) - message = {"from_accid": "user_acc", "content": "Hello"} + clock.advance(3) + await service.poll_and_process_messages() + assert [item["content"] for item in boxim.sent] == ["合并回复"] - result = await service.execute_takeover(mock_auth, message) - - assert result is True - mock_boxim.get_credentials.assert_called_once_with("owner_huihui_123") - mock_boxim.send_p2p_message.assert_called_once() + db = session_factory() + try: + tasks = db.query(TakeoverReplyTask).order_by(TakeoverReplyTask.created_at).all() + assert [task.status for task in tasks] == ["cancelled", "sent"] + assert tasks[0].cancel_reason == "newer_incoming_message" + finally: + db.close() -@pytest.mark.asyncio -async def test_execute_takeover_fails_when_avatar_not_found(mock_db, mock_boxim, mock_auth): - """execute_takeover should return False when Avatar is not found.""" - avatar_filter = MagicMock() - avatar_filter.first.return_value = None - avatar_query = MagicMock() - avatar_query.filter.return_value = avatar_filter - mock_db.query.return_value = avatar_query - - service = TakeoverService(mock_db, mock_boxim) - message = {"from_accid": "user_acc", "content": "Hello"} - - result = await service.execute_takeover(mock_auth, message) - - assert result is False - mock_boxim.get_credentials.assert_not_called() - - -@pytest.mark.asyncio -async def test_execute_takeover_fails_when_no_credentials(mock_db, mock_boxim, mock_auth, mock_avatar): - """execute_takeover should return False when boxim.get_credentials returns None.""" - avatar_filter = MagicMock() - avatar_filter.first.return_value = mock_avatar - avatar_query = MagicMock() - avatar_query.filter.return_value = avatar_filter - mock_db.query.return_value = avatar_query - - mock_boxim.get_credentials.return_value = None - service = TakeoverService(mock_db, mock_boxim) - message = {"from_accid": "user_acc", "content": "Hello"} - - result = await service.execute_takeover(mock_auth, message) - - assert result is False - - -# --- enqueue_delayed_message --- - - -def test_enqueue_delayed_message_with_redis(mock_db, mock_boxim, mock_auth, mock_avatar): - mock_redis = MagicMock() - avatar_filter = MagicMock() - avatar_filter.first.return_value = mock_avatar - avatar_query = MagicMock() - avatar_query.filter.return_value = avatar_filter - mock_db.query.return_value = avatar_query - - service = TakeoverService(mock_db, mock_boxim, mock_redis) - message = {"msg_id": "msg_1", "from_accid": "user_acc", "content": "Hello"} - - service.enqueue_delayed_message(mock_auth, message) - - mock_redis.setex.assert_called_once() - call_args = mock_redis.setex.call_args - value = call_args[0][1] - import json - payload = json.loads(call_args[0][2]) - assert payload["owner_huihui_id"] == "owner_huihui_123" - - -def test_enqueue_delayed_message_without_redis_logs_warning(mock_db, mock_boxim, mock_auth, mock_avatar): - """When Redis is not configured, enqueue_delayed_message should log a warning and not crash.""" - avatar_filter = MagicMock() - avatar_filter.first.return_value = mock_avatar - avatar_query = MagicMock() - avatar_query.filter.return_value = avatar_filter - mock_db.query.return_value = avatar_query - - service = TakeoverService(mock_db, mock_boxim) - message = {"msg_id": "msg_1", "from_accid": "user_acc", "content": "Hello"} - - service.enqueue_delayed_message(mock_auth, message) - - -# --- process_delayed_queue --- - - -@pytest.mark.asyncio -async def test_process_delayed_queue_no_redis(mock_db, mock_boxim): - """process_delayed_queue should return immediately without Redis.""" - service = TakeoverService(mock_db, mock_boxim) - await service.process_delayed_queue() - mock_db.query.assert_not_called() - - -@pytest.mark.asyncio -async def test_process_delayed_queue_processes_messages(mock_db, mock_boxim, mock_auth, mock_avatar): - """process_delayed_queue should read from Redis, resolve auth, and execute takeover.""" - mock_redis = MagicMock() - mock_redis.keys.return_value = ["takeover:delayed:target_user_123:msg_1"] - mock_redis.get.return_value = '{"from_accid": "user_acc", "content": "Hello"}' - - avatar_filter = MagicMock() - avatar_filter.first.return_value = mock_avatar - avatar_query = MagicMock() - avatar_query.filter.return_value = avatar_filter - - auth_filter = MagicMock() - auth_filter.filter.return_value = auth_filter - auth_filter.first.return_value = mock_auth - - def query_side_effect(model): - if model == Avatar: - return avatar_query - return auth_filter - - mock_db.query.side_effect = query_side_effect - - with patch("services.takeover_service.httpx.AsyncClient") as mock_client_class: - mock_response = MagicMock() - mock_response.json.return_value = {"code": 200, "data": {"answer": "Hello back"}} - mock_client_class.return_value.__aenter__.return_value.post.return_value = mock_response - - service = TakeoverService(mock_db, mock_boxim, mock_redis) - await service.process_delayed_queue() - - mock_boxim.send_p2p_message.assert_called_once() - mock_redis.delete.assert_called_once() +def test_plain_text_reply_removes_markdown_and_empty_lines(): + assert _plain_text_reply("## 建议\n\n**不能自行用药**\n`必要时就医`") == "建议\n不能自行用药\n必要时就医" diff --git a/digital-avatar-app/src/api/index.ts b/digital-avatar-app/src/api/index.ts index 4ad37cd..59f0166 100644 --- a/digital-avatar-app/src/api/index.ts +++ b/digital-avatar-app/src/api/index.ts @@ -145,7 +145,7 @@ export const chargeToken = (planId: string) => // ==================== 授权管理 API ==================== -export type AvatarPermission = 'friend' | 'chat' | 'publish' | 'browse' | 'interact' +export type AvatarPermission = 'friend' | 'chat' | 'publish' | 'browse' | 'interact' | 'takeover' export interface AvatarPermissionSettings { avatarId: string @@ -158,6 +158,17 @@ export const getAvatarPermissionSettings = (avatarId: string) => export const updateAvatarPermissionSettings = (avatarId: string, permissions: AvatarPermission[]) => request.put(`/avatar/${avatarId}/permission-settings`, { permissions }) +export interface TakeoverStatus { + enabled: boolean + status: 'disabled' | 'connecting' | 'ready' | 'needs_login' | 'error' + message: string + pendingCount: number + lastPolledAt: string | null +} + +export const getTakeoverStatus = (avatarId: string) => + request.get(`/avatar/${avatarId}/takeover/status`) + export interface Authorization { id: string avatarId: string diff --git a/digital-avatar-app/src/views/AuthorizationManage.vue b/digital-avatar-app/src/views/AuthorizationManage.vue index fa44e9d..6817329 100644 --- a/digital-avatar-app/src/views/AuthorizationManage.vue +++ b/digital-avatar-app/src/views/AuthorizationManage.vue @@ -27,6 +27,7 @@ v-for="item in permissionItems" :key="item.key" class="permission-row" + :class="{ takeover: item.key === 'takeover' }" type="button" role="switch" :aria-checked="permissionState[item.key]" @@ -51,14 +52,18 @@ - + + + + + {{ item.title }} - {{ item.description }} + {{ item.key === 'takeover' ? takeoverDescription : item.description }}