fix: sync Huihui app user homepage profile

PUT /api/avatar/{avatar_id}/authorizations/takeover to update authorization
takeover settings (enabled, mode, delay_seconds) with validation.

backend/app/api/endpoints/avatars.py

@@ -1,8 +1,14 @@
 """数字分身管理 API 端点"""
-from fastapi import APIRouter, Query, HTTPException
+from typing import Optional
+import mimetypes
+import uuid
+from fastapi import APIRouter, Query, HTTPException, UploadFile, File
+from fastapi.responses import JSONResponse
 
 from app.schemas import ApiResponse
 from app.services.avatar_service import avatar_service, get_session, is_available
+from app.core.database import AsyncSessionLocal
+from app.core.config import settings
 
 router = APIRouter()
 
@@ -69,3 +75,127 @@ def update_avatar_status(avatar_id: str, body: dict):
         return ApiResponse(data=data, message=msg)
     finally:
         db.close()
+
+
+async def _upload_to_filecenter(file_bytes: bytes, filename: str) -> str:
+    """利用会会平台 filecenter 上传头像，返回 URL"""
+    import httpx
+    import hashlib
+    import random
+    from datetime import datetime
+
+    async with AsyncSessionLocal() as db:
+        # 读取平台配置
+        from sqlalchemy import select
+        from app.models import SystemConfig
+        result = await db.execute(select(SystemConfig))
+        configs = {row.config_key: row.config_value for row in result.scalars().all()}
+
+        cfg = {
+            "appId": configs.get("platform_app_id", ""),
+            "accessId": configs.get("platform_access_id", ""),
+            "accessSecret": configs.get("platform_access_secret", ""),
+        }
+        biz_url = configs.get("news_platform_base_url", "http://192.168.1.200:63120")
+
+    if not cfg["accessSecret"]:
+        raise RuntimeError("平台 accessSecret 未配置")
+
+    # 构建签名参数
+    nonce = str(random.random())[2:][: random.randint(8, 12)]
+    timestamp = datetime.now().strftime("%Y%m%d%I%M%S")
+    sign_params = {
+        "appId": cfg["appId"],
+        "accessId": cfg["accessId"],
+        "timestamp": timestamp,
+        "nonce": nonce,
+        "module": "userInfo",
+        "service": "kccloud",
+    }
+
+    # 计算签名
+    keys = sorted(sign_params.keys())
+    sign_parts = []
+    for k in keys:
+        v = sign_params.get(k)
+        if v and v != "" and v != []:
+            sign_parts.append(f"{k}={v}")
+    sign_str = "&".join(sign_parts) + f"&accessSecret={cfg['accessSecret']}"
+    signature = hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()
+
+    # 构建 filecenter URL
+    filecenter_url = biz_url.replace("/huihuibusiness", "/filecenter")
+    if "/api/" in filecenter_url:
+        filecenter_url = filecenter_url.split("/api/", 1)[0] + "/api/filecenter"
+    else:
+        filecenter_url = filecenter_url.rstrip("/") + "/filecenter"
+
+    # 确定 MIME 类型
+    mime = mimetypes.guess_type(filename)[0] or "image/jpeg"
+    files = {"file": (filename, file_bytes, mime)}
+
+    # 发送请求
+    async with httpx.AsyncClient(timeout=30) as client:
+        r = await client.post(
+            f"{filecenter_url}/fileUpload",
+            files=files,
+            data={**sign_params, "signature": signature},
+        )
+    d = r.json()
+    if d.get("code") in [0, 200]:
+        url = d.get("data") or d.get("url") or ""
+        if isinstance(url, dict):
+            url = url.get("url") or url.get("path") or ""
+        if not url:
+            raise RuntimeError(f"filecenter 返回空 URL: {d}")
+        return url
+    raise RuntimeError(f"filecenter 上传失败: {d.get('message', '未知错误')}")
+
+
+@router.post("/{avatar_id}/upload-photo")
+async def upload_avatar_photo(
+    avatar_id: str,
+    file: UploadFile = File(...),
+):
+    """上传数字分身头像（通过会会平台 filecenter）"""
+    if not is_available():
+        raise HTTPException(status_code=503, detail="数字分身数据库尚未初始化")
+
+    # 验证文件
+    if not file.content_type or not file.content_type.startswith("image/"):
+        raise HTTPException(status_code=400, detail="仅支持图片文件")
+
+    file_bytes = await file.read()
+    if len(file_bytes) > 5 * 1024 * 1024:
+        raise HTTPException(status_code=400, detail="头像文件不能超过5MB")
+
+    # 确保文件扩展名正确
+    filename = file.filename or "avatar.jpg"
+    ext = filename.split(".")[-1].lower() if "." in filename else "jpg"
+    if ext not in ("jpg", "jpeg", "png", "gif", "webp"):
+        filename = f"avatar.{ext}"
+
+    try:
+        photo_url = await _upload_to_filecenter(file_bytes, filename)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"头像上传失败: {str(e)}")
+
+    # 更新数据库
+    db = get_session()
+    try:
+        data = avatar_service.get_avatar(db, avatar_id)
+        if not data:
+            raise HTTPException(status_code=404, detail="数字分身不存在")
+
+        from sqlalchemy import text
+        db.execute(
+            text("UPDATE avatars SET photo_url = :url, updated_at = datetime('now') WHERE id = :id"),
+            {"url": photo_url, "id": avatar_id},
+        )
+        db.commit()
+
+        # 刷新数据
+        updated = avatar_service.get_avatar(db, avatar_id)
+        return ApiResponse(data=updated, message="头像上传成功")
+    finally:
+        db.close()

backend/app/api/endpoints/users.py

@@ -355,13 +355,15 @@ async def upload_avatar(
         # 未登录用户本地落盘，避免 base64 超过 avatar_url 字段长度
         avatar_url = await _save_local_avatar(file_bytes, file.filename or "", file.content_type)
 
-    # 更新数据库
-    await db.execute(update(_VU).where(_VU.id == user_id).values(avatar_url=avatar_url))
-    await db.commit()
-
-    # 如果已同步到平台，再调用 update_user_profile 更新头像字段
+    # 已登录用户必须同时写入会会当前资料和“TA 的主页”。
+    # 任一接口失败都不得返回“头像更新成功”。
     if sync_to_platform and user.status == 2 and avatar_url:
-        await news_service.update_user_profile(db, user, avatar=avatar_url)
+        ok, err = await news_service.update_user_profile(db, user, avatar=avatar_url)
+        if not ok:
+            return ApiResponse(code=502, message=f"头像已上传，但同步到会会失败: {err}")
+    else:
+        await db.execute(update(_VU).where(_VU.id == user_id).values(avatar_url=avatar_url))
+        await db.commit()
 
     return ApiResponse(data={"avatar_url": avatar_url}, message="头像更新成功")
 @router.post("/logout-all")

backend/app/services/news_service.py

@@ -228,6 +228,40 @@ class NewsPlatformService:
                         "avatar":       sync_avatar,
                     }, expire=86400)
 
