feat: 新增样本中心对外API接口及认证限流1. 新增认证接口。2. 新增知识库对外API（列表/详情/批量入库/任务查询）。3. 修复 Milvus 入库字段缺失问题

src/app/api/v1/knowledge_base_api_view.py

@@ -0,0 +1,132 @@
+"""
+知识库对外API接口路由
+"""
+from fastapi import APIRouter, Header
+from fastapi.responses import JSONResponse
+from app.schemas.base import ResponseSchema
+from app.services.api.knowledge_base_api_service import KnowledgeBaseApiService
+from app.services.api.token_api_service import TokenApiService
+
+router = APIRouter(tags=["样本中心对外API-知识库"])
+knowledge_service = KnowledgeBaseApiService()
+token_service = TokenApiService()
+
+
+async def _check_token(authorization: str = Header(None)) -> JSONResponse | None:
+    """验证 API token，失败返回 401 响应，通过返回 None"""
+    if not authorization:
+        return JSONResponse(
+            status_code=401,
+            content=ResponseSchema(
+                code="000401",
+                message="认证失败：缺少 Authorization 请求头",
+                data=None
+            ).model_dump(mode='json')
+        )
+
+    # 支持 "Bearer <token>" 格式
+    token = authorization
+    if authorization.startswith("Bearer "):
+        token = authorization[7:]
+
+    payload = await token_service.verify_api_token(token)
+    if not payload:
+        return JSONResponse(
+            status_code=401,
+            content=ResponseSchema(
+                code="000401",
+                message="认证失败：Token无效或已过期",
+                data=None
+            ).model_dump(mode='json')
+        )
+    return None
+
+
+@router.get("/knowledge-bases")
+async def list_knowledge_bases(
+    page: int = 1,
+    page_size: int = 20,
+    authorization: str = Header(None)
+):
+    """查询知识库列表"""
+    err = await _check_token(authorization)
+    if err:
+        return err
+
+    total, items = await knowledge_service.get_knowledge_base_list(page, page_size)
+    return ResponseSchema(
+        code="000000",
+        message="success",
+        data={
+            "total": total,
+            "page": page,
+            "page_size": page_size,
+            "items": items
+        }
+    )
+
+
+@router.get("/knowledge-bases/{kb_id}")
+async def get_knowledge_base_detail(
+    kb_id: str,
+    authorization: str = Header(None)
+):
+    """查询知识库详情"""
+    err = await _check_token(authorization)
+    if err:
+        return err
+
+    detail = await knowledge_service.get_knowledge_base_detail(kb_id)
+    if not detail:
+        return ResponseSchema(code="001001", message="知识库不存在", data=None)
+    return ResponseSchema(code="000000", message="success", data=detail)
+
+
+@router.get("/knowledge-bases/batch-import/{task_id}")
+async def get_batch_import_task(
+    task_id: str,
+    authorization: str = Header(None)
+):
+    """批量入库任务查询"""
+    err = await _check_token(authorization)
+    if err:
+        return err
+
+    task_info = await knowledge_service.get_batch_import_task(task_id)
+    if not task_info:
+        return ResponseSchema(code="002002", message="任务不存在", data=None)
+    return ResponseSchema(code="000000", message="success", data=task_info)
+
+
+@router.post("/knowledge-bases/{kb_id}/batch-import")
+async def create_batch_import_task(
+    kb_id: str,
+    request_data: dict,
+    authorization: str = Header(None)
+):
+    """批量入库提交"""
+    err = await _check_token(authorization)
+    if err:
+        return err
+
+    task_no = request_data.get("task_no")
+    if not task_no:
+        return ResponseSchema(code="002001", message="task_no不能为空", data=None)
+
+    task_id, status = await knowledge_service.create_batch_import_task(
+        kb_id,
+        task_no,
+        request_data.get("callback_url"),
+        request_data.get("parents", []),
+        request_data.get("children", [])
+    )
+    if not task_id:
+        return ResponseSchema(code="001001", message="知识库不存在或未启用", data=None)
+    return ResponseSchema(
+        code="000000",
+        message="success",
+        data={
+            "task_id": task_id,
+            "status": status
+        }
+    )

