334 lines
14 KiB
Python
334 lines
14 KiB
Python
import json
|
|
from typing import Any, Optional
|
|
|
|
from fastapi import APIRouter, Depends, File, Form, HTTPException, Request, UploadFile
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.api.reviews import require_current_user
|
|
from app.clients.shared_ai_platform import SharedAIPlatformClient, get_shared_ai_platform_client
|
|
from app.core.response import success_response
|
|
from app.db import get_db_session
|
|
from app.models.knowledge_base_document import KnowledgeBaseDocument
|
|
from app.repositories.knowledge_base_document_repository import KnowledgeBaseDocumentRepository
|
|
from app.services.audit_log_service import AuditLogService
|
|
|
|
router = APIRouter(prefix="/knowledge-bases", tags=["knowledge-bases"])
|
|
ROUTE_PERMISSION_SCOPE = "knowledge_base_management"
|
|
ROUTE_PERMISSION_ACTIONS = {
|
|
"list": "knowledge_bases:list",
|
|
"create": "knowledge_bases:create",
|
|
"documents": "knowledge_bases:documents:list",
|
|
"upload_document": "knowledge_bases:documents:upload",
|
|
"retry_document": "knowledge_bases:documents:retry",
|
|
}
|
|
|
|
|
|
def get_audit_log_service(db: Session = Depends(get_db_session)) -> AuditLogService:
|
|
return AuditLogService(db)
|
|
|
|
|
|
def _parse_tags_json(tags_json: Optional[str]) -> Optional[list[str]]:
|
|
if tags_json is None or not str(tags_json).strip():
|
|
return None
|
|
try:
|
|
payload = json.loads(tags_json)
|
|
except json.JSONDecodeError as exc:
|
|
raise HTTPException(status_code=400, detail="tags_json must be a valid JSON array") from exc
|
|
if not isinstance(payload, list) or not all(isinstance(item, str) for item in payload):
|
|
raise HTTPException(status_code=400, detail="tags_json must be a JSON array of strings")
|
|
normalized = [item.strip() for item in payload if item and item.strip()]
|
|
return normalized or None
|
|
|
|
|
|
def _build_metadata_override(
|
|
*,
|
|
source_type: Optional[str],
|
|
authority: Optional[str],
|
|
effective_date: Optional[str],
|
|
region: Optional[str],
|
|
tags: Optional[list[str]],
|
|
) -> dict[str, Any]:
|
|
payload: dict[str, Any] = {}
|
|
if source_type and source_type.strip():
|
|
payload["source_type"] = source_type.strip()
|
|
if authority and authority.strip():
|
|
payload["authority"] = authority.strip()
|
|
if effective_date and effective_date.strip():
|
|
payload["effective_date"] = effective_date.strip()
|
|
if region and region.strip():
|
|
payload["region"] = region.strip()
|
|
if tags is not None:
|
|
payload["tags"] = tags
|
|
return payload
|
|
|
|
|
|
def _merge_local_document_overlay(remote_item: dict[str, Any], local_item: KnowledgeBaseDocument) -> dict[str, Any]:
|
|
merged = dict(remote_item)
|
|
if local_item.source_type:
|
|
merged["source_type"] = local_item.source_type
|
|
if local_item.authority:
|
|
merged["authority"] = local_item.authority
|
|
if local_item.effective_date:
|
|
merged["effective_date"] = local_item.effective_date
|
|
if local_item.region:
|
|
merged["region"] = local_item.region
|
|
if local_item.tags_json is not None:
|
|
merged["tags"] = local_item.tags_json
|
|
merged["tags_json"] = local_item.tags_json
|
|
return merged
|
|
|
|
|
|
def _upsert_local_document_overlay(
|
|
db: Session,
|
|
*,
|
|
current_user: dict,
|
|
remote_item: dict[str, Any],
|
|
metadata_override: Optional[dict[str, Any]] = None,
|
|
) -> dict[str, Any]:
|
|
repository = KnowledgeBaseDocumentRepository(db)
|
|
tenant_id = str(current_user.get("tenant_id") or "")
|
|
knowledge_base_id = str(remote_item.get("knowledge_base_id") or "")
|
|
document_id = str(remote_item.get("id") or "")
|
|
if not tenant_id or not knowledge_base_id or not document_id:
|
|
return remote_item
|
|
|
|
existing = repository.get_by_id(document_id, tenant_id=tenant_id, knowledge_base_id=knowledge_base_id)
|
|
overlay = metadata_override or {}
|
|
|
|
source_type = overlay.get("source_type") or remote_item.get("source_type") or (existing.source_type if existing else None)
|
|
authority = overlay.get("authority") or remote_item.get("authority") or (existing.authority if existing else None)
|
|
effective_date = overlay.get("effective_date") or remote_item.get("effective_date") or (existing.effective_date if existing else None)
|
|
region = overlay.get("region") or remote_item.get("region") or (existing.region if existing else None)
|
|
tags = overlay.get("tags")
|
|
if tags is None:
|
|
tags = remote_item.get("tags")
|
|
if tags is None:
|
|
tags = remote_item.get("tags_json")
|
|
if tags is None and existing is not None:
|
|
tags = existing.tags_json
|
|
|
|
if existing is None:
|
|
existing = KnowledgeBaseDocument(
|
|
id=document_id,
|
|
knowledge_base_id=knowledge_base_id,
|
|
tenant_id=tenant_id,
|
|
file_name=str(remote_item.get("file_name") or ""),
|
|
mime_type=remote_item.get("mime_type"),
|
|
file_size=int(remote_item.get("file_size") or 0),
|
|
status=str(remote_item.get("status") or "indexed"),
|
|
source_type=source_type,
|
|
authority=authority,
|
|
effective_date=effective_date,
|
|
region=region,
|
|
tags_json=tags,
|
|
content_excerpt=remote_item.get("content_excerpt"),
|
|
error_message=remote_item.get("error_message"),
|
|
)
|
|
repository.create(existing)
|
|
else:
|
|
existing.file_name = str(remote_item.get("file_name") or existing.file_name)
|
|
existing.mime_type = remote_item.get("mime_type") or existing.mime_type
|
|
existing.file_size = int(remote_item.get("file_size") or existing.file_size or 0)
|
|
existing.status = str(remote_item.get("status") or existing.status or "indexed")
|
|
existing.source_type = source_type
|
|
existing.authority = authority
|
|
existing.effective_date = effective_date
|
|
existing.region = region
|
|
existing.tags_json = tags
|
|
existing.content_excerpt = remote_item.get("content_excerpt")
|
|
existing.error_message = remote_item.get("error_message")
|
|
repository.update(existing)
|
|
|
|
return _merge_local_document_overlay(remote_item, existing)
|
|
|
|
|
|
@router.get("")
|
|
def list_knowledge_bases(
|
|
page: int = 1,
|
|
page_size: int = 20,
|
|
shared_ai_platform_client: SharedAIPlatformClient = Depends(get_shared_ai_platform_client),
|
|
current_user: dict = Depends(require_current_user),
|
|
):
|
|
return success_response(shared_ai_platform_client.list_knowledge_bases(current_user, page, page_size))
|
|
|
|
|
|
@router.post("")
|
|
def create_knowledge_base(
|
|
name: str = Form(...),
|
|
description: Optional[str] = Form(default=None),
|
|
shared_ai_platform_client: SharedAIPlatformClient = Depends(get_shared_ai_platform_client),
|
|
audit_log_service: AuditLogService = Depends(get_audit_log_service),
|
|
current_user: dict = Depends(require_current_user),
|
|
):
|
|
result = shared_ai_platform_client.create_knowledge_base(current_user, name, description)
|
|
audit_log_service.record_event(
|
|
tenant_id=current_user.get("tenant_id"),
|
|
user_id=current_user.get("id"),
|
|
action="create_knowledge_base",
|
|
resource_type="knowledge_base",
|
|
resource_id=result["id"],
|
|
detail_json={
|
|
"name": result["name"],
|
|
"description": result["description"],
|
|
"status": result["status"],
|
|
"document_count": result["document_count"],
|
|
},
|
|
)
|
|
return success_response(result)
|
|
|
|
|
|
@router.get("/{knowledge_base_id}/documents")
|
|
def list_knowledge_base_documents(
|
|
knowledge_base_id: str,
|
|
page: int = 1,
|
|
page_size: int = 20,
|
|
shared_ai_platform_client: SharedAIPlatformClient = Depends(get_shared_ai_platform_client),
|
|
current_user: dict = Depends(require_current_user),
|
|
db: Session = Depends(get_db_session),
|
|
):
|
|
payload = shared_ai_platform_client.list_documents(current_user, knowledge_base_id, page, page_size)
|
|
items = payload.get("items", [])
|
|
if not isinstance(items, list) or not items:
|
|
return success_response(payload)
|
|
|
|
repository = KnowledgeBaseDocumentRepository(db)
|
|
document_ids = [str(item.get("id") or "") for item in items if item.get("id")]
|
|
overlays = repository.list_by_ids(str(current_user.get("tenant_id") or ""), document_ids)
|
|
overlay_map = {overlay.id: overlay for overlay in overlays}
|
|
|
|
payload["items"] = [
|
|
_merge_local_document_overlay(item, overlay_map[item["id"]]) if item.get("id") in overlay_map else item
|
|
for item in items
|
|
]
|
|
return success_response(payload)
|
|
|
|
|
|
@router.post("/{knowledge_base_id}/documents")
|
|
async def upload_knowledge_base_document(
|
|
knowledge_base_id: str,
|
|
request: Request,
|
|
file: UploadFile = File(...),
|
|
note: Optional[str] = Form(default=None),
|
|
source_type: Optional[str] = Form(default=None),
|
|
authority: Optional[str] = Form(default=None),
|
|
effective_date: Optional[str] = Form(default=None),
|
|
region: Optional[str] = Form(default=None),
|
|
tags_json: Optional[str] = Form(default=None),
|
|
shared_ai_platform_client: SharedAIPlatformClient = Depends(get_shared_ai_platform_client),
|
|
audit_log_service: AuditLogService = Depends(get_audit_log_service),
|
|
current_user: dict = Depends(require_current_user),
|
|
db: Session = Depends(get_db_session),
|
|
):
|
|
parsed_tags = _parse_tags_json(tags_json)
|
|
content = await file.read()
|
|
result = shared_ai_platform_client.upload_document(current_user, knowledge_base_id, file.filename, file.content_type, content)
|
|
result = _upsert_local_document_overlay(
|
|
db,
|
|
current_user=current_user,
|
|
remote_item=result,
|
|
metadata_override=_build_metadata_override(
|
|
source_type=source_type,
|
|
authority=authority,
|
|
effective_date=effective_date,
|
|
region=region,
|
|
tags=parsed_tags,
|
|
),
|
|
)
|
|
audit_log_service.record_event(
|
|
tenant_id=current_user.get("tenant_id"),
|
|
user_id=current_user.get("id"),
|
|
action="upload_knowledge_document",
|
|
resource_type="knowledge_base_document",
|
|
resource_id=result["id"],
|
|
detail_json={
|
|
"knowledge_base_id": knowledge_base_id,
|
|
"file_name": result["file_name"],
|
|
"mime_type": result["mime_type"],
|
|
"file_size": result["file_size"],
|
|
"status": result.get("status"),
|
|
"current_stage": result.get("current_stage"),
|
|
"parse_status": result.get("parse_status"),
|
|
"index_status": result.get("index_status"),
|
|
"error_message": result.get("error_message"),
|
|
"note": note,
|
|
"source_type": result.get("source_type"),
|
|
"authority": result.get("authority"),
|
|
"effective_date": result.get("effective_date"),
|
|
"region": result.get("region"),
|
|
"tags": result.get("tags"),
|
|
},
|
|
ip_address=request.client.host if request.client else None,
|
|
user_agent=request.headers.get("user-agent"),
|
|
)
|
|
return success_response(result, message="文档已入库" if result["status"] == "indexed" else "文档入库失败")
|
|
|
|
|
|
@router.post("/{knowledge_base_id}/documents/{document_id}/retry")
|
|
async def retry_knowledge_base_document(
|
|
knowledge_base_id: str,
|
|
document_id: str,
|
|
request: Request,
|
|
file: Optional[UploadFile] = File(default=None),
|
|
note: Optional[str] = Form(default=None),
|
|
source_type: Optional[str] = Form(default=None),
|
|
authority: Optional[str] = Form(default=None),
|
|
effective_date: Optional[str] = Form(default=None),
|
|
region: Optional[str] = Form(default=None),
|
|
tags_json: Optional[str] = Form(default=None),
|
|
shared_ai_platform_client: SharedAIPlatformClient = Depends(get_shared_ai_platform_client),
|
|
audit_log_service: AuditLogService = Depends(get_audit_log_service),
|
|
current_user: dict = Depends(require_current_user),
|
|
db: Session = Depends(get_db_session),
|
|
):
|
|
parsed_tags = _parse_tags_json(tags_json)
|
|
content = await file.read() if file is not None else None
|
|
result = shared_ai_platform_client.retry_document(
|
|
current_user=current_user,
|
|
knowledge_base_id=knowledge_base_id,
|
|
document_id=document_id,
|
|
file_name=file.filename if file is not None else None,
|
|
mime_type=file.content_type if file is not None else None,
|
|
content=content,
|
|
)
|
|
result = _upsert_local_document_overlay(
|
|
db,
|
|
current_user=current_user,
|
|
remote_item=result,
|
|
metadata_override=_build_metadata_override(
|
|
source_type=source_type,
|
|
authority=authority,
|
|
effective_date=effective_date,
|
|
region=region,
|
|
tags=parsed_tags,
|
|
),
|
|
)
|
|
audit_log_service.record_event(
|
|
tenant_id=current_user.get("tenant_id"),
|
|
user_id=current_user.get("id"),
|
|
action="retry_knowledge_document",
|
|
resource_type="knowledge_base_document",
|
|
resource_id=result["id"],
|
|
detail_json={
|
|
"knowledge_base_id": knowledge_base_id,
|
|
"original_document_id": document_id,
|
|
"file_name": result["file_name"],
|
|
"mime_type": result["mime_type"],
|
|
"file_size": result["file_size"],
|
|
"status": result.get("status"),
|
|
"current_stage": result.get("current_stage"),
|
|
"parse_status": result.get("parse_status"),
|
|
"index_status": result.get("index_status"),
|
|
"error_message": result.get("error_message"),
|
|
"replacement_file": file is not None,
|
|
"note": note,
|
|
"source_type": result.get("source_type"),
|
|
"authority": result.get("authority"),
|
|
"effective_date": result.get("effective_date"),
|
|
"region": result.get("region"),
|
|
"tags": result.get("tags"),
|
|
},
|
|
ip_address=request.client.host if request.client else None,
|
|
user_agent=request.headers.get("user-agent"),
|
|
)
|
|
return success_response(result, message="文档重试已完成" if result["status"] == "indexed" else "文档重试仍需处理")
|