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 的升级路径）。
72 lines
2.5 KiB
Python
72 lines
2.5 KiB
Python
"""素材路由（FR-ADM-03）：图像资产登记与审核。
|
||
|
||
素材由运营流水线高频登记（T-P2-07/15），审核是唯一能把 ``tryon_ready``
|
||
打开的入口——服务层已强制「tryon_ready 只能随 approved 打开」，这里只做
|
||
透传，约束在 HTTP 层的表现是 400（BAD_REQUEST）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
from fastapi import APIRouter, Depends, Query
|
||
|
||
from apps.admin.deps import KbAssetServiceDep, admin_auth
|
||
from packages.kb.schemas import AssetCreate, AssetOut, AssetReview
|
||
from packages.shared.models import KbAsset
|
||
from packages.shared.response import ok
|
||
|
||
router = APIRouter(prefix="/assets", tags=["admin:assets"], dependencies=[Depends(admin_auth)])
|
||
|
||
|
||
def _serialize(asset: KbAsset) -> dict[str, Any]:
|
||
return AssetOut(
|
||
asset_id=str(asset.asset_id),
|
||
entity_id=str(asset.entity_id),
|
||
asset_type=asset.asset_type,
|
||
cos_key=asset.cos_key,
|
||
spec_check=dict(asset.spec_check),
|
||
tryon_ready=asset.tryon_ready,
|
||
review_status=asset.review_status,
|
||
license=asset.license,
|
||
created_at=asset.created_at,
|
||
).model_dump()
|
||
|
||
|
||
@router.post("", summary="登记素材（登记时 tryon_ready 恒为 False，与请求体取值无关）")
|
||
async def register_asset(payload: AssetCreate, assets: KbAssetServiceDep) -> dict[str, Any]:
|
||
asset = await assets.register_asset(payload)
|
||
return ok(_serialize(asset))
|
||
|
||
|
||
@router.get("", summary="素材列表：entity_id / tryon_ready 过滤 + 分页")
|
||
async def list_assets(
|
||
assets: KbAssetServiceDep,
|
||
entity_id: int | None = Query(default=None, description="按实体过滤（雪花 ID）"),
|
||
tryon_ready: bool | None = Query(default=None, description="true=仅可换装素材"),
|
||
limit: int = Query(default=50, ge=1, le=200),
|
||
offset: int = Query(default=0, ge=0),
|
||
) -> dict[str, Any]:
|
||
rows, total = await assets.list_assets(
|
||
entity_id=entity_id, tryon_ready=tryon_ready, limit=limit, offset=offset
|
||
)
|
||
return ok(
|
||
{
|
||
"items": [_serialize(row) for row in rows],
|
||
"total": total,
|
||
"limit": limit,
|
||
"offset": offset,
|
||
}
|
||
)
|
||
|
||
|
||
@router.post(
|
||
"/{asset_id}/review",
|
||
summary="审核素材（tryon_ready=true 仅在 review_status=approved 时被接受）",
|
||
)
|
||
async def review_asset(
|
||
asset_id: int, payload: AssetReview, assets: KbAssetServiceDep
|
||
) -> dict[str, Any]:
|
||
asset = await assets.review_asset(asset_id, payload)
|
||
return ok(_serialize(asset))
|