Olympus/backend/hermes/app/repositories/knowledge_repository.py

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

from app.models.knowledge import (
    KnowledgeDevice,
    KnowledgeDocument,
    KnowledgeManufacturer,
    KnowledgeNote,
)


def paginate(query, limit: int, offset: int):
    return query.limit(max(1, min(limit, 100))).offset(max(0, offset))


class KnowledgeRepository:
    @staticmethod
    def list_manufacturers(db: Session, *, q: str = "", limit: int = 50, offset: int = 0) -> list[KnowledgeManufacturer]:
        query = select(KnowledgeManufacturer).order_by(KnowledgeManufacturer.name)
        if q:
            pattern = f"%{q.lower()}%"
            query = query.where(func.lower(KnowledgeManufacturer.name).like(pattern))
        return list(db.scalars(paginate(query, limit, offset)))

    @staticmethod
    def get_manufacturer(db: Session, manufacturer_id: int) -> KnowledgeManufacturer | None:
        return db.get(KnowledgeManufacturer, manufacturer_id)

    @staticmethod
    def get_manufacturer_by_name(db: Session, name: str, exclude_id: int | None = None) -> KnowledgeManufacturer | None:
        query = select(KnowledgeManufacturer).where(func.lower(KnowledgeManufacturer.name) == name.lower())
        if exclude_id is not None:
            query = query.where(KnowledgeManufacturer.id != exclude_id)
        return db.scalar(query)

    @staticmethod
    def get_manufacturer_by_slug(db: Session, slug: str, exclude_id: int | None = None) -> KnowledgeManufacturer | None:
        query = select(KnowledgeManufacturer).where(KnowledgeManufacturer.slug == slug)
        if exclude_id is not None:
            query = query.where(KnowledgeManufacturer.id != exclude_id)
        return db.scalar(query)

    @staticmethod
    def list_devices(
        db: Session,
        *,
        q: str = "",
        manufacturer_id: int | None = None,
        device_type: str = "",
        limit: int = 50,
        offset: int = 0,
    ) -> list[KnowledgeDevice]:
        query = select(KnowledgeDevice).options(joinedload(KnowledgeDevice.manufacturer)).order_by(KnowledgeDevice.name)
        if manufacturer_id is not None:
            query = query.where(KnowledgeDevice.manufacturer_id == manufacturer_id)
        if device_type:
            query = query.where(KnowledgeDevice.device_type == device_type)
        if q:
            pattern = f"%{q.lower()}%"
            query = query.where(
                or_(
                    func.lower(KnowledgeDevice.name).like(pattern),
                    func.lower(KnowledgeDevice.model_number).like(pattern),
                )
            )
        return list(db.scalars(paginate(query, limit, offset)))

    @staticmethod
    def get_device(db: Session, device_id: int) -> KnowledgeDevice | None:
        return db.scalar(
            select(KnowledgeDevice)
            .where(KnowledgeDevice.id == device_id)
            .options(joinedload(KnowledgeDevice.manufacturer))
        )

    @staticmethod
    def get_device_by_slug(
        db: Session,
        *,
        manufacturer_id: int,
        slug: str,
        exclude_id: int | None = None,
    ) -> KnowledgeDevice | None:
        query = select(KnowledgeDevice).where(KnowledgeDevice.manufacturer_id == manufacturer_id, KnowledgeDevice.slug == slug)
        if exclude_id is not None:
            query = query.where(KnowledgeDevice.id != exclude_id)
        return db.scalar(query)

    @staticmethod
    def list_documents(
        db: Session,
        *,
        q: str = "",
        manufacturer_id: int | None = None,
        device_id: int | None = None,
        document_type: str = "",
        language: str = "",
        tag: str = "",
        limit: int = 50,
        offset: int = 0,
    ) -> list[KnowledgeDocument]:
        query = (
            select(KnowledgeDocument)
            .options(joinedload(KnowledgeDocument.manufacturer), joinedload(KnowledgeDocument.device).joinedload(KnowledgeDevice.manufacturer))
            .order_by(KnowledgeDocument.created_at.desc())
        )
        if manufacturer_id is not None:
            query = query.where(KnowledgeDocument.manufacturer_id == manufacturer_id)
        if device_id is not None:
            query = query.where(KnowledgeDocument.device_id == device_id)
        if document_type:
            query = query.where(KnowledgeDocument.document_type == document_type)
        if language:
            query = query.where(KnowledgeDocument.language == language)
        if q:
            pattern = f"%{q.lower()}%"
            query = query.where(func.lower(KnowledgeDocument.title).like(pattern))
        documents = list(db.scalars(paginate(query, limit, offset)))
        if tag:
            tag_lower = tag.lower()
            documents = [document for document in documents if any(item.lower() == tag_lower for item in document.tags)]
        return documents

    @staticmethod
    def get_document(db: Session, document_id: int) -> KnowledgeDocument | None:
        return db.scalar(
            select(KnowledgeDocument)
            .where(KnowledgeDocument.id == document_id)
            .options(joinedload(KnowledgeDocument.manufacturer), joinedload(KnowledgeDocument.device).joinedload(KnowledgeDevice.manufacturer))
        )

    @staticmethod
    def get_document_by_slug(db: Session, slug: str, exclude_id: int | None = None) -> KnowledgeDocument | None:
        query = select(KnowledgeDocument).where(KnowledgeDocument.slug == slug)
        if exclude_id is not None:
            query = query.where(KnowledgeDocument.id != exclude_id)
        return db.scalar(query)

    @staticmethod
    def list_notes(
        db: Session,
        *,
        q: str = "",
        manufacturer_id: int | None = None,
        device_id: int | None = None,
        note_type: str = "",
        tag: str = "",
        limit: int = 50,
        offset: int = 0,
    ) -> list[KnowledgeNote]:
        query = (
            select(KnowledgeNote)
            .options(joinedload(KnowledgeNote.manufacturer), joinedload(KnowledgeNote.device).joinedload(KnowledgeDevice.manufacturer))
            .order_by(KnowledgeNote.created_at.desc())
        )
        if manufacturer_id is not None:
            query = query.where(KnowledgeNote.manufacturer_id == manufacturer_id)
        if device_id is not None:
            query = query.where(KnowledgeNote.device_id == device_id)
        if note_type:
            query = query.where(KnowledgeNote.note_type == note_type)
        if q:
            pattern = f"%{q.lower()}%"
            query = query.where(or_(func.lower(KnowledgeNote.title).like(pattern), func.lower(KnowledgeNote.content).like(pattern)))
        notes = list(db.scalars(paginate(query, limit, offset)))
        if tag:
            tag_lower = tag.lower()
            notes = [note for note in notes if any(item.lower() == tag_lower for item in note.tags)]
        return notes

    @staticmethod
    def get_note(db: Session, note_id: int) -> KnowledgeNote | None:
        return db.scalar(
            select(KnowledgeNote)
            .where(KnowledgeNote.id == note_id)
            .options(joinedload(KnowledgeNote.manufacturer), joinedload(KnowledgeNote.device).joinedload(KnowledgeDevice.manufacturer))
        )