"""Restart-safe automatic replies over Huihui's self-hosted BOXIM.""" import asyncio import hashlib import logging import os import re import secrets import time from datetime import datetime, timedelta from typing import Callable from sqlalchemy.orm import Session from models import ( Avatar, ChatAttachment, TakeoverCursor, TakeoverMessage, TakeoverReplyTask, User, ) from services.boxim_client import BoxIMClient, BoxIMError from services.boxim_image_service import ( BoxIMImageError, download_boxim_image, parse_boxim_image_url, ) from services.vision_service import ImageValidationError logger = logging.getLogger(__name__) ACTIVE_TASK_STATUSES = ("pending", "generating", "ready", "sending") GENERATABLE_TASK_STATUSES = ("pending",) MAX_PROMPT_LENGTH = 4000 DEFAULT_MAX_MESSAGE_AGE_SECONDS = 600 MAX_SEND_OVERDUE_SECONDS = 120 STUCK_LOCK_SECONDS = 90 TAKEOVER_PERMISSION = "takeover" TAKEOVER_DELAY_KEY = "takeoverReplyDelaySeconds" DEFAULT_REPLY_DELAY_SECONDS = 180 MIN_REPLY_DELAY_SECONDS = 3 MAX_REPLY_DELAY_SECONDS = 86_400 HUMAN_PAUSE_SECONDS = 600 RATE_LIMIT_WINDOW_SECONDS = 300 RATE_LIMIT_MAX_REPLIES = 5 AVATAR_LOCAL_ID_PREFIX = "880" BOXIM_TEXT_MESSAGE_TYPE = 0 BOXIM_IMAGE_MESSAGE_TYPE = 1 BOXIM_IMAGE_PROMPT = "请看看这张图片。" BOXIM_IMAGE_UNAVAILABLE_REPLY = "这张图片我暂时没看清，麻烦重新发送一张清晰的原图。" IMAGE_CONTEXT_LOOKBACK_SECONDS = 1800 IMAGE_REFERENCE_LOOKBACK_SECONDS = 172_800 MAX_RECENT_IMAGE_CONTEXTS = 3 _IMAGE_REFERENCE_PATTERN = re.compile( r"(?:图片|图像|照片|截图|这张图|刚才.{0,8}图|病例|病历|检查单|检验单|化验单|报告|影像|" r"\b(?:image|photo|picture|screenshot|scan|report)\b)", re.IGNORECASE, ) 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"(? str: """Build a deterministic BOXIM idempotency key that also marks avatar traffic.""" digest = hashlib.sha256(f"{owner_id}:{trigger_message_id}".encode("utf-8")).digest() suffix = int.from_bytes(digest[:8], "big") % (10**15) return f"{AVATAR_LOCAL_ID_PREFIX}{suffix:015d}" def _is_avatar_local_id(value: str | None) -> bool: local_id = str(value or "").strip() return len(local_id) == 18 and local_id.isdigit() and local_id.startswith(AVATAR_LOCAL_ID_PREFIX) def _configured_reply_delay(avatar: Avatar, fallback: int | None = None) -> int: raw = (avatar.config or {}).get( TAKEOVER_DELAY_KEY, fallback if fallback is not None else DEFAULT_REPLY_DELAY_SECONDS, ) if isinstance(raw, bool): return DEFAULT_REPLY_DELAY_SECONDS try: delay = int(raw) except (TypeError, ValueError): return DEFAULT_REPLY_DELAY_SECONDS if not MIN_REPLY_DELAY_SECONDS <= delay <= MAX_REPLY_DELAY_SECONDS: return DEFAULT_REPLY_DELAY_SECONDS return delay def _event_prompt(event: TakeoverMessage) -> str: if event.message_type == BOXIM_TEXT_MESSAGE_TYPE: return event.content.strip() if event.message_type == BOXIM_IMAGE_MESSAGE_TYPE: return BOXIM_IMAGE_PROMPT return "" def _references_recent_image(value: str) -> bool: return bool(_IMAGE_REFERENCE_PATTERN.search(value or "")) class TakeoverService: """Poll BOXIM, honor the owner grace period, then generate and send one reply.""" def __init__( self, session_factory: Callable[[], Session], boxim_client: BoxIMClient, *, reply_delay_seconds: int | None = None, poll_concurrency: int = 8, max_message_age_seconds: int = DEFAULT_MAX_MESSAGE_AGE_SECONDS, now: Callable[[], datetime] = _utcnow, ): self.session_factory = session_factory self.boxim = boxim_client self.reply_delay_seconds = reply_delay_seconds self.poll_concurrency = max(1, min(int(poll_concurrency), 64)) self.max_message_age_seconds = max(60, int(max_message_age_seconds)) self.now = now self._sessions: dict[str, dict] = {} self._poll_lock = asyncio.Lock() self._process_lock = asyncio.Lock() self._persist_lock = asyncio.Lock() async def poll_and_process_messages(self): """Run one complete cycle for callers that do not use the split scheduler.""" await self.poll_messages() await self.process_reply_tasks() async def poll_messages(self): """Fetch BOXIM events without blocking reply generation and dispatch.""" if self._poll_lock.locked(): return async with self._poll_lock: self._recover_stuck_tasks() avatar_ids = self._enabled_avatar_ids() self._cancel_disabled_tasks(set(avatar_ids)) self._ensure_takeover_cursors(avatar_ids) semaphore = asyncio.Semaphore(self.poll_concurrency) async def sync(avatar_id: str): async with semaphore: return await self._sync_avatar(avatar_id) results = await asyncio.gather( *(sync(avatar_id) for avatar_id in avatar_ids), return_exceptions=True, ) for avatar_id, result in zip(avatar_ids, results): if isinstance(result, Exception): logger.warning("BOXIM poll crashed for avatar %s: %s", avatar_id, result) def _ensure_takeover_cursors(self, avatar_ids: list[str]): """Create durable cursors before concurrent network polling starts.""" if not avatar_ids: return db = self.session_factory() try: existing = { row[0] for row in db.query(TakeoverCursor.avatar_id) .filter(TakeoverCursor.avatar_id.in_(avatar_ids)) .all() } avatars = ( db.query(Avatar.id, Avatar.owner_id) .filter( Avatar.id.in_( [avatar_id for avatar_id in avatar_ids if avatar_id not in existing] ) ) .all() ) for avatar_id, owner_id in avatars: db.add(TakeoverCursor(avatar_id=avatar_id, owner_id=owner_id)) if avatars: db.commit() finally: db.close() async def process_reply_tasks(self): """Generate and send replies independently from BOXIM's long poll.""" if self._process_lock.locked(): return async with self._process_lock: self._recover_stuck_tasks() avatar_ids = set(self._enabled_avatar_ids()) self._cancel_disabled_tasks(avatar_ids) await self._prepare_replies() await self._dispatch_ready_replies() def _enabled_avatar_ids(self) -> list[str]: db = self.session_factory() try: avatars = ( db.query(Avatar) .filter(Avatar.status == "active") .order_by(Avatar.updated_at.desc(), Avatar.created_at.desc()) .all() ) selected = {} for avatar in avatars: if _takeover_enabled(avatar) and avatar.owner_id not in selected: selected[avatar.owner_id] = avatar.id return list(selected.values()) finally: db.close() 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() 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() 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) def _record_connection_failure( self, db: Session, avatar: Avatar, cursor: TakeoverCursor, message: str, *, disable_takeover: bool, ): cursor.last_error = message cursor.last_polled_at = self.now() if not disable_takeover: return permissions = (avatar.config or {}).get("authorizationPermissions", []) avatar.config = { **(avatar.config or {}), "authorizationPermissions": [ permission for permission in permissions if permission != TAKEOVER_PERMISSION ], } tasks = ( db.query(TakeoverReplyTask) .filter( TakeoverReplyTask.avatar_id == avatar.id, TakeoverReplyTask.status.in_(ACTIVE_TASK_STATUSES), ) .all() ) for task in tasks: task.status = "cancelled" task.cancel_reason = "connection_failed" task.locked_at = 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.commit() else: # Release SQLite's read transaction before the long network poll. db.commit() if not user or not user.huihui_token: self._record_connection_failure( db, avatar, cursor, "请重新登录会会生产账号后再开启主动接管", disable_takeover=True, ) 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) message = "BOXIM 授权已失效，请重新登录会会生产账号" disable_takeover = True else: message = f"BOXIM 暂时连接失败：{str(exc)[:160]}" disable_takeover = False self._record_connection_failure( db, avatar, cursor, message, disable_takeover=disable_takeover, ) db.commit() logger.warning( "BOXIM sync failed for avatar %s (will_retry=%s): %s", avatar.id, not disable_takeover, 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) read_receipts: dict[str, int] = {} for message in messages: message_id = _numeric_id(message.get("id")) max_message_id = max(max_message_id, message_id) send_id = str(message.get("sendId") or "") recv_id = str(message.get("recvId") or "") if recv_id == cursor.boxim_owner_id and send_id and message_id: read_receipts[send_id] = max(read_receipts.get(send_id, 0), message_id) # BOXIM publishes this HTTP state change to connected socket clients. # Do it before advancing the cursor so a failed receipt is retried. for peer_id, message_id in read_receipts.items(): await self.boxim.mark_private_messages_read( session["access_token"], peer_id, message_id ) # Keep SQLite write transactions short. The read-receipt request above # can block on the network and must not hold the database write lock. async with self._persist_lock: for message in messages: self._record_message( db, avatar, cursor.boxim_owner_id, message, schedule_reply=not priming, ) 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 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: return if not peer_id: return now = self.now() send_time = _boxim_time(message.get("sendTime"), now) is_avatar = _is_avatar_local_id(local_id) if not is_avatar and local_id: is_avatar = bool( db.query(TakeoverReplyTask) .filter( TakeoverReplyTask.boxim_local_id == local_id, TakeoverReplyTask.status.in_(("ready", "sending", "sent")), ) .first() ) if not is_avatar: is_avatar = bool( db.query(TakeoverReplyTask) .filter( TakeoverReplyTask.boxim_sent_message_id == message_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 not in { BOXIM_TEXT_MESSAGE_TYPE, BOXIM_IMAGE_MESSAGE_TYPE, }: return if event.message_type == BOXIM_TEXT_MESSAGE_TYPE and not event.content.strip(): return if event.message_type == BOXIM_IMAGE_MESSAGE_TYPE: try: parse_boxim_image_url( event.content, base_url=getattr(self.boxim, "im_base_url", ""), ) except BoxIMImageError as exc: logger.warning( "Ignored invalid BOXIM image message %s for avatar %s: %s", message_id, avatar.id, exc, ) return if (now - send_time).total_seconds() > self.max_message_age_seconds: logger.info( "Ignored stale BOXIM message %s for avatar %s (age=%ss)", message_id, avatar.id, int((now - send_time).total_seconds()), ) return if is_avatar: self._cancel_conversation(db, avatar.owner_id, peer_id, "peer_avatar_message") logger.info( "Skipped BOXIM reply for avatar %s message %s: peer_avatar_message", avatar.id, message_id, ) return if self._human_pause_active(db, avatar.owner_id, peer_id, now): self._cancel_conversation(db, avatar.owner_id, peer_id, "owner_active") logger.info( "Skipped BOXIM reply for avatar %s message %s: owner_active", avatar.id, message_id, ) return if self._conversation_rate_limited(db, avatar.owner_id, peer_id, now): self._cancel_conversation(db, avatar.owner_id, peer_id, "rate_limited") logger.info( "Skipped BOXIM reply for avatar %s message %s: rate_limited", avatar.id, message_id, ) return self._schedule_reply(db, avatar, event) @staticmethod def _human_pause_active(db: Session, owner_id: str, peer_id: str, now: datetime) -> bool: threshold = now - timedelta(seconds=HUMAN_PAUSE_SECONDS) return bool( db.query(TakeoverMessage.id) .filter( TakeoverMessage.owner_id == owner_id, TakeoverMessage.peer_id == peer_id, TakeoverMessage.direction == "outgoing", TakeoverMessage.is_avatar.is_(False), TakeoverMessage.send_time >= threshold, ) .first() ) @staticmethod def _conversation_rate_limited( db: Session, owner_id: str, peer_id: str, now: datetime, ) -> bool: threshold = now - timedelta(seconds=RATE_LIMIT_WINDOW_SECONDS) return ( db.query(TakeoverReplyTask.id) .filter( TakeoverReplyTask.owner_id == owner_id, TakeoverReplyTask.peer_id == peer_id, TakeoverReplyTask.status == "sent", TakeoverReplyTask.sent_at >= threshold, ) .count() >= RATE_LIMIT_MAX_REPLIES ) @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 if event.message_type == BOXIM_TEXT_MESSAGE_TYPE: for image_event in self._recent_unhandled_images( db, avatar, event, source_ids, ): prompt_parts.append(_event_prompt(image_event)) source_ids.append(image_event.boxim_message_id) prompt_parts.append(_event_prompt(event)) source_ids.append(event.boxim_message_id) prompt = "\n".join(part for part in prompt_parts if part).strip()[-MAX_PROMPT_LENGTH:] due_at = max( event.send_time + timedelta(seconds=_configured_reply_delay(avatar, self.reply_delay_seconds)), self.now(), ) task_id = secrets.token_hex(16) local_id = _avatar_local_id(avatar.owner_id, event.boxim_message_id) 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), ) ) @staticmethod def _recent_unhandled_images( db: Session, avatar: Avatar, event: TakeoverMessage, current_source_ids: list[str], ) -> list[TakeoverMessage]: """Recover missed images, or reuse a referenced image from the last two days.""" references_image = _references_recent_image(event.content) lookback_seconds = ( IMAGE_REFERENCE_LOOKBACK_SECONDS if references_image else IMAGE_CONTEXT_LOOKBACK_SECONDS ) threshold = event.send_time - timedelta(seconds=lookback_seconds) candidates = ( db.query(TakeoverMessage) .filter( TakeoverMessage.avatar_id == avatar.id, TakeoverMessage.owner_id == avatar.owner_id, TakeoverMessage.peer_id == event.peer_id, TakeoverMessage.direction == "incoming", TakeoverMessage.message_type == BOXIM_IMAGE_MESSAGE_TYPE, TakeoverMessage.is_avatar.is_(False), TakeoverMessage.send_time >= threshold, TakeoverMessage.send_time <= event.send_time, ) .order_by(TakeoverMessage.send_time.desc()) .limit(MAX_RECENT_IMAGE_CONTEXTS) .all() ) if not candidates: return [] current_ids = set(current_source_ids) if references_image: return [ image for image in reversed(candidates) if image.boxim_message_id not in current_ids ] handled_ids = set(current_ids) task_sources = ( db.query(TakeoverReplyTask.source_message_ids) .filter( TakeoverReplyTask.avatar_id == avatar.id, TakeoverReplyTask.owner_id == avatar.owner_id, TakeoverReplyTask.peer_id == event.peer_id, TakeoverReplyTask.created_at >= threshold, ) .all() ) for (source_message_ids,) in task_sources: handled_ids.update(source_message_ids or []) return [ image for image in reversed(candidates) if image.boxim_message_id not in handled_ids ] 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 == "", TakeoverReplyTask.scheduled_at <= self.now(), ) .order_by(TakeoverReplyTask.created_at.asc()) .limit(10) .all() ) ] finally: db.close() if not task_ids: return 0 # Each conversation owns its task, so unrelated contacts can generate in # parallel instead of one slow model response delaying every other peer. semaphore = asyncio.Semaphore(4) async def generate(task_id: str) -> bool: async with semaphore: return await asyncio.to_thread(self._generate_reply, task_id) results = await asyncio.gather(*(generate(task_id) for task_id in task_ids)) return sum(bool(result) for result in results) def _takeover_image_attachment( self, db: Session, avatar: Avatar, event: TakeoverMessage, ) -> ChatAttachment: now = self.now() if event.attachment_id: cached = db.get(ChatAttachment, event.attachment_id) if cached and cached.status == "ready" and cached.expires_at > now: cached.used_at = now db.commit() return cached downloaded = download_boxim_image( event.content, base_url=getattr( self.boxim, "im_base_url", os.getenv("BOXIM_API_BASE_URL", "https://im.99hui.com/api"), ), ) from routers.chat import _analyze_image_bytes attachment = _analyze_image_bytes( db, avatar, downloaded.content, filename=downloaded.filename, mime_type=downloaded.mime_type, uploader_kind="boxim", ) event.attachment_id = attachment.id attachment.used_at = now db.commit() logger.info( "BOXIM image analyzed message=%s attachment=%s avatar=%s category=%s", event.boxim_message_id, attachment.id, avatar.id, attachment.category, ) return attachment 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 []) source_events = { event.boxim_message_id: event for event in ( db.query(TakeoverMessage) .filter( TakeoverMessage.owner_id == task.owner_id, TakeoverMessage.peer_id == task.peer_id, TakeoverMessage.avatar_id == task.avatar_id, TakeoverMessage.boxim_message_id.in_(excluded_ids), ) .all() if excluded_ids else [] ) } events = ( db.query(TakeoverMessage) .filter( TakeoverMessage.owner_id == task.owner_id, TakeoverMessage.peer_id == task.peer_id, TakeoverMessage.avatar_id == task.avatar_id, ) .order_by(TakeoverMessage.send_time.desc()) .limit(30) .all() ) image_attachments = [] image_failed = False for message_id in (task.source_message_ids or [])[-3:]: event = source_events.get(message_id) if not event or event.message_type != BOXIM_IMAGE_MESSAGE_TYPE: continue try: image_attachments.append( self._takeover_image_attachment(db, avatar, event) ) except (BoxIMImageError, ImageValidationError) as exc: image_failed = True logger.warning( "BOXIM image unavailable message=%s avatar=%s: %s", event.boxim_message_id, avatar.id, exc, ) history = [] for event in reversed(events): if ( event.boxim_message_id in excluded_ids or event.message_type != BOXIM_TEXT_MESSAGE_TYPE or not event.content.strip() ): continue if event.direction == "incoming" and event.is_avatar: continue history.append( { "role": "user" if event.direction == "incoming" else "assistant", "content": event.content.strip(), } ) history = history[-10:] from routers.chat import _attachment_contexts, _resolve_reply image_contexts = _attachment_contexts(image_attachments) if image_failed and not image_contexts: answer = BOXIM_IMAGE_UNAVAILABLE_REPLY else: result = _resolve_reply( db, avatar, task.prompt, history, usage_source="takeover", image_contexts=image_contexts, ) 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() if task_ids: await asyncio.gather(*(self._send_task(task_id) for task_id in task_ids)) 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_SEND_OVERDUE_SECONDS: task.status = "cancelled" task.cancel_reason = "stale_reply" db.commit() return False cursor = ( db.query(TakeoverCursor) .filter(TakeoverCursor.avatar_id == task.avatar_id) .first() ) if not cursor or not cursor.last_polled_at or cursor.last_polled_at < task.scheduled_at: # Do not race the owner's final seconds of the grace period. A # completed poll at/after the due time must confirm no human reply. 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()