huihuiSquare/backend/app/services/avatar_service.py

"""数字分身管理服务层 — 同步连接数字分身应用的 SQLite 数据库"""
import json
import os
from datetime import datetime, timedelta
from typing import Optional, Tuple

from sqlalchemy import create_engine, select, text
from sqlalchemy.orm import sessionmaker, Session

from app.core.config import settings
from app.core.logger import logger
from app.models import UserPersonality, VirtualUser


_engine = None
_SessionLocal: Optional[sessionmaker] = None

AVATAR_ACCOUNT_PREFIX = "__avatar__:"
SQUARE_INTERACTION_PERMISSION = "interact"
SQUARE_INTERACTION_ACTIONS = frozenset({"like", "collect", "comment", "reply"})


def _get_engine_and_session():
    global _engine, _SessionLocal
    if _engine is None:
        db_path = settings.AVATAR_DB_PATH
        if not db_path:
            # 默认路径：从 backend/app/core/ 向上三级
            base = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
            db_path = os.path.join(base, "digital-avatar-app", "backend", "avatar.db")
        if not os.path.isabs(db_path):
            db_path = os.path.abspath(db_path)
        if not os.path.exists(db_path):
            # 数据库不存在时返回 None，由调用方处理
            return None, None
        _engine = create_engine(
            f"sqlite:///{db_path}",
            connect_args={"check_same_thread": False, "timeout": 30},
        )
        _SessionLocal = sessionmaker(bind=_engine, autoflush=False, expire_on_commit=False)
    return _engine, _SessionLocal()


def get_session() -> Optional[Session]:
    _, session = _get_engine_and_session()
    return session


def is_available() -> bool:
    """检查数字分身数据库是否可用"""
    engine, _ = _get_engine_and_session()
    return engine is not None


def _resolve_photo_url(photo_url: str) -> str:
    """将相对路径的头像 URL 补全为绝对路径"""
    if not photo_url:
        return ""
    if photo_url.startswith(("http://", "https://")):
        return photo_url
    base = settings.AVATAR_BACKEND_URL
    if base:
        base = base.rstrip("/")
        return f"{base}{photo_url}"
    return photo_url


def _get_global_token_balance(db: Session) -> int:
    """获取全局 token_account 余额（单行表）"""
    try:
        result = db.execute(text("SELECT balance FROM token_account LIMIT 1")).fetchone()
        return result.balance if result and result.balance else 0
    except Exception:
        return 0


def _decode_config(value) -> dict:
    if isinstance(value, dict):
        return value
    if isinstance(value, str):
        try:
            decoded = json.loads(value)
            return decoded if isinstance(decoded, dict) else {}
        except (json.JSONDecodeError, ValueError):
            return {}
    return {}


def is_delegated_avatar_user(user: VirtualUser | None) -> bool:
    return bool(user and (user.account or "").startswith(AVATAR_ACCOUNT_PREFIX))


def delegated_avatar_id(user: VirtualUser | None) -> str:
    if not is_delegated_avatar_user(user):
        return ""
    return (user.account or "")[len(AVATAR_ACCOUNT_PREFIX):]


def _list_square_interaction_authorizations(db: Session) -> list[dict]:
    """读取已明确授权分身参与广场互动的身份与会会令牌。"""
    rows = db.execute(text("""
        SELECT
            a.id AS avatar_id,
            a.name AS avatar_name,
            a.display_name AS avatar_display_name,
            a.description AS avatar_description,
            a.photo_url AS avatar_photo_url,
            a.config AS avatar_config,
            u.huihui_user_id,
            u.nickname AS owner_nickname,
            u.avatar_url AS owner_avatar_url,
            u.huihui_token
        FROM avatars a
        JOIN users u ON u.huihui_user_id = a.owner_id
        WHERE a.status = 'active'
    """)).fetchall()

    authorized = []
    for row in rows:
        config = _decode_config(row.avatar_config)
        permissions = config.get("authorizationPermissions", [])
        if not isinstance(permissions, list) or SQUARE_INTERACTION_PERMISSION not in permissions:
            continue
        platform_uid = str(row.huihui_user_id or "").strip()
        token = str(row.huihui_token or "").strip()
        if not platform_uid or not token:
            continue
        authorized.append({
            "avatar_id": str(row.avatar_id),
            "avatar_name": row.avatar_display_name or row.avatar_name or row.owner_nickname or "数字分身",
            "avatar_description": row.avatar_description or "",
            "avatar_url": _resolve_photo_url(row.avatar_photo_url or row.owner_avatar_url or ""),
            "config": config,
            "platform_uid": platform_uid,
            "token": token,
        })
    return authorized