+                    # 导入用户的昵称/头像先保存在本地；首次或后续登录时，
+                    # 如果会会端仍是旧值，必须补写“当前资料 + TA 的主页”。
+                    desired_nickname = preferred_nickname
+                    desired_real_name = (user.real_name or desired_nickname or "").strip()
+                    desired_avatar = (user.avatar_url or sync_avatar or "").strip()
+                    needs_profile_sync = any([
+                        desired_nickname and desired_nickname != sync_nickname,
+                        desired_real_name and desired_real_name != sync_real_name,
+                        desired_avatar and desired_avatar != sync_avatar,
+                    ])
+                    # usercenter 与 App 的“TA 的主页”不是同一数据源。即使
+                    # usercenter 已一致，也必须检查 huihuiuserextend 的主页资料。
+                    if not needs_profile_sync:
+                        home_ok, home_data = await self.get_huihui_user_home(db, user)
+                        needs_profile_sync = (
+                            not home_ok
+                            or (desired_nickname and home_data.get("name") != desired_nickname)
+                            or (desired_avatar and home_data.get("avatar") != desired_avatar)
+                        )
+                    if needs_profile_sync:
+                        ok, err = await self.update_user_profile(
+                            db,
+                            user,
+                            nick_name=desired_nickname or None,
+                            real_name=desired_real_name or None,
+                            avatar=desired_avatar or None,
+                        )
+                        if not ok:
+                            await delete_session(user.id)
+                            raise ValueError(f"会会用户资料同步失败: {err}")
+                        sync_nickname = desired_nickname
+                        sync_real_name = desired_real_name
+                        sync_avatar = desired_avatar
+
                     # 更新本地数据库，同步平台用户信息
                     update_vals = dict(
                         status=2, session_token=access_token,
@@ -1247,7 +1281,11 @@ class NewsPlatformService:
         if description is not None: body["description"] = description
         if email       is not None: body["email"]       = email
 
-        # 使用 PATCH /v2/users/current 接口（支持修改昵称）
+        # 会会实际有三份用户资料：
+        # 1. /v2/users/current 更新当前用户资料；
+        # 2. /users/page/{id} 更新 usercenter 公开资料；
+        # 3. /huihuiuserextend/user 更新 App“TA 的主页”。
+        # 三处都成功且 App 主页回读一致后才允许标记为成功。
         headers = dict(self._bearer(token))
         headers["Content-Type"] = "application/json"
 
@@ -1259,27 +1297,193 @@ class NewsPlatformService:
                     headers=headers,
                 )
             d = r.json()
-            if d.get("code") in [0, 200]:
-                # 同步到本地数据库
-                local_vals = {}
-                if nick_name   is not None: local_vals["nickname"]  = nick_name
-                if real_name   is not None: local_vals["real_name"] = real_name
-                if sex         is not None: local_vals["sex"]       = sex
-                if avatar      is not None: local_vals["avatar_url"] = avatar
-                if local_vals:
-                    from sqlalchemy import update
-                    await db.execute(update(VirtualUser).where(
-                        VirtualUser.id == user.id).values(**local_vals))
-                    await db.commit()
-                logger.info(f"✅ 用户 {user.account} 信息已同步到目标系统")
-                return True, ""
-            err = d.get("message") or f"code={d.get('code')}"
-            logger.warning(f"[修改用户信息] {user.account} 失败: {err} body={r.text[:200]}")
-            return False, err
+            if d.get("code") not in [0, 200]:
+                err = d.get("message") or f"code={d.get('code')}"
+                logger.warning(f"[修改用户信息] {user.account} 失败: {err} body={r.text[:200]}")
+                return False, err
+
+            page_ok, page_err = await self.update_public_user_page(
+                db,
+                user,
+                nick_name=nick_name,
+                avatar=avatar,
+            )
+            if not page_ok:
+                logger.warning(f"[同步TA的主页] {user.account} 失败: {page_err}")
+                return False, f"TA的主页同步失败: {page_err}"
+
+            home_ok, home_err = await self.update_huihui_user_home(
+                db,
+                user,
+                name=nick_name,
+                avatar=avatar,
+            )
+            if not home_ok:
+                logger.warning(f"[同步App用户主页] {user.account} 失败: {home_err}")
+                return False, f"App用户主页同步失败: {home_err}"
+
+            # 三套会会资料均成功并通过 App 主页回读后再同步本地数据库。
+            local_vals = {}
+            if nick_name   is not None: local_vals["nickname"]  = nick_name
+            if real_name   is not None: local_vals["real_name"] = real_name
+            if sex         is not None: local_vals["sex"]       = sex
+            if avatar      is not None: local_vals["avatar_url"] = avatar
+            if local_vals:
+                await db.execute(update(VirtualUser).where(
+                    VirtualUser.id == user.id).values(**local_vals))
+                await db.commit()
+            logger.info(f"✅ 用户 {user.account} 三套资料与App用户主页均已同步")
+            return True, ""
         except Exception as e:
             logger.warning(f"[修改用户信息] {user.account} 异常: {e}")
             return False, str(e)
 
+    async def update_public_user_page(
+        self, db: AsyncSession, user: VirtualUser,
+        nick_name: str = None, avatar: str = None,
+    ) -> tuple[bool, str]:
+        """同步会会 App“TA 的主页”展示的公开昵称和头像。"""
+        sess = await get_session(user.id)
+        if not sess:
+            return False, "用户未登录，请先登录"
+
+        platform_uid = sess.get("platform_uid") or user.platform_uid or ""
+        if not platform_uid:
+            return False, "缺少平台用户ID，请重新登录"
+
+        cfg = await self._client(db)
+        auth = await self._auth_url(db)
+        params = {"userId": platform_uid}
+        if nick_name is not None:
+            params["nickName"] = nick_name
+        if avatar is not None:
+            params["icon"] = avatar
+        signed_params = self._build_form(params, cfg)
+
+        try:
+            async with httpx.AsyncClient(timeout=15) as c:
+                r = await c.patch(
+                    f"{auth}/users/page/{platform_uid}",
+                    params=signed_params,
+                    headers=self._bearer(sess.get("token", "")),
+                )
+            d = r.json()
+            if r.status_code == 200 and d.get("code") in [0, 200] and d.get("data") is not False:
+                return True, ""
+            return False, d.get("message") or f"HTTP={r.status_code}, code={d.get('code')}"
+        except Exception as e:
+            return False, str(e)
+
+    async def get_public_user_profile(
+        self, db: AsyncSession, user: VirtualUser,
+    ) -> tuple[bool, dict | str]:
+        """通过公开用户详情接口回读“TA 的主页”数据。"""
+        sess = await get_session(user.id)
+        if not sess:
+            return False, "用户未登录"
+        platform_uid = sess.get("platform_uid") or user.platform_uid or ""
+        if not platform_uid:
+            return False, "缺少平台用户ID"
+        cfg = await self._client(db)
+        auth = await self._auth_url(db)
+        params = self._build_form({"userId": platform_uid}, cfg)
+        try:
+            async with httpx.AsyncClient(timeout=15) as c:
+                r = await c.get(
+                    f"{auth}/users/{platform_uid}",
+                    params=params,
+                    headers=self._bearer(sess.get("token", "")),
+                )
+            d = r.json()
+            if r.status_code == 200 and d.get("code") in [0, 200] and isinstance(d.get("data"), dict):
+                return True, d["data"]
+            return False, d.get("message") or f"HTTP={r.status_code}, code={d.get('code')}"
+        except Exception as e:
+            return False, str(e)
+
+    async def get_huihui_user_home(
+        self, db: AsyncSession, user: VirtualUser,
+    ) -> tuple[bool, dict | str]:
+        """回读会会 App `/otherIndex` 实际使用的“TA 的主页”资料。"""
+        sess = await get_session(user.id)
+        if not sess:
+            return False, "用户未登录"
+        platform_uid = sess.get("platform_uid") or user.platform_uid or ""
+        if not platform_uid:
+            return False, "缺少平台用户ID"
+
+        cfg = await self._client(db)
+        api_root = self._api_root(await self._biz_url(db))
+        extra = {"userId": platform_uid}
+        org_id = sess.get("org_id") or cfg.get("orgId") or ""
+        if org_id:
+            extra["orgId"] = org_id
+        params = self._build_form(extra, cfg)
+        try:
+            async with httpx.AsyncClient(timeout=15) as c:
+                r = await c.get(
+                    f"{api_root}/huihuiuserextend/user/home/{platform_uid}",
+                    params=params,
+                    headers=self._bearer(sess.get("token", "")),
+                )
+            d = r.json()
+            if r.status_code == 200 and d.get("code") in [0, 200] and isinstance(d.get("data"), dict):
+                return True, d["data"]
+            return False, d.get("message") or f"HTTP={r.status_code}, code={d.get('code')}"
+        except Exception as e:
+            return False, str(e)
+
+    async def update_huihui_user_home(
+        self, db: AsyncSession, user: VirtualUser,
+        name: str = None, avatar: str = None,
+    ) -> tuple[bool, str]:
+        """写入并回读验证会会 App 真正使用的用户扩展主页资料。"""
+        sess = await get_session(user.id)
+        if not sess:
+            return False, "用户未登录，请先登录"
+        platform_uid = sess.get("platform_uid") or user.platform_uid or ""
+        if not platform_uid:
+            return False, "缺少平台用户ID，请重新登录"
+
+        current_ok, current = await self.get_huihui_user_home(db, user)
+        if not current_ok:
+            return False, f"主页资料回读失败: {current}"
+        extend_id = current.get("id")
+        if not extend_id:
+            return False, "会会用户扩展资料缺少记录ID"
+
+        desired_name = name if name is not None else (user.nickname or "")
+        desired_avatar = avatar if avatar is not None else (user.avatar_url or "")
+        body = {
+            "id": extend_id,
+            "userId": platform_uid,
+            "name": desired_name,
+            "avatar": desired_avatar,
+        }
+        cfg = await self._client(db)
+        api_root = self._api_root(await self._biz_url(db))
+        params = self._build_form({"userId": platform_uid}, cfg)
+        try:
+            async with httpx.AsyncClient(timeout=15) as c:
+                r = await c.patch(
+                    f"{api_root}/huihuiuserextend/user",
+                    params=params,
+                    json=body,
+                    headers={**self._bearer(sess.get("token", "")), "Content-Type": "application/json"},
+                )
+            d = r.json()
+            if r.status_code != 200 or d.get("code") not in [0, 200]:
+                return False, d.get("message") or f"HTTP={r.status_code}, code={d.get('code')}"
+
+            verify_ok, verified = await self.get_huihui_user_home(db, user)
+            if not verify_ok:
+                return False, f"写入后回读失败: {verified}"
+            if verified.get("name") != desired_name or verified.get("avatar") != desired_avatar:
+                return False, "写入后App用户主页昵称或头像不一致"
+            return True, ""
+        except Exception as e:
+            return False, str(e)
+
     async def upload_avatar(
         self, db: AsyncSession, user: VirtualUser, file_bytes: bytes, filename: str
     ) -> tuple[bool, str]:

