diff --git a/backend/python/app/api/admin/knowledge_admin.py b/backend/python/app/api/admin/knowledge_admin.py index 68ec5d8..79580cf 100644 --- a/backend/python/app/api/admin/knowledge_admin.py +++ b/backend/python/app/api/admin/knowledge_admin.py @@ -1,1330 +1,1274 @@ -""" -知识库管理 API（管理员） -Phase 2.4: AI PaaS Knowledge Base Management + External KB Integration -""" -from fastapi import APIRouter, Depends, HTTPException, Query, Body -from sqlalchemy import or_, func, desc, asc, text -from sqlalchemy.orm import Session -from pydantic import BaseModel, Field -from typing import Optional, List, Dict, Any -from datetime import datetime -import uuid -import json -import logging -import httpx - -from ...database import get_db -from ...models import ( - KnowledgeBase, KnowledgeDocument, KnowledgeDirectory, - KnowledgeGraphNode, KnowledgeGraphEdge, KnowledgeGraphBuildLog, - User, gen_uuid -) -from ...auth import get_current_admin -from ...middleware.log_service import LogService - -router = APIRouter() -logger = logging.getLogger("aiflow") - - -# ============================================================ -# Pydantic 模型 -# ============================================================ - -class ExternalServiceConfig(BaseModel): - name: str = Field(min_length=1, max_length=200) - service_type: str = Field(..., description="fastgpt / dify / ragflow") - api_url: str = Field(min_length=1, max_length=500) - api_key: Optional[str] = None - config: Optional[Dict[str, Any]] = None - - -class ExternalServiceUpdate(BaseModel): - name: Optional[str] = None - api_url: Optional[str] = None - api_key: Optional[str] = None - config: Optional[Dict[str, Any]] = None - is_enabled: Optional[bool] = None - - -class KBCreate(BaseModel): - name: str = Field(min_length=1, max_length=200) - description: Optional[str] = None - is_official: bool = False - external_service_id: Optional[str] = None - external_config: Optional[Dict[str, Any]] = None - - -class KBUpdate(BaseModel): - name: Optional[str] = None - description: Optional[str] = None - is_official: Optional[bool] = None - status: Optional[str] = None # active / inactive / archived - external_service_id: Optional[str] = None - external_config: Optional[Dict[str, Any]] = None - - -class SyncRequest(BaseModel): - external_service_id: Optional[str] = None # override KB's linked service - sync_direction: str = "pull" # pull / push / bidirectional - options: Optional[Dict[str, Any]] = None - - -# ============================================================ -# 辅助函数 -# ============================================================ - -def kb_to_dict(kb: KnowledgeBase, db: Session = None) -> dict: - """将 KnowledgeBase 转换为字典""" - result = { - "id": kb.id, - "user_id": kb.user_id, - "name": kb.name, - "description": kb.description, - "directory_path": kb.directory_path, - "document_count": kb.document_count or 0, - "knowledge_graph_count": kb.knowledge_graph_count or 0, - "created_at": kb.created_at.isoformat() if kb.created_at else None, - "updated_at": kb.updated_at.isoformat() if kb.updated_at else None, - } - - # 新增字段（Phase 2.4） - # 通过 getattr 安全获取，因为旧记录可能没有这些字段 - result["is_official"] = getattr(kb, "is_official", False) - result["status"] = getattr(kb, "status", "active") - result["external_service_id"] = getattr(kb, "external_service_id", None) - result["external_id"] = getattr(kb, "external_id", None) - result["external_sync_status"] = getattr(kb, "external_sync_status", "none") - result["external_config"] = None - ext_config_raw = getattr(kb, "external_config", None) - if ext_config_raw: - try: - result["external_config"] = json.loads(ext_config_raw) if isinstance(ext_config_raw, str) else ext_config_raw - except Exception: - result["external_config"] = {} - - # 查询关联用户名 - if db and kb.user_id: - user = db.query(User).filter(User.id == kb.user_id).first() - result["owner_name"] = user.username if user else "未知" - - # 计算存储占用 - if db: - total_size = db.query(func.sum(KnowledgeDocument.file_size)).filter( - KnowledgeDocument.knowledge_base_id == kb.id - ).scalar() or 0 - result["total_storage_bytes"] = int(total_size) - result["total_storage_mb"] = round(int(total_size) / (1024 * 1024), 2) - - return result - - -def doc_to_dict(doc: KnowledgeDocument) -> dict: - """将 KnowledgeDocument 转换为字典""" - return { - "id": doc.id, - "knowledge_base_id": doc.knowledge_base_id, - "directory_id": doc.directory_id, - "filename": doc.filename, - "file_type": doc.file_type, - "file_size": doc.file_size or 0, - "file_path": doc.file_path, - "status": doc.status, - "chunk_count": doc.chunk_count or 0, - "error_message": doc.error_message, - "created_at": doc.created_at.isoformat() if doc.created_at else None, - "updated_at": doc.updated_at.isoformat() if doc.updated_at else None, - } - - -def format_file_size(size_bytes: int) -> str: - """格式化文件大小""" - if size_bytes < 1024: - return f"{size_bytes} B" - elif size_bytes < 1024 * 1024: - return f"{size_bytes / 1024:.1f} KB" - elif size_bytes < 1024 * 1024 * 1024: - return f"{size_bytes / (1024 * 1024):.1f} MB" - else: - return f"{size_bytes / (1024 * 1024 * 1024):.1f} GB" - - -# ============================================================ -# 统计概览 -# ============================================================ - -@router.get("/stats/overview", summary="知识库统计概览") -async def get_kb_stats( - admin_user=Depends(get_current_admin), - db: Session = Depends(get_db), -): - """获取知识库全局统计数据""" - # 基础统计 - total_kb = db.query(func.count(KnowledgeBase.id)).scalar() or 0 - total_docs = db.query(func.count(KnowledgeDocument.id)).scalar() or 0 - total_storage = db.query(func.sum(KnowledgeDocument.file_size)).scalar() or 0 - total_nodes = db.query(func.count(KnowledgeGraphNode.id)).scalar() or 0 - total_edges = db.query(func.count(KnowledgeGraphEdge.id)).scalar() or 0 - - # 官方 vs 用户知识库 - official_kb = 0 - user_kb = 0 - try: - official_kb = db.query(func.count(KnowledgeBase.id)).filter( - getattr(KnowledgeBase, "is_official", False) == True - ).scalar() or 0 - except Exception: - official_kb = 0 - user_kb = total_kb - official_kb - - # 文档状态分布 - doc_status_dist = {} - statuses = ["uploading", "processing", "completed", "failed"] - for s in statuses: - cnt = db.query(func.count(KnowledgeDocument.id)).filter( - KnowledgeDocument.status == s - ).scalar() or 0 - doc_status_dist[s] = cnt - - # 文件类型分布 - file_type_dist = {} - type_rows = db.query( - KnowledgeDocument.file_type, - func.count(KnowledgeDocument.id) - ).group_by(KnowledgeDocument.file_type).all() - for ft, cnt in type_rows: - file_type_dist[ft] = cnt - - # 最近7天创建的知识库数 - from datetime import timedelta - week_ago = datetime.now() - timedelta(days=7) - recent_kb = db.query(func.count(KnowledgeBase.id)).filter( - KnowledgeBase.created_at >= week_ago - ).scalar() or 0 - - # 活跃知识库（有文档的知识库） - active_kb = db.query(func.count(func.distinct(KnowledgeDocument.knowledge_base_id))).scalar() or 0 - - # 知识图谱统计 - kb_with_graph = db.query(func.count(func.distinct(KnowledgeGraphNode.knowledge_base_id))).scalar() or 0 - - # 外部集成统计 - external_kb = 0 - try: - external_kb = db.query(func.count(KnowledgeBase.id)).filter( - getattr(KnowledgeBase, "external_service_id", None) != None, - getattr(KnowledgeBase, "external_service_id", None) != "" - ).scalar() or 0 - except Exception: - external_kb = 0 - - return { - "code": 0, - "data": { - "total_knowledge_bases": total_kb, - "total_documents": total_docs, - "total_storage_bytes": int(total_storage), - "total_storage_display": format_file_size(int(total_storage)), - "official_kb": official_kb, - "user_kb": user_kb, - "active_kb": active_kb, - "kb_with_graph": kb_with_graph, - "external_kb": external_kb, - "recent_kb_7d": recent_kb, - "total_graph_nodes": int(total_nodes), - "total_graph_edges": int(total_edges), - "doc_status_distribution": doc_status_dist, - "file_type_distribution": file_type_dist, - } - } - - -# ============================================================ -# 知识库 CRUD -# ============================================================ - -@router.get("/", summary="获取知识库列表") -async def list_knowledge_bases( - keyword: str = Query(None, description="搜索关键词"), - status: str = Query(None, description="状态: active/inactive/archived"), - is_official: bool = Query(None, description="是否官方知识库"), - has_external: bool = Query(None, description="是否有外部集成"), - page: int = Query(1, ge=1), - page_size: int = Query(15, ge=1, le=100), - admin_user=Depends(get_current_admin), - db: Session = Depends(get_db), -): - """获取所有知识库列表（分页、筛选）""" - query = db.query(KnowledgeBase) - - if keyword: - query = query.filter( - or_( - KnowledgeBase.name.ilike(f"%{keyword}%"), - KnowledgeBase.description.ilike(f"%{keyword}%"), - ) - ) - if status: - try: - query = query.filter(getattr(KnowledgeBase, "status", None) == status) - except Exception: - pass # 字段可能不存在 - if is_official is not None: - try: - query = query.filter(getattr(KnowledgeBase, "is_official", None) == is_official) - except Exception: - pass - - total = query.count() - kbs = query.order_by(desc(KnowledgeBase.created_at)).offset( - (page - 1) * page_size - ).limit(page_size).all() - - return { - "code": 0, - "data": { - "total": total, - "page": page, - "page_size": page_size, - "items": [kb_to_dict(kb, db) for kb in kbs], - } - } - - -@router.get("/{kb_id}", summary="获取知识库详情") -async def get_knowledge_base( - kb_id: str, - admin_user=Depends(get_current_admin), - db: Session = Depends(get_db), -): - """获取知识库详细信息，包含文档和图谱数据""" - kb = db.query(KnowledgeBase).filter(KnowledgeBase.id == kb_id).first() - if not kb: - raise HTTPException(status_code=404, detail="知识库不存在") - - result = kb_to_dict(kb, db) - - # 文档统计 - doc_stats = db.query( - KnowledgeDocument.status, - func.count(KnowledgeDocument.id), - func.sum(KnowledgeDocument.file_size) - ).filter(KnowledgeDocument.knowledge_base_id == kb_id).group_by( - KnowledgeDocument.status - ).all() - - result["doc_stats"] = { - s: {"count": int(cnt), "size": int(sz or 0)} - for s, cnt, sz in doc_stats - } - - # 图谱统计 - graph_nodes = db.query(func.count(KnowledgeGraphNode.id)).filter( - KnowledgeGraphNode.knowledge_base_id == kb_id - ).scalar() or 0 - graph_edges = db.query(func.count(KnowledgeGraphEdge.id)).filter( - KnowledgeGraphEdge.knowledge_base_id == kb_id - ).scalar() or 0 - result["graph_stats"] = { - "nodes": int(graph_nodes), - "edges": int(graph_edges), - } - - # 外部服务信息 - ext_service_id = getattr(kb, "external_service_id", None) - if ext_service_id: - # 查询外部服务配置 - try: - ext_service = db.query(ExternalKBService).filter( - ExternalKBService.id == ext_service_id - ).first() - if ext_service: - result["external_service"] = { - "id": ext_service.id, - "name": ext_service.name, - "service_type": ext_service.service_type, - "api_url": ext_service.api_url, - "status": getattr(ext_service, "status", "unknown"), - } - except Exception: - result["external_service"] = {"id": ext_service_id, "note": "外部服务表尚未初始化"} - - return {"code": 0, "data": result} - - -@router.post("/", summary="创建知识库") -async def create_knowledge_base( - req: KBCreate, - admin_user=Depends(get_current_admin), - db: Session = Depends(get_db), -): - """创建新的知识库（支持官方平台知识库）""" - # 检查名称唯一性 - existing = db.query(KnowledgeBase).filter(KnowledgeBase.name == req.name).first() - if existing: - raise HTTPException(status_code=400, detail="知识库名称已存在") - - kb = KnowledgeBase( - id=str(uuid.uuid4()), - user_id=admin_user.id, - name=req.name, - description=req.description or "", - directory_path="/", - document_count=0, - knowledge_graph_count=0, - created_at=datetime.now(), - updated_at=datetime.now(), - ) - - # Phase 2.4 新增字段 - try: - setattr(kb, "is_official", req.is_official) - setattr(kb, "status", "active") - if req.external_service_id: - setattr(kb, "external_service_id", req.external_service_id) - if req.external_config: - setattr(kb, "external_config", json.dumps(req.external_config, ensure_ascii=False)) - setattr(kb, "external_sync_status", "none") - except Exception: - logger.warning("KnowledgeBase 新增字段设置失败，可能表结构尚未更新") - - db.add(kb) - db.commit() - db.refresh(kb) - - LogService.log_audit( - admin_id=admin_user.id, - admin_name=admin_user.username, - action="create_knowledge_base", - target_type="knowledge_base", - target_id=kb.id, - detail=f"创建知识库: {kb.name} (官方={req.is_official})", - ) - - return {"code": 0, "message": "知识库创建成功", "data": kb_to_dict(kb, db)} - - -@router.put("/{kb_id}", summary="更新知识库") -async def update_knowledge_base( - kb_id: str, - req: KBUpdate, - admin_user=Depends(get_current_admin), - db: Session = Depends(get_db), -): - """更新知识库配置""" - kb = db.query(KnowledgeBase).filter(KnowledgeBase.id == kb_id).first() - if not kb: - raise HTTPException(status_code=404, detail="知识库不存在") - - if req.name is not None: - dup = db.query(KnowledgeBase).filter( - KnowledgeBase.name == req.name, - KnowledgeBase.id != kb_id - ).first() - if dup: - raise HTTPException(status_code=400, detail="知识库名称已存在") - kb.name = req.name - - if req.description is not None: - kb.description = req.description - - # Phase 2.4 新增字段更新 - try: - if req.is_official is not None: - setattr(kb, "is_official", req.is_official) - if req.status is not None: - setattr(kb, "status", req.status) - if req.external_service_id is not None: - setattr(kb, "external_service_id", req.external_service_id) - if req.external_config is not None: - setattr(kb, "external_config", json.dumps(req.external_config, ensure_ascii=False)) - except Exception: - logger.warning("KnowledgeBase 新增字段更新失败") - - kb.updated_at = datetime.now() - db.commit() - db.refresh(kb) - - LogService.log_audit( - admin_id=admin_user.id, - admin_name=admin_user.username, - action="update_knowledge_base", - target_type="knowledge_base", - target_id=kb_id, - detail=f"更新知识库: {kb.name}", - ) - - return {"code": 0, "message": "知识库已更新", "data": kb_to_dict(kb, db)} - - -@router.delete("/{kb_id}", summary="删除知识库") -async def delete_knowledge_base( - kb_id: str, - admin_user=Depends(get_current_admin), - db: Session = Depends(get_db), -): - """删除知识库及其所有文档""" - kb = db.query(KnowledgeBase).filter(KnowledgeBase.id == kb_id).first() - if not kb: - raise HTTPException(status_code=404, detail="知识库不存在") - - kb_name = kb.name - - # 删除关联文档 - docs = db.query(KnowledgeDocument).filter( - KnowledgeDocument.knowledge_base_id == kb_id - ).all() - for d in docs: - db.delete(d) - - # 删除关联图谱 - nodes = db.query(KnowledgeGraphNode).filter( - KnowledgeGraphNode.knowledge_base_id == kb_id - ).all() - for n in nodes: - db.delete(n) - - edges = db.query(KnowledgeGraphEdge).filter( - KnowledgeGraphEdge.knowledge_base_id == kb_id - ).all() - for e in edges: - db.delete(e) - - build_logs = db.query(KnowledgeGraphBuildLog).filter( - KnowledgeGraphBuildLog.knowledge_base_id == kb_id - ).all() - for bl in build_logs: - db.delete(bl) - - # 删除关联目录 - dirs = db.query(KnowledgeDirectory).filter( - KnowledgeDirectory.knowledge_base_id == kb_id - ).all() - for d in dirs: - db.delete(d) - - # 删除知识库本身 - db.delete(kb) - db.commit() - - LogService.log_audit( - admin_id=admin_user.id, - admin_name=admin_user.username, - action="delete_knowledge_base", - target_type="knowledge_base", - target_id=kb_id, - detail=f"删除知识库: {kb_name} (含 {len(docs)} 文档, {len(nodes)} 节点, {len(edges)} 边)", - ) - - return {"code": 0, "message": "知识库已删除"} - - -@router.put("/{kb_id}/toggle", summary="启用/禁用知识库") -async def toggle_knowledge_base( - kb_id: str, - is_active: bool = Query(..., description="是否启用"), - admin_user=Depends(get_current_admin), - db: Session = Depends(get_db), -): - """启用或禁用知识库""" - kb = db.query(KnowledgeBase).filter(KnowledgeBase.id == kb_id).first() - if not kb: - raise HTTPException(status_code=404, detail="知识库不存在") - - new_status = "active" if is_active else "inactive" - try: - setattr(kb, "status", new_status) - except Exception: - pass - - kb.updated_at = datetime.now() - db.commit() - - LogService.log_audit( - admin_id=admin_user.id, - admin_name=admin_user.username, - action="toggle_knowledge_base", - target_type="knowledge_base", - target_id=kb_id, - detail=f"{'启用' if is_active else '禁用'}知识库: {kb.name}", - ) - - return { - "code": 0, - "message": f"知识库已{'启用' if is_active else '禁用'}", - "data": kb_to_dict(kb, db), - } - - -# ============================================================ -# 文档管理 -# ============================================================ - -@router.get("/{kb_id}/documents", summary="获取知识库文档列表") -async def get_kb_documents( - kb_id: str, - status: str = Query(None, description="文档状态筛选"), - file_type: str = Query(None, description="文件类型筛选"), - keyword: str = Query(None, description="文件名搜索"), - page: int = Query(1, ge=1), - page_size: int = Query(20, ge=1, le=100), - admin_user=Depends(get_current_admin), - db: Session = Depends(get_db), -): - """获取知识库中的文档列表""" - kb = db.query(KnowledgeBase).filter(KnowledgeBase.id == kb_id).first() - if not kb: - raise HTTPException(status_code=404, detail="知识库不存在") - - query = db.query(KnowledgeDocument).filter( - KnowledgeDocument.knowledge_base_id == kb_id - ) - - if status: - query = query.filter(KnowledgeDocument.status == status) - if file_type: - query = query.filter(KnowledgeDocument.file_type == file_type) - if keyword: - query = query.filter(KnowledgeDocument.filename.ilike(f"%{keyword}%")) - - total = query.count() - docs = query.order_by(desc(KnowledgeDocument.created_at)).offset( - (page - 1) * page_size - ).limit(page_size).all() - - return { - "code": 0, - "data": { - "total": total, - "page": page, - "page_size": page_size, - "items": [doc_to_dict(d) for d in docs], - } - } - - -@router.delete("/{kb_id}/documents/{doc_id}", summary="删除知识库文档") -async def delete_kb_document( - kb_id: str, - doc_id: str, - admin_user=Depends(get_current_admin), - db: Session = Depends(get_db), -): - """删除知识库中的单个文档""" - doc = db.query(KnowledgeDocument).filter( - KnowledgeDocument.id == doc_id, - KnowledgeDocument.knowledge_base_id == kb_id, - ).first() - if not doc: - raise HTTPException(status_code=404, detail="文档不存在") - - filename = doc.filename - db.delete(doc) - - # 更新知识库文档计数 - kb = db.query(KnowledgeBase).filter(KnowledgeBase.id == kb_id).first() - if kb: - kb.document_count = db.query(func.count(KnowledgeDocument.id)).filter( - KnowledgeDocument.knowledge_base_id == kb_id - ).scalar() or 0 - kb.updated_at = datetime.now() - - db.commit() - - LogService.log_audit( - admin_id=admin_user.id, - admin_name=admin_user.username, - action="delete_knowledge_document", - target_type="knowledge_document", - target_id=doc_id, - detail=f"删除文档: {filename} (知识库: {kb.name})", - ) - - return {"code": 0, "message": "文档已删除"} - - -# ============================================================ -# 知识图谱管理 -# ============================================================ - -@router.get("/{kb_id}/graph/stats", summary="获取知识图谱统计") -async def get_kb_graph_stats( - kb_id: str, - admin_user=Depends(get_current_admin), - db: Session = Depends(get_db), -): - """获取知识库的知识图谱统计信息""" - kb = db.query(KnowledgeBase).filter(KnowledgeBase.id == kb_id).first() - if not kb: - raise HTTPException(status_code=404, detail="知识库不存在") - - nodes = db.query(func.count(KnowledgeGraphNode.id)).filter( - KnowledgeGraphNode.knowledge_base_id == kb_id - ).scalar() or 0 - - edges = db.query(func.count(KnowledgeGraphEdge.id)).filter( - KnowledgeGraphEdge.knowledge_base_id == kb_id - ).scalar() or 0 - - # 最近构建日志 - recent_logs = db.query(KnowledgeGraphBuildLog).filter( - KnowledgeGraphBuildLog.knowledge_base_id == kb_id - ).order_by(desc(KnowledgeGraphBuildLog.created_at)).limit(5).all() - - # 节点类型分布 - node_types = db.query( - KnowledgeGraphNode.label, - func.count(KnowledgeGraphNode.id) - ).filter( - KnowledgeGraphNode.knowledge_base_id == kb_id - ).group_by(KnowledgeGraphNode.label).limit(20).all() - - return { - "code": 0, - "data": { - "total_nodes": int(nodes), - "total_edges": int(edges), - "node_type_distribution": [ - {"label": lbl, "count": int(cnt)} for lbl, cnt in node_types - ], - "recent_build_logs": [ - { - "id": bl.id, - "status": getattr(bl, "status", "unknown"), - "created_at": bl.created_at.isoformat() if bl.created_at else None, - } - for bl in recent_logs - ], - } - } - - -# ============================================================ -# 外部知识库服务管理 -# ============================================================ - -# ExternalKBService 模型定义（如果表不存在会在初始化时创建） -# 这里用原生 SQL 操作，确保即使 ORM 模型未注册也能工作 - -def _get_external_services_table(db: Session): - """安全地获取外部知识库服务列表（兼容表不存在的情况）""" - try: - # 尝试使用 ORM 模型查询 - from ...models import ExternalKBService - return db.query(ExternalKBService).all() - except (AttributeError, ImportError): - # ORM 模型不存在，用原生 SQL - try: - result = db.execute( - text("SELECT id, name, service_type, api_url, api_key, config, is_enabled, status, created_at, updated_at FROM external_kb_services ORDER BY created_at DESC") - ) - rows = result.fetchall() - return [ - { - "id": r[0], - "name": r[1], - "service_type": r[2], - "api_url": r[3], - "api_key": r[4], - "config": json.loads(r[5]) if r[5] else {}, - "is_enabled": bool(r[6]), - "status": r[7], - "created_at": r[8].isoformat() if r[8] else None, - "updated_at": r[9].isoformat() if r[9] else None, - } - for r in rows - ] - except Exception as e: - logger.warning(f"查询外部知识库服务表失败: {e}") - return [] - - -def _check_service_connection(service_type: str, api_url: str, api_key: str) -> dict: - """测试外部知识库服务连接状态""" - try: - headers = {} - if api_key: - headers["Authorization"] = f"Bearer {api_key}" - - if service_type == "fastgpt": - # FastGPT 健康检查 - check_url = f"{api_url.rstrip('/')}/api/v1/status" - resp = httpx.get(check_url, headers=headers, timeout=10) - if resp.status_code == 200: - return {"connected": True, "status": "connected", "detail": "FastGPT 连接成功"} - else: - return {"connected": False, "status": "error", "detail": f"FastGPT 返回状态码 {resp.status_code}"} - - elif service_type == "dify": - # Dify 健康检查 - check_url = f"{api_url.rstrip('/')}/v1/datasets" - resp = httpx.get(check_url, headers=headers, timeout=10) - if resp.status_code in [200, 401]: # 401说明服务存在但需要认证 - return {"connected": True, "status": "connected", "detail": "Dify 连接成功"} - else: - return {"connected": False, "status": "error", "detail": f"Dify 返回状态码 {resp.status_code}"} - - elif service_type == "ragflow": - # RAGFlow 健康检查 - check_url = f"{api_url.rstrip('/')}/api/v1/datasets" - resp = httpx.get(check_url, headers=headers, timeout=10) - if resp.status_code in [200, 401]: - return {"connected": True, "status": "connected", "detail": "RAGFlow 连接成功"} - else: - return {"connected": False, "status": "error", "detail": f"RAGFlow 返回状态码 {resp.status_code}"} - - else: - # 通用检查 - resp = httpx.get(api_url, headers=headers, timeout=10) - return {"connected": resp.status_code < 500, "status": "connected" if resp.status_code < 500 else "error"} - - except httpx.TimeoutException: - return {"connected": False, "status": "timeout", "detail": "连接超时"} - except Exception as e: - return {"connected": False, "status": "error", "detail": str(e)} - - -@router.get("/external/services", summary="获取外部知识库服务列表") -async def list_external_services( - service_type: str = Query(None, description="服务类型: fastgpt/dify/ragflow"), - admin_user=Depends(get_current_admin), - db: Session = Depends(get_db), -): - """获取所有外部知识库服务配置""" - services = _get_external_services_table(db) - - if service_type: - services = [s for s in services if s.get("service_type") == service_type] - - return { - "code": 0, - "data": { - "total": len(services), - "items": services, - "supported_types": [ - {"type": "fastgpt", "name": "FastGPT", "description": "开源 AI 知识库平台，支持多种向量模型"}, - {"type": "dify", "name": "Dify", "description": "LLM 应用开发平台，内置知识库管理"}, - {"type": "ragflow", "name": "RAGFlow", "description": "基于深度文档理解的开源 RAG 引擎"}, - ] - } - } - - -@router.post("/external/services", summary="创建外部知识库服务配置") -async def create_external_service( - req: ExternalServiceConfig, - admin_user=Depends(get_current_admin), - db: Session = Depends(get_db), -): - """创建新的外部知识库服务连接配置""" - # 验证服务类型 - valid_types = ["fastgpt", "dify", "ragflow"] - if req.service_type not in valid_types: - raise HTTPException(status_code=400, detail=f"不支持的服务类型: {req.service_type}") - - # 检查连接 - conn_result = _check_service_connection(req.service_type, req.api_url, req.api_key or "") - - service_id = str(uuid.uuid4()) - config_json = json.dumps(req.config or {}, ensure_ascii=False) - conn_status = conn_result.get("status", "disconnected") - - # 使用原生 SQL 插入（兼容 ORM 模型可能未注册） - try: - db.execute( - text(f"INSERT INTO external_kb_services (id, name, service_type, api_url, api_key, config, is_enabled, status, created_at, updated_at) " - f"VALUES (:id, :name, :stype, :url, :key, :cfg, 1, :status, NOW(), NOW())"), - { - "id": service_id, - "name": req.name, - "stype": req.service_type, - "url": req.api_url, - "key": req.api_key or "", - "cfg": config_json, - "status": conn_status, - } - ) - db.commit() - except Exception as e: - # 表可能不存在，尝试创建 - logger.warning(f"插入外部服务失败: {e}, 尝试创建表") - db.execute(text(""" - CREATE TABLE IF NOT EXISTS external_kb_services ( - id VARCHAR(36) PRIMARY KEY, - name VARCHAR(200) NOT NULL, - service_type VARCHAR(20) NOT NULL, - api_url VARCHAR(500) NOT NULL, - api_key VARCHAR(255) DEFAULT '', - config JSON DEFAULT NULL, - is_enabled BOOLEAN DEFAULT TRUE, - status VARCHAR(20) DEFAULT 'disconnected', - created_at DATETIME DEFAULT CURRENT_TIMESTAMP, - updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP - ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 - """)) - db.commit() - # 再次尝试插入 - db.execute( - text(f"INSERT INTO external_kb_services (id, name, service_type, api_url, api_key, config, is_enabled, status, created_at, updated_at) " - f"VALUES (:id, :name, :stype, :url, :key, :cfg, 1, :status, NOW(), NOW())"), - { - "id": service_id, - "name": req.name, - "stype": req.service_type, - "url": req.api_url, - "key": req.api_key or "", - "cfg": config_json, - "status": conn_status, - } - ) - db.commit() - - LogService.log_audit( - admin_id=admin_user.id, - admin_name=admin_user.username, - action="create_external_kb_service", - target_type="external_kb_service", - target_id=service_id, - detail=f"创建外部知识库服务: {req.name} ({req.service_type})", - ) - - return { - "code": 0, - "message": "外部知识库服务配置已创建", - "data": { - "id": service_id, - "name": req.name, - "service_type": req.service_type, - "api_url": req.api_url, - "connection": conn_result, - } - } - - -@router.put("/external/services/{service_id}", summary="更新外部知识库服务配置") -async def update_external_service( - service_id: str, - req: ExternalServiceUpdate, - admin_user=Depends(get_current_admin), - db: Session = Depends(get_db), -): - """更新外部知识库服务配置""" - # 查找现有服务 - try: - result = db.execute( - text("SELECT id, name, service_type, api_url, api_key FROM external_kb_services WHERE id = :sid"), - {"sid": service_id} - ) - row = result.fetchone() - except Exception: - raise HTTPException(status_code=404, detail="外部知识库服务不存在或表未创建") - - if not row: - raise HTTPException(status_code=404, detail="外部知识库服务不存在") - - # 构建更新字段 - updates = {} - if req.name is not None: - updates["name"] = req.name - if req.api_url is not None: - updates["api_url"] = req.api_url - if req.api_key is not None: - updates["api_key"] = req.api_key - if req.config is not None: - updates["config"] = json.dumps(req.config, ensure_ascii=False) - if req.is_enabled is not None: - updates["is_enabled"] = 1 if req.is_enabled else 0 - - # 如果 URL 或 API Key 变了，重新测试连接 - new_url = req.api_url or row[3] - new_key = req.api_key or row[4] - if req.api_url or req.api_key: - service_type = row[2] - conn_result = _check_service_connection(service_type, new_url, new_key) - updates["status"] = conn_result.get("status", "disconnected") - - updates["updated_at"] = "NOW()" - - # 执行更新 - set_clause = ", ".join([f"{k} = :{k}" for k in updates.keys() if k != "updated_at"]) - set_clause += ", updated_at = NOW()" - params = {k: v for k, v in updates.items() if k != "updated_at"} - params["sid"] = service_id - - db.execute( - text(f"UPDATE external_kb_services SET {set_clause} WHERE id = :sid"), - params - ) - db.commit() - - LogService.log_audit( - admin_id=admin_user.id, - admin_name=admin_user.username, - action="update_external_kb_service", - target_type="external_kb_service", - target_id=service_id, - detail=f"更新外部知识库服务: {row[1]}", - ) - - return {"code": 0, "message": "外部知识库服务配置已更新"} - - -@router.delete("/external/services/{service_id}", summary="删除外部知识库服务配置") -async def delete_external_service( - service_id: str, - admin_user=Depends(get_current_admin), - db: Session = Depends(get_db), -): - """删除外部知识库服务配置""" - try: - result = db.execute( - text("SELECT name FROM external_kb_services WHERE id = :sid"), - {"sid": service_id} - ) - row = result.fetchone() - except Exception: - raise HTTPException(status_code=404, detail="外部知识库服务不存在") - - if not row: - raise HTTPException(status_code=404, detail="外部知识库服务不存在") - - name = row[0] - db.execute(text("DELETE FROM external_kb_services WHERE id = :sid"), {"sid": service_id}) - db.commit() - - LogService.log_audit( - admin_id=admin_user.id, - admin_name=admin_user.username, - action="delete_external_kb_service", - target_type="external_kb_service", - target_id=service_id, - detail=f"删除外部知识库服务: {name}", - ) - - return {"code": 0, "message": "外部知识库服务配置已删除"} - - -@router.get("/external/services/{service_id}/status", summary="检查外部知识库服务连接状态") -async def check_external_service_status( - service_id: str, - admin_user=Depends(get_current_admin), - db: Session = Depends(get_db), -): - """实时检查外部知识库服务的连接状态""" - try: - result = db.execute( - text("SELECT service_type, api_url, api_key FROM external_kb_services WHERE id = :sid"), - {"sid": service_id} - ) - row = result.fetchone() - except Exception: - raise HTTPException(status_code=404, detail="外部知识库服务不存在") - - if not row: - raise HTTPException(status_code=404, detail="外部知识库服务不存在") - - conn_result = _check_service_connection(row[0], row[1], row[2]) - - # 更新状态 - try: - db.execute( - text("UPDATE external_kb_services SET status = :status, updated_at = NOW() WHERE id = :sid"), - {"status": conn_result.get("status", "disconnected"), "sid": service_id} - ) - db.commit() - except Exception: - pass - - return {"code": 0, "data": conn_result} - - -# ============================================================ -# 知识库同步（与外部服务） -# ============================================================ - -@router.post("/external/sync/{kb_id}", summary="同步知识库与外部服务") -async def sync_knowledge_base( - kb_id: str, - req: SyncRequest, - admin_user=Depends(get_current_admin), - db: Session = Depends(get_db), -): - """将知识库与外部知识库服务进行数据同步""" - kb = db.query(KnowledgeBase).filter(KnowledgeBase.id == kb_id).first() - if not kb: - raise HTTPException(status_code=404, detail="知识库不存在") - - # 确定同步目标服务 - ext_service_id = req.external_service_id or getattr(kb, "external_service_id", None) - if not ext_service_id: - raise HTTPException(status_code=400, detail="未指定外部知识库服务，请先关联或传入 external_service_id") - - # 查询外部服务配置 - try: - result = db.execute( - text("SELECT name, service_type, api_url, api_key, config FROM external_kb_services WHERE id = :sid"), - {"sid": ext_service_id} - ) - row = result.fetchone() - except Exception: - raise HTTPException(status_code=404, detail="外部知识库服务不存在") - - if not row: - raise HTTPException(status_code=404, detail="外部知识库服务不存在") - - service_name, service_type, api_url, api_key, config_raw = row - service_config = json.loads(config_raw) if config_raw else {} - - # 执行同步 - sync_result = {"synced": False, "detail": "", "items_synced": 0} - - try: - headers = {} - if api_key: - headers["Authorization"] = f"Bearer {api_key}" - - if service_type == "fastgpt" and req.sync_direction == "push": - # 向 FastGPT 推送知识库数据 - # 1. 在 FastGPT 创建/获取对应的 dataset - ext_kb_id = getattr(kb, "external_id", None) - - if not ext_kb_id: - # 创建新 dataset - create_resp = httpx.post( - f"{api_url.rstrip('/')}/api/v1/datasets", - headers=headers, - json={"name": kb.name, "description": kb.description or ""}, - timeout=30, - ) - if create_resp.status_code == 200: - ext_data = create_resp.json() - ext_kb_id = ext_data.get("data", {}).get("id") or ext_data.get("id") - # 更新 KB 的 external_id - try: - setattr(kb, "external_id", ext_kb_id) - except Exception: - db.execute( - text("UPDATE knowledge_bases SET external_id = :eid WHERE id = :kid"), - {"eid": ext_kb_id, "kid": kb_id} - ) - else: - sync_result["detail"] = f"FastGPT 创建 dataset 失败: {create_resp.status_code}" - sync_result["synced"] = False - - if ext_kb_id: - # 推送文档内容到 FastGPT - docs = db.query(KnowledgeDocument).filter( - KnowledgeDocument.knowledge_base_id == kb_id, - KnowledgeDocument.status == "completed" - ).all() - - synced_count = 0 - for doc in docs: - content = doc.content or "" - if not content: - continue - - # 分段推送 - push_resp = httpx.post( - f"{api_url.rstrip('/')}/api/v1/datasets/{ext_kb_id}/documents", - headers=headers, - json={ - "name": doc.filename, - "content": content[:50000], # 限制单次推送大小 - }, - timeout=30, - ) - if push_resp.status_code in [200, 201]: - synced_count += 1 - - sync_result["synced"] = True - sync_result["detail"] = f"已推送 {synced_count}/{len(docs)} 文档到 FastGPT" - sync_result["items_synced"] = synced_count - - elif service_type == "dify" and req.sync_direction == "push": - # 向 Dify 推送知识库数据 - ext_kb_id = getattr(kb, "external_id", None) - - if not ext_kb_id: - create_resp = httpx.post( - f"{api_url.rstrip('/')}/v1/datasets", - headers=headers, - json={"name": kb.name}, - timeout=30, - ) - if create_resp.status_code in [200, 201]: - ext_data = create_resp.json() - ext_kb_id = ext_data.get("id") - try: - setattr(kb, "external_id", ext_kb_id) - except Exception: - db.execute( - text("UPDATE knowledge_bases SET external_id = :eid WHERE id = :kid"), - {"eid": ext_kb_id, "kid": kb_id} - ) - - if ext_kb_id: - docs = db.query(KnowledgeDocument).filter( - KnowledgeDocument.knowledge_base_id == kb_id, - KnowledgeDocument.status == "completed" - ).all() - - synced_count = 0 - for doc in docs: - content = doc.content or "" - if not content: - continue - - # Dify 文档创建 - doc_resp = httpx.post( - f"{api_url.rstrip('/')}/v1/datasets/{ext_kb_id}/document/create_by_text", - headers=headers, - json={ - "name": doc.filename, - "text": content[:50000], - "indexing_technique": "high_quality", - }, - timeout=30, - ) - if doc_resp.status_code in [200, 201]: - synced_count += 1 - - sync_result["synced"] = True - sync_result["detail"] = f"已推送 {synced_count}/{len(docs)} 文档到 Dify" - sync_result["items_synced"] = synced_count - - elif service_type == "ragflow" and req.sync_direction == "push": - # 向 RAGFlow 推送知识库数据 - ext_kb_id = getattr(kb, "external_id", None) - - if not ext_kb_id: - create_resp = httpx.post( - f"{api_url.rstrip('/')}/api/v1/datasets", - headers=headers, - json={"name": kb.name}, - timeout=30, - ) - if create_resp.status_code in [200, 201]: - ext_data = create_resp.json() - ext_kb_id = ext_data.get("id") - try: - setattr(kb, "external_id", ext_kb_id) - except Exception: - db.execute( - text("UPDATE knowledge_bases SET external_id = :eid WHERE id = :kid"), - {"eid": ext_kb_id, "kid": kb_id} - ) - - if ext_kb_id: - docs = db.query(KnowledgeDocument).filter( - KnowledgeDocument.knowledge_base_id == kb_id, - KnowledgeDocument.status == "completed" - ).all() - - synced_count = 0 - for doc in docs: - content = doc.content or "" - if not content: - continue - - doc_resp = httpx.post( - f"{api_url.rstrip('/')}/api/v1/datasets/{ext_kb_id}/documents", - headers=headers, - json={"name": doc.filename, "content": content[:50000]}, - timeout=30, - ) - if doc_resp.status_code in [200, 201]: - synced_count += 1 - - sync_result["synced"] = True - sync_result["detail"] = f"已推送 {synced_count}/{len(docs)} 文档到 RAGFlow" - sync_result["items_synced"] = synced_count - - elif req.sync_direction == "pull": - # 从外部服务拉取知识库数据 - sync_result["detail"] = f"从 {service_type} 拉取数据（暂未实现完整拉取逻辑）" - sync_result["synced"] = True - sync_result["items_synced"] = 0 - - elif req.sync_direction == "bidirectional": - sync_result["detail"] = f"双向同步（暂未实现完整双向同步逻辑）" - sync_result["synced"] = True - sync_result["items_synced"] = 0 - - # 更新同步状态 - try: - setattr(kb, "external_sync_status", "synced" if sync_result["synced"] else "error") - except Exception: - db.execute( - text("UPDATE knowledge_bases SET external_sync_status = :es WHERE id = :kid"), - {"es": "synced" if sync_result["synced"] else "error", "kid": kb_id} - ) - - kb.updated_at = datetime.now() - db.commit() - - except Exception as e: - logger.error(f"知识库同步失败: {e}") - sync_result["detail"] = f"同步异常: {str(e)}" - sync_result["synced"] = False - - try: - setattr(kb, "external_sync_status", "error") - except Exception: - db.execute( - text("UPDATE knowledge_bases SET external_sync_status = 'error' WHERE id = :kid"), - {"kid": kb_id} - ) - db.commit() - - LogService.log_audit( - admin_id=admin_user.id, - admin_name=admin_user.username, - action="sync_knowledge_base", - target_type="knowledge_base", - target_id=kb_id, - detail=f"同步知识库 '{kb.name}' → {service_name} ({service_type}), 方向={req.sync_direction}, 结果={sync_result['detail']}", - ) - - return {"code": 0, "data": sync_result} - - -# ============================================================ -# 批量操作 -# ============================================================ - -@router.post("/batch/delete", summary="批量删除知识库") -async def batch_delete_kbs( - kb_ids: List[str] = Body(..., description="知识库ID列表"), - admin_user=Depends(get_current_admin), - db: Session = Depends(get_db), -): - """批量删除知识库""" - deleted = 0 - failed = 0 - for kb_id in kb_ids: - kb = db.query(KnowledgeBase).filter(KnowledgeBase.id == kb_id).first() - if not kb: - failed += 1 - continue - - # 删除关联数据 - db.query(KnowledgeDocument).filter(KnowledgeDocument.knowledge_base_id == kb_id).delete() - db.query(KnowledgeGraphNode).filter(KnowledgeGraphNode.knowledge_base_id == kb_id).delete() - db.query(KnowledgeGraphEdge).filter(KnowledgeGraphEdge.knowledge_base_id == kb_id).delete() - db.query(KnowledgeGraphBuildLog).filter(KnowledgeGraphBuildLog.knowledge_base_id == kb_id).delete() - db.query(KnowledgeDirectory).filter(KnowledgeDirectory.knowledge_base_id == kb_id).delete() - db.delete(kb) - deleted += 1 - - db.commit() - - LogService.log_audit( - admin_id=admin_user.id, - admin_name=admin_user.username, - action="batch_delete_knowledge_bases", - target_type="knowledge_base", - target_id=",".join(kb_ids[:10]), - detail=f"批量删除知识库: 成功 {deleted}, 失败 {failed}", - ) - - return {"code": 0, "data": {"deleted": deleted, "failed": failed}} +""" +知识库管理 API（管理员） +Phase 2.4: AI PaaS Knowledge Base Management + External KB Integration +""" +from fastapi import APIRouter, Depends, HTTPException, Query, Body +from sqlalchemy import or_, func, desc, asc, text +from sqlalchemy.orm import Session +from pydantic import BaseModel, Field +from typing import Optional, List, Dict, Any +from datetime import datetime +import uuid +import json +import logging +import httpx + +from ...database import get_db +from ...models import ( + KnowledgeBase, KnowledgeDocument, KnowledgeDirectory, + KnowledgeGraphNode, KnowledgeGraphEdge, KnowledgeGraphBuildLog, + User, gen_uuid +) +from ...auth import get_current_admin +from ...middleware.log_service import LogService + +router = APIRouter() +logger = logging.getLogger("aiflow") + + +# ============================================================ +# Pydantic 模型 +# ============================================================ + +class ExternalServiceConfig(BaseModel): + name: str = Field(min_length=1, max_length=200) + service_type: str = Field(..., description="dify") + api_url: str = Field(min_length=1, max_length=500) + api_key: Optional[str] = None + config: Optional[Dict[str, Any]] = None + + +class ExternalServiceUpdate(BaseModel): + name: Optional[str] = None + api_url: Optional[str] = None + api_key: Optional[str] = None + config: Optional[Dict[str, Any]] = None + is_enabled: Optional[bool] = None + + +class KBCreate(BaseModel): + name: str = Field(min_length=1, max_length=200) + description: Optional[str] = None + is_official: bool = False + external_service_id: Optional[str] = None + external_config: Optional[Dict[str, Any]] = None + + +class KBUpdate(BaseModel): + name: Optional[str] = None + description: Optional[str] = None + is_official: Optional[bool] = None + status: Optional[str] = None # active / inactive / archived + external_service_id: Optional[str] = None + external_config: Optional[Dict[str, Any]] = None + + +class SyncRequest(BaseModel): + external_service_id: Optional[str] = None # override KB's linked service + sync_direction: str = "pull" # pull / push / bidirectional + options: Optional[Dict[str, Any]] = None + + +# ============================================================ +# 辅助函数 +# ============================================================ + +def kb_to_dict(kb: KnowledgeBase, db: Session = None) -> dict: + """将 KnowledgeBase 转换为字典""" + result = { + "id": kb.id, + "user_id": kb.user_id, + "name": kb.name, + "description": kb.description, + "directory_path": kb.directory_path, + "document_count": kb.document_count or 0, + "knowledge_graph_count": kb.knowledge_graph_count or 0, + "created_at": kb.created_at.isoformat() if kb.created_at else None, + "updated_at": kb.updated_at.isoformat() if kb.updated_at else None, + } + + # 新增字段（Phase 2.4） + # 通过 getattr 安全获取，因为旧记录可能没有这些字段 + result["is_official"] = getattr(kb, "is_official", False) + result["status"] = getattr(kb, "status", "active") + result["external_service_id"] = getattr(kb, "external_service_id", None) + result["external_id"] = getattr(kb, "external_id", None) + result["external_sync_status"] = getattr(kb, "external_sync_status", "none") + result["external_config"] = None + ext_config_raw = getattr(kb, "external_config", None) + if ext_config_raw: + try: + result["external_config"] = json.loads(ext_config_raw) if isinstance(ext_config_raw, str) else ext_config_raw + except Exception: + result["external_config"] = {} + + # 查询关联用户名 + if db and kb.user_id: + user = db.query(User).filter(User.id == kb.user_id).first() + result["owner_name"] = user.username if user else "未知" + + # 计算存储占用 + if db: + total_size = db.query(func.sum(KnowledgeDocument.file_size)).filter( + KnowledgeDocument.knowledge_base_id == kb.id + ).scalar() or 0 + result["total_storage_bytes"] = int(total_size) + result["total_storage_mb"] = round(int(total_size) / (1024 * 1024), 2) + + return result + + +def doc_to_dict(doc: KnowledgeDocument) -> dict: + """将 KnowledgeDocument 转换为字典""" + return { + "id": doc.id, + "knowledge_base_id": doc.knowledge_base_id, + "directory_id": doc.directory_id, + "filename": doc.filename, + "file_type": doc.file_type, + "file_size": doc.file_size or 0, + "file_path": doc.file_path, + "status": doc.status, + "chunk_count": doc.chunk_count or 0, + "error_message": doc.error_message, + "created_at": doc.created_at.isoformat() if doc.created_at else None, + "updated_at": doc.updated_at.isoformat() if doc.updated_at else None, + } + + +def format_file_size(size_bytes: int) -> str: + """格式化文件大小""" + if size_bytes < 1024: + return f"{size_bytes} B" + elif size_bytes < 1024 * 1024: + return f"{size_bytes / 1024:.1f} KB" + elif size_bytes < 1024 * 1024 * 1024: + return f"{size_bytes / (1024 * 1024):.1f} MB" + else: + return f"{size_bytes / (1024 * 1024 * 1024):.1f} GB" + + +# ============================================================ +# 统计概览 +# ============================================================ + +@router.get("/stats/overview", summary="知识库统计概览") +async def get_kb_stats( + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """获取知识库全局统计数据""" + # 基础统计 + total_kb = db.query(func.count(KnowledgeBase.id)).scalar() or 0 + total_docs = db.query(func.count(KnowledgeDocument.id)).scalar() or 0 + total_storage = db.query(func.sum(KnowledgeDocument.file_size)).scalar() or 0 + total_nodes = db.query(func.count(KnowledgeGraphNode.id)).scalar() or 0 + total_edges = db.query(func.count(KnowledgeGraphEdge.id)).scalar() or 0 + + # 官方 vs 用户知识库 + official_kb = 0 + user_kb = 0 + try: + official_kb = db.query(func.count(KnowledgeBase.id)).filter( + getattr(KnowledgeBase, "is_official", False) == True + ).scalar() or 0 + except Exception: + official_kb = 0 + user_kb = total_kb - official_kb + + # 文档状态分布 + doc_status_dist = {} + statuses = ["uploading", "processing", "completed", "failed"] + for s in statuses: + cnt = db.query(func.count(KnowledgeDocument.id)).filter( + KnowledgeDocument.status == s + ).scalar() or 0 + doc_status_dist[s] = cnt + + # 文件类型分布 + file_type_dist = {} + type_rows = db.query( + KnowledgeDocument.file_type, + func.count(KnowledgeDocument.id) + ).group_by(KnowledgeDocument.file_type).all() + for ft, cnt in type_rows: + file_type_dist[ft] = cnt + + # 最近7天创建的知识库数 + from datetime import timedelta + week_ago = datetime.now() - timedelta(days=7) + recent_kb = db.query(func.count(KnowledgeBase.id)).filter( + KnowledgeBase.created_at >= week_ago + ).scalar() or 0 + + # 活跃知识库（有文档的知识库） + active_kb = db.query(func.count(func.distinct(KnowledgeDocument.knowledge_base_id))).scalar() or 0 + + # 知识图谱统计 + kb_with_graph = db.query(func.count(func.distinct(KnowledgeGraphNode.knowledge_base_id))).scalar() or 0 + + # 外部集成统计 + external_kb = 0 + try: + external_kb = db.query(func.count(KnowledgeBase.id)).filter( + getattr(KnowledgeBase, "external_service_id", None) != None, + getattr(KnowledgeBase, "external_service_id", None) != "" + ).scalar() or 0 + except Exception: + external_kb = 0 + + return { + "code": 0, + "data": { + "total_knowledge_bases": total_kb, + "total_documents": total_docs, + "total_storage_bytes": int(total_storage), + "total_storage_display": format_file_size(int(total_storage)), + "official_kb": official_kb, + "user_kb": user_kb, + "active_kb": active_kb, + "kb_with_graph": kb_with_graph, + "external_kb": external_kb, + "recent_kb_7d": recent_kb, + "total_graph_nodes": int(total_nodes), + "total_graph_edges": int(total_edges), + "doc_status_distribution": doc_status_dist, + "file_type_distribution": file_type_dist, + } + } + + +# ============================================================ +# 知识库 CRUD +# ============================================================ + +@router.get("/", summary="获取知识库列表") +async def list_knowledge_bases( + keyword: str = Query(None, description="搜索关键词"), + status: str = Query(None, description="状态: active/inactive/archived"), + is_official: bool = Query(None, description="是否官方知识库"), + has_external: bool = Query(None, description="是否有外部集成"), + page: int = Query(1, ge=1), + page_size: int = Query(15, ge=1, le=100), + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """获取所有知识库列表（分页、筛选）""" + query = db.query(KnowledgeBase) + + if keyword: + query = query.filter( + or_( + KnowledgeBase.name.ilike(f"%{keyword}%"), + KnowledgeBase.description.ilike(f"%{keyword}%"), + ) + ) + if status: + try: + query = query.filter(getattr(KnowledgeBase, "status", None) == status) + except Exception: + pass # 字段可能不存在 + if is_official is not None: + try: + query = query.filter(getattr(KnowledgeBase, "is_official", None) == is_official) + except Exception: + pass + + total = query.count() + kbs = query.order_by(desc(KnowledgeBase.created_at)).offset( + (page - 1) * page_size + ).limit(page_size).all() + + return { + "code": 0, + "data": { + "total": total, + "page": page, + "page_size": page_size, + "items": [kb_to_dict(kb, db) for kb in kbs], + } + } + + +@router.get("/{kb_id}", summary="获取知识库详情") +async def get_knowledge_base( + kb_id: str, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """获取知识库详细信息，包含文档和图谱数据""" + kb = db.query(KnowledgeBase).filter(KnowledgeBase.id == kb_id).first() + if not kb: + raise HTTPException(status_code=404, detail="知识库不存在") + + result = kb_to_dict(kb, db) + + # 文档统计 + doc_stats = db.query( + KnowledgeDocument.status, + func.count(KnowledgeDocument.id), + func.sum(KnowledgeDocument.file_size) + ).filter(KnowledgeDocument.knowledge_base_id == kb_id).group_by( + KnowledgeDocument.status + ).all() + + result["doc_stats"] = { + s: {"count": int(cnt), "size": int(sz or 0)} + for s, cnt, sz in doc_stats + } + + # 图谱统计 + graph_nodes = db.query(func.count(KnowledgeGraphNode.id)).filter( + KnowledgeGraphNode.knowledge_base_id == kb_id + ).scalar() or 0 + graph_edges = db.query(func.count(KnowledgeGraphEdge.id)).filter( + KnowledgeGraphEdge.knowledge_base_id == kb_id + ).scalar() or 0 + result["graph_stats"] = { + "nodes": int(graph_nodes), + "edges": int(graph_edges), + } + + # 外部服务信息 + ext_service_id = getattr(kb, "external_service_id", None) + if ext_service_id: + # 查询外部服务配置 + try: + ext_service = db.query(ExternalKBService).filter( + ExternalKBService.id == ext_service_id + ).first() + if ext_service: + result["external_service"] = { + "id": ext_service.id, + "name": ext_service.name, + "service_type": ext_service.service_type, + "api_url": ext_service.api_url, + "status": getattr(ext_service, "status", "unknown"), + } + except Exception: + result["external_service"] = {"id": ext_service_id, "note": "外部服务表尚未初始化"} + + return {"code": 0, "data": result} + + +@router.post("/", summary="创建知识库") +async def create_knowledge_base( + req: KBCreate, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """创建新的知识库（支持官方平台知识库）""" + # 检查名称唯一性 + existing = db.query(KnowledgeBase).filter(KnowledgeBase.name == req.name).first() + if existing: + raise HTTPException(status_code=400, detail="知识库名称已存在") + + kb = KnowledgeBase( + id=str(uuid.uuid4()), + user_id=admin_user.id, + name=req.name, + description=req.description or "", + directory_path="/", + document_count=0, + knowledge_graph_count=0, + created_at=datetime.now(), + updated_at=datetime.now(), + ) + + # Phase 2.4 新增字段 + try: + setattr(kb, "is_official", req.is_official) + setattr(kb, "status", "active") + if req.external_service_id: + setattr(kb, "external_service_id", req.external_service_id) + if req.external_config: + setattr(kb, "external_config", json.dumps(req.external_config, ensure_ascii=False)) + setattr(kb, "external_sync_status", "none") + except Exception: + logger.warning("KnowledgeBase 新增字段设置失败，可能表结构尚未更新") + + db.add(kb) + db.commit() + db.refresh(kb) + + LogService.log_audit( + admin_id=admin_user.id, + admin_name=admin_user.username, + action="create_knowledge_base", + target_type="knowledge_base", + target_id=kb.id, + detail=f"创建知识库: {kb.name} (官方={req.is_official})", + ) + + return {"code": 0, "message": "知识库创建成功", "data": kb_to_dict(kb, db)} + + +@router.put("/{kb_id}", summary="更新知识库") +async def update_knowledge_base( + kb_id: str, + req: KBUpdate, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """更新知识库配置""" + kb = db.query(KnowledgeBase).filter(KnowledgeBase.id == kb_id).first() + if not kb: + raise HTTPException(status_code=404, detail="知识库不存在") + + if req.name is not None: + dup = db.query(KnowledgeBase).filter( + KnowledgeBase.name == req.name, + KnowledgeBase.id != kb_id + ).first() + if dup: + raise HTTPException(status_code=400, detail="知识库名称已存在") + kb.name = req.name + + if req.description is not None: + kb.description = req.description + + # Phase 2.4 新增字段更新 + try: + if req.is_official is not None: + setattr(kb, "is_official", req.is_official) + if req.status is not None: + setattr(kb, "status", req.status) + if req.external_service_id is not None: + setattr(kb, "external_service_id", req.external_service_id) + if req.external_config is not None: + setattr(kb, "external_config", json.dumps(req.external_config, ensure_ascii=False)) + except Exception: + logger.warning("KnowledgeBase 新增字段更新失败") + + kb.updated_at = datetime.now() + db.commit() + db.refresh(kb) + + LogService.log_audit( + admin_id=admin_user.id, + admin_name=admin_user.username, + action="update_knowledge_base", + target_type="knowledge_base", + target_id=kb_id, + detail=f"更新知识库: {kb.name}", + ) + + return {"code": 0, "message": "知识库已更新", "data": kb_to_dict(kb, db)} + + +@router.delete("/{kb_id}", summary="删除知识库") +async def delete_knowledge_base( + kb_id: str, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """删除知识库及其所有文档""" + kb = db.query(KnowledgeBase).filter(KnowledgeBase.id == kb_id).first() + if not kb: + raise HTTPException(status_code=404, detail="知识库不存在") + + kb_name = kb.name + + # 删除关联文档 + docs = db.query(KnowledgeDocument).filter( + KnowledgeDocument.knowledge_base_id == kb_id + ).all() + for d in docs: + db.delete(d) + + # 删除关联图谱 + nodes = db.query(KnowledgeGraphNode).filter( + KnowledgeGraphNode.knowledge_base_id == kb_id + ).all() + for n in nodes: + db.delete(n) + + edges = db.query(KnowledgeGraphEdge).filter( + KnowledgeGraphEdge.knowledge_base_id == kb_id + ).all() + for e in edges: + db.delete(e) + + build_logs = db.query(KnowledgeGraphBuildLog).filter( + KnowledgeGraphBuildLog.knowledge_base_id == kb_id + ).all() + for bl in build_logs: + db.delete(bl) + + # 删除关联目录 + dirs = db.query(KnowledgeDirectory).filter( + KnowledgeDirectory.knowledge_base_id == kb_id + ).all() + for d in dirs: + db.delete(d) + + # 删除知识库本身 + db.delete(kb) + db.commit() + + LogService.log_audit( + admin_id=admin_user.id, + admin_name=admin_user.username, + action="delete_knowledge_base", + target_type="knowledge_base", + target_id=kb_id, + detail=f"删除知识库: {kb_name} (含 {len(docs)} 文档, {len(nodes)} 节点, {len(edges)} 边)", + ) + + return {"code": 0, "message": "知识库已删除"} + + +@router.put("/{kb_id}/toggle", summary="启用/禁用知识库") +async def toggle_knowledge_base( + kb_id: str, + is_active: bool = Query(..., description="是否启用"), + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """启用或禁用知识库""" + kb = db.query(KnowledgeBase).filter(KnowledgeBase.id == kb_id).first() + if not kb: + raise HTTPException(status_code=404, detail="知识库不存在") + + new_status = "active" if is_active else "inactive" + try: + setattr(kb, "status", new_status) + except Exception: + pass + + kb.updated_at = datetime.now() + db.commit() + + LogService.log_audit( + admin_id=admin_user.id, + admin_name=admin_user.username, + action="toggle_knowledge_base", + target_type="knowledge_base", + target_id=kb_id, + detail=f"{'启用' if is_active else '禁用'}知识库: {kb.name}", + ) + + return { + "code": 0, + "message": f"知识库已{'启用' if is_active else '禁用'}", + "data": kb_to_dict(kb, db), + } + + +# ============================================================ +# 文档管理 +# ============================================================ + +@router.get("/{kb_id}/documents", summary="获取知识库文档列表") +async def get_kb_documents( + kb_id: str, + status: str = Query(None, description="文档状态筛选"), + file_type: str = Query(None, description="文件类型筛选"), + keyword: str = Query(None, description="文件名搜索"), + page: int = Query(1, ge=1), + page_size: int = Query(20, ge=1, le=100), + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """获取知识库中的文档列表""" + kb = db.query(KnowledgeBase).filter(KnowledgeBase.id == kb_id).first() + if not kb: + raise HTTPException(status_code=404, detail="知识库不存在") + + query = db.query(KnowledgeDocument).filter( + KnowledgeDocument.knowledge_base_id == kb_id + ) + + if status: + query = query.filter(KnowledgeDocument.status == status) + if file_type: + query = query.filter(KnowledgeDocument.file_type == file_type) + if keyword: + query = query.filter(KnowledgeDocument.filename.ilike(f"%{keyword}%")) + + total = query.count() + docs = query.order_by(desc(KnowledgeDocument.created_at)).offset( + (page - 1) * page_size + ).limit(page_size).all() + + return { + "code": 0, + "data": { + "total": total, + "page": page, + "page_size": page_size, + "items": [doc_to_dict(d) for d in docs], + } + } + + +@router.delete("/{kb_id}/documents/{doc_id}", summary="删除知识库文档") +async def delete_kb_document( + kb_id: str, + doc_id: str, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """删除知识库中的单个文档""" + doc = db.query(KnowledgeDocument).filter( + KnowledgeDocument.id == doc_id, + KnowledgeDocument.knowledge_base_id == kb_id, + ).first() + if not doc: + raise HTTPException(status_code=404, detail="文档不存在") + + filename = doc.filename + db.delete(doc) + + # 更新知识库文档计数 + kb = db.query(KnowledgeBase).filter(KnowledgeBase.id == kb_id).first() + if kb: + kb.document_count = db.query(func.count(KnowledgeDocument.id)).filter( + KnowledgeDocument.knowledge_base_id == kb_id + ).scalar() or 0 + kb.updated_at = datetime.now() + + db.commit() + + LogService.log_audit( + admin_id=admin_user.id, + admin_name=admin_user.username, + action="delete_knowledge_document", + target_type="knowledge_document", + target_id=doc_id, + detail=f"删除文档: {filename} (知识库: {kb.name})", + ) + + return {"code": 0, "message": "文档已删除"} + + +# ============================================================ +# 知识图谱管理 +# ============================================================ + +@router.get("/{kb_id}/graph/stats", summary="获取知识图谱统计") +async def get_kb_graph_stats( + kb_id: str, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """获取知识库的知识图谱统计信息""" + kb = db.query(KnowledgeBase).filter(KnowledgeBase.id == kb_id).first() + if not kb: + raise HTTPException(status_code=404, detail="知识库不存在") + + nodes = db.query(func.count(KnowledgeGraphNode.id)).filter( + KnowledgeGraphNode.knowledge_base_id == kb_id + ).scalar() or 0 + + edges = db.query(func.count(KnowledgeGraphEdge.id)).filter( + KnowledgeGraphEdge.knowledge_base_id == kb_id + ).scalar() or 0 + + # 最近构建日志 + recent_logs = db.query(KnowledgeGraphBuildLog).filter( + KnowledgeGraphBuildLog.knowledge_base_id == kb_id + ).order_by(desc(KnowledgeGraphBuildLog.created_at)).limit(5).all() + + # 节点类型分布 + node_types = db.query( + KnowledgeGraphNode.label, + func.count(KnowledgeGraphNode.id) + ).filter( + KnowledgeGraphNode.knowledge_base_id == kb_id + ).group_by(KnowledgeGraphNode.label).limit(20).all() + + return { + "code": 0, + "data": { + "total_nodes": int(nodes), + "total_edges": int(edges), + "node_type_distribution": [ + {"label": lbl, "count": int(cnt)} for lbl, cnt in node_types + ], + "recent_build_logs": [ + { + "id": bl.id, + "status": getattr(bl, "status", "unknown"), + "created_at": bl.created_at.isoformat() if bl.created_at else None, + } + for bl in recent_logs + ], + } + } + + +# ============================================================ +# 外部知识库服务管理 +# ============================================================ + +# ExternalKBService 模型定义（如果表不存在会在初始化时创建） +# 这里用原生 SQL 操作，确保即使 ORM 模型未注册也能工作 + +def _get_external_services_table(db: Session): + """安全地获取外部知识库服务列表（兼容表不存在的情况）""" + try: + # 尝试使用 ORM 模型查询 + from ...models import ExternalKBService + return db.query(ExternalKBService).all() + except (AttributeError, ImportError): + # ORM 模型不存在，用原生 SQL + try: + result = db.execute( + text("SELECT id, name, service_type, api_url, api_key, config, is_enabled, status, created_at, updated_at FROM external_kb_services ORDER BY created_at DESC") + ) + rows = result.fetchall() + return [ + { + "id": r[0], + "name": r[1], + "service_type": r[2], + "api_url": r[3], + "api_key": r[4], + "config": json.loads(r[5]) if r[5] else {}, + "is_enabled": bool(r[6]), + "status": r[7], + "created_at": r[8].isoformat() if r[8] else None, + "updated_at": r[9].isoformat() if r[9] else None, + } + for r in rows + ] + except Exception as e: + logger.warning(f"查询外部知识库服务表失败: {e}") + return [] + + +def _check_service_connection(service_type: str, api_url: str, api_key: str) -> dict: + """测试外部知识库服务连接状态（当前仅支持 Dify）""" + try: + if service_type == "dify": + from ...dify_kb import dify_test + return dify_test(api_url, api_key) + else: + # 未知类型（如历史遗留的 fastgpt/ragflow），按通用方式探测可达性 + headers = {} + if api_key: + headers["Authorization"] = f"Bearer {api_key}" + resp = httpx.get(api_url, headers=headers, timeout=10) + return { + "connected": resp.status_code < 500, + "usable": resp.status_code < 500, + "status": "connected" if resp.status_code < 500 else "error", + "detail": f"通用探测返回状态码 {resp.status_code}", + } + except httpx.TimeoutException: + return {"connected": False, "usable": False, "status": "timeout", "detail": "连接超时"} + except Exception as e: + return {"connected": False, "usable": False, "status": "error", "detail": str(e)} + + +@router.get("/external/services", summary="获取外部知识库服务列表") +async def list_external_services( + service_type: str = Query(None, description="服务类型: fastgpt/dify/ragflow"), + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """获取所有外部知识库服务配置""" + services = _get_external_services_table(db) + + if service_type: + services = [s for s in services if s.get("service_type") == service_type] + + return { + "code": 0, + "data": { + "total": len(services), + "items": services, + "supported_types": [ + {"type": "dify", "name": "Dify", "description": "LLM 应用开发平台，内置知识库管理（RAG）"}, + ] + } + } + + +@router.post("/external/services", summary="创建外部知识库服务配置") +async def create_external_service( + req: ExternalServiceConfig, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """创建新的外部知识库服务连接配置""" + # 验证服务类型（当前仅支持 Dify） + valid_types = ["dify"] + if req.service_type not in valid_types: + raise HTTPException(status_code=400, detail=f"不支持的服务类型: {req.service_type}") + + # 检查连接 + conn_result = _check_service_connection(req.service_type, req.api_url, req.api_key or "") + + service_id = str(uuid.uuid4()) + config_json = json.dumps(req.config or {}, ensure_ascii=False) + conn_status = conn_result.get("status", "disconnected") + + # 使用原生 SQL 插入（兼容 ORM 模型可能未注册） + try: + db.execute( + text(f"INSERT INTO external_kb_services (id, name, service_type, api_url, api_key, config, is_enabled, status, created_at, updated_at) " + f"VALUES (:id, :name, :stype, :url, :key, :cfg, 1, :status, NOW(), NOW())"), + { + "id": service_id, + "name": req.name, + "stype": req.service_type, + "url": req.api_url, + "key": req.api_key or "", + "cfg": config_json, + "status": conn_status, + } + ) + db.commit() + except Exception as e: + # 表可能不存在，尝试创建 + logger.warning(f"插入外部服务失败: {e}, 尝试创建表") + db.execute(text(""" + CREATE TABLE IF NOT EXISTS external_kb_services ( + id VARCHAR(36) PRIMARY KEY, + name VARCHAR(200) NOT NULL, + service_type VARCHAR(20) NOT NULL, + api_url VARCHAR(500) NOT NULL, + api_key VARCHAR(255) DEFAULT '', + config JSON DEFAULT NULL, + is_enabled BOOLEAN DEFAULT TRUE, + status VARCHAR(20) DEFAULT 'disconnected', + created_at DATETIME DEFAULT CURRENT_TIMESTAMP, + updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 + """)) + db.commit() + # 再次尝试插入 + db.execute( + text(f"INSERT INTO external_kb_services (id, name, service_type, api_url, api_key, config, is_enabled, status, created_at, updated_at) " + f"VALUES (:id, :name, :stype, :url, :key, :cfg, 1, :status, NOW(), NOW())"), + { + "id": service_id, + "name": req.name, + "stype": req.service_type, + "url": req.api_url, + "key": req.api_key or "", + "cfg": config_json, + "status": conn_status, + } + ) + db.commit() + + LogService.log_audit( + admin_id=admin_user.id, + admin_name=admin_user.username, + action="create_external_kb_service", + target_type="external_kb_service", + target_id=service_id, + detail=f"创建外部知识库服务: {req.name} ({req.service_type})", + ) + + return { + "code": 0, + "message": "外部知识库服务配置已创建", + "data": { + "id": service_id, + "name": req.name, + "service_type": req.service_type, + "api_url": req.api_url, + "connection": conn_result, + } + } + + +@router.put("/external/services/{service_id}", summary="更新外部知识库服务配置") +async def update_external_service( + service_id: str, + req: ExternalServiceUpdate, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """更新外部知识库服务配置""" + # 查找现有服务 + try: + result = db.execute( + text("SELECT id, name, service_type, api_url, api_key FROM external_kb_services WHERE id = :sid"), + {"sid": service_id} + ) + row = result.fetchone() + except Exception: + raise HTTPException(status_code=404, detail="外部知识库服务不存在或表未创建") + + if not row: + raise HTTPException(status_code=404, detail="外部知识库服务不存在") + + # 构建更新字段 + updates = {} + if req.name is not None: + updates["name"] = req.name + if req.api_url is not None: + updates["api_url"] = req.api_url + if req.api_key is not None: + updates["api_key"] = req.api_key + if req.config is not None: + updates["config"] = json.dumps(req.config, ensure_ascii=False) + if req.is_enabled is not None: + updates["is_enabled"] = 1 if req.is_enabled else 0 + + # 如果 URL 或 API Key 变了，重新测试连接 + new_url = req.api_url or row[3] + new_key = req.api_key or row[4] + if req.api_url or req.api_key: + service_type = row[2] + conn_result = _check_service_connection(service_type, new_url, new_key) + updates["status"] = conn_result.get("status", "disconnected") + + updates["updated_at"] = "NOW()" + + # 执行更新 + set_clause = ", ".join([f"{k} = :{k}" for k in updates.keys() if k != "updated_at"]) + set_clause += ", updated_at = NOW()" + params = {k: v for k, v in updates.items() if k != "updated_at"} + params["sid"] = service_id + + db.execute( + text(f"UPDATE external_kb_services SET {set_clause} WHERE id = :sid"), + params + ) + db.commit() + + LogService.log_audit( + admin_id=admin_user.id, + admin_name=admin_user.username, + action="update_external_kb_service", + target_type="external_kb_service", + target_id=service_id, + detail=f"更新外部知识库服务: {row[1]}", + ) + + return {"code": 0, "message": "外部知识库服务配置已更新"} + + +@router.delete("/external/services/{service_id}", summary="删除外部知识库服务配置") +async def delete_external_service( + service_id: str, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """删除外部知识库服务配置""" + try: + result = db.execute( + text("SELECT name FROM external_kb_services WHERE id = :sid"), + {"sid": service_id} + ) + row = result.fetchone() + except Exception: + raise HTTPException(status_code=404, detail="外部知识库服务不存在") + + if not row: + raise HTTPException(status_code=404, detail="外部知识库服务不存在") + + name = row[0] + db.execute(text("DELETE FROM external_kb_services WHERE id = :sid"), {"sid": service_id}) + db.commit() + + LogService.log_audit( + admin_id=admin_user.id, + admin_name=admin_user.username, + action="delete_external_kb_service", + target_type="external_kb_service", + target_id=service_id, + detail=f"删除外部知识库服务: {name}", + ) + + return {"code": 0, "message": "外部知识库服务配置已删除"} + + +@router.get("/external/services/{service_id}/status", summary="检查外部知识库服务连接状态") +async def check_external_service_status( + service_id: str, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """实时检查外部知识库服务的连接状态""" + try: + result = db.execute( + text("SELECT service_type, api_url, api_key FROM external_kb_services WHERE id = :sid"), + {"sid": service_id} + ) + row = result.fetchone() + except Exception: + raise HTTPException(status_code=404, detail="外部知识库服务不存在") + + if not row: + raise HTTPException(status_code=404, detail="外部知识库服务不存在") + + conn_result = _check_service_connection(row[0], row[1], row[2]) + + # 更新状态 + try: + db.execute( + text("UPDATE external_kb_services SET status = :status, updated_at = NOW() WHERE id = :sid"), + {"status": conn_result.get("status", "disconnected"), "sid": service_id} + ) + db.commit() + except Exception: + pass + + return {"code": 0, "data": conn_result} + + +# ============================================================ +# 知识库同步（与外部服务） +# ============================================================ + +@router.post("/external/sync/{kb_id}", summary="同步知识库与外部服务") +async def sync_knowledge_base( + kb_id: str, + req: SyncRequest, + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """将知识库与外部知识库服务进行数据同步""" + kb = db.query(KnowledgeBase).filter(KnowledgeBase.id == kb_id).first() + if not kb: + raise HTTPException(status_code=404, detail="知识库不存在") + + # 确定同步目标服务 + ext_service_id = req.external_service_id or getattr(kb, "external_service_id", None) + if not ext_service_id: + raise HTTPException(status_code=400, detail="未指定外部知识库服务，请先关联或传入 external_service_id") + + # 查询外部服务配置 + try: + result = db.execute( + text("SELECT name, service_type, api_url, api_key, config FROM external_kb_services WHERE id = :sid"), + {"sid": ext_service_id} + ) + row = result.fetchone() + except Exception: + raise HTTPException(status_code=404, detail="外部知识库服务不存在") + + if not row: + raise HTTPException(status_code=404, detail="外部知识库服务不存在") + + service_name, service_type, api_url, api_key, config_raw = row + service_config = json.loads(config_raw) if config_raw else {} + + # 执行同步 + sync_result = {"synced": False, "detail": "", "items_synced": 0} + + try: + headers = {} + if api_key: + headers["Authorization"] = f"Bearer {api_key}" + + if service_type == "dify" and req.sync_direction == "push": + # 向 Dify 推送知识库数据（使用 app.dify_kb 客户端，按 dataset_name 解析） + from ...dify_kb import dify_ingest_text, resolve_dataset_id + dataset_name = service_config.get("dataset_name") if isinstance(service_config, dict) else None + ds_id = resolve_dataset_id(api_url, api_key, dataset_name) + if not ds_id: + sync_result["detail"] = "无法解析 Dify 知识库 ID（请检查 API Key 是否绑定到指定数据集）" + sync_result["synced"] = False + else: + docs = db.query(KnowledgeDocument).filter( + KnowledgeDocument.knowledge_base_id == kb_id, + KnowledgeDocument.status == "completed" + ).all() + synced_count = 0 + for doc in docs: + content = doc.content or "" + if not content: + continue + if dify_ingest_text(api_url, api_key, content[:50000], name=doc.filename, dataset_name=dataset_name): + synced_count += 1 + sync_result["synced"] = True + sync_result["detail"] = f"已推送 {synced_count}/{len(docs)} 文档到 Dify" + sync_result["items_synced"] = synced_count + + elif req.sync_direction == "pull": + # 从外部服务拉取知识库数据 + sync_result["detail"] = f"从 {service_type} 拉取数据（暂未实现完整拉取逻辑）" + sync_result["synced"] = True + sync_result["items_synced"] = 0 + + elif req.sync_direction == "bidirectional": + sync_result["detail"] = f"双向同步（暂未实现完整双向同步逻辑）" + sync_result["synced"] = True + sync_result["items_synced"] = 0 + + # 更新同步状态 + try: + setattr(kb, "external_sync_status", "synced" if sync_result["synced"] else "error") + except Exception: + db.execute( + text("UPDATE knowledge_bases SET external_sync_status = :es WHERE id = :kid"), + {"es": "synced" if sync_result["synced"] else "error", "kid": kb_id} + ) + + kb.updated_at = datetime.now() + db.commit() + + except Exception as e: + logger.error(f"知识库同步失败: {e}") + sync_result["detail"] = f"同步异常: {str(e)}" + sync_result["synced"] = False + + try: + setattr(kb, "external_sync_status", "error") + except Exception: + db.execute( + text("UPDATE knowledge_bases SET external_sync_status = 'error' WHERE id = :kid"), + {"kid": kb_id} + ) + db.commit() + + LogService.log_audit( + admin_id=admin_user.id, + admin_name=admin_user.username, + action="sync_knowledge_base", + target_type="knowledge_base", + target_id=kb_id, + detail=f"同步知识库 '{kb.name}' → {service_name} ({service_type}), 方向={req.sync_direction}, 结果={sync_result['detail']}", + ) + + return {"code": 0, "data": sync_result} + + +# ============================================================ +# 平台内容批量入库到 Dify（AI聊天产出 + 用户上传文档） +# ============================================================ + +@router.post("/external/dify/ingest", summary="将平台内容批量入库到 Dify 知识库") +async def ingest_platform_to_dify( + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """把平台中「可量化的内容」推送到 Dify 知识库（qihuai）: + - AI 聊天的助手回答（文本、非工具/非媒体、长度 > 120） + - 用户上传的文档资产（assets, category='file' 且含文本内容） + 返回各来源的入库数量。 + """ + from ...dify_kb import dify_ingest_text, resolve_dataset_id + + # 读取已启用的 Dify 服务 + try: + row = db.execute( + text("SELECT api_url, api_key, config FROM external_kb_services " + "WHERE service_type='dify' AND is_enabled=1 LIMIT 1") + ).fetchone() + except Exception: + row = None + if not row: + raise HTTPException(status_code=400, detail="未配置已启用的 Dify 知识库服务，请先添加 Dify 服务并测试连接") + api_url, api_key, config_raw = row + dataset_name = None + if config_raw: + try: + cfg = config_raw if isinstance(config_raw, dict) else json.loads(config_raw) + dataset_name = cfg.get("dataset_name") + except Exception: + pass + ds_id = resolve_dataset_id(api_url, api_key, dataset_name) + if not ds_id: + raise HTTPException(status_code=400, detail="无法解析 Dify 知识库 ID（请检查 API Key 与 dataset_name）") + + stats = {"chat_answers": 0, "user_docs": 0, "skipped": 0, "failed": 0} + + # 1. AI 聊天助手回答 + try: + from ...models import Message + answers = db.query(Message).filter( + Message.role == "assistant" + ).order_by(Message.created_at.asc()).limit(2000).all() + for m in answers: + content = getattr(m, "content", None) or "" + content = content.strip() + # 跳过媒体/工具类消息与过短内容 + if len(content) < 120: + stats["skipped"] += 1 + continue + if content.startswith("[") and ("MEDIA" in content or "IMAGE" in content or "VIDEO" in content): + stats["skipped"] += 1 + continue + conv_id = getattr(m, "conversation_id", None) or "unknown" + if dify_ingest_text(api_url, api_key, content, name=f"chat_{conv_id}_{m.id}", dataset_name=dataset_name): + stats["chat_answers"] += 1 + else: + stats["failed"] += 1 + except Exception as e: + logger.error(f"ingest chat answers failed: {e}") + + # 2. 用户上传的文档资产 + try: + from ...models import Asset + assets = db.query(Asset).filter( + getattr(Asset, "category", "file") == "file" + ).order_by(getattr(Asset, "created_at", text("id")).asc()).limit(2000).all() + for a in assets: + content = getattr(a, "content", None) or "" + content = content.strip() + if len(content) < 50: + stats["skipped"] += 1 + continue + fname = getattr(a, "filename", None) or getattr(a, "name", None) or "asset" + if dify_ingest_text(api_url, api_key, content[:50000], name=f"asset_{a.id}_{fname}", dataset_name=dataset_name): + stats["user_docs"] += 1 + else: + stats["failed"] += 1 + except Exception as e: + logger.error(f"ingest assets failed: {e}") + + LogService.log_audit( + admin_id=admin_user.id, + admin_name=admin_user.username, + action="ingest_platform_to_dify", + target_type="external_kb_service", + target_id=ds_id, + detail=f"平台内容入库 Dify: 聊天回答 {stats['chat_answers']} 条, 用户文档 {stats['user_docs']} 条, 跳过 {stats['skipped']}, 失败 {stats['failed']}", + ) + + return { + "code": 0, + "message": f"入库完成：聊天回答 {stats['chat_answers']} 条，用户文档 {stats['user_docs']} 条", + "data": stats, + } + + +# ============================================================ +# 批量操作 +# ============================================================ + +@router.post("/batch/delete", summary="批量删除知识库") +async def batch_delete_kbs( + kb_ids: List[str] = Body(..., description="知识库ID列表"), + admin_user=Depends(get_current_admin), + db: Session = Depends(get_db), +): + """批量删除知识库""" + deleted = 0 + failed = 0 + for kb_id in kb_ids: + kb = db.query(KnowledgeBase).filter(KnowledgeBase.id == kb_id).first() + if not kb: + failed += 1 + continue + + # 删除关联数据 + db.query(KnowledgeDocument).filter(KnowledgeDocument.knowledge_base_id == kb_id).delete() + db.query(KnowledgeGraphNode).filter(KnowledgeGraphNode.knowledge_base_id == kb_id).delete() + db.query(KnowledgeGraphEdge).filter(KnowledgeGraphEdge.knowledge_base_id == kb_id).delete() + db.query(KnowledgeGraphBuildLog).filter(KnowledgeGraphBuildLog.knowledge_base_id == kb_id).delete() + db.query(KnowledgeDirectory).filter(KnowledgeDirectory.knowledge_base_id == kb_id).delete() + db.delete(kb) + deleted += 1 + + db.commit() + + LogService.log_audit( + admin_id=admin_user.id, + admin_name=admin_user.username, + action="batch_delete_knowledge_bases", + target_type="knowledge_base", + target_id=",".join(kb_ids[:10]), + detail=f"批量删除知识库: 成功 {deleted}, 失败 {failed}", + ) + + return {"code": 0, "data": {"deleted": deleted, "failed": failed}} diff --git a/backend/python/app/api/chat.py b/backend/python/app/api/chat.py index 78b6cdf..982054c 100644 --- a/backend/python/app/api/chat.py +++ b/backend/python/app/api/chat.py @@ -1,1506 +1,1558 @@ -""" -增强版 Chat API — Phase 2.5 -包含: Skill ID统一(MarketApp映射), 真实Skill执行, Agent工作流注入, - 多轮上下文保持, BM25 KB检索, Function Calling循环增强 -""" -from fastapi import APIRouter, Depends, HTTPException, Query -from sqlalchemy.orm import Session -from sqlalchemy import and_, desc, asc, func -from pydantic import BaseModel, Field -from typing import Optional -import uuid -from datetime import datetime, timezone, timedelta -import re -import json -import asyncio -import math -import httpx -import os -import logging - -from ..database import get_db, SessionLocal -from ..models import ( - User, Conversation, Message, AIModel, ModelProvider, TokenUsageLog, - KnowledgeBase, KnowledgeDocument, KnowledgeDocumentChunk, EvolutionSkill, - EvolutionTrajectory, MarketApp, AppPurchase, AppInstallation, - gen_uuid -) -from ..auth import get_current_user -from ..middleware.log_service import LogService -from ..capability_registry import resolve_auto_model - -router = APIRouter() - -logger = logging.getLogger(__name__) - - -def detect_media_intent(text: str) -> str | None: - """根据用户输入判断是否为 图像/视频/语音 生成请求，返回对应能力或 None。 - - 用于 Auto 模式：用户未指定具体模型时，依据意图自动匹配平台各类能力的默认模型。 - 疑问句（含「吗/呢/怎么/能否」或结尾问号）不触发，避免「你能生成视频吗」被误判。 - """ - if not text: - return None - t = text.lower().strip() - # 疑问句不触发媒体生成 - if t.endswith(("？", "?")) or any(q in t for q in ("吗", "呢", "怎么", "如何", "能否", "是否可以", "可以吗", "会不会", "why", "how")): - return None - video_kw = ("视频", "短片", "影片", "vide", "动画视频", "一段视频", "做个视频", "生成视频", - "视频生成", "来段视频", "拍个视频", "做段视频", "视频剪辑") - image_kw = ("图片", "图像", "插画", "海报", "照片", "头像", "壁纸", "漫画", "表情包", "logo", "图标", - "imag", "画一张", "画张", "画一只", "画只", "画个", "画幅", - "来张", "来一张", "来幅", "发张图", "发个图", - "生成图片", "生成一张", "生成一幅", "生成图", "图像生成", - "做一张", "做张图", "做张", "出张图", "出一张图", - "帮我画", "给我画", "请画", "绘制", "作画", "ai绘画", "ai画图", - "画个图", "画张图", "画一幅图", "一张图", "一幅画", "生成一幅") - audio_kw = ("语音", "配音", "朗读", "播报", "念一段", "有声", "tts", "音频", "生成语音", - "读一下", "念一下", "语音合成") - if any(k in t for k in video_kw): - return "video" - if any(k in t for k in image_kw): - return "image" - if any(k in t for k in audio_kw): - return "audio" - return None - -# CST timezone (UTC+8) for consistent time display -CST_TZ = timezone(timedelta(hours=8)) - - -def cst_now(): - """Get current time in CST timezone""" - return datetime.now(CST_TZ) - - -def _html_escape(s: str) -> str: - return (str(s).replace("&", "&").replace("<", "<").replace(">", ">") - .replace('"', """).replace("'", "'")) - - -def _outbound_proxy() -> str: - """出网代理地址。容器内直连 apihub.agnes-ai.com 会被对端重置 - (httpx RemoteProtocolError: Server disconnected without sending a response)， - 必须经 172.18.0.1:7890 智能路由代理(其对国内 AGNES 域名走直连)。""" - import os - return os.environ.get("HTTPS_PROXY") or os.environ.get("HTTP_PROXY") or "http://172.18.0.1:7890" - - -async def _generate_media(db, ai_model, prompt: str, user_id: str = None) -> str: - """图像/视频生成模型统一入口：按 capabilities 调用 AGNES 等媒体生成 API。 - 返回可直接在 AI 聊天中渲染的富文本(含 /