LQAdminPlatform/src/app/sample/schemas/knowledge_base.py

from typing import Optional, List
from pydantic import BaseModel, Field
from app.schemas.base import BaseModelSchema

class MetadataField(BaseModel):
    """元数据字段定义"""
    field_zh_name: str = Field(..., description="中文名称")
    field_en_name: str = Field(..., description="英文名称")
    field_type: str = Field(..., description="字段类型: text/num")
    remark: Optional[str] = Field(None, description="备注")

class KnowledgeBaseBase(BaseModel):
    name: str = Field(..., description="知识库名称")
    collection_name_parent: Optional[str] = Field(None, description="Milvus集合名称(父)")
    collection_name_children: str = Field(..., description="Milvus集合名称(子)")
    description: Optional[str] = Field(None, description="描述")
    # 默认状态改为禁用，只有同步成功后才置为 normal
    status: Optional[str] = Field("disabled", description="状态")

class CustomSchemaField(BaseModel):
    """自定义Schema字段定义"""
    field_name: str = Field(..., description="字段名称(英文)")
    field_type: str = Field(..., description="字段类型: BOOL/INT8/INT16/INT32/INT64/FLOAT/DOUBLE/VARCHAR/JSON")
    max_length: Optional[int] = Field(None, description="最大长度(VARCHAR需要)")
    is_primary: bool = Field(False, description="是否主键(通常不需要用户指定)")
    description: Optional[str] = Field(None, description="描述")

class KnowledgeBaseCreate(KnowledgeBaseBase):
    """创建知识库请求参数"""
    dimension: int = Field(4096, description="向量维度，默认4096")
    metadata_fields: Optional[List[MetadataField]] = Field(None, description="元数据字段列表")
    custom_schemas: Optional[List[CustomSchemaField]] = Field(None, description="自定义Schema字段列表")

class KnowledgeBaseUpdate(BaseModel):
    """更新知识库请求参数"""
    name: Optional[str] = None
    description: Optional[str] = None
    status: Optional[str] = None
    metadata_fields: Optional[List[MetadataField]] = Field(None, description="元数据字段列表")

class DescriptionUpdate(BaseModel):
    """仅更新描述（保留原有兼容性）"""
    description: str

from datetime import datetime

class KnowledgeBaseResponse(BaseModelSchema):
    """知识库响应模型"""
    id: str
    name: str
    collection_name_parent: Optional[str] = None
    collection_name_children: Optional[str]
    description: Optional[str]
    status: str
    document_count: int
    is_synced: bool = Field(False, description="是否已同步到Milvus")
    created_by: Optional[str] = None
    updated_by: Optional[str] = None
    created_at: Optional[str] = None
    updated_at: Optional[str] = None
    created_time: Optional[str] = None
    updated_time: Optional[str] = None

    class Config:
        from_attributes = True
        # json_encoders = {
        #     datetime: lambda v: v.strftime("%Y-%m-%d %H:%M:%S") if v else None
        # }