def get_square_interaction_permissions(avatar_id: str) -> frozenset[str]:
    """实时复核授权；数据库不可用、令牌失效或撤权时一律拒绝执行。"""
    avatar_db = get_session()
    if avatar_db is None:
        return frozenset()
    try:
        authorized_ids = {
            item["avatar_id"] for item in _list_square_interaction_authorizations(avatar_db)
        }
        return SQUARE_INTERACTION_ACTIONS if avatar_id in authorized_ids else frozenset()
    except Exception as exc:
        logger.error(f"读取数字分身广场互动授权失败: {exc}")
        return frozenset()
    finally:
        avatar_db.close()


def _word_count_range(config: dict) -> tuple[int, int]:
    ranges = {
        "short": (10, 35),
        "medium": (20, 60),
        "long": (30, 80),
    }
    return ranges.get(str(config.get("responseLength") or "medium"), (20, 60))


async def sync_square_interaction_users(db) -> set[str]:
    """把已授权分身同步为调度器身份，并刷新其会会会话。"""
    avatar_db = get_session()
    if avatar_db is None:
        logger.warning("数字分身数据库不可用，跳过广场互动授权同步")
        return set()
    try:
        authorized = _list_square_interaction_authorizations(avatar_db)
    except Exception as exc:
        logger.error(f"同步数字分身广场互动授权失败: {exc}")
        return set()
    finally:
        avatar_db.close()

    from app.core.redis_client import delete_session, set_session

    result = await db.execute(
        select(VirtualUser).where(VirtualUser.account.like(f"{AVATAR_ACCOUNT_PREFIX}%"))
    )
    existing_users = {delegated_avatar_id(user): user for user in result.scalars().all()}
    authorized_ids = {item["avatar_id"] for item in authorized}

    for avatar_id, user in existing_users.items():
        if avatar_id not in authorized_ids:
            user.is_enabled = 0
            user.status = 0
            user.session_token = None
            user.session_expires_at = None
            await delete_session(user.id)

    for item in authorized:
        avatar_id = item["avatar_id"]
        user = existing_users.get(avatar_id)
        if user is None:
            user = VirtualUser(
                nickname=item["avatar_name"],
                account=f"{AVATAR_ACCOUNT_PREFIX}{avatar_id}",
                password_enc="",
                status=2,
                is_enabled=1,
                platform_uid=item["platform_uid"],
                remark="用户授权的数字分身广场互动身份",
            )
            db.add(user)
            await db.flush()

        expires_at = datetime.now() + timedelta(days=1)
        user.nickname = item["avatar_name"]
        user.real_name = item["avatar_name"]
        user.avatar_url = item["avatar_url"]
        user.platform_uid = item["platform_uid"]
        user.session_token = item["token"]
        user.session_expires_at = expires_at
        user.last_login_at = datetime.now()
        user.status = 2
        user.is_enabled = 1

        config = item["config"]
        personality_result = await db.execute(
            select(UserPersonality).where(UserPersonality.user_id == user.id)
        )
        personality = personality_result.scalar_one_or_none()
        word_min, word_max = _word_count_range(config)
        prompt_parts = [item["avatar_description"], str(config.get("systemPrompt") or "")]
        style_prompt = "\n".join(part.strip() for part in prompt_parts if part and part.strip())
        if personality is None:
            personality = UserPersonality(user_id=user.id)
            db.add(personality)
        personality.language_style = str(config.get("replyStyle") or "professional")
        personality.personality_desc = item["avatar_description"]
        personality.comment_style_prompt = style_prompt
        personality.word_count_min = word_min
        personality.word_count_max = word_max

        await set_session(user.id, {
            "token": item["token"],
            "session_id": f"avatar:{avatar_id}",
            "platform_uid": item["platform_uid"],
            "org_id": "",
            "login_time": datetime.now().isoformat(),
            "nickname": item["avatar_name"],
            "real_name": item["avatar_name"],
            "avatar": item["avatar_url"],
            "delegated_avatar_id": avatar_id,
        }, expire=86400)

    await db.commit()
    return authorized_ids


