179 lines
6.6 KiB
Python
179 lines
6.6 KiB
Python
# 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.accounts.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.UUIDField(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.accounts.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):
|
|
"""验证知识库类型与创建者权限是否匹配"""
|
|
try:
|
|
user = User.objects.get(id=self.user_id)
|
|
if user.role == 'member' and self.type != 'private':
|
|
raise ValidationError('组员只能创建私人知识库')
|
|
if user.role == 'leader' and self.type not in ['member', 'private']:
|
|
raise ValidationError('组长只能创建成员级或私人知识库')
|
|
except User.DoesNotExist:
|
|
raise ValidationError('创建者不存在')
|
|
|
|
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}" |