src/app/api/v1/token_api_view.py

@@ -0,0 +1,60 @@
+"""
+对外API Token认证接口路由
+"""
+from fastapi import APIRouter
+from fastapi.responses import JSONResponse
+from app.schemas.base import ResponseSchema
+from app.services.api.token_api_service import TokenApiService
+from app.middleware.rate_limiter import check_rate_limit, get_rate_limit_headers
+from datetime import datetime, timezone
+
+router = APIRouter(tags=["对外API-Token认证"])
+service = TokenApiService()
+
+
+@router.post("/auth/token")
+async def get_access_token(request_data: dict):
+    """
+    获取访问令牌
+
+    请求参数：
+    - app_id: 应用标识
+    - app_secret: 应用密钥
+    """
+    app_id = request_data.get("app_id", "")
+    app_secret = request_data.get("app_secret", "")
+
+    from app.services.api.token_api_service import logger as svc_logger
+    svc_logger.info(f"[Token] received app_id={app_id}, app_secret_len={len(app_secret)}, app_secret_first3={app_secret[:3] if app_secret else 'empty'}")
+
+    if not app_id or not app_secret:
+        return ResponseSchema(
+            code="000400",
+            message="app_id和app_secret不能为空",
+            data=None
+        )
+
+    # 限流检查: 10 次/分钟/app_id
+    allowed, remaining, reset_seconds = check_rate_limit("token", app_id)
+    headers = get_rate_limit_headers("token", remaining, reset_seconds)
+
+    if not allowed:
+        response = ResponseSchema(code="000429", message="请求过于频繁，请稍后重试", data=None)
+        return JSONResponse(status_code=429, content=response.model_dump(mode='json'), headers=headers)
+
+    access_token, error_code, error_message, expires_in = await service.generate_token(app_id, app_secret)
+
+    if error_code:
+        response = ResponseSchema(code=error_code, message=error_message, data=None)
+        return JSONResponse(status_code=200, content=response.model_dump(mode='json'), headers=headers)
+
+    response = ResponseSchema(
+        code="000000",
+        message="success",
+        data={
+            "access_token": access_token,
+            "expires_in": expires_in,
+            "token_type": "Bearer"
+        }
+    )
+    return JSONResponse(status_code=200, content=response.model_dump(mode='json'), headers=headers)

src/app/core/exceptions.py

@@ -103,15 +103,15 @@ class ConflictError(BaseAPIException):
 
 class RateLimitError(BaseAPIException):
     """频率限制错误"""
-    
+
     def __init__(
         self,
-        message: str = "请求过于频繁",
+        message: str = "请求过于频繁，请稍后重试",
         details: Optional[Dict[str, Any]] = None
     ):
         super().__init__(
             message=message,
-            code=100006,
+            code="000429",
             status_code=429,
             details=details
         )

src/app/middleware/rate_limiter.py