class AvatarService:

    @staticmethod
    def list_avatars(
        db: Session,
        page: int = 1,
        page_size: int = 20,
        keyword: Optional[str] = None,
        status: Optional[str] = None,
    ) -> Tuple[int, list]:
        """分页查询所有数字分身（含归属用户信息）"""
        where_clauses = []
        params = {}

        if keyword:
            where_clauses.append(
                "(a.name LIKE :kw OR a.display_name LIKE :kw)"
            )
            params["kw"] = f"%{keyword}%"

        if status:
            where_clauses.append("a.status = :status")
            params["status"] = status

        where_sql = ""
        if where_clauses:
            where_sql = "WHERE " + " AND ".join(where_clauses)

        # 计数
        count_sql = f"SELECT COUNT(*) FROM avatars a {where_sql}"
        total = db.execute(text(count_sql), params).scalar() or 0

        # 分页查询
        offset = (page - 1) * page_size
        params["limit"] = page_size
        params["offset"] = offset

        query = text(f"""
            SELECT
                a.id, a.name, a.display_name, a.description,
                a.photo_url, a.emoji, a.status, a.token_balance,
                a.config, a.created_at, a.updated_at,
                u.nickname AS owner_nickname,
                u.phone AS owner_phone
            FROM avatars a
            LEFT JOIN users u ON a.owner_id = u.huihui_user_id
            {where_sql}
            ORDER BY a.created_at DESC
            LIMIT :limit OFFSET :offset
        """)
        rows = db.execute(query, params).fetchall()

        items = []
        global_token = _get_global_token_balance(db)
        for row in rows:
            config = {}
            if row.config:
                if isinstance(row.config, str):
                    import json
                    try:
                        config = json.loads(row.config)
                    except (json.JSONDecodeError, ValueError):
                        config = {}
                elif isinstance(row.config, dict):
                    config = row.config
            items.append({
                "id": row.id,
                "name": row.name,
                "display_name": row.display_name,
                "description": row.description or "",
                "photo_url": _resolve_photo_url(row.photo_url),
                "emoji": row.emoji or "🤖",
                "status": row.status or "active",
                "token_balance": global_token or (row.token_balance or 0),
                "config": config,
                "owner_nickname": row.owner_nickname or "",
                "owner_phone": row.owner_phone or "",
                "created_at": str(row.created_at) if row.created_at else "",
                "updated_at": str(row.updated_at) if row.updated_at else "",
            })

        return total, items

    @staticmethod
    def get_avatar(db: Session, avatar_id: str) -> Optional[dict]:
        """获取单个数字分身详情"""
        query = text("""
            SELECT
                a.id, a.name, a.display_name, a.description,
                a.photo_url, a.emoji, a.status, a.token_balance,
                a.config, a.created_at, a.updated_at,
                u.nickname AS owner_nickname,
                u.phone AS owner_phone
            FROM avatars a
            LEFT JOIN users u ON a.owner_id = u.huihui_user_id
            WHERE a.id = :avatar_id
        """)
        row = db.execute(query, {"avatar_id": avatar_id}).fetchone()
        if not row:
            return None

        config = {}
        if row.config:
            if isinstance(row.config, str):
                import json
                try:
                    config = json.loads(row.config)
                except (json.JSONDecodeError, ValueError):
                    config = {}
            elif isinstance(row.config, dict):
                config = row.config

        global_token = _get_global_token_balance(db)
        return {
            "id": row.id,
            "name": row.name,
            "display_name": row.display_name,
            "description": row.description or "",
            "photo_url": _resolve_photo_url(row.photo_url),
            "emoji": row.emoji or "🤖",
            "status": row.status or "active",
            "token_balance": global_token or (row.token_balance or 0),
            "config": config,
            "owner_nickname": row.owner_nickname or "",
            "owner_phone": row.owner_phone or "",
            "created_at": str(row.created_at) if row.created_at else "",
            "updated_at": str(row.updated_at) if row.updated_at else "",
        }

    @staticmethod
    def update_status(db: Session, avatar_id: str, status: str) -> Optional[dict]:
        """更新数字分身状态（开关机）"""
        query = text("""
            UPDATE avatars SET status = :status, updated_at = datetime('now')
            WHERE id = :avatar_id
        """)
        result = db.execute(query, {"status": status, "avatar_id": avatar_id})
        db.commit()
        if result.rowcount == 0:
            return None
        return AvatarService.get_avatar(db, avatar_id)


avatar_service = AvatarService()