82 lines
2.5 KiB
Python
82 lines
2.5 KiB
Python
"""世谱数据访问。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from sqlalchemy import or_, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.lorebook import Lorebook, LorebookEntry
|
|
|
|
|
|
async def list_active_books(
|
|
session: AsyncSession,
|
|
character_id: str,
|
|
owner_scope: str,
|
|
) -> list[Lorebook]:
|
|
"""查角色的 active 世谱：character 作用域（按人物隔离）+ global 作用域（跨人物共享）。"""
|
|
result = await session.execute(
|
|
select(Lorebook).where(
|
|
Lorebook.status == "active",
|
|
Lorebook.owner_scope == owner_scope,
|
|
or_(Lorebook.character_id == character_id, Lorebook.scope == "global"),
|
|
)
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def list_entries_for_books(
|
|
session: AsyncSession,
|
|
lorebook_ids: list[str],
|
|
) -> list[LorebookEntry]:
|
|
"""批量查多本书的启用条目，按 insertion_order 排序。"""
|
|
if not lorebook_ids:
|
|
return []
|
|
result = await session.execute(
|
|
select(LorebookEntry)
|
|
.where(
|
|
LorebookEntry.lorebook_id.in_(lorebook_ids),
|
|
LorebookEntry.enabled.is_(True),
|
|
)
|
|
.order_by(LorebookEntry.insertion_order)
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def get_book(session: AsyncSession, lorebook_id: str) -> Lorebook | None:
|
|
return await session.get(Lorebook, lorebook_id)
|
|
|
|
|
|
async def list_books(
|
|
session: AsyncSession,
|
|
character_id: str | None,
|
|
owner_scope: str,
|
|
*,
|
|
scope: str | None = None,
|
|
) -> list[Lorebook]:
|
|
"""列出 active 世谱（archived 不返回）。"""
|
|
conditions = [
|
|
Lorebook.status == "active",
|
|
Lorebook.owner_scope == owner_scope,
|
|
]
|
|
if character_id is not None:
|
|
conditions.append(or_(Lorebook.character_id == character_id, Lorebook.scope == "global"))
|
|
if scope is not None:
|
|
conditions.append(Lorebook.scope == scope)
|
|
result = await session.execute(
|
|
select(Lorebook).where(*conditions).order_by(Lorebook.created_at.desc())
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def list_book_entries(session: AsyncSession, lorebook_id: str) -> list[LorebookEntry]:
|
|
result = await session.execute(
|
|
select(LorebookEntry)
|
|
.where(LorebookEntry.lorebook_id == lorebook_id)
|
|
.order_by(LorebookEntry.insertion_order)
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def get_entry(session: AsyncSession, entry_id: str) -> LorebookEntry | None:
|
|
return await session.get(LorebookEntry, entry_id)
|