127 lines
5.5 KiB
Python
127 lines
5.5 KiB
Python
import uuid
|
|
from sqlalchemy.orm import Session
|
|
from app.models.user_model import User
|
|
from app.models.knowledge_model import Knowledge
|
|
from app.schemas.knowledge_schema import KnowledgeCreate, KnowledgeUpdate
|
|
from app.repositories import knowledge_repository
|
|
from app.core.logging_config import get_business_logger
|
|
|
|
# Obtain a dedicated logger for business logic
|
|
business_logger = get_business_logger()
|
|
|
|
|
|
def get_knowledges_paginated(
|
|
db: Session,
|
|
current_user: User,
|
|
filters: list,
|
|
page: int,
|
|
pagesize: int,
|
|
orderby: str = None,
|
|
desc: bool = False
|
|
) -> tuple[int, list]:
|
|
business_logger.debug(f"Query knowledge base in pages: username={current_user.username}, page={page}, pagesize={pagesize}, orderby={orderby}, desc={desc}")
|
|
|
|
try:
|
|
total, items = knowledge_repository.get_knowledges_paginated(
|
|
db=db,
|
|
filters=filters,
|
|
page=page,
|
|
pagesize=pagesize,
|
|
orderby=orderby,
|
|
desc=desc
|
|
)
|
|
business_logger.info(f"The knowledge base paging query has been successful: username={current_user.username}, total={total}, Number of current page={len(items)}")
|
|
return total, items
|
|
except Exception as e:
|
|
business_logger.error(f"Querying knowledge base pagination failed: username={current_user.username} - {str(e)}")
|
|
raise
|
|
|
|
|
|
def get_chunked_knowledgeids(
|
|
db: Session,
|
|
current_user: User,
|
|
filters: list
|
|
) -> list:
|
|
business_logger.debug(f"Query the list of vectorized knowledge base IDs: username={current_user.username}")
|
|
|
|
try:
|
|
items = knowledge_repository.get_chunked_knowledgeids(
|
|
db=db,
|
|
filters=filters
|
|
)
|
|
business_logger.info(f"Querying the vectorized knowledge base id list succeeded: username={current_user.username} count={len(items)}")
|
|
return items
|
|
except Exception as e:
|
|
business_logger.error(f"Querying the vectorized knowledge base id list failed: username={current_user.username} - {str(e)}")
|
|
raise
|
|
|
|
|
|
def create_knowledge(
|
|
db: Session, knowledge: KnowledgeCreate, current_user: User
|
|
) -> Knowledge:
|
|
business_logger.info(f"Create a knowledge base: {knowledge.name}, creator: {current_user.username}")
|
|
|
|
try:
|
|
knowledge.created_by = current_user.id
|
|
if knowledge.workspace_id is None:
|
|
knowledge.workspace_id = current_user.current_workspace_id
|
|
if knowledge.parent_id is None:
|
|
knowledge.parent_id = knowledge.workspace_id
|
|
business_logger.debug(f"Start creating the knowledge base: {knowledge.name}")
|
|
db_knowledge = knowledge_repository.create_knowledge(
|
|
db=db, knowledge=knowledge
|
|
)
|
|
business_logger.info(f"The knowledge base has been successfully created: {knowledge.name} (ID: {db_knowledge.id}), creator: {current_user.username}")
|
|
return db_knowledge
|
|
except Exception as e:
|
|
business_logger.error(f"Failed to create a knowledge base: {knowledge.name} - {str(e)}")
|
|
raise
|
|
|
|
|
|
def get_knowledge_by_id(db: Session, knowledge_id: uuid.UUID, current_user: User) -> Knowledge | None:
|
|
business_logger.debug(f"Query knowledge base based on ID: knowledge_id={knowledge_id}, username: {current_user.username}")
|
|
|
|
try:
|
|
knowledge = knowledge_repository.get_knowledge_by_id(db=db, knowledge_id=knowledge_id)
|
|
if knowledge:
|
|
business_logger.info(f"knowledge base query successful: {knowledge.name} (ID: {knowledge_id})")
|
|
else:
|
|
business_logger.warning(f"knowledge base does not exist: knowledge_id={knowledge_id}")
|
|
return knowledge
|
|
except Exception as e:
|
|
business_logger.error(f"Failed to query the knowledge base based on the ID: knowledge_id={knowledge_id} - {str(e)}")
|
|
raise
|
|
|
|
|
|
def get_knowledge_by_name(db: Session, name: str, current_user: User) -> Knowledge | None:
|
|
business_logger.debug(f"Query knowledge base based on name: name={name}, username: {current_user.username}")
|
|
|
|
try:
|
|
knowledge = knowledge_repository.get_knowledge_by_name(db=db, name=name, workspace_id=current_user.current_workspace_id)
|
|
if knowledge:
|
|
business_logger.info(f"knowledge base query successful: {name} (ID: {knowledge.id})")
|
|
else:
|
|
business_logger.warning(f"knowledge base does not exist: name={name}")
|
|
return knowledge
|
|
except Exception as e:
|
|
business_logger.error(f"Failed to query the knowledge base based on the name: name={name} - {str(e)}")
|
|
raise
|
|
|
|
|
|
def delete_knowledge_by_id(db: Session, knowledge_id: uuid.UUID, current_user: User) -> None:
|
|
business_logger.info(f"Delete knowledge base: knowledge_id={knowledge_id}, operator: {current_user.username}")
|
|
|
|
try:
|
|
# First, query the knowledge base information for logging purposes
|
|
knowledge = knowledge_repository.get_knowledge_by_id(db=db, knowledge_id=knowledge_id)
|
|
if knowledge:
|
|
business_logger.debug(f"Execute knowledge base deletion: {knowledge.name} (ID: {knowledge_id})")
|
|
else:
|
|
business_logger.warning(f"The knowledge base to be deleted does not exist: knowledge_id={knowledge_id}")
|
|
|
|
knowledge_repository.delete_knowledge_by_id(db=db, knowledge_id=knowledge_id)
|
|
business_logger.info(f"knowledge base record deleted successfully: knowledge_id={knowledge_id}, operator: {current_user.username}")
|
|
except Exception as e:
|
|
business_logger.error(f"Failed to delete knowledge base: knowledge_id={knowledge_id} - {str(e)}")
|
|
raise
|