@@ -0,0 +1,86 @@
+"""
+API 请求限流模块
+基于 Redis 滑动窗口计数实现，支持按 app_id 和接口维度限流
+"""
+import time
+import logging
+from typing import Dict, Tuple
+from app.base.async_redis_connection import get_redis_connection
+
+logger = logging.getLogger(__name__)
+
+# 限流规则: key -> (最大请求数, 窗口秒数)
+RATE_LIMITS: Dict[str, Tuple[int, int]] = {
+    "token":        (10,   60),   # 获取 Token: 10 次/分钟/app_id
+    "kb_list":      (60,   60),   # 知识库列表: 60 次/分钟/app_id
+    "kb_detail":    (60,   60),   # 知识库详情: 60 次/分钟/app_id
+    "batch_import": (20,   60),   # 批量入库提交: 20 次/分钟/app_id
+    "task_query":   (120,  60),   # 任务查询: 120 次/分钟/app_id
+}
+
+
+def check_rate_limit(rule_key: str, app_id: str) -> Tuple[bool, int, int]:
+    """
+    检查是否超出限流阈值
+
+    Args:
+        rule_key: 限流规则 key（如 "token", "kb_list"）
+        app_id: 应用标识
+
+    Returns:
+        (是否允许请求, 剩余次数, 重置时间秒)
+    """
+    if rule_key not in RATE_LIMITS:
+        logger.warning(f"[RateLimit] 未知限流规则: {rule_key}")
+        return True, 0, 0
+
+    max_requests, window_seconds = RATE_LIMITS[rule_key]
+    redis_key = f"rate_limit:{rule_key}:{app_id}"
+
+    redis_client = get_redis_connection()
+    if not redis_client:
+        logger.warning(f"[RateLimit] Redis不可用，跳过限流检查: {rule_key}/{app_id}")
+        return True, 0, 0
+
+    try:
+        now = time.time()
+        window_start = now - window_seconds
+
+        # 删除窗口外的旧记录
+        redis_client.zremrangebyscore(redis_key, 0, window_start)
+
+        # 统计窗口内请求数
+        current_count = redis_client.zcard(redis_key)
+
+        if current_count >= max_requests:
+            # 获取最早一条记录的时间，计算重置时间
+            earliest = redis_client.zrange(redis_key, 0, 0, withscores=True)
+            reset_at = earliest[0][1] + window_seconds if earliest else now + window_seconds
+            remaining = 0
+            logger.warning(
+                f"[RateLimit] 超限: rule={rule_key}, app_id={app_id}, "
+                f"count={current_count}/{max_requests}"
+            )
+            return False, remaining, int(reset_at - now)
+
+        # 记录本次请求
+        redis_client.zadd(redis_key, {f"{now}:{current_count}": now})
+        redis_client.expire(redis_key, window_seconds + 1)
+
+        remaining = max_requests - current_count - 1
+        return True, remaining, window_seconds
+
+    except Exception as e:
+        logger.error(f"[RateLimit] 限流检查失败: {e}")
+        return True, 0, 0
+
+
+def get_rate_limit_headers(rule_key: str, remaining: int, reset_seconds: int) -> Dict[str, str]:
+    """生成限流响应头"""
+    if rule_key in RATE_LIMITS:
+        return {
+            "X-RateLimit-Limit": str(RATE_LIMITS[rule_key][0]),
+            "X-RateLimit-Remaining": str(remaining),
+            "X-RateLimit-Reset": str(int(time.time()) + reset_seconds),
+        }
+    return {}

src/app/server/app.py

@@ -62,6 +62,8 @@ from views.dict_item_view import router as dict_item_router
 
 # 导入对外API路由
 from app.api.v1.system_api_view import router as system_api_router
+from app.api.v1.knowledge_base_api_view import router as kb_api_router
+from app.api.v1.token_api_view import router as token_api_router
 
 # 导入现有API路由
 # from app.api.v1.api_router import api_router
@@ -94,7 +96,7 @@ async def lifespan(app: FastAPI):
     logger.info(f"📍 服务地址: http://{config_handler.get('admin_app', 'HOST', '0.0.0.0')}:{config_handler.get_int('admin_app', 'PORT', 8000)}")
     logger.info(f"📚 API文档: http://{config_handler.get('admin_app', 'HOST', '0.0.0.0')}:{config_handler.get_int('admin_app', 'PORT', 8000)}/docs")
     logger.info("=" * 60)
-    
+
     yield
     
     # 关闭时执行
