from sqlalchemy import or_
|
|
from app.models import KnowledgeModel, group_knowledge_table
|
from app.models.user_model import UserModel
|
from Log import logger
|
|
|
async def get_knowledge_list(db, user_id, keyword, page_size, page_index, status, knowledge, location):
|
user = db.query(UserModel).filter(UserModel.id == user_id).first()
|
if user is None:
|
return {"rows": []}
|
query = db.query(KnowledgeModel).filter(KnowledgeModel.knowledge_type==knowledge)
|
if user.permission != "admin":
|
klg_list = [j.id for i in user.groups for j in i.knowledges]
|
query = query.filter(or_(KnowledgeModel.tenant_id == user_id, KnowledgeModel.id.in_(klg_list)))
|
if location:
|
query = query.filter(KnowledgeModel.permission == "team")
|
if keyword:
|
query = query.filter(KnowledgeModel.name.like('%{}%'.format(keyword)))
|
total = query.count()
|
if page_size:
|
query = query.limit(page_size).offset((page_index - 1) * page_size)
|
return {"total": total, "rows": [kld.to_json() for kld in query.all()]}
|
|
|
async def create_knowledge_service(db, klg_id, name, description, icon, klg_type, user_id):
|
try:
|
dialog_model = KnowledgeModel(id=klg_id,name=name, description=description,icon=icon, knowledge_type=klg_type, tenant_id=user_id)
|
db.add(dialog_model)
|
db.commit()
|
db.refresh(dialog_model)
|
except Exception as e:
|
logger.error(e)
|
db.rollback()
|
return False
|
return True
|
