1051 lines
38 KiB
Python
1051 lines
38 KiB
Python
"""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"(?<!\*)\*([^*\n]+)\*(?!\*)", r"\1", text)
|
|
text = re.sub(r"`([^`]+)`", r"\1", text)
|
|
text = re.sub(r"^\s{0,3}#{1,6}\s*", "", text, flags=re.M)
|
|
lines = [line.strip() for line in text.split("\n")]
|
|
return "\n".join(line for line in lines if line).strip()
|
|
|
|
|
|
def _avatar_local_id(owner_id: str, trigger_message_id: str) -> 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()
|