@@ -162,6 +164,7 @@ app.add_middleware(
         "/auth/refresh",
         "/auth/sso/authorize",
         "/api/oauth/exchange-code",
+        "/api/v1/auth/token",
         "/docs",
         "/openapi.json",
         "/redoc",
@@ -310,7 +313,13 @@ async def root():
 # app.include_router(api_router, prefix="/api/v1")
 # 对外API路由
 app.include_router(system_api_router, prefix="/api/external/v1")
+
 # 新的模块化视图路由
+app.include_router(kb_api_router, prefix="/api/v1")
+
+# Token认证路由（对外API）
+app.include_router(token_api_router, prefix="/api/v1")
+
 app.include_router(system_router, prefix="/api/v1")
 app.include_router(auth_router, prefix="/api/v1")
 app.include_router(sso_router, prefix="")

src/app/services/api/knowledge_base_api_service.py

@@ -0,0 +1,401 @@
+"""
+知识库对外API业务逻辑
+"""
+import logging
+import json
+from typing import Tuple, List, Dict, Any, Optional
+from app.base.async_mysql_connection import get_db_connection
+
+logger = logging.getLogger(__name__)
+
+
+class KnowledgeBaseApiService:
+    """知识库对外API服务"""
+
+    async def get_knowledge_base_list(self, page: int, page_size: int) -> Tuple[int, List[Dict[str, Any]]]:
+        """查询知识库列表（分页）"""
+        conn = get_db_connection()
+        if not conn:
+            return 0, []
+
+        cursor = conn.cursor()
+        try:
+            cursor.execute(
+                "SELECT COUNT(*) as cnt FROM t_samp_knowledge_base "
+                "WHERE status = 'normal' AND is_deleted = 0"
+            )
+            total = cursor.fetchone()['cnt']
+
+            offset = (page - 1) * page_size
+            cursor.execute(
+                "SELECT id, name, "
+                "collection_name_parent as parent_table, "
+                "collection_name_children as child_table, "
+                "document_count, status, "
+                "created_time as created_at, created_by "
+                "FROM t_samp_knowledge_base "
+                "WHERE status = 'normal' AND is_deleted = 0 "
+                "ORDER BY created_time DESC LIMIT %s OFFSET %s",
+                (page_size, offset)
+            )
+            kbs = cursor.fetchall()
+
+            items = []
+            for kb in kbs:
+                cursor.execute(
+                    "SELECT field_zh_name as field_name_cn, field_en_name as field_name_en, "
+                    "field_type, remark as description "
+                    "FROM t_samp_metadata WHERE knowledge_base_id = %s",
+                    (kb['id'],)
+                )
+                metadata_schema = cursor.fetchall()
+                kb['metadata_schema'] = metadata_schema
+                items.append(kb)
+
+            return total, items
+
+        except Exception as e:
+            logger.error(f"查询知识库列表失败: {e}")
+            return 0, []
+        finally:
+            cursor.close()
+            conn.close()
+
+    async def get_knowledge_base_detail(self, kb_id: str) -> Optional[Dict[str, Any]]:
+        """查询知识库详情"""
+        conn = get_db_connection()
+        if not conn:
+            return None
+
+        cursor = conn.cursor()
+        try:
+            cursor.execute(
+                "SELECT id, name, description, "
+                "collection_name_parent as parent_table, "
+                "collection_name_children as child_table, "
+                "document_count, status, "
+                "created_time as created_at, created_by, "
+                "updated_time as updated_at "
+                "FROM t_samp_knowledge_base WHERE id = %s",
+                (kb_id,)
+            )
+            kb = cursor.fetchone()
+            if not kb:
+                return None
+
+            cursor.execute(
+                "SELECT field_zh_name as field_name_cn, field_en_name as field_name_en, "
+                "field_type, remark as description "
+                "FROM t_samp_metadata WHERE knowledge_base_id = %s",
+                (kb_id,)
+            )
+            kb['metadata_schema'] = cursor.fetchall()
+
+            return kb
+
+        except Exception as e:
+            logger.error(f"查询知识库详情失败: {e}")
+            return None
+        finally:
+            cursor.close()
+            conn.close()
+
+
+
+
+
+
+
+    async def create_batch_import_task(
+        self,
+        kb_id: str,
+        task_no: str,
+        callback_url: str,
+        parents: list,
+        children: list
+    ) -> Tuple[Optional[str], Optional[str]]:
+        """创建批量入库任务"""
+        conn = get_db_connection()
+        if not conn:
+            return None, None
+
+        cursor = conn.cursor()
+        try:
+            cursor.execute(
+                "SELECT id, name, collection_name_parent, collection_name_children "
+                "FROM t_samp_knowledge_base WHERE id = %s AND status = 'normal'",
+                (kb_id,)
+            )
+            kb = cursor.fetchone()
+            if not kb:
+                return None, None
+
+            if not task_no:
+                logger.warning(f"批量入库任务缺少task_no, kb_id: {kb_id}")
+                return None, None
+
+            import uuid, time
+            task_id = f"task_{time.strftime('%Y%m%d')}{uuid.uuid4().hex[:12]}"
+
+            task_params = json.dumps({
+                "kb_id": kb_id,
+                "parents": parents,
+                "children": children
+            }, ensure_ascii=False)
+
+            cursor.execute(
+                "INSERT INTO t_samp_task_management "
+                "(task_id, task_no, task_type, task_params, task_source, callback_url, status) "
+                "VALUES (%s, %s, %s, %s, %s, %s, %s)",
+                (task_id, task_no, "bi", task_params, "col", callback_url, "pending")
+            )
+            conn.commit()
+
+            import asyncio
+            asyncio.create_task(
+                self._process_batch_import(task_id, task_no, kb, parents, children, callback_url)
+            )
+
+            return task_id, "pending"
+
+        except Exception as e:
+            logger.error(f"创建入库任务失败: {e}")
+            conn.rollback()
+            return None, None
+        finally:
+            cursor.close()
+            conn.close()
+
+    async def _process_batch_import(
+        self,
+        task_id: str,
+        task_no: str,
+        kb: dict,
+        parents: list,
+        children: list,
+        callback_url: str
+    ):
+        """异步处理批量入库（后台执行）"""
+        conn = get_db_connection()
+        if not conn:
+            return
+
+        cursor = conn.cursor()
+        total = len(parents) + len(children)
+        succeeded = 0
+        failed = 0
+        failures = []
+
+        try:
+            cursor.execute(
+                "UPDATE t_samp_task_management SET status = %s, updated_time = NOW() WHERE task_id = %s",
+                ("processing", task_id)
+            )
+            conn.commit()
+
+            for item in parents:
+                try:
+                    self._insert_to_milvus(kb, item, is_parent=True)
+                    succeeded += 1
+                except Exception as e:
+                    failed += 1
+                    failures.append({
+                        "index": item.get("index", 0),
+                        "parent_id": item.get("parent_id"),
+                        "error": str(e)
+                    })
+
+            for item in children:
+                try:
+                    self._insert_to_milvus(kb, item, is_parent=False)
+                    succeeded += 1
+                except Exception as e:
+                    failed += 1
+                    failures.append({
+                        "index": item.get("index", 0),
+                        "parent_id": item.get("parent_id"),
+                        "error": str(e)
+                    })
+
+            cursor.execute(
+                "UPDATE t_samp_task_management "
+                "SET status = %s, error_message = %s, completed_time = NOW(), updated_time = NOW() "
+                "WHERE task_id = %s",
+                ("completed", json.dumps(failures, ensure_ascii=False) if failures else None, task_id)
+            )
+            conn.commit()
+
+            if callback_url:
+                await self._send_callback(callback_url, task_id, task_no, kb['id'], "completed", total, succeeded, failed, failures)
+
+        except Exception as e:
+            cursor.execute(
+                "UPDATE t_samp_task_management "
+                "SET status = %s, error_message = %s, completed_time = NOW(), updated_time = NOW() "
+                "WHERE task_id = %s",
+                ("failed", str(e), task_id)
+            )
+            conn.commit()
+
+            if callback_url:
+                await self._send_callback(callback_url, task_id, task_no, kb['id'], "failed", total, 0, 0, [])
+        finally:
+            cursor.close()
+            conn.close()
+
+    def _insert_to_milvus(self, kb: dict, item: dict, is_parent: bool):
+        """将单条数据写入Milvus"""
+        from app.services.milvus_service import milvus_service
+        import time as _time
+
+        coll_name = kb['collection_name_parent'] if is_parent else kb['collection_name_children']
+        if not coll_name:
+            raise ValueError("集合名称为空")
+
+        # 如果集合不存在，自动创建
+        milvus_service.ensure_collection_exists(coll_name)
+
+        text = item.get("text", "")
+        if not text:
+            raise ValueError("文本内容为空")
+
+        from app.base.embedding_connection import get_embedding_model
+        model = get_embedding_model()
+        vector = model.embed_query(text)
+
+        now_ms = int(_time.time() * 1000)
+        record = {
+            "text": text,
+            "dense": vector,
+            "document_id": str(item.get("doc_id", item.get("parent_id", ""))),
+            "parent_id": str(item.get("parent_id", "")),
+            "index": item.get("index", 0),
+            "hierarchy": item.get("hierarchy", ""),
+            "metadata": json.dumps(item.get("metadata", {}), ensure_ascii=False),
+            "tag_list": json.dumps(item.get("tag_list", []), ensure_ascii=False),
+            "permission": json.dumps(item.get("permission", {}), ensure_ascii=False),
+            "is_deleted": False,
+            "created_by": "api_import",
+            "created_time": now_ms,
+            "updated_by": "api_import",
+            "updated_time": now_ms,
+        }
+
+        milvus_service.client.insert(collection_name=coll_name, data=[record])
+
+    async def _send_callback(self, callback_url: str, task_id: str, task_no: str, kb_id: str, status: str, total: int, succeeded: int, failed: int, failures: list):
+        """发送回调通知"""
+        import httpx
+        payload = {
+            "task_id": task_id,
+            "task_no": task_no,
+            "kb_id": kb_id,
+            "status": status,
+            "progress": {
+                "total": total,
+                "processed": total,
+                "succeeded": succeeded,
+                "failed": failed
+            },
+            "failures": failures
+        }
+
+        max_retries = 3
+        delays = [10, 30, 60]
+
+        for i in range(max_retries):
+            try:
+                async with httpx.AsyncClient(timeout=10) as client:
+                    resp = await client.post(callback_url, json=payload)
+                    if resp.status_code == 200:
+                        return
+            except Exception as e:
+                logger.warning(f"回调失败(第{i+1}次): {e}")
+                if i < max_retries - 1:
+                    import asyncio
+                    await asyncio.sleep(delays[i])
+
+        logger.error(f"回调最终失败: {callback_url}")
+
+
+
+
+
+        
+
+    async def get_batch_import_task(self, task_id: str) -> Optional[Dict[str, Any]]:
+        """查询批量入库任务状态"""
+        conn = get_db_connection()
+        if not conn:
+            return None
+
+        cursor = conn.cursor()
+        try:
+            task_id = task_id.strip()
+
+            cursor.execute(
+                "SELECT task_id, task_no, status, task_params, error_message, "
+                "completed_time, created_time, updated_time "
+                "FROM t_samp_task_management WHERE task_id = %s",
+                (task_id,)
+            )
+            task = cursor.fetchone()
+            if not task:
+                return None
+
+            status = task['status']
+
+            params = json.loads(task['task_params']) if task['task_params'] else {}
+            total = len(params.get('parents', [])) + len(params.get('children', []))
+
+            failures = []
+            if task['error_message']:
+                try:
+                    failures = json.loads(task['error_message'])
+                    if not isinstance(failures, list):
+                        failures = []
+                except:
+                    failures = []
+
+            result = {
+                "task_id": task['task_id'],
+                "task_no": task['task_no'],
+                "status": status
+            }
+
+            if status in ('pending', 'processing'):
+                succeeded = max(0, total - len(failures))
+                result["progress"] = {
+                    "total": total,
+                    "processed": succeeded + len(failures),
+                    "succeeded": succeeded,
+                    "failed": len(failures)
+                }
+                result["created_at"] = task['created_time']
+                result["updated_at"] = task['updated_time']
+
+            elif status == 'completed':
+                succeeded = max(0, total - len(failures))
+                result["progress"] = {
+                    "total": total,
+                    "processed": total,
+                    "succeeded": succeeded,
+                    "failed": len(failures)
+                }
+                result["created_at"] = task['created_time']
+                result["completed_at"] = task['completed_time']
+                result["failures"] = failures
+
+            elif status == 'failed':
+                result["error"] = task['error_message'] if task['error_message'] else "任务执行失败"
+                result["created_at"] = task['created_time']
+                result["completed_at"] = task['completed_time']
+
+            return result
+
+        except Exception as e:
+            logger.error(f"查询任务状态失败: {e}")
+            return None
+        finally:
+            cursor.close()
+            conn.close()

src/app/services/api/token_api_service.py

@@ -0,0 +1,171 @@
+"""
+对外API Token认证业务逻辑
+"""
+import logging
+import secrets
+from typing import Optional, Tuple
+from datetime import datetime, timedelta, timezone
+from app.base.async_mysql_connection import get_db_connection
+from app.services.jwt_token import create_access_token, verify_token
+from app.core.config import config_handler
+
+logger = logging.getLogger(__name__)
+
+
+class TokenApiService:
+    """对外API Token认证服务"""
+
+    async def generate_token(
+        self,
+        app_id: str,
+        app_secret: str
+    ) -> Tuple[Optional[str], Optional[str], Optional[str]]:
+        """
+        验证应用凭证并生成访问令牌
+
+        Returns:
+            (access_token, error_code, error_message, expires_in)
+            成功时返回 (token, None, None, expires_in)
+            失败时返回 (None, error_code, error_message, None)
+        """
+        logger.info(f"[Token] generate_token called: app_id={app_id}, app_secret_len={len(app_secret) if app_secret else 0}")
+        conn = get_db_connection()
+        if not conn:
+            return None, "000500", "数据库连接失败", None
+
+        cursor = conn.cursor()
+        try:
+            # 1. 查询应用信息
+            cursor.execute(
+                "SELECT id, name, app_key, app_secret, is_active, access_token_expires "
+                "FROM t_sys_app WHERE app_key = %s",
+                (app_id,)
+            )
+            app = cursor.fetchone()
+            logger.info(f"[Token] 查询app结果: {app}")
+            if not app:
+                logger.warning(f"[Token] app不存在: app_id={app_id}")
+                return None, "000401", f"应用不存在: app_id={app_id}", None
+
+            # 2. 验证应用是否激活
+            if not app['is_active']:
+                logger.warning(f"[Token] app未激活: app_id={app_id}")
+                return None, "000401", f"应用已禁用: app_id={app_id}", None
+
+            # 3. 验证 app_secret (明文比较，strip 避免复制带入空白字符)
+            stored_secret = app['app_secret'].strip()
+            provided_secret = app_secret.strip()
+            if stored_secret != provided_secret:
+                logger.warning(f"[Token] app_secret不匹配: app_id={app_id}")
+                # 详细对比：找出第一个不同的字符位置
+                for i in range(min(len(stored_secret), len(provided_secret))):
+                    if stored_secret[i] != provided_secret[i]:
+                        logger.warning(f"[Token] 差异位置[{i}]: 存储值='{stored_secret[i]}'(0x{ord(stored_secret[i]):02X}), 提供值='{provided_secret[i]}'(0x{ord(provided_secret[i]):02X})")
+                        break
+                if len(stored_secret) != len(provided_secret):
+                    logger.warning(f"[Token] 长度差异: 存储值={len(stored_secret)}, 提供值={len(provided_secret)}")
+                return None, "000401", "app_secret不正确", None
+
+            # 4. 检查该应用当前有效 Token 数量（最多 3 个）
+            app_db_id = app['id']
+            cursor.execute(
+                "SELECT COUNT(*) as cnt FROM t_oauth_access_tokens "
+                "WHERE app_id = %s AND revoked = 0 AND expires_at > NOW()",
+                (app_db_id,)
+            )
+            active_count = cursor.fetchone()['cnt']
+            if active_count >= 3:
+                logger.warning(f"[Token] token数量已达上限: app_id={app_id}, count={active_count}")
+                return None, "000401", "该应用有效Token数量已达上限(3个)", None
+
+            # 5. 生成访问令牌
+            expires_in = app.get('access_token_expires') or 7200
+            token_data = {
+                "sub": app_db_id,
+                "app_id": app['app_key'],
+                "app_name": app['name'],
+                "type": "api"
+            }
+            access_token = create_access_token(token_data, expires_delta=timedelta(seconds=expires_in))
+
+            # 6. 存储令牌到数据库
+            expires_at = datetime.now(timezone.utc) + timedelta(seconds=expires_in)
+            cursor.execute(
+                "INSERT INTO t_oauth_access_tokens "
+                "(id, app_id, token, token_type, expires_at, scope, revoked) "
+                "VALUES (UUID(), %s, %s, 'Bearer', %s, 'api', 0)",
+                (app_db_id, access_token, expires_at)
+            )
+            conn.commit()
+
+            return access_token, None, None, expires_in
+
+        except Exception as e:
+            logger.error(f"生成Token失败: {e}")
+            conn.rollback()
+            return None, "000500", f"服务器内部错误: {str(e)}", None
+        finally:
+            cursor.close()
+            conn.close()
+
+    async def verify_api_token(self, token: str) -> Optional[dict]:
+        """
+        验证 API 访问令牌是否有效
+
+        1. 先校验 JWT 签名和有效期
+        2. 再查库确认 token 未被吊销（双保险）
+
+        Returns:
+            验证通过返回 payload dict，失败返回 None
+        """
+        # 先验证 JWT
+        payload = verify_token(token)
+        if not payload:
+            return None
+
+        # 再查库确认未被吊销且未过期
+        conn = get_db_connection()
+        if not conn:
+            return None
+        cursor = conn.cursor()
+        try:
+            cursor.execute(
+                "SELECT id, app_id, revoked FROM t_oauth_access_tokens WHERE token = %s",
+                (token,)
+            )
+            row = cursor.fetchone()
+            if not row or row['revoked']:
+                return None
+            return payload
+        except Exception as e:
+            logger.error(f"[Token] 验证API token失败: {e}")
+            return None
+        finally:
+            cursor.close()
+            conn.close()
+
+    async def revoke_token(self, app_id: str, token: str) -> bool:
+        """吊销指定Token"""
+        conn = get_db_connection()
+        if not conn:
+            return False
+
+        cursor = conn.cursor()
+        try:
+            # 验证token属于该app
+            cursor.execute(
+                "UPDATE t_oauth_access_tokens SET revoked = 1 "
+                "WHERE token = %s AND app_id IN ("
+                "  SELECT id FROM t_sys_app WHERE app_key = %s"
+                ")",
+                (token, app_id)
+            )
+            conn.commit()
+            return cursor.rowcount > 0
+        except Exception as e:
+            logger.error(f"吊销Token失败: {e}")
+            conn.rollback()
+            return False
+        finally:
+            cursor.close()
+            conn.close()

src/app/services/milvus_service.py

@@ -214,7 +214,7 @@ class MilvusService:
             "index": index,
             "tag_list": tags,
             "permission": {},
-            "is_deleted": 0,
+            "is_deleted": False,
             "created_by": user_id,
             "created_time": int(time.time() * 1000),
             "updated_by": user_id,
@@ -237,10 +237,11 @@ class MilvusService:
             schema.add_field("document_id", DataType.VARCHAR, max_length=256)
             schema.add_field("parent_id", DataType.VARCHAR, max_length=256)
             schema.add_field("index", DataType.INT64)
+            schema.add_field("hierarchy", DataType.VARCHAR, max_length=65535)
             schema.add_field("tag_list", DataType.VARCHAR, max_length=2048)
             schema.add_field("permission", DataType.JSON, nullable=True)
             schema.add_field("metadata", DataType.JSON, nullable=True)
-            schema.add_field("is_deleted", DataType.BOOL, default_value=0)
+            schema.add_field("is_deleted", DataType.BOOL, default_value=False)
             schema.add_field("created_by", DataType.VARCHAR, max_length=256, nullable=True)
             schema.add_field("created_time", DataType.INT64)
             schema.add_field("updated_by", DataType.VARCHAR, max_length=256, nullable=True)