backend/tests/test_huihui_profile_sync.py

@@ -0,0 +1,143 @@
+import unittest
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+from app.services.news_service import NewsPlatformService
+
+
+class _Response:
+    def __init__(self, payload, status_code=200):
+        self._payload = payload
+        self.status_code = status_code
+        self.text = str(payload)
+
+    def json(self):
+        return self._payload
+
+
+class _Client:
+    responses = []
+    calls = []
+
+    def __init__(self, *args, **kwargs):
+        pass
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb):
+        return False
+
+    async def patch(self, url, **kwargs):
+        self.__class__.calls.append(("PATCH", url, kwargs))
+        return self.__class__.responses.pop(0)
+
+    async def get(self, url, **kwargs):
+        self.__class__.calls.append(("GET", url, kwargs))
+        return self.__class__.responses.pop(0)
+
+
+class HuihuiProfileSyncTests(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.service = NewsPlatformService()
+        self.service._auth_url = AsyncMock(return_value="https://99hui.com/api/usercenter")
+        self.service._biz_url = AsyncMock(return_value="https://99hui.com/api/huihuibusiness")
+        self.service._client = AsyncMock(return_value={
+            "appId": "app", "accessId": "access", "accessSecret": "secret",
+            "clientCode": "", "orgId": "",
+        })
+        self.db = SimpleNamespace(execute=AsyncMock(), commit=AsyncMock())
+        self.user = SimpleNamespace(
+            id=51, account="13721560046", platform_uid="platform-51",
+            nickname="黎佳怡", real_name="黎佳怡", sex=0, avatar_url="https://img/avatar.jpg",
+        )
+        _Client.calls = []
+
+    async def test_updates_all_profiles_and_verifies_app_home(self):
+        _Client.responses = [
+            _Response({"code": 0, "data": True}),
+            _Response({"code": 0, "data": True}),
+            _Response({"code": 0, "data": {
+                "id": "extend-51", "userId": "platform-51", "name": "", "avatar": "",
+            }}),
+            _Response({"code": 0, "data": None}),
+            _Response({"code": 0, "data": {
+                "id": "extend-51", "userId": "platform-51", "name": "黎佳怡",
+                "avatar": "https://img/avatar.jpg",
+            }}),
+        ]
+        session = {"token": "token", "platform_uid": "platform-51", "org_id": "org-1"}
+        with patch("app.services.news_service.get_session", AsyncMock(return_value=session)), \
+             patch("app.services.news_service.httpx.AsyncClient", _Client):
+            ok, err = await self.service.update_user_profile(
+                self.db, self.user,
+                nick_name="黎佳怡", real_name="黎佳怡",
+                avatar="https://img/avatar.jpg",
+            )
+
+        self.assertTrue(ok, err)
+        self.assertEqual(_Client.calls[0][1], "https://99hui.com/api/usercenter/v2/users/current")
+        self.assertEqual(_Client.calls[1][1], "https://99hui.com/api/usercenter/users/page/platform-51")
+        public_params = _Client.calls[1][2]["params"]
+        self.assertEqual(public_params["nickName"], "黎佳怡")
+        self.assertEqual(public_params["icon"], "https://img/avatar.jpg")
+        self.assertEqual(_Client.calls[2][0], "GET")
+        self.assertEqual(
+            _Client.calls[2][1],
+            "https://99hui.com/api/huihuiuserextend/user/home/platform-51",
+        )
+        self.assertEqual(_Client.calls[3][0], "PATCH")
+        self.assertEqual(
+            _Client.calls[3][1],
+            "https://99hui.com/api/huihuiuserextend/user",
+        )
+        self.assertEqual(_Client.calls[3][2]["json"]["id"], "extend-51")
+        self.assertEqual(_Client.calls[3][2]["json"]["name"], "黎佳怡")
+        self.assertEqual(_Client.calls[3][2]["json"]["avatar"], "https://img/avatar.jpg")
+        self.assertEqual(_Client.calls[4][0], "GET")
+        self.db.commit.assert_awaited_once()
+
+    async def test_public_page_failure_is_not_reported_as_success(self):
+        _Client.responses = [
+            _Response({"code": 0, "data": True}),
+            _Response({"code": 500, "message": "page update failed"}),
+        ]
+        session = {"token": "token", "platform_uid": "platform-51"}
+        with patch("app.services.news_service.get_session", AsyncMock(return_value=session)), \
+             patch("app.services.news_service.httpx.AsyncClient", _Client):
+            ok, err = await self.service.update_user_profile(
+                self.db, self.user,
+                nick_name="黎佳怡", avatar="https://img/avatar.jpg",
+            )
+
+        self.assertFalse(ok)
+        self.assertIn("TA的主页同步失败", err)
+        self.db.commit.assert_not_awaited()
+
+    async def test_app_home_mismatch_is_not_reported_as_success(self):
+        _Client.responses = [
+            _Response({"code": 0, "data": True}),
+            _Response({"code": 0, "data": True}),
+            _Response({"code": 0, "data": {
+                "id": "extend-51", "userId": "platform-51", "name": "", "avatar": "",
+            }}),
+            _Response({"code": 0, "data": None}),
+            _Response({"code": 0, "data": {
+                "id": "extend-51", "userId": "platform-51", "name": "", "avatar": "",
+            }}),
+        ]
+        session = {"token": "token", "platform_uid": "platform-51", "org_id": "org-1"}
+        with patch("app.services.news_service.get_session", AsyncMock(return_value=session)), \
+             patch("app.services.news_service.httpx.AsyncClient", _Client):
+            ok, err = await self.service.update_user_profile(
+                self.db, self.user,
+                nick_name="黎佳怡", avatar="https://img/avatar.jpg",
+            )
+
+        self.assertFalse(ok)
+        self.assertIn("App用户主页同步失败", err)
+        self.db.commit.assert_not_awaited()
+
+
+if __name__ == "__main__":
+    unittest.main()

digital-avatar-app/backend/database.py

@@ -35,6 +35,9 @@ def init_db():
         ("knowledge_docs", "chunk_count", "INTEGER DEFAULT 0"),
         ("knowledge_docs", "vectorized_at", "TIMESTAMP"),
         ("avatars", "owner_id", "VARCHAR DEFAULT ''"),
+        ("authorizations", "takeover_enabled", "BOOLEAN DEFAULT 0"),
+        ("authorizations", "takeover_mode", "VARCHAR DEFAULT 'immediate'"),
+        ("authorizations", "takeover_delay_seconds", "INTEGER DEFAULT 30"),
     )
 
 

