vent_ai_platform/app/db/repositories/knowledge.py

from __future__ import annotations

from sqlalchemy import or_, select
from sqlalchemy.orm import Session

from app.db.models.knowledge import KnowledgeItem
from app.db.repositories.base import paginate


def list_knowledge(
    db: Session,
    *,
    mine_id: str | None = None,
    knowledge_type: str | None = None,
    category: str | None = None,
    keyword: str | None = None,
    page: int = 1,
    page_size: int = 20,
) -> tuple[list[KnowledgeItem], int]:
    statement = select(KnowledgeItem)
    if mine_id:
        statement = statement.where(
            or_(KnowledgeItem.mine_id == mine_id, KnowledgeItem.mine_id.is_(None))
        )
    if knowledge_type:
        statement = statement.where(KnowledgeItem.knowledge_type == knowledge_type)
    if category:
        statement = statement.where(KnowledgeItem.category == category)
    if keyword:
        query = f"%{keyword.strip()}%"
        statement = statement.where(
            or_(
                KnowledgeItem.title.like(query),
                KnowledgeItem.content.like(query),
                KnowledgeItem.summary.like(query),
                KnowledgeItem.code.like(query),
            )
        )
    statement = statement.order_by(KnowledgeItem.updated_at.desc())
    return paginate(db, statement, page, page_size)


def get_knowledge(db: Session, knowledge_id: str) -> KnowledgeItem | None:
    return db.get(KnowledgeItem, knowledge_id)