Step 11：百问百答（FAQ / Knowledge Base）

backend/app/crud/faq_item.py

@@ -0,0 +1,90 @@
+import uuid
+from typing import Sequence
+
+from sqlalchemy import 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,
+        keywords=item_in.keywords,
+        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) | (FaqItem.study_id.is_(None)))
+
+    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