Cheng Zhou
126 lines
4.1 KiB
Python
126 lines
4.1 KiB
Python
import uuid
|
|
from typing import Sequence
|
|
|
|
from sqlalchemy import func, or_, select, update as sa_update
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.faq_category import FaqCategory
|
|
from app.models.faq_item import FaqItem
|
|
from app.schemas.faq import FaqCreate, FaqUpdate
|
|
|
|
|
|
async def create_item(db: AsyncSession, item_in: FaqCreate, *, created_by: uuid.UUID) -> FaqItem:
|
|
# ensure category exists and matches study scope
|
|
category = await db.get(FaqCategory, item_in.category_id)
|
|
if not category:
|
|
raise ValueError("Category not found")
|
|
if category.study_id != item_in.study_id:
|
|
# allow global category if item is global (None) or project if matches
|
|
if not (category.study_id is None and item_in.study_id is None):
|
|
raise ValueError("Category scope mismatch")
|
|
|
|
item = FaqItem(
|
|
study_id=item_in.study_id,
|
|
category_id=item_in.category_id,
|
|
question=item_in.question,
|
|
answer=item_in.answer or "",
|
|
status="PENDING",
|
|
version=1,
|
|
is_active=True,
|
|
created_by=created_by,
|
|
)
|
|
db.add(item)
|
|
await db.commit()
|
|
await db.refresh(item)
|
|
return item
|
|
|
|
|
|
async def get_item(db: AsyncSession, item_id: uuid.UUID) -> FaqItem | None:
|
|
result = await db.execute(select(FaqItem).where(FaqItem.id == item_id))
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def list_items(
|
|
db: AsyncSession,
|
|
*,
|
|
study_id: uuid.UUID | None,
|
|
category_id: uuid.UUID | None = None,
|
|
keyword: str | None = None,
|
|
is_active: bool | None = True,
|
|
study_scope: str | None = "project",
|
|
) -> Sequence[FaqItem]:
|
|
stmt = select(FaqItem).join(FaqCategory, FaqCategory.id == FaqItem.category_id)
|
|
|
|
if study_scope == "all":
|
|
pass
|
|
elif study_scope == "global":
|
|
stmt = stmt.where(FaqItem.study_id.is_(None))
|
|
else: # project (default)
|
|
stmt = stmt.where(FaqItem.study_id == study_id)
|
|
|
|
if category_id:
|
|
stmt = stmt.where(FaqItem.category_id == category_id)
|
|
if keyword:
|
|
like_pattern = f"%{keyword}%"
|
|
stmt = stmt.where(
|
|
or_(
|
|
FaqItem.question.ilike(like_pattern),
|
|
FaqItem.answer.ilike(like_pattern),
|
|
FaqItem.keywords.ilike(like_pattern),
|
|
)
|
|
)
|
|
if is_active is not None:
|
|
stmt = stmt.where(FaqItem.is_active == is_active)
|
|
result = await db.execute(stmt)
|
|
return result.scalars().all()
|
|
|
|
|
|
async def update_item(db: AsyncSession, item: FaqItem, item_in: FaqUpdate) -> FaqItem:
|
|
update_data = item_in.model_dump(exclude_unset=True)
|
|
if "question" in update_data or "answer" in update_data or "keywords" in update_data:
|
|
update_data["version"] = item.version + 1
|
|
if update_data:
|
|
await db.execute(
|
|
sa_update(FaqItem)
|
|
.where(FaqItem.id == item.id)
|
|
.values(**update_data)
|
|
)
|
|
await db.commit()
|
|
await db.refresh(item)
|
|
return item
|
|
|
|
|
|
async def touch_item(db: AsyncSession, item_id: uuid.UUID) -> None:
|
|
await db.execute(
|
|
sa_update(FaqItem)
|
|
.where(FaqItem.id == item_id)
|
|
.values(updated_at=func.now())
|
|
)
|
|
await db.commit()
|
|
|
|
|
|
async def set_best_reply(db: AsyncSession, item_id: uuid.UUID, reply_id: uuid.UUID | None) -> None:
|
|
await db.execute(
|
|
sa_update(FaqItem)
|
|
.where(FaqItem.id == item_id)
|
|
.values(best_reply_id=reply_id, updated_at=func.now())
|
|
)
|
|
await db.commit()
|
|
|
|
|
|
async def set_status(db: AsyncSession, item_id: uuid.UUID, status: str, resolved_by_confirm: bool | None = None) -> None:
|
|
values: dict[str, object] = {"status": status, "updated_at": func.now()}
|
|
if resolved_by_confirm is not None:
|
|
values["resolved_by_confirm"] = resolved_by_confirm
|
|
await db.execute(
|
|
sa_update(FaqItem)
|
|
.where(FaqItem.id == item_id)
|
|
.values(**values)
|
|
)
|
|
await db.commit()
|
|
|
|
|
|
async def count_items_by_category(db: AsyncSession, category_id: uuid.UUID) -> int:
|
|
result = await db.execute(select(func.count()).select_from(FaqItem).where(FaqItem.category_id == category_id))
|
|
return int(result.scalar_one())
|