huihuiSquare/digital-avatar-app/backend/routers/takeover.py

"""数字分身 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 (
    DEFAULT_TAKEOVER_DELAY_SECONDS,
    MAX_TAKEOVER_DELAY_SECONDS,
    MIN_TAKEOVER_DELAY_SECONDS,
    _require_authorization,
    _stored_takeover_delay,
)
from routers.avatars import _require_owned_avatar

router = APIRouter(tags=["分身接管"])
BOXIM_STATUS_FRESH_SECONDS = 60


def _delay_label(seconds: int) -> str:
    if seconds % 60 == 0:
        return f"{seconds // 60} 分钟"
    return f"{seconds} 秒"


@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
    reply_delay_seconds = _stored_takeover_delay(avatar)
    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 cursor and cursor.last_error:
        status, message = "error", cursor.last_error
    elif not enabled:
        status, message = "disabled", "主动接管未开启"
    elif not user or not user.huihui_token:
        status, message = "needs_login", "请重新登录会会生产账号以连接 BOXIM"
    elif (
        cursor
        and cursor.initialized
        and cursor.last_polled_at
        # BOXIM offline-message reads can long-poll for about 20 seconds.
        and cursor.last_polled_at
        >= datetime.utcnow() - timedelta(seconds=BOXIM_STATUS_FRESH_SECONDS)
    ):
        status, message = (
            "ready",
            f"BOXIM 已连接，收到私聊消息 {_delay_label(reply_delay_seconds)}后自动回复",
        )
    else:
        status, message = "connecting", "正在连接 BOXIM"

    return ok(
        {
            "enabled": enabled,
            "status": status,
            "message": message,
            "pendingCount": pending_count,
            "takeoverReplyDelaySeconds": reply_delay_seconds,
            "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


def _read(payload: dict, camel_key: str, snake_key: str, default=None):
    if camel_key in payload:
        return payload[camel_key]
    if snake_key in payload:
        return payload[snake_key]
    return default


@router.put("/avatar/{avatar_id}/authorizations/takeover")
def update_takeover_config(
    avatar_id: str,
    payload: dict = Body(...),
    authorization: str = Header(None),
    db: Session = Depends(get_db),
):
    _require_owned_avatar(db, avatar_id, authorization)
    auth_id = _read(payload, "authorizationId", "authorization_id")
    if not auth_id:
        return fail("缺少 authorization_id", 400)

    auth = _require_authorization(db, avatar_id, str(auth_id))
    enabled = bool(auth.takeover_enabled)
    mode = auth.takeover_mode or "immediate"
    delay = auth.takeover_delay_seconds or DEFAULT_TAKEOVER_DELAY_SECONDS

    if _has(payload, "takeoverEnabled", "takeover_enabled"):
        raw_enabled = _read(payload, "takeoverEnabled", "takeover_enabled")
        if not isinstance(raw_enabled, bool):
            return fail("takeover_enabled 必须是布尔值", 400)
        enabled = raw_enabled

    if _has(payload, "takeoverMode", "takeover_mode"):
        mode = _read(payload, "takeoverMode", "takeover_mode")
        if mode not in ("immediate", "delayed"):
            return fail("takeover_mode 必须是 immediate 或 delayed", 400)

    if _has(payload, "takeoverDelaySeconds", "takeover_delay_seconds"):
        delay = _read(payload, "takeoverDelaySeconds", "takeover_delay_seconds")
        if (
            isinstance(delay, bool)
            or not isinstance(delay, int)
            or not MIN_TAKEOVER_DELAY_SECONDS <= delay <= MAX_TAKEOVER_DELAY_SECONDS
        ):
            return fail("延迟时间需在 3 秒到 24 小时之间", 400)

    if enabled and auth.target_type != "user":
        return fail("本期仅支持对会会用户开启单聊接管", 400)
    if enabled and auth.status != "active":
        return fail("请先启用该授权，再开启聊天接管", 400)

    permissions = list(auth.permissions or [])
    if enabled:
        if "chat" not in permissions and "reply" not in permissions:
            permissions.append("chat")
        if "takeover" not in permissions:
            permissions.append("takeover")
    else:
        permissions = [permission for permission in permissions if permission != "takeover"]

    auth.permissions = permissions
    auth.takeover_enabled = enabled
    auth.takeover_mode = mode
    auth.takeover_delay_seconds = delay
    db.commit()
    db.refresh(auth)
    return ok(auth.to_dict(), "接管配置已保存")