xsl
3961b6fbfa
apps/admin 独立应用：X-Admin-Token 鉴权（空=全部路由不注册=404）、实体/规则
维护与状态机流转、发布编排（publish/archive → 规则缓存 invalidate +
retrieval invalidate_cache 双失效）、素材登记与审核（tryon_ready 门禁透传）、
缺口看板（T-P2-09，FR-ADM-01~05）。V1 共享令牌鉴权，无角色无审计字段，
操作留证靠 structlog request_id。
测试：单测 tests/unit/test_admin_app.py（10 例：空令牌 404、错/缺/非 ASCII
令牌 40100、正确令牌 200 假服务、发布编排接线、幂等重放无副作用）；
集成 tests/integration/test_admin_console.py（5 例：顾问全流程不碰客户端
API、双失效副作用可观察、非法跃迁 40900、素材审核门禁、缺口看板排序、
令牌 401）。483 单测 + 43 集成全绿；ruff/mypy/lint-imports 全绿。
偏差说明：
1. packages/retrieval.repo.list_gaps 有缺陷：对 select(KbGap) 的结果调
.mappings()，行键是实体名而非列名，GapSummary(row) 取 row.gap_id 必然
AttributeError（该路径无既有测试覆盖，接真库即暴露）。packages/** 本
任务禁改，故缺口查询在 apps/admin/routers/gaps.py 以等价 SQL 自实现
（hit_count 降序、同频次按 last_seen_at 降序、limit/offset 语义一致）；
建议合并阶段修复上游后切回 retrieval.list_gaps。
2. 实体 transition 的 target 收全量四态（任务文档示例仅列 reviewing/
published/archived）：按任务测试要求 published→draft 必须返回 40900
（状态机语义冲突）而非 schema 422；draft 作为目标由服务层跃迁矩阵恒拒。
3. 规则路由以 rule_key 为路径参数（任务文档写 /rules/{id}，但 RuleService
全程以 rule_key 寻址，无按数字 id 的接口，以代码为准）。
4. 发布编排触发面含 archived（撤稿）：依据 retrieval.invalidate_cache 的
自身契约「发布/归档知识后调用」，比文档字面的「发布成功后」多了撤稿
方向；reviewing 等其余状态不触发。
5. updated_by/author 记 "admin-console"：V1 共享令牌无操作者身份，字段
留渠道来源；运营账号体系属 V2（见包 docstring 的升级路径）。
133 lines
4.8 KiB
Python
133 lines
4.8 KiB
Python
"""知识实体路由（FR-ADM-01）：维护 + 状态机流转 + 图谱查看。
|
||
|
||
与客户端 ``apps.api`` 的 kb 路由关键差异：
|
||
|
||
- 鉴权走 ``X-Admin-Token``（共享令牌），操作者没有用户身份，``updated_by``
|
||
固定记录为 ``admin-console``——V1 的操作留证靠 request_id 关联访问日志；
|
||
- 多出 ``POST /items/{id}/transition``：把状态机流转从 PATCH 的「内容+状态
|
||
可混用」里独立出来，发布/撤稿动作在这里挂上跨域缓存失效编排（T-P2-10
|
||
的「自然失效」入口）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
from fastapi import APIRouter, Depends, Query
|
||
|
||
from apps.admin.deps import KbServiceDep, admin_auth
|
||
from apps.admin.publishing import CACHE_AFFECTING_TARGETS, invalidate_downstream_caches
|
||
from apps.admin.schemas import TransitionRequest
|
||
from packages.kb.schemas import (
|
||
MAX_GRAPH_HOPS,
|
||
EntityCreate,
|
||
EntityOut,
|
||
EntityStatus,
|
||
EntityType,
|
||
EntityUpdate,
|
||
)
|
||
from packages.shared.models import KbEntity
|
||
from packages.shared.response import ok
|
||
|
||
router = APIRouter(prefix="/kb", tags=["admin:kb"], dependencies=[Depends(admin_auth)])
|
||
|
||
#: V1 共享令牌鉴权没有操作者身份，字段统一记渠道来源（见模块 docstring）。
|
||
ADMIN_OPERATOR = "admin-console"
|
||
|
||
|
||
def _serialize(entity: KbEntity) -> dict[str, Any]:
|
||
return EntityOut(
|
||
entity_id=str(entity.entity_id),
|
||
entity_type=entity.entity_type,
|
||
name=entity.name,
|
||
attrs=dict(entity.attrs),
|
||
status=entity.status,
|
||
version=entity.version,
|
||
source=entity.source,
|
||
license=entity.license,
|
||
updated_by=entity.updated_by,
|
||
created_at=entity.created_at,
|
||
updated_at=entity.updated_at,
|
||
).model_dump()
|
||
|
||
|
||
@router.get("/items", summary="实体列表：entity_type / status 过滤 + 名称模糊 + 分页")
|
||
async def list_items(
|
||
kb: KbServiceDep,
|
||
entity_type: EntityType | None = Query(default=None, description="12 类本体之一"),
|
||
status: EntityStatus | None = Query(default=None),
|
||
name: str | None = Query(default=None, max_length=128, description="名称模糊匹配"),
|
||
limit: int = Query(default=20, ge=1, le=100),
|
||
offset: int = Query(default=0, ge=0),
|
||
) -> dict[str, Any]:
|
||
rows, total = await kb.search_entities(
|
||
entity_type=entity_type,
|
||
status=status,
|
||
name_ilike=name,
|
||
limit=limit,
|
||
offset=offset,
|
||
)
|
||
return ok(
|
||
{
|
||
"items": [_serialize(row) for row in rows],
|
||
"total": total,
|
||
"limit": limit,
|
||
"offset": offset,
|
||
}
|
||
)
|
||
|
||
|
||
@router.post("/items", summary="新建实体（draft / version=1 起步）")
|
||
async def create_item(payload: EntityCreate, kb: KbServiceDep) -> dict[str, Any]:
|
||
entity = await kb.create_entity(payload=payload, updated_by=ADMIN_OPERATOR)
|
||
return ok(_serialize(entity))
|
||
|
||
|
||
@router.get("/items/{entity_id}", summary="实体详情")
|
||
async def get_item(entity_id: int, kb: KbServiceDep) -> dict[str, Any]:
|
||
entity = await kb.get_entity(entity_id=entity_id)
|
||
return ok(_serialize(entity))
|
||
|
||
|
||
@router.patch("/items/{entity_id}", summary="修改内容（draft/reviewing；published 只读）")
|
||
async def update_item(entity_id: int, payload: EntityUpdate, kb: KbServiceDep) -> dict[str, Any]:
|
||
entity = await kb.update_entity(entity_id=entity_id, payload=payload, updated_by=ADMIN_OPERATOR)
|
||
return ok(_serialize(entity))
|
||
|
||
|
||
@router.post(
|
||
"/items/{entity_id}/transition",
|
||
summary="状态机流转：draft→reviewing→published→archived（发布/撤稿触发缓存失效编排）",
|
||
)
|
||
async def transition_item(
|
||
entity_id: int, payload: TransitionRequest, kb: KbServiceDep
|
||
) -> dict[str, Any]:
|
||
# 先读当前状态：与目标相同的重放是幂等 no-op（与 service 语义一致），
|
||
# 不触发失效编排——缓存重建是全量动作，重放不该有副作用。
|
||
current = await kb.get_entity(entity_id=entity_id)
|
||
if current.status == payload.target:
|
||
return ok(_serialize(current))
|
||
|
||
entity = await kb.update_entity(
|
||
entity_id=entity_id,
|
||
payload=EntityUpdate(status=payload.target),
|
||
updated_by=ADMIN_OPERATOR,
|
||
)
|
||
if payload.target in CACHE_AFFECTING_TARGETS:
|
||
invalidate_downstream_caches(
|
||
domain="kb_entity", key=str(entity.entity_id), target=payload.target
|
||
)
|
||
return ok(_serialize(entity))
|
||
|
||
|
||
@router.get("/items/{entity_id}/graph", summary="从该实体出发的 1~3 跳可解释路径")
|
||
async def get_item_graph(
|
||
entity_id: int,
|
||
kb: KbServiceDep,
|
||
max_hops: int = Query(
|
||
default=MAX_GRAPH_HOPS, ge=1, le=MAX_GRAPH_HOPS, description="遍历深度 1~3，超界 422"
|
||
),
|
||
) -> dict[str, Any]:
|
||
result = await kb.traverse_graph(entity_id=entity_id, max_hops=max_hops)
|
||
return ok(result.model_dump())
|