digital-avatar-app/backend/main.py

@@ -2,6 +2,11 @@ from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 
 import os
+import logging
+
+from apscheduler.schedulers.background import BackgroundScheduler
+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
@@ -13,8 +18,11 @@ import routers.organizations
 import routers.knowledge
 import routers.huihui_auth
 import routers.chat
+import routers.takeover
 from responses import ok
 
+logger = logging.getLogger(__name__)
+
 app = FastAPI(title="会会数字分身 API", version="1.0.0")
 
 app.add_middleware(
@@ -32,6 +40,7 @@ app.include_router(routers.organizations.router, prefix="/api")
 app.include_router(routers.knowledge.router, prefix="/api")
 app.include_router(routers.huihui_auth.router, prefix="/api")
 app.include_router(routers.chat.router, prefix="/api")
+app.include_router(routers.takeover.router, prefix="/api")
 
 UPLOAD_DIR = routers.knowledge.UPLOAD_DIR
 os.makedirs(UPLOAD_DIR, exist_ok=True)
@@ -103,3 +112,41 @@ def seed():
 def on_startup():
     init_db()
     seed()
+
+    # --- 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
+        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_APP_ID": os.getenv("HUIHUI_APP_ID", ""),
+            "HUIHUI_ACCESS_ID": os.getenv("HUIHUI_ACCESS_ID", ""),
+            "HUIHUI_ACCESS_SECRET": os.getenv("HUIHUI_ACCESS_SECRET", ""),
+        }
+        boxim_client = BoxIMClient(boxim_config)
+
+        # Initialize takeover service
+        from services.takeover_service import TakeoverService
+        takeover_service = TakeoverService(SessionLocal(), boxim_client, redis_client)
+
+        # Start periodic polling job
+        scheduler = BackgroundScheduler()
+        scheduler.add_job(
+            takeover_service.poll_and_process_messages,
+            trigger=IntervalTrigger(seconds=10),
+            id="takeover_message_poll",
+        )
+        scheduler.start()
+        logger.info("Takeover message polling scheduler started (interval=10s)")
+    except Exception as e:
+        logger.warning(f"Failed to initialize takeover scheduler, app will continue without it: {e}")

digital-avatar-app/backend/models.py

@@ -51,6 +51,9 @@ class Authorization(Base):
     target_name = Column(String, default="")
     permissions = Column(JSON, default=list)
     status = Column(String, default="active")  # active | inactive
+    takeover_enabled = Column(Boolean, default=False)  # 是否开启分身接管
+    takeover_mode = Column(String, default="immediate")  # immediate | delayed
+    takeover_delay_seconds = Column(Integer, default=30)  # 延迟秒数
     created_at = Column(DateTime, server_default=func.now())
 
     def to_dict(self):
@@ -62,6 +65,9 @@ class Authorization(Base):
             "targetName": self.target_name,
             "permissions": self.permissions or [],
             "status": self.status,
+            "takeoverEnabled": self.takeover_enabled,
+            "takeoverMode": self.takeover_mode,
+            "takeoverDelaySeconds": self.takeover_delay_seconds,
             "createdAt": _iso(self.created_at),
         }
 

digital-avatar-app/backend/requirements.txt

@@ -7,3 +7,5 @@ httpx
 pypdf
 python-docx
 openpyxl
+redis>=5.0
+apscheduler>=3.10

digital-avatar-app/backend/routers/avatars.py

@@ -1,7 +1,11 @@
-from fastapi import APIRouter, Depends, Body, Header
+from fastapi import APIRouter, Depends, Body, Header, UploadFile, File
 from sqlalchemy.orm import Session
+import os
+import uuid
+import mimetypes
 
 from database import get_db
+from routers.knowledge import UPLOAD_DIR
 from models import Avatar, KnowledgeDoc, KnowledgeChunk, QAPair, Authorization, User
 from responses import ok, fail
 
@@ -93,3 +97,41 @@ def delete_avatar(avatar_id: str, db: Session = Depends(get_db)):
     db.delete(a)
     db.commit()
     return ok({"success": True})
+
+
+@router.post("/avatar/{avatar_id}/photo")
+async def upload_avatar_photo(
+    avatar_id: str,
+    file: UploadFile = File(...),
+    db: Session = Depends(get_db),
+):
+    """上传数字分身头像"""
+    a = db.query(Avatar).filter(Avatar.id == avatar_id).first()
+    if not a:
+        return fail("分身不存在", 404)
+
+    # 验证文件类型
+    if not file.content_type or not file.content_type.startswith("image/"):
+        return fail("仅支持图片文件", 400)
+
+    file_bytes = await file.read()
+    if len(file_bytes) > 5 * 1024 * 1024:
+        return fail("头像文件不能超过5MB", 400)
+
+    # 保存到 uploads 目录
+    ext = mimetypes.guess_extension(file.content_type) or ".jpg"
+    filename = f"avatar-{uuid.uuid4().hex}{ext}"
+    avatar_dir = os.path.join(UPLOAD_DIR, avatar_id)
+    os.makedirs(avatar_dir, exist_ok=True)
+    file_path = os.path.join(avatar_dir, filename)
+
+    with open(file_path, "wb") as f:
+        f.write(file_bytes)
+
+    # 更新数据库
+    photo_url = f"/api/files/{avatar_id}/{filename}"
+    a.photo_url = photo_url
+    db.commit()
+    db.refresh(a)
+
+    return ok(a.to_dict())

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

@@ -0,0 +1,41 @@
+"""分身接管配置 API"""
+from fastapi import APIRouter, Depends, Body
+from sqlalchemy.orm import Session
+
+from database import get_db
+from models import Authorization
+from responses import ok, fail
+
+router = APIRouter(tags=["分身接管"])
+
+
+@router.put("/avatar/{avatar_id}/authorizations/takeover")
+def update_takeover_config(
+    avatar_id: str,
+    payload: dict = Body(...),
+    db: Session = Depends(get_db),
+):
+    """更新分身接管配置"""
+    auth_id = payload.get("authorizationId") or payload.get("authorization_id")
+    if not auth_id:
+        return fail("缺少 authorization_id", 400)
+
+    auth = db.query(Authorization).filter(Authorization.id == auth_id).first()
+    if not auth:
+        return fail("授权不存在", 404)
+
+    if "takeover_enabled" in payload:
+        auth.takeover_enabled = payload["takeover_enabled"]
+    if "takeover_mode" in payload:
+        mode = payload["takeover_mode"]
+        if mode not in ("immediate", "delayed"):
+            return fail("takeover_mode 必须是 immediate 或 delayed", 400)
+        auth.takeover_mode = mode
+    if "takeover_delay_seconds" in payload:
+        delay = payload["takeover_delay_seconds"]
+        if not isinstance(delay, int) or delay < 5:
+            return fail("takeover_delay_seconds 必须 >= 5", 400)
+        auth.takeover_delay_seconds = delay
+
+    db.commit()
+    return ok(auth.to_dict())

digital-avatar-app/backend/services/boxim_client.py

