daren/apps/knowledge_base/models.py

# apps/knowledge_base/models.py
from django.db import models
from django.utils import timezone
from django.core.exceptions import ValidationError
import uuid
from apps.user.models import User

class KnowledgeBase(models.Model):
    """知识库模型"""
    KNOWLEDGE_BASE_TYPES = [
        ('admin', '管理级知识库'),
        ('leader', '部门级知识库'),
        ('member', '成员级知识库'),
        ('private', '私有知识库'),
        ('secret', '公司级别私密知识库'),
    ]

    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
    user_id = models.BigIntegerField(verbose_name='创建者ID')
    name = models.CharField(max_length=100, unique=True, verbose_name='知识库名称')
    desc = models.TextField(verbose_name='知识库描述', null=True, blank=True)
    type = models.CharField(
        max_length=20, 
        choices=KNOWLEDGE_BASE_TYPES,
        default='private',
        verbose_name='知识库类型'
    )
    department = models.CharField(max_length=50, null=True, blank=True)
    group = models.CharField(max_length=50, null=True, blank=True)
    documents = models.JSONField(default=list)
    char_length = models.IntegerField(default=0)
    document_count = models.IntegerField(default=0)
    external_id = models.UUIDField(null=True, blank=True)
    create_time = models.DateTimeField(auto_now_add=True)
    update_time = models.DateTimeField(auto_now=True)

    # def is_owner(self, user):
    #     """检查用户是否是所有者（通过权限表检查）"""
    #     from apps.permissions.models import KnowledgeBasePermission
    #     return str(user.id) == str(self.user_id) or KnowledgeBasePermission.objects.filter(
    #         knowledge_base=self,
    #         user=user,
    #         can_read=True,
    #         can_edit=True,
    #         can_delete=True,
    #         status='active'
    #     ).exists()

    # def get_owners(self):
    #     """获取所有所有者（包括创建者和具有完整权限的用户）"""
    #     from apps.user.models import User
    #     from apps.permissions.models import KnowledgeBasePermission
    #     # 获取创建者
    #     owners = [self.user_id]
    #     # 获取具有完整权限的用户
    #     permission_owners = KnowledgeBasePermission.objects.filter(
    #         knowledge_base=self,
    #         can_read=True,
    #         can_edit=True,
    #         can_delete=True,
    #         status='active'
    #     ).values_list('user_id', flat=True)
    #     owners.extend(permission_owners)
    #     return User.objects.filter(id__in=set(owners))

    def calculate_stats(self):
        """计算文档统计信息"""
        total_chars = 0
        doc_count = 0
        
        if self.documents:
            doc_count = len(self.documents)
            for doc in self.documents:
                if 'paragraphs' in doc:
                    for para in doc['paragraphs']:
                        if 'content' in para:
                            total_chars += len(para['content'])
                        if 'title' in para:
                            total_chars += len(para['title'])
        
        return doc_count, total_chars

    def save(self, *args, **kwargs):
        """重写保存方法"""
        # 只在创建时计算统计信息
        if not self.pk:  # 如果是新实例
            doc_count, char_count = self.calculate_stats()
            self.document_count = doc_count
            self.char_length = char_count
        
        # 直接调用Model的save方法
        models.Model.save(self, *args, **kwargs)

    @classmethod
    def update_external_id(cls, instance_id, external_id):
        """更新外部ID的静态方法"""
        cls.objects.filter(id=instance_id).update(external_id=external_id)

    class Meta:
        indexes = [
            models.Index(fields=['type']),
            models.Index(fields=['department']),
            models.Index(fields=['group'])
        ]

    def __str__(self):
        return f"{self.name} ({self.get_type_display()})"

    def clean(self):
        """验证知识库类型与创建者权限是否匹配"""
        # 移除所有权限验证，允许任何用户创建任何类型知识库
        pass

    def to_response_dict(self):
        """转换为API响应格式"""
        return {
            "create_time": self.create_time.isoformat(),
            "update_time": self.update_time.isoformat(),
            "id": str(self.id),
            "name": self.name,
            "desc": self.desc,
            "type": self.type,
            "user_id": str(self.user_id),
            "char_length": self.char_length,
            "document_count": self.document_count
        }

    def to_external_format(self):
        """转换为外部接口格式"""
        return {
            "name": self.name,
            "desc": self.desc,
            "documents": self.documents
        }


class KnowledgeBaseDocument(models.Model):
    """知识库文档关联模型"""
    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
    knowledge_base = models.ForeignKey(
        KnowledgeBase, 
        on_delete=models.CASCADE,
        related_name='kb_documents',
        verbose_name='知识库'
    )
    document_id = models.CharField(max_length=100, verbose_name='文档ID')
    document_name = models.CharField(max_length=255, verbose_name='文档名称')
    external_id = models.CharField(max_length=100, verbose_name='外部文档ID')
    uploader_name = models.CharField(max_length=100, default="未知用户", verbose_name='上传者姓名')
    status = models.CharField(
        max_length=20, 
        default='active',
        choices=[
            ('active', '有效'),
            ('deleted', '已删除')
        ],
        verbose_name='状态'
    )
    create_time = models.DateTimeField(auto_now_add=True, verbose_name='创建时间')
    update_time = models.DateTimeField(auto_now=True, verbose_name='更新时间')

    class Meta:
        unique_together = ['knowledge_base', 'document_id']
        indexes = [
            models.Index(fields=['knowledge_base', 'status']),
            models.Index(fields=['document_id']),
            models.Index(fields=['external_id'])
        ]
        verbose_name = '知识库文档'
        verbose_name_plural = '知识库文档'

    def __str__(self):
        return f"{self.knowledge_base.name} - {self.document_name}"