85 lines
3.3 KiB
Python
85 lines
3.3 KiB
Python
"""知识库服务：CRUD + Token 管理。"""
|
|
|
|
from app.core.errors import NotFoundError, PermissionDeniedError
|
|
from app.models.knowledge_base import KnowledgeBase
|
|
from app.models.user import User
|
|
from app.repositories.kb_repo import KnowledgeBaseRepository
|
|
from app.services.token_service import TokenService
|
|
from sqlalchemy.orm import Session
|
|
|
|
|
|
class KnowledgeBaseService:
|
|
def __init__(self, session: Session) -> None:
|
|
self._session = session
|
|
self._kb_repo = KnowledgeBaseRepository(session)
|
|
self._token_svc = TokenService()
|
|
|
|
def create(self, user: User, name: str, description: str | None) -> tuple[KnowledgeBase, str]:
|
|
"""创建知识库。返回 (kb, full_token)。
|
|
|
|
full_token 仅此一次返回，用于构建完整 AI URL。
|
|
"""
|
|
token, token_hash, token_encrypted, token_hint = self._token_svc.create_token_pair()
|
|
kb = self._kb_repo.create(
|
|
user_id=user.id,
|
|
name=name,
|
|
description=description,
|
|
token_hash=token_hash,
|
|
token_encrypted=token_encrypted,
|
|
token_hint=token_hint,
|
|
)
|
|
self._session.commit()
|
|
return kb, token
|
|
|
|
def get_or_404(self, kb_id: str, user: User) -> KnowledgeBase:
|
|
"""获取知识库，校验所有权。不存在或无权 → 404。"""
|
|
kb = self._kb_repo.get_by_id(kb_id)
|
|
if kb is None or kb.user_id != user.id or kb.status == "DELETED":
|
|
raise NotFoundError("知识库不存在。")
|
|
return kb
|
|
|
|
def list_by_user(self, user: User, *, page: int = 1, page_size: int = 20):
|
|
"""分页列表。"""
|
|
return self._kb_repo.list_by_user(user.id, page=page, page_size=page_size)
|
|
|
|
def update(self, kb_id: str, user: User, name: str | None, description: str | None) -> KnowledgeBase:
|
|
kb = self.get_or_404(kb_id, user)
|
|
update_fields = {}
|
|
if name is not None:
|
|
update_fields["name"] = name
|
|
if description is not None:
|
|
update_fields["description"] = description
|
|
if update_fields:
|
|
self._kb_repo.update(kb, **update_fields)
|
|
self._session.commit()
|
|
return kb
|
|
|
|
def delete(self, kb_id: str, user: User) -> None:
|
|
kb = self.get_or_404(kb_id, user)
|
|
self._kb_repo.delete(kb)
|
|
self._session.commit()
|
|
|
|
def regenerate_token(self, kb_id: str, user: User) -> tuple[KnowledgeBase, str]:
|
|
"""重新生成 Token。旧链接立即失效。"""
|
|
kb = self.get_or_404(kb_id, user)
|
|
token, token_hash, token_encrypted, token_hint = self._token_svc.create_token_pair()
|
|
self._kb_repo.update(
|
|
kb,
|
|
token_hash=token_hash,
|
|
token_encrypted=token_encrypted,
|
|
token_hint=token_hint,
|
|
)
|
|
self._session.commit()
|
|
return kb, token
|
|
|
|
def set_enabled(self, kb_id: str, user: User, enabled: bool) -> KnowledgeBase:
|
|
kb = self.get_or_404(kb_id, user)
|
|
self._kb_repo.update(kb, enabled=enabled)
|
|
self._session.commit()
|
|
return kb
|
|
|
|
def get_full_token(self, kb: KnowledgeBase) -> str | None:
|
|
"""解密 token 原文（供后台显示完整链接）。"""
|
|
if kb.token_encrypted:
|
|
return self._token_svc.decrypt_token(kb.token_encrypted)
|
|
return None |