@@ -0,0 +1,76 @@
+"""盒子 IM 客户端 — 封装网易云信 IM 接口调用"""
+import hashlib
+import random
+import string
+from datetime import datetime
+from typing import Optional
+
+import httpx
+
+
+class BoxIMClient:
+    """盒子 IM 客户端，通过会会平台网关调用网易云信 IM"""
+
+    def __init__(self, config: dict):
+        self.base_url = config.get("HUIHUI_IM_BASE_URL", "http://192.168.1.200:60040")
+        self.app_id = config.get("HUIHUI_APP_ID", "")
+        self.access_id = config.get("HUIHUI_ACCESS_ID", "")
+        self.access_secret = config.get("HUIHUI_ACCESS_SECRET", "")
+
+    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小时制
+        params = {
+            "appId": self.app_id,
+            "accessId": self.access_id,
+            "nonce": nonce,
+            "timestamp": timestamp,
+            **extra,
+        }
+        # 计算签名 — 排序 key, 过滤空值, 拼接后加 accessSecret, MD5 大写
+        keys = sorted(params.keys())
+        sign_parts = []
+        for k in keys:
+            if k in ("signature", "accessSecret"):
+                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"
+        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
+
+    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,
+            )
+        data = r.json()
+        return data.get("code") in (0, 200)

digital-avatar-app/backend/services/takeover_service.py

@@ -0,0 +1,174 @@
+"""Takeover service — message listening, decision, reply execution."""
+import json
+import logging
+import os
+from typing import Optional
+
+import httpx
+from sqlalchemy.orm import Session
+
+from models import Avatar, Authorization
+from services.boxim_client import BoxIMClient
+
+logger = logging.getLogger(__name__)
+
+
+class TakeoverService:
+    """Service for handling avatar takeover — generating replies and sending them via IM."""
+
+    def __init__(
+        self,
+        db: Session,
+        boxim_client: BoxIMClient,
+        redis_client=None,
+    ):
+        self.db = db
+        self.boxim = boxim_client
+        self.redis = redis_client
+        self._chat_api_base = os.getenv(
+            "TAKEOVER_CHAT_API_BASE", "http://localhost:8000/api"
+        )
+
+    def check_takeover_enabled(
+        self, owner_huihui_id: str, from_user_id: str
+    ) -> Optional[Authorization]:
+        """Check whether takeover is enabled for the given target user."""
+        avatar = self.db.query(Avatar).filter(Avatar.owner_id == owner_huihui_id).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.takeover_enabled == True)
+            .first()
+        )
+        return auth if auth and auth.takeover_enabled 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}")
+
+    async def poll_and_process_messages(self):
+        """Periodic polling job: fetch unread messages and process each."""
+        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}")
+
+    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 []
+
+    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", "")
+
+        auth = self.check_takeover_enabled(owner_id, from_id)
+        if not auth:
+            return
+
+        if auth.takeover_mode == "immediate":
+            await self.execute_takeover(auth, message)
+        else:
+            self.enqueue_delayed_message(auth, message)

digital-avatar-app/backend/tests/conftest.py

@@ -0,0 +1,26 @@
+import pytest
+from database import init_db, SessionLocal
+from models import Authorization
+
+
+@pytest.fixture(scope="session", autouse=True)
+def setup_database():
+    """Initialize DB tables and seed an Authorization row."""
+    init_db()
+    db = SessionLocal()
+    try:
+        existing = db.query(Authorization).first()
+        if existing is None:
+            auth = Authorization(
+                id="test-auth-1",
+                avatar_id="test-avatar-1",
+                target_type="user",
+                target_id="test-user-1",
+                target_name="Test User",
+                permissions=["read", "write"],
+                status="active",
+            )
+            db.add(auth)
+            db.commit()
+    finally:
+        db.close()

digital-avatar-app/backend/tests/test_boxim_client.py

