rag-gateway/app/models/knowledge_model.py

import json
from datetime import datetime
from typing import Optional

from pydantic import BaseModel
from sqlalchemy import Column, Integer, String, Table, ForeignKey, DateTime, BigInteger, Text, Float, Boolean
from sqlalchemy.orm import relationship, backref

from app.models.base_model import Base

class KnowledgeModel(Base):
    __tablename__ = 'knowledgebases'
    __mapper_args__ = {
        # "order_by": 'SEQ'
    }
    id = Column(String(36), primary_key=True)  # id
    name = Column(String(128))                 # 名称
    create_date = Column(DateTime, default=datetime.now())             # 创建时间
    update_date = Column(DateTime, default=datetime.now(), onupdate=datetime.now())             # 更新时间
    permission = Column(String(32), default="me")
    tenant_id = Column(String(32))              # 创建人id
    description = Column(Text)                 # 说明
    status = Column(String(1))                    # 状态
    documents = Column(Integer, default=0)                    # 文档
    icon = Column(String(128))                    # 文档
    knowledge_type = Column(String(1))                  # 知识库平台
    embd_id = Column(String(128))


    def get_id(self):
        return str(self.id)

    def to_json(self, user_id=""):
        return {
            'id': self.id,
            'name': self.name,
            'create_time': self.create_date.strftime('%Y-%m-%d %H:%M:%S'),
            'update_time': self.update_date.strftime('%Y-%m-%d %H:%M:%S'),
            'permission': self.permission,
            'user_id': self.tenant_id,
            'description': self.description,
            'status': self.status,
            'documents': self.documents if self.documents else 0,
            'icon': self.icon,
            'embd_id': self.embd_id,
            'link': True if (self.tenant_id == user_id or user_id == "admin" ) and self.permission == "team" else False,
        }
    def __repr__(self):
        return '<Knowledge name:%r url:%r>\n' % (self.name, self.id)


class klgParameter(BaseModel):
    id: str
    name: str
    description: Optional[str] = ""
    icon: str
    klgType: str



class klgIcon(BaseModel):
    id: str
    icon: str



class KnowledgeUserModel(Base):
    __tablename__ = 'knowledgebase_user'

    id = Column(Integer, primary_key=True)  # id
    kb_id = Column(String(36))
    user_id = Column(Integer)
    create_date = Column(DateTime, default=datetime.now())             # 创建时间
    update_date = Column(DateTime, default=datetime.now(), onupdate=datetime.now())             # 更新时间
    status = Column(Integer)  # 状态



    def get_id(self):
        return str(self.id)

    def to_json(self, user_id=""):
        return {
            'id': self.id,
            'kb_id': self.kb_id,
            'create_time': self.create_date.strftime('%Y-%m-%d %H:%M:%S'),
            'update_time': self.update_date.strftime('%Y-%m-%d %H:%M:%S'),
            'user_id': self.user_id,
            'status': self.status,
            'owner': True if str(self.user_id) == user_id else False,
        }
    def __repr__(self):
        return '<Knowledge name:%r url:%r>\n' % (self.name, self.id)



class KlgOwner(BaseModel):
    knowledgeId: str
    UserId: int