130 lines
4.1 KiB
Python
130 lines
4.1 KiB
Python
"""人物数据访问。"""
|
|
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.character import Character
|
|
from app.models.xirang import Memory
|
|
|
|
|
|
async def get_active_character(session: AsyncSession, owner_scope: str) -> Character | None:
|
|
result = await session.execute(
|
|
select(Character)
|
|
.where(
|
|
Character.owner_scope == owner_scope,
|
|
Character.status.in_(["active", "initialized"]),
|
|
Character.is_system_template.is_(False),
|
|
)
|
|
.order_by(Character.created_at.asc())
|
|
.limit(1)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def get_character(session: AsyncSession, character_id: str) -> Character | None:
|
|
return await session.get(Character, character_id)
|
|
|
|
|
|
async def get_system_template_by_key(session: AsyncSession, template_key: str) -> Character | None:
|
|
result = await session.execute(
|
|
select(Character)
|
|
.where(
|
|
Character.is_system_template.is_(True),
|
|
Character.template_key == template_key,
|
|
)
|
|
.order_by(Character.created_at.asc())
|
|
.limit(1)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def get_character_by_source_template(
|
|
session: AsyncSession, source_template_id: str, owner_scope: str
|
|
) -> Character | None:
|
|
"""按来源模板 ID 查找已存在的用户人物，用于启动时跳过重复创建。"""
|
|
result = await session.execute(
|
|
select(Character)
|
|
.where(
|
|
Character.source_template_id == source_template_id,
|
|
Character.owner_scope == owner_scope,
|
|
Character.is_system_template.is_(False),
|
|
)
|
|
.order_by(Character.created_at.asc())
|
|
.limit(1)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def add_character(session: AsyncSession, character: Character) -> Character:
|
|
session.add(character)
|
|
await session.flush()
|
|
return character
|
|
|
|
|
|
async def list_characters(
|
|
session: AsyncSession,
|
|
owner_scope: str,
|
|
*,
|
|
status: str | None = None,
|
|
page: int = 1,
|
|
page_size: int = 20,
|
|
) -> tuple[list[Character], int]:
|
|
"""列出某归属范围下的非模板人物，按创建时间倒序分页。"""
|
|
conditions = [
|
|
Character.owner_scope == owner_scope,
|
|
Character.is_system_template.is_(False),
|
|
]
|
|
if status is not None:
|
|
conditions.append(Character.status == status)
|
|
|
|
total = (
|
|
await session.execute(select(func.count()).select_from(Character).where(*conditions))
|
|
).scalar_one()
|
|
|
|
items_result = await session.execute(
|
|
select(Character)
|
|
.where(*conditions)
|
|
.order_by(Character.created_at.desc())
|
|
.offset(max(page - 1, 0) * page_size)
|
|
.limit(page_size)
|
|
)
|
|
items = list(items_result.scalars().all())
|
|
return items, int(total or 0)
|
|
|
|
|
|
async def count_memories(session: AsyncSession, character_id: str) -> int:
|
|
"""统计人物关联的未删除记忆数量（列表项 memory_count）。"""
|
|
total = (
|
|
await session.execute(
|
|
select(func.count())
|
|
.select_from(Memory)
|
|
.where(Memory.character_id == character_id, Memory.status != "tombstoned")
|
|
)
|
|
).scalar_one()
|
|
return int(total or 0)
|
|
|
|
|
|
async def list_active_in_scope(
|
|
session: AsyncSession, owner_scope: str, *, except_id: str
|
|
) -> list[Character]:
|
|
"""列出同归属范围下处于激活/已初始化态的其他人物（A1：激活互斥用）。"""
|
|
result = await session.execute(
|
|
select(Character).where(
|
|
Character.owner_scope == owner_scope,
|
|
Character.is_system_template.is_(False),
|
|
Character.status.in_(["active", "initialized"]),
|
|
Character.id != except_id,
|
|
)
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def list_system_templates(session: AsyncSession) -> list[Character]:
|
|
"""列出所有系统模板（is_system_template=True），按 template_key 排序。"""
|
|
result = await session.execute(
|
|
select(Character)
|
|
.where(Character.is_system_template.is_(True))
|
|
.order_by(Character.template_key.asc())
|
|
)
|
|
return list(result.scalars().all())
|