@@ -0,0 +1,134 @@
+"""Tests for the Box IM client (Netease Yunxin gateway wrapper)."""
+import pytest
+from unittest.mock import AsyncMock, MagicMock, patch
+
+
+@pytest.fixture
+def mock_config():
+    return {
+        "HUIHUI_IM_BASE_URL": "http://192.168.1.200:60040",
+        "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 _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)
+
+
+@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
+
+        client = BoxIMClient(mock_config)
+        result = await client.get_credentials("user123")
+
+        assert result["accid"] == "user123"
+        assert result["token"] == "tok_xyz"
+
+
+@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
+
+        client = BoxIMClient(mock_config)
+        result = await client.send_p2p_message("owner_acc", "target_acc", "Hello")
+
+        assert result is True
+
+
+@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
+
+        client = BoxIMClient(mock_config)
+        result = await client.send_p2p_message("owner_acc", "target_acc", "Hello")
+
+        assert result is False
+
+
+@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
+
+
+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
+    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

digital-avatar-app/backend/tests/test_takeover_api.py

@@ -0,0 +1,105 @@
+"""Tests for PUT /api/avatar/{avatar_id}/authorizations/takeover endpoint."""
+from fastapi.testclient import TestClient
+from main import app
+from database import SessionLocal, Base, engine
+from models import Authorization, Avatar
+
+
+def setup_test_db():
+    Base.metadata.create_all(bind=engine)
+    db = SessionLocal()
+    avatar = Avatar(name="test", status="active", config={})
+    db.add(avatar)
+    db.commit()
+    db.refresh(avatar)
+    auth = Authorization(avatar_id=avatar.id, target_id="user1", target_name="测试用户")
+    db.add(auth)
+    db.commit()
+    db.refresh(auth)
+    return db, auth.id
+
+
+def test_update_takeover_config():
+    db, auth_id = setup_test_db()
+    try:
+        client = TestClient(app)
+        response = client.put(
+            f"/api/avatar/test_avatar_id/authorizations/takeover",
+            json={
+                "authorization_id": auth_id,
+                "takeover_enabled": True,
+                "takeover_mode": "delayed",
+                "takeover_delay_seconds": 60,
+            },
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert data["code"] == 200
+        assert data["data"]["takeoverEnabled"] is True
+        assert data["data"]["takeoverMode"] == "delayed"
+        assert data["data"]["takeoverDelaySeconds"] == 60
+        # 验证数据库已更新
+        auth = db.query(Authorization).filter(Authorization.id == auth_id).first()
+        assert auth.takeover_enabled is True
+        assert auth.takeover_mode == "delayed"
+        assert auth.takeover_delay_seconds == 60
+    finally:
+        db.close()
+
+
+def test_update_takeover_invalid_mode():
+    db, auth_id = setup_test_db()
+    try:
+        client = TestClient(app)
+        response = client.put(
+            f"/api/avatar/test/authorizations/takeover",
+            json={
+                "authorization_id": auth_id,
+                "takeover_mode": "invalid_mode",
+            },
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert data["code"] == 400
+    finally:
+        db.close()
+
+
+def test_update_takeover_invalid_delay():
+    db, auth_id = setup_test_db()
+    try:
+        client = TestClient(app)
+        response = client.put(
+            f"/api/avatar/test/authorizations/takeover",
+            json={
+                "authorization_id": auth_id,
+                "takeover_delay_seconds": 2,
+            },
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert data["code"] == 400
+    finally:
+        db.close()
+
+
+def test_update_takeover_missing_auth_id():
+    client = TestClient(app)
+    response = client.put(
+        f"/api/avatar/test/authorizations/takeover",
+        json={"takeover_enabled": True},
+    )
+    assert response.status_code == 200
+    data = response.json()
+    assert data["code"] == 400
+
+
+def test_update_takeover_not_found():
+    client = TestClient(app)
+    response = client.put(
+        f"/api/avatar/test/authorizations/takeover",
+        json={"authorization_id": "nonexistent"},
+    )
+    assert response.status_code == 200
+    data = response.json()
+    assert data["code"] == 404

digital-avatar-app/backend/tests/test_takeover_model.py

@@ -0,0 +1,30 @@
+from database import SessionLocal
+from models import Authorization
+
+
+def test_authorization_takeover_fields():
+    db = SessionLocal()
+    try:
+        auth = db.query(Authorization).first()
+        assert auth is not None
+        # Check new fields exist and have default values
+        assert hasattr(auth, 'takeover_enabled')
+        assert hasattr(auth, 'takeover_mode')
+        assert hasattr(auth, 'takeover_delay_seconds')
+        assert auth.takeover_enabled == False
+        assert auth.takeover_mode == 'immediate'
+        assert auth.takeover_delay_seconds == 30
+    finally:
+        db.close()
+
+
+def test_authorization_to_dict_includes_takeover():
+    db = SessionLocal()
+    try:
+        auth = db.query(Authorization).first()
+        d = auth.to_dict()
+        assert 'takeoverEnabled' in d
+        assert 'takeoverMode' in d
+        assert 'takeoverDelaySeconds' in d
+    finally:
+        db.close()

digital-avatar-app/backend/tests/test_takeover_scheduler.py

@@ -0,0 +1,197 @@
+"""Tests for the scheduled takeover message polling."""
+import json
+import pytest
+from unittest.mock import MagicMock, patch, AsyncMock
+
+
+def test_app_has_startup_event():
+    """Verify the app has a startup event configured."""
+    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.BackgroundScheduler")
+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_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()
+    call_kwargs = mock_scheduler_class.return_value.add_job.call_args[1]
+    assert call_kwargs["id"] == "takeover_message_poll"
+
+
+@patch("services.takeover_service.TakeoverService")
+@patch("services.boxim_client.BoxIMClient")
+@patch("main.BackgroundScheduler")
+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_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.BackgroundScheduler")
+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_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.BackgroundScheduler")
+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
+
+
+# --- 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)
+    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_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()

digital-avatar-app/backend/tests/test_takeover_service.py

@@ -0,0 +1,318 @@
+"""Tests for the TakeoverService — message listening, decision, reply execution."""
+import pytest
+from unittest.mock import AsyncMock, patch, MagicMock
+from services.takeover_service import TakeoverService
+from models import Authorization, Avatar
+
+
+@pytest.fixture
+def mock_db():
+    db = MagicMock()
+    return db
+
+
+@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"
+    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
+    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 ---
+
+
+@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
+
+        service = TakeoverService(mock_db, mock_boxim)
+        result = await service.generate_reply("avatar_123", "Hello")
+        assert result == "Hello back"
+
+
+@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
+
+        service = TakeoverService(mock_db, mock_boxim)
+        result = await service.generate_reply("avatar_123", "Hello")
+        assert result == ""
+
+
+@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
+
+        service = TakeoverService(mock_db, mock_boxim)
+        result = await service.generate_reply("avatar_123", "Hello")
+        assert result == ""
+
+
+# --- execute_takeover ---
+
+
+@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
+
+    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)
+        message = {"from_accid": "user_acc", "content": "Hello"}
+
+        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()
+
+
+@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()

digital-avatar-app/src/api/index.ts

@@ -299,4 +299,13 @@ export const getCurrentUser = () =>
 export const logoutUser = () =>
   request.post('/huihui/logout')
 
+// 更新接管配置
+export const updateTakeoverConfig = (avatarId: string, data: {
+  authorizationId: string
+  takeoverEnabled: boolean
+  takeoverMode?: 'immediate' | 'delayed'
+  takeoverDelaySeconds?: number
+}) =>
+  request.put(`/avatar/${avatarId}/authorizations/takeover`, data)
+
 export default request

digital-avatar-app/src/views/AuthorizationManage.vue

@@ -30,6 +30,46 @@
             {{ auth.status === 'active' ? '撤销' : '授权' }}
           </button>
         </div>
+        <!-- 接管配置 -->
+        <div class="takeover-section">
+          <button class="takeover-toggle-btn" @click="toggleTakeoverEdit(auth)">
+            {{ editingTakeover[auth.id] ? '收起配置' : '配置接管' }}
+          </button>
+          <div class="takeover-form" v-if="editingTakeover[auth.id]">
+            <div class="takeover-row">
+              <label class="takeover-label">启用接管</label>
+              <label class="toggle-switch">
+                <input type="checkbox" v-model="auth.takeoverEnabled" />
+                <span class="toggle-slider"></span>
+              </label>
+            </div>
+            <div class="takeover-row">
+              <label class="takeover-label">接管模式</label>
+              <select class="takeover-select" v-model="auth.takeoverMode" :disabled="!auth.takeoverEnabled">
+                <option value="immediate">立即接管</option>
+                <option value="delayed">延迟接管</option>
+              </select>
+            </div>
+            <div class="takeover-row" v-if="auth.takeoverMode === 'delayed' && auth.takeoverEnabled">
+              <label class="takeover-label">延迟时间（秒）</label>
+              <input
+                type="number"
+                class="takeover-input"
+                v-model.number="auth.takeoverDelaySeconds"
+                min="5"
+                placeholder="30"
+              />
+            </div>
+            <div class="takeover-actions">
+              <button class="takeover-cancel-btn" @click="cancelTakeoverEdit(auth)">取消</button>
+              <button class="takeover-save-btn" @click="saveTakeoverConfig(auth)">保存</button>
+            </div>
+          </div>
+          <div class="takeover-summary" v-else-if="auth.takeoverEnabled">
+            <span class="takeover-badge enabled">已启用</span>
+            <span class="takeover-mode-text">{{ auth.takeoverMode === 'immediate' ? '立即接管' : '延迟 ' + (auth.takeoverDelaySeconds || 30) + 's' }}</span>
+          </div>
+        </div>
       </div>
     </section>
 
@@ -47,7 +87,7 @@
 import { ref, onMounted } from 'vue'
 import { useRouter } from 'vue-router'
 import { useAvatarStore } from '@/store/avatar'
-import { getAuthorizationList, updateAuthorization } from '@/api'
+import { getAuthorizationList, updateAuthorization, updateTakeoverConfig } from '@/api'
 import { pickAvatarId, unwrapListData } from '@/utils/avatar-page-data.js'
 
 const router = useRouter()
@@ -60,8 +100,14 @@ const authList = ref<Array<{
   targetName: string
   permissions: string[]
   status: 'active' | 'inactive'
+  takeoverEnabled?: boolean
+  takeoverMode?: 'immediate' | 'delayed'
+  takeoverDelaySeconds?: number
 }>>([])
 
+// 接管配置编辑状态（每行一个）
+const editingTakeover = ref<Record<string, boolean>>({})
+
 // 从后端加载授权列表
 const loadAuth = async () => {
   try {
@@ -137,6 +183,38 @@ const goBack = () => {
   router.back()
 }
 
+// 切换接管配置编辑状态
+const toggleTakeoverEdit = (auth: any) => {
+  editingTakeover.value[auth.id] = !editingTakeover.value[auth.id]
+}
+
+// 保存接管配置
+const saveTakeoverConfig = async (auth: any) => {
+  const avatarId = pickAvatarId(avatarStore.currentAvatarId, avatarStore.avatars)
+  if (!avatarId) return
+
+  const delay = auth.takeoverMode === 'delayed' ? (auth.takeoverDelaySeconds ?? 30) : undefined
+
+  try {
+    await updateTakeoverConfig(avatarId, {
+      authorizationId: auth.id,
+      takeoverEnabled: auth.takeoverEnabled ?? false,
+      takeoverMode: auth.takeoverMode || 'immediate',
+      takeoverDelaySeconds: delay,
+    })
+    editingTakeover.value[auth.id] = false
+  } catch (e: any) {
+    alert('保存接管配置失败：' + (e?.message || '请重试'))
+  }
+}
+
+// 取消编辑接管配置
+const cancelTakeoverEdit = (auth: any) => {
+  // 重新加载以恢复原始值
+  loadAuth()
+  editingTakeover.value[auth.id] = false
+}
+
 onMounted(() => {
   loadAuth()
 })
@@ -321,4 +399,172 @@ onMounted(() => {
   font-weight: 600;
   cursor: pointer;
 }
+
+/* 接管配置 */
+.takeover-section {
+  margin-top: 8px;
+  padding-top: 12px;
+  border-top: 1px solid #F3F4F6;
+  width: 100%;
+}
+
+.takeover-toggle-btn {
+  background: none;
+  border: none;
+  color: #6366F1;
+  font-size: 12px;
+  font-weight: 500;
+  cursor: pointer;
+  padding: 4px 0;
+}
+
+.takeover-form {
+  margin-top: 10px;
+  display: flex;
+  flex-direction: column;
+  gap: 10px;
+  background: #FAFBFC;
+  border-radius: 8px;
+  padding: 12px;
+}
+
+.takeover-row {
+  display: flex;
+  align-items: center;
+  gap: 10px;
+}
+
+.takeover-label {
+  font-size: 13px;
+  color: #374151;
+  font-weight: 500;
+  min-width: 72px;
+  flex-shrink: 0;
+}
+
+/* 切换开关 */
+.toggle-switch {
+  position: relative;
+  display: inline-block;
+  width: 40px;
+  height: 22px;
+  flex-shrink: 0;
+}
+
+.toggle-switch input {
+  opacity: 0;
+  width: 0;
+  height: 0;
+}
+
+.toggle-slider {
+  position: absolute;
+  cursor: pointer;
+  inset: 0;
+  background: #D1D5DB;
+  border-radius: 22px;
+  transition: background 0.2s;
+}
+
+.toggle-slider::before {
+  content: '';
+  position: absolute;
+  height: 16px;
+  width: 16px;
+  left: 3px;
+  bottom: 3px;
+  background: white;
+  border-radius: 50%;
+  transition: transform 0.2s;
+}
+
+.toggle-switch input:checked + .toggle-slider {
+  background: #6366F1;
+}
+
+.toggle-switch input:checked + .toggle-slider::before {
+  transform: translateX(18px);
+}
+
+.takeover-select {
+  flex: 1;
+  padding: 6px 10px;
+  border: 1px solid #E5E7EB;
+  border-radius: 6px;
+  font-size: 13px;
+  color: #374151;
+  background: white;
+  outline: none;
+}
+
+.takeover-select:disabled {
+  background: #F3F4F6;
+  color: #9CA3AF;
+}
+
+.takeover-input {
+  flex: 1;
+  padding: 6px 10px;
+  border: 1px solid #E5E7EB;
+  border-radius: 6px;
+  font-size: 13px;
+  color: #374151;
+  background: white;
+  outline: none;
+  width: 80px;
+}
+
+.takeover-actions {
+  display: flex;
+  gap: 8px;
+  justify-content: flex-end;
+  margin-top: 4px;
+}
+
+.takeover-cancel-btn {
+  padding: 6px 16px;
+  border-radius: 6px;
+  font-size: 12px;
+  font-weight: 500;
+  cursor: pointer;
+  border: 1px solid #E5E7EB;
+  background: white;
+  color: #6B7280;
+}
+
+.takeover-save-btn {
+  padding: 6px 16px;
+  border-radius: 6px;
+  font-size: 12px;
+  font-weight: 500;
+  cursor: pointer;
+  border: none;
+  background: #6366F1;
+  color: white;
+}
+
+/* 接管配置摘要 */
+.takeover-summary {
+  display: flex;
+  align-items: center;
+  gap: 8px;
+  margin-top: 6px;
+}
+
+.takeover-badge {
+  padding: 2px 8px;
+  border-radius: 4px;
+  font-size: 11px;
+  font-weight: 600;
+}
+
+.takeover-badge.enabled {
+  background: #EEF2FF;
+  color: #6366F1;
+}
+
+.takeover-mode-text {
+  font-size: 12px;
+  color: #6B7280;
+}
 </style>

docs/superpowers/specs/2026-08-07-avatar-takeover-chat-design.md

@@ -0,0 +1,284 @@
+# 数字分身接管聊天功能 — 设计文档
+
+## 1. 概述
+
+### 1.1 背景
+当前数字分身应用中，分身只能与主人一对一聊天。用户希望分身能在主人授权下，接管主人在会会平台上的聊天（单聊和群聊），替主人与他人对话。
+
+### 1.2 目标
+- 主人可在授权管理页面开启"分身接管聊天"功能
+- 支持两种接管模式：立即接管、延迟接管（N 秒后）
+- 分身回复优先使用知识库（QA 问答对 + 文档向量搜索）
+- 支持单聊和群聊两种场景
+- 以主人身份发送回复，对方无感知
+
+### 1.3 技术基础
+- 会会平台使用网易云信作为 IM 底层
+- 已有接口：`/api/im/netease`（获取云信凭证）、`/api/v2message/push/record/*`（消息推送）
+- 已有分身聊天接口：`POST /api/avatar/{id}/chat`
+
+---
+
+## 2. 架构设计
+
+### 2.1 整体架构
+
+```
+┌─────────────────────────────────────────────────────────┐
+│                    会会平台 (网易云信)                     │
+│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────┐  │
+│  │  单聊 P2P   │  │  群聊 Group │  │  消息推送 Push  │  │
+│  └─────────────┘  └─────────────┘  └─────────────────┘  │
+└──────────────────────┬──────────────────────────────────┘
+                       │ HTTP API
+──────────────────────▼──────────────────────────────────┐
+│              分身接管服务 (Avatar Takeover Service)       │
+│  ┌──────────────┐  ┌──────────────┐  ┌───────────────┐ │
+│  │  消息监听器  │  │  接管决策器  │  │  回复执行器   │ │
+│  └──────────────┘  ──────────────┘  └───────────────┘ │
+│  ┌──────────────────────────────────────────────────┐  │
+│  │              Redis 延迟队列                       │  │
+│  └──────────────────────────────────────────────────┘  │
+└──────────────────────┬──────────────────────────────────┘
+                       │ 调用现有接口
+┌──────────────────────▼──────────────────────────────────┐
+│                   数字分身应用                            │
+│  ┌──────────────┐  ┌──────────────┐  ┌───────────────┐ │
+│  │  授权管理 UI │  │  分身聊天 API │  │  知识库       │ │
+│  └──────────────┘  └──────────────┘  └───────────────┘ │
+└─────────────────────────────────────────────────────────┘
+```
+
+### 2.2 模块划分
+
+| 模块 | 位置 | 职责 |
+|------|------|------|
+| 接管配置 | 授权管理页面 + authorizations 表 | 存储接管开关、模式、延迟秒数 |
+| 消息监听器 | 后端定时任务 | 轮询网易云信消息，发现新消息 |
+| 接管决策器 | 后端服务 | 判断是否接管、立即还是延迟 |
+| 回复执行器 | 后端服务 | 调用分身聊天接口生成回复，以主人身份发送 |
+| 延迟队列 | Redis | 存储待处理消息，实现延迟接管 |
+
+---
+
+## 3. 数据模型
+
+### 3.1 扩展 Authorization 表
+
+在现有 `authorizations` 表基础上新增字段：
+
+| 字段 | 类型 | 说明 |
+|------|------|------|
+| `takeover_enabled` | BOOLEAN | 是否开启分身接管聊天 |
+| `takeover_mode` | VARCHAR | 接管模式：`immediate`（立即）/ `delayed`（延迟） |
+| `takeover_delay_seconds` | INTEGER | 延迟秒数（仅 delayed 模式有效），默认 30 |
+
+**权限关联**：
+- `permissions` 数组中包含 `"takeover"` 时，表示该授权允许分身接管
+- `target_type = "user"` 且 `permissions` 含 `"takeover"` → 单聊接管
+- `target_type = "organization"` 或群聊相关 → 群聊接管
+
+### 3.2 Redis 数据结构
+
+```
+# 延迟消息队列
+takeover:delayed:{owner_accid}:{message_id} = {
+  "avatar_id": "...",
+  "from_accid": "...",
+  "content": "...",
+  "chat_type": "p2p|group",
+  "chat_id": "...",
+  "timestamp": 1234567890
+}
+TTL = takeover_delay_seconds + 10
+```
+
+---
+
+## 4. 核心流程
+
+### 4.1 消息监听流程
+
+```
+┌──────────┐     ┌──────────────┐     ┌──────────────┐
+│ 定时任务  │────►│ 轮询 Push API │────►│ 获取未读消息 │
+└──────────┘     └──────────────┘     └──────┬───────┘
+                                              │
+                                    ┌─────────▼─────────┐
+                                    │ 遍历每条新消息     │
+                                    └─────────┬─────────┘
+                                              │
+                                    ┌─────────▼─────────┐
+                                    │ 查询主人授权配置   │
+                                    └─────────┬─────────┘
+                                              │
+                          ┌───────────────────┼───────────────────┐
+                          │                   │                   │
+                    ┌─────▼─────┐      ┌──────▼──────    ┌──────▼──────┐
+                    │ 未开启接管 │      │ 立即接管模式 │    │ 延迟接管模式 │
+                    └───────────┘      └──────┬──────┘    └──────┬──────┘
+                                              │                  │
+                                    ┌─────────▼─────────┐ ┌────▼────
+                                    │ 立即调用分身回复   │ │ 写入延迟 │
+                                    ───────────────────┘ │ 队列    │
+                                                          └─────────┘
+```
+
+### 4.2 延迟接管流程
+
+```
+┌──────────────────────────────────────────────────────────┐
+│ 消息写入 Redis 延迟队列 (TTL = delay_seconds + 10)         │
+└────────────────────────┬─────────────────────────────────┘
+                         │
+                         │ 延迟期间
+                         │
+               ┌─────────▼─────────┐
+               │ 主人是否有回复？   │
+               └─────────┬─────────┘
+                    ────┴────┐
+                    │         │
+              ┌─────▼──┐ ┌───▼────
+              │ 有回复  │ │ 无回复  │
+              └──┬─────┘ └───┬────┘
+                 │           │
+           ┌─────▼─────┐ ───▼──────────────┐
+           │ 删除队列   │ │ TTL 到期触发      │
+           │ 不接管     │ │ 调用分身回复      │
+           ───────────┘ └──────────────────
+```
+
+### 4.3 回复生成流程
+
+```
+┌────────────────────┐
+│ 收到他人消息内容    │
+└─────────┬──────────┘
+          │
+┌─────────▼──────────┐
+│ 查找主人的分身      │
+│ (按 owner_id 匹配) │
+└─────────┬──────────
+          │
+┌─────────▼──────────┐
+│ 调用分身聊天接口    │
+│ POST /api/avatar/  │
+│   {id}/chat        │
+│ 优先：QA 问答对     │
+│ 其次：知识库文档    │
+│ 最后：Qwen 生成    │
+└─────────┬──────────┘
+          │
+┌─────────▼──────────
+│ 以主人身份发送回复  │
+│ 通过网易云信 API    │
+└────────────────────┘
+```
+
+---
+
+## 5. API 设计
+
+### 5.1 授权管理增强
+
+**PUT /api/avatar/{avatar_id}/authorizations/takeover**
+
+请求体：
+```json
+{
+  "authorization_id": "授权记录ID",
+  "takeover_enabled": true,
+  "takeover_mode": "delayed",
+  "takeover_delay_seconds": 30
+}
+```
+
+### 5.2 消息监听（内部接口）
+
+**POST /api/internal/takeover/process**
+
+由定时任务调用，无需认证。
+
+请求体：
+```json
+{
+  "owner_accid": "主人网易云信账号",
+  "messages": [
+    {
+      "msg_id": "消息ID",
+      "from_accid": "发送者账号",
+      "content": "消息内容",
+      "chat_type": "p2p",
+      "chat_id": "会话ID",
+      "timestamp": 1234567890
+    }
+  ]
+}
+```
+
+### 5.3 网易云信消息发送
+
+复用现有网易云信 API 封装，新增单聊消息发送方法：
+
+**POST /im/netease/message/send/p2p**
+
+请求体：
+```json
+{
+  "from_accid": "主人账号",
+  "to_accid": "对方账号",
+  "content": "回复内容",
+  "msg_type": "text"
+}
+```
+
+---
+
+## 6. 错误处理
+
+| 场景 | 处理方式 |
+|------|---------|
+| 分身聊天接口调用失败 | 记录日志，不回复，下次继续监听 |
+| 网易云信发送失败 | 重试 3 次，失败后记录日志 |
+| 主人有多个分身 | 取第一个 active 状态的分身 |
+| 知识库无匹配内容 | 使用 Qwen 兜底生成回复 |
+| Redis 连接失败 | 降级为立即接管模式 |
+
+---
+
+## 7. 配置项
+
+在 `SystemConfig` 表中新增配置：
+
+| 配置 key | 说明 | 默认值 |
+|---------|------|-------|
+| `takeover_poll_interval` | 消息轮询间隔（秒） | 10 |
+| `takeover_default_delay` | 默认延迟秒数 | 30 |
+| `takeover_retry_times` | 发送重试次数 | 3 |
+
+---
+
+## 8. 测试要点
+
+1. **功能测试**
+   - 立即接管：收到消息后立即由分身回复
+   - 延迟接管：主人 N 秒内回复则不分身接管，超时则分身回复
+   - 知识库优先：QA 问答对精确匹配优先返回
+
+2. **边界测试**
+   - 主人有多个分身时的选择逻辑
+   - 知识库为空时的 Qwen 兜底
+   - 网易云信 API 异常时的降级
+
+3. **性能测试**
+   - 轮询频率对服务器负载的影响
+   - Redis 延迟队列的内存占用
+
+---
+
+## 9. 后续扩展
+
+- Webhook 回调替代轮询（需要会会平台支持）
+- 多分身协同接管
+- 接管历史记录和统计
+- 接管效果评估和优化

frontend/src/api/index.js

@@ -90,6 +90,9 @@ export const tailLogFile = (filename, lines = 100) => request.get(`/logs/files/$
 export const getAvatars = (params) => request.get('/avatars', { params })
 export const getAvatarDetail = (id) => request.get(`/avatars/${id}`)
 export const updateAvatarStatus = (id, status) => request.put(`/avatars/${id}/status`, { status })
+export const uploadAvatarPhoto = (id, formData) => request.post(`/avatars/${id}/upload-photo`, formData, {
+  headers: { 'Content-Type': 'multipart/form-data' }
+})
 
 export default request
 export const uploadAvatar = (userId, formData) => request.post(`/users/${userId}/upload-avatar`, formData, { headers: { "Content-Type": "multipart/form-data" } })

frontend/src/views/Avatars.vue

@@ -76,9 +76,14 @@
           </el-tooltip>
         </template>
       </el-table-column>
-      <el-table-column label="操作" width="100" fixed="right">
+      <el-table-column label="操作" width="160" fixed="right">
         <template #default="{ row }">
-          <el-button size="small" @click="openDetail(row)">详情</el-button>
+          <div style="display:flex;gap:4px">
+            <el-upload :show-file-list="false" accept="image/*" :before-upload="(file) => handleUploadPhoto(row, file)">
+              <el-button size="small" :loading="row._uploading"> 上传</el-button>
+            </el-upload>
+            <el-button size="small" @click="openDetail(row)">详情</el-button>
+          </div>
         </template>
       </el-table-column>
     </el-table>
@@ -195,7 +200,7 @@
 <script setup>
 import { ref, computed, onMounted, onUnmounted } from 'vue'
 import { ElMessage } from 'element-plus'
-import { getAvatars, updateAvatarStatus } from '@/api'
+import { getAvatars, updateAvatarStatus, uploadAvatarPhoto } from '@/api'
 
 const avatars = ref([])
 const total = ref(0)
@@ -255,6 +260,27 @@ function openDetail(row) {
   detailVisible.value = true
 }
 
+async function handleUploadPhoto(row, file) {
+  const formData = new FormData()
+  formData.append('file', file)
+  row._uploading = true
+  try {
+    const res = await uploadAvatarPhoto(row.id, formData)
+    if (res.code === 200) {
+      row.photo_url = res.data?.photo_url || row.photo_url
+      ElMessage.success('头像上传成功')
+      loadAvatars()
+    } else {
+      ElMessage.error(res.message || '上传失败')
+    }
+  } catch (e) {
+    ElMessage.error('上传失败：' + (e.message || '未知错误'))
+  } finally {
+    row._uploading = false
+  }
+  return false // prevent el-upload default behavior
+}
+
 function handleAvatarError(e) {
   // 图片加载失败时隐藏 img，只保留 emoji fallback
   e.target.style.display = 'none'