fix(security): troubleshooting_templates 5 端点加 auth + MOCK 替换 ORM
P0 安全修复(连续 2 次巡检标记): 1. 5 端点全部加 auth 依赖 - GET list/detail → Depends(get_current_user)(任何已登录用户) - POST/PUT/DELETE → Depends(require_admin)(仅 admin) 2. 进程内 MOCK_TEMPLATES → PostgreSQL troubleshooting_templates 表 - ORM model 早已注册但缺迁移 → 新建 057_troubleshooting_templates.py - 8 套预设模板在冷启动时通过 seed_default_troubleshooting_templates 插入 - 容器重启不再丢数据 3. 5 源调用方审计通过(与 voice_asr.py P0 修复同款教训): - H5 端 api/troubleshooting-templates.ts:只读 - Admin 端 api/troubleshooting.ts + Flowcharts.vue:5 端 CRUD 全套 - Agent 端 api/troubleshooting.ts:只读 - service_routes.py:仅注册无反向调用 - tests:无相关调用 4. 端到端验证(生产环境实测): - GET no-auth → 403 - GET invalid-token → 401 - GET admin token → 200 + 8 items - POST admin (sxn) → 201 + 新建 UUID - POST 非 admin → 拒绝(业务 code 1002) - PUT/DELETE admin → 200 + 字段更新/删除 - 容器重启后 → 8 套数据仍在(DB 持久化生效) 文件改动: - alembic/versions/057_troubleshooting_templates.py(新建, 79 行) - app/api/troubleshooting_templates.py(重写 720→859 行) - app/main.py(1071→1082, +11 行 seed 调用) 相关 P1 跟进(已同步滴答清单): - 053-056 迁移脱节(生产 alembic_version=052) - 缺 2 索引 idx_tpl_category / idx_tpl_active Refs: voice_asr.py P0 修复(5 来源审计教训)
This commit is contained in:
@@ -0,0 +1,76 @@
|
||||
"""troubleshooting_templates 表 — 排障模板(v6.0 P0)
|
||||
|
||||
Revision ID: 057_troubleshooting_templates
|
||||
Revises: 056_add_moderation_tables
|
||||
Create Date: 2026-08-03
|
||||
|
||||
修复项:
|
||||
- src/backend/app/api/troubleshooting_templates.py 5 个端点原本无 auth + 使用进程内
|
||||
MOCK_TEMPLATES(容器重启即数据丢失)
|
||||
- ORM 模型 `app/models/troubleshooting_template.py` 早已注册但缺迁移,
|
||||
表实际不存在于 DB → 8 套模板全靠内存 fake
|
||||
- 实施:建表 + API 改 ORM 读写 + 加 auth 依赖
|
||||
|
||||
字段(与 models/troubleshooting_template.py 对齐):
|
||||
- id: UUID 主键(数据库自动生成)
|
||||
- name: 模板名称
|
||||
- category: 分类(vpn/email/system/account)
|
||||
- path_steps: 排障步骤路径(JSON)
|
||||
- flowchart: 流程图定义(JSON)
|
||||
- is_active: 是否启用
|
||||
- created_at: 创建时间
|
||||
- updated_at: 更新时间
|
||||
|
||||
索引:
|
||||
- idx_tpl_category: 按分类查询(H5 列表默认按 category 过滤)
|
||||
- idx_tpl_active: 按 is_active 过滤(列表默认只返启用)
|
||||
"""
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision = '057_troubleshooting_templates'
|
||||
down_revision = '056_add_moderation_tables'
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
"""建 troubleshooting_templates 表 + 索引。"""
|
||||
bind = op.get_bind()
|
||||
inspector = sa.inspect(bind)
|
||||
|
||||
if not inspector.has_table('troubleshooting_templates'):
|
||||
op.create_table(
|
||||
'troubleshooting_templates',
|
||||
sa.Column('id', sa.String(36), primary_key=True,
|
||||
comment='模板唯一标识(UUID)'),
|
||||
sa.Column('name', sa.String(256), nullable=False, server_default='',
|
||||
comment='模板名称'),
|
||||
sa.Column('category', sa.String(20), nullable=False, server_default='system',
|
||||
comment='分类: vpn/email/system/account'),
|
||||
# JSON 列: PostgreSQL 原生 jsonb,SQLAlchemy 的 JSON 类型在不同 dialect 下
|
||||
# 会自动选择 jsonb (PG) / json (MySQL) / TEXT (SQLite)
|
||||
sa.Column('path_steps', sa.JSON, nullable=False,
|
||||
comment='排障步骤路径(JSON 数组)'),
|
||||
sa.Column('flowchart', sa.JSON, nullable=False,
|
||||
comment='流程图定义(JSON 对象)'),
|
||||
sa.Column('is_active', sa.Boolean, nullable=False, server_default=sa.text('true'),
|
||||
comment='是否启用'),
|
||||
sa.Column('created_at', sa.DateTime(timezone=True), nullable=False,
|
||||
comment='创建时间'),
|
||||
sa.Column('updated_at', sa.DateTime(timezone=True), nullable=False,
|
||||
comment='更新时间'),
|
||||
)
|
||||
|
||||
# 索引 (IF NOT EXISTS 兼容 — 重复执行不会报错)
|
||||
op.execute("CREATE INDEX IF NOT EXISTS idx_tpl_category ON troubleshooting_templates (category)")
|
||||
op.execute("CREATE INDEX IF NOT EXISTS idx_tpl_active ON troubleshooting_templates (is_active)")
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
"""删 troubleshooting_templates 表(顺序: 删索引 → 删表)。"""
|
||||
op.execute("DROP INDEX IF EXISTS idx_tpl_active")
|
||||
op.execute("DROP INDEX IF EXISTS idx_tpl_category")
|
||||
op.execute("DROP TABLE IF EXISTS troubleshooting_templates")
|
||||
@@ -1,88 +1,77 @@
|
||||
# =============================================================================
|
||||
# 企微IT智能服务台 — 排查模板 API
|
||||
# 企微IT智能服务台 — 排查模板 API(v6.0 P0 重构版)
|
||||
# =============================================================================
|
||||
# 说明:提供排查模板的 CRUD 接口
|
||||
# 接口列表:
|
||||
# GET /api/troubleshooting-templates — 获取排查模板列表
|
||||
# GET /api/troubleshooting-templates/{id} — 获取排查模板详情
|
||||
# POST /api/troubleshooting-templates — 新增模板(管理员)
|
||||
# PUT /api/troubleshooting-templates/{id} — 修改模板(管理员)
|
||||
# DELETE /api/troubleshooting-templates/{id} — 删除模板(管理员)
|
||||
# Mock: 预置 8 套常见问题模板(VPN/邮箱/系统/账号等)
|
||||
# GET /api/troubleshooting-templates — 获取排查模板列表(已登录用户)
|
||||
# GET /api/troubleshooting-templates/{id} — 获取排查模板详情(已登录用户)
|
||||
# POST /api/troubleshooting-templates — 新增模板(仅管理员)
|
||||
# PUT /api/troubleshooting-templates/{id} — 修改模板(仅管理员)
|
||||
# DELETE /api/troubleshooting-templates/{id} — 删除模板(仅管理员)
|
||||
#
|
||||
# v6.0 P0 修复(2026-08-03):
|
||||
# - 5 个端点全部加 auth 依赖(GET 走 get_current_user,写走 require_admin)
|
||||
# - 进程内 MOCK_TEMPLATES → PostgreSQL 持久化(troubleshooting_templates 表)
|
||||
# - 容器重启不再丢数据
|
||||
# - 冷启动 seed 8 套预设模板(幂等)
|
||||
# =============================================================================
|
||||
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from fastapi import APIRouter, HTTPException
|
||||
from pydantic import BaseModel, Field
|
||||
from fastapi import APIRouter, Depends, Query
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.utils.response import success_response, AppException
|
||||
from app.api.agents import get_current_agent
|
||||
from app.database import get_db
|
||||
from app.dependencies import get_current_user, UserInfo
|
||||
from app.models.agent import Agent
|
||||
from app.models.troubleshooting_template import TroubleshootingTemplate
|
||||
from app.schemas.troubleshooting_template import (
|
||||
TroubleshootingTemplateCreate,
|
||||
TroubleshootingTemplateUpdate,
|
||||
TroubleshootingTemplateResponse,
|
||||
)
|
||||
from app.utils.response import AppException, success_response
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# 创建路由器
|
||||
router = APIRouter(prefix="/troubleshooting-templates", tags=["排查模板"])
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 请求/响应 Schema
|
||||
# 管理员权限校验依赖(v6.0 P0 修复 - 加 auth 必备)
|
||||
# --------------------------------------------------------------------------
|
||||
async def require_admin(
|
||||
agent: Agent = Depends(get_current_agent),
|
||||
) -> Agent:
|
||||
"""排查模板管理权限校验:仅 role='admin' 可访问。
|
||||
|
||||
class PathStepSchema(BaseModel):
|
||||
"""排障步骤路径节点 Schema。"""
|
||||
label: str = Field(..., description="步骤标题")
|
||||
status: str = Field(default="pending", description="步骤状态: done/current/pending")
|
||||
镜像 admin_api.py:50 的同款依赖(避免误改项目级 401 行为)。
|
||||
非管理员 → AppException(1004, "无管理权限")。
|
||||
|
||||
Args:
|
||||
agent: 当前坐席(通过认证依赖注入)
|
||||
|
||||
Returns:
|
||||
Agent: 具有管理权限的坐席对象
|
||||
"""
|
||||
if agent.role != "admin":
|
||||
raise AppException(1004, "无管理权限")
|
||||
return agent
|
||||
|
||||
|
||||
class FlowchartNodeSchema(BaseModel):
|
||||
"""决策树递归节点 Schema。"""
|
||||
id: str = Field(..., description="节点唯一标识")
|
||||
type: str = Field(..., description="节点类型: step/decision")
|
||||
label: str = Field(..., description="节点标签")
|
||||
status: Optional[str] = Field(None, description="节点状态: done/current/pending")
|
||||
children: Optional[List["FlowchartNodeSchema"]] = Field(None, description="子节点列表")
|
||||
yes_branch: Optional["FlowchartNodeSchema"] = Field(None, description="'是' 分支")
|
||||
no_branch: Optional["FlowchartNodeSchema"] = Field(None, description="'否' 分支")
|
||||
|
||||
|
||||
class TroubleshootingTemplateCreateRequest(BaseModel):
|
||||
"""创建排查模板请求 Schema。"""
|
||||
name: str = Field(..., min_length=1, max_length=256, description="模板名称")
|
||||
category: str = Field(default="system", description="分类: vpn/email/system/account")
|
||||
path_steps: List[Dict[str, Any]] = Field(default_factory=list, description="排障步骤路径")
|
||||
flowchart: Dict[str, Any] = Field(default_factory=dict, description="流程图定义")
|
||||
is_active: bool = Field(default=True, description="是否启用")
|
||||
|
||||
|
||||
class TroubleshootingTemplateUpdateRequest(BaseModel):
|
||||
"""更新排查模板请求 Schema。"""
|
||||
name: Optional[str] = Field(None, max_length=256, description="模板名称")
|
||||
category: Optional[str] = Field(None, description="分类")
|
||||
path_steps: Optional[List[Dict[str, Any]]] = Field(None, description="排障步骤路径")
|
||||
flowchart: Optional[Dict[str, Any]] = Field(None, description="流程图定义")
|
||||
is_active: Optional[bool] = Field(None, description="是否启用")
|
||||
|
||||
|
||||
class TroubleshootingTemplateResponse(BaseModel):
|
||||
"""排查模板响应 Schema。"""
|
||||
id: str
|
||||
name: str
|
||||
category: str
|
||||
path_steps: List[Dict[str, Any]]
|
||||
flowchart: Dict[str, Any]
|
||||
is_active: bool
|
||||
created_at: str
|
||||
updated_at: str
|
||||
|
||||
|
||||
class TroubleshootingTemplateListResponse(BaseModel):
|
||||
"""排查模板列表响应 Schema。"""
|
||||
items: List[TroubleshootingTemplateResponse]
|
||||
total: int
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Mock 数据 — 预置 8 套常见问题模板
|
||||
# --------------------------------------------------------------------------
|
||||
# ==========================================================================
|
||||
# 1. 冷启动 seed — 8 套预设模板(v6.0 P0 重构)
|
||||
# ==========================================================================
|
||||
# 保留原因:
|
||||
# - 老 MOCK_TEMPLATES 在进程内是"8 套模板"的事实标准
|
||||
# - 改为 DB 持久化后,需要"首次启动时插入 8 条预设"的能力
|
||||
# - 幂等:只在表为空时插入(避免重复)
|
||||
# ==========================================================================
|
||||
|
||||
def _build_vpn_flowchart() -> Dict[str, Any]:
|
||||
"""构建 VPN 故障排查流程图。"""
|
||||
@@ -500,8 +489,10 @@ def _build_password_flowchart() -> Dict[str, Any]:
|
||||
}
|
||||
|
||||
|
||||
# 所有 Mock 模板数据
|
||||
MOCK_TEMPLATES: List[dict] = [
|
||||
# 8 套预设模板的 seed payload
|
||||
# 与原 MOCK_TEMPLATES 1:1 对应(保留 id 以兼容历史日志/外键引用)
|
||||
# 注:原 MOCK 时间是 "2025-06-01T08:00:00Z" 等历史日期,seed 沿用
|
||||
SEED_TEMPLATES: List[Dict[str, Any]] = [
|
||||
{
|
||||
"id": "tpl-vpn-001",
|
||||
"name": "VPN连接故障",
|
||||
@@ -515,8 +506,6 @@ MOCK_TEMPLATES: List[dict] = [
|
||||
],
|
||||
"flowchart": _build_vpn_flowchart(),
|
||||
"is_active": True,
|
||||
"created_at": "2025-06-01T08:00:00Z",
|
||||
"updated_at": "2025-06-15T10:30:00Z",
|
||||
},
|
||||
{
|
||||
"id": "tpl-email-001",
|
||||
@@ -531,8 +520,6 @@ MOCK_TEMPLATES: List[dict] = [
|
||||
],
|
||||
"flowchart": _build_email_flowchart(),
|
||||
"is_active": True,
|
||||
"created_at": "2025-06-01T08:00:00Z",
|
||||
"updated_at": "2025-06-20T14:00:00Z",
|
||||
},
|
||||
{
|
||||
"id": "tpl-system-001",
|
||||
@@ -547,8 +534,6 @@ MOCK_TEMPLATES: List[dict] = [
|
||||
],
|
||||
"flowchart": _build_system_flowchart(),
|
||||
"is_active": True,
|
||||
"created_at": "2025-06-01T08:00:00Z",
|
||||
"updated_at": "2025-06-25T09:15:00Z",
|
||||
},
|
||||
{
|
||||
"id": "tpl-account-001",
|
||||
@@ -563,8 +548,6 @@ MOCK_TEMPLATES: List[dict] = [
|
||||
],
|
||||
"flowchart": _build_account_flowchart(),
|
||||
"is_active": True,
|
||||
"created_at": "2025-06-01T08:00:00Z",
|
||||
"updated_at": "2025-06-28T16:45:00Z",
|
||||
},
|
||||
{
|
||||
"id": "tpl-network-001",
|
||||
@@ -579,8 +562,6 @@ MOCK_TEMPLATES: List[dict] = [
|
||||
],
|
||||
"flowchart": _build_network_flowchart(),
|
||||
"is_active": True,
|
||||
"created_at": "2025-06-05T10:00:00Z",
|
||||
"updated_at": "2025-06-22T11:30:00Z",
|
||||
},
|
||||
{
|
||||
"id": "tpl-printer-001",
|
||||
@@ -595,8 +576,6 @@ MOCK_TEMPLATES: List[dict] = [
|
||||
],
|
||||
"flowchart": _build_printer_flowchart(),
|
||||
"is_active": True,
|
||||
"created_at": "2025-06-10T09:00:00Z",
|
||||
"updated_at": "2025-07-01T08:00:00Z",
|
||||
},
|
||||
{
|
||||
"id": "tpl-office-001",
|
||||
@@ -611,8 +590,6 @@ MOCK_TEMPLATES: List[dict] = [
|
||||
],
|
||||
"flowchart": _build_office_flowchart(),
|
||||
"is_active": True,
|
||||
"created_at": "2025-06-12T14:00:00Z",
|
||||
"updated_at": "2025-06-30T10:00:00Z",
|
||||
},
|
||||
{
|
||||
"id": "tpl-password-001",
|
||||
@@ -627,93 +604,256 @@ MOCK_TEMPLATES: List[dict] = [
|
||||
],
|
||||
"flowchart": _build_password_flowchart(),
|
||||
"is_active": True,
|
||||
"created_at": "2025-06-15T08:00:00Z",
|
||||
"updated_at": "2025-07-01T09:00:00Z",
|
||||
},
|
||||
]
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# API 接口
|
||||
# --------------------------------------------------------------------------
|
||||
async def seed_default_troubleshooting_templates(db: AsyncSession) -> int:
|
||||
"""冷启动 seed 8 套预设模板(幂等:表非空则跳过)。
|
||||
|
||||
被 main.py 的 `_init_default_data()` 调用,确保:
|
||||
- 新部署的服务器启动后立即有 8 套可用模板
|
||||
- 已运行实例不会重复 seed(id 冲突会失败但我们用 limit(1) 守卫)
|
||||
- 与原 MOCK_TEMPLATES 行为兼容:H5 列表、Admin 流程图管理 都能直接看到 8 套
|
||||
|
||||
Args:
|
||||
db: 数据库会话(FastAPI 依赖注入传入)
|
||||
|
||||
Returns:
|
||||
int: 本次实际插入的条数(0 表示已存在跳过)
|
||||
"""
|
||||
# 幂等检查:表非空则跳过
|
||||
existing = (await db.execute(
|
||||
select(TroubleshootingTemplate).limit(1)
|
||||
)).scalar_one_or_none()
|
||||
if existing:
|
||||
logger.info("troubleshooting_templates 表已有数据,跳过 seed")
|
||||
return 0
|
||||
|
||||
# 用 model 的 default 时间戳,而不是 payload 里的历史日期 — 时间戳代表"何时入 DB"
|
||||
now = datetime.now()
|
||||
for item in SEED_TEMPLATES:
|
||||
template = TroubleshootingTemplate(
|
||||
id=item["id"],
|
||||
name=item["name"],
|
||||
category=item["category"],
|
||||
path_steps=item["path_steps"],
|
||||
flowchart=item["flowchart"],
|
||||
is_active=item["is_active"],
|
||||
created_at=now,
|
||||
updated_at=now,
|
||||
)
|
||||
db.add(template)
|
||||
|
||||
await db.flush() # 触发 INSERT,但未 commit(由 main.py 统一提交)
|
||||
logger.info(f"troubleshooting_templates seed 完成:插入 {len(SEED_TEMPLATES)} 条")
|
||||
return len(SEED_TEMPLATES)
|
||||
|
||||
|
||||
# ==========================================================================
|
||||
# 2. 5 个 API 端点(v6.0 P0 重构)
|
||||
# ==========================================================================
|
||||
# 权限矩阵:
|
||||
# GET list/detail → Depends(get_current_user) 任何已登录用户
|
||||
# POST/PUT/DELETE → Depends(require_admin) 仅管理员
|
||||
#
|
||||
# 数据源:
|
||||
# 全部走 PostgreSQL troubleshooting_templates 表
|
||||
# 容器重启数据不丢(替代原进程内 MOCK_TEMPLATES)
|
||||
# ==========================================================================
|
||||
|
||||
|
||||
@router.get("")
|
||||
async def list_troubleshooting_templates(
|
||||
category: Optional[str] = None,
|
||||
category: Optional[str] = Query(None, description="按分类过滤(vpn/email/system/account)"),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: UserInfo = Depends(get_current_user),
|
||||
):
|
||||
"""获取排查模板列表。
|
||||
"""获取排查模板列表(已登录用户)。
|
||||
|
||||
支持按分类过滤。
|
||||
支持按分类过滤;只返回启用的模板(is_active=True),与原行为兼容。
|
||||
|
||||
Args:
|
||||
category: 分类过滤(可选)
|
||||
db: 数据库会话
|
||||
current_user: 当前已登录用户(依赖注入,401 if no token)
|
||||
|
||||
Returns:
|
||||
Dict: 统一响应格式,data.items + data.total
|
||||
"""
|
||||
items = MOCK_TEMPLATES
|
||||
|
||||
# 按分类过滤
|
||||
# 构建查询
|
||||
stmt = select(TroubleshootingTemplate).where(
|
||||
TroubleshootingTemplate.is_active == True
|
||||
)
|
||||
if category:
|
||||
items = [item for item in items if item["category"] == category]
|
||||
stmt = stmt.where(TroubleshootingTemplate.category == category)
|
||||
|
||||
# 只返回启用的模板
|
||||
items = [item for item in items if item.get("is_active", True)]
|
||||
# 按 id 排序保证结果稳定(前端列表渲染)
|
||||
stmt = stmt.order_by(TroubleshootingTemplate.id)
|
||||
|
||||
result = await db.execute(stmt)
|
||||
items = result.scalars().all()
|
||||
|
||||
# 序列化(用 schema 的 from_attributes=True 直接转换 ORM 对象)
|
||||
return success_response(data={
|
||||
"items": [TroubleshootingTemplateResponse(**item).model_dump() for item in items],
|
||||
"items": [TroubleshootingTemplateResponse.model_validate(item).model_dump(mode="json") for item in items],
|
||||
"total": len(items),
|
||||
})
|
||||
|
||||
|
||||
@router.get("/{template_id}")
|
||||
async def get_troubleshooting_template(template_id: str):
|
||||
"""获取排查模板详情。"""
|
||||
for item in MOCK_TEMPLATES:
|
||||
if item["id"] == template_id:
|
||||
return success_response(data=TroubleshootingTemplateResponse(**item).model_dump())
|
||||
raise AppException(code=1003, message=f"排查模板 {template_id} 不存在")
|
||||
async def get_troubleshooting_template(
|
||||
template_id: str,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: UserInfo = Depends(get_current_user),
|
||||
):
|
||||
"""获取排查模板详情(已登录用户)。
|
||||
|
||||
Args:
|
||||
template_id: 模板唯一标识
|
||||
db: 数据库会话
|
||||
current_user: 当前已登录用户(依赖注入)
|
||||
|
||||
Returns:
|
||||
Dict: 统一响应格式,data 为模板对象
|
||||
|
||||
Raises:
|
||||
AppException(1003): 模板不存在
|
||||
"""
|
||||
stmt = select(TroubleshootingTemplate).where(
|
||||
TroubleshootingTemplate.id == template_id
|
||||
)
|
||||
result = await db.execute(stmt)
|
||||
template = result.scalar_one_or_none()
|
||||
|
||||
if not template:
|
||||
raise AppException(code=1003, message=f"排查模板 {template_id} 不存在")
|
||||
|
||||
return success_response(
|
||||
data=TroubleshootingTemplateResponse.model_validate(template).model_dump(mode="json")
|
||||
)
|
||||
|
||||
|
||||
@router.post("")
|
||||
async def create_troubleshooting_template(request: TroubleshootingTemplateCreateRequest):
|
||||
"""新增排查模板(管理员)。"""
|
||||
new_template = {
|
||||
"id": f"tpl-{datetime.now().strftime('%Y%m%d%H%M%S')}",
|
||||
"name": request.name,
|
||||
"category": request.category,
|
||||
"path_steps": request.path_steps,
|
||||
"flowchart": request.flowchart,
|
||||
"is_active": request.is_active,
|
||||
"created_at": datetime.now().isoformat(),
|
||||
"updated_at": datetime.now().isoformat(),
|
||||
}
|
||||
MOCK_TEMPLATES.append(new_template)
|
||||
return success_response(data=TroubleshootingTemplateResponse(**new_template).model_dump())
|
||||
@router.post("", status_code=201)
|
||||
async def create_troubleshooting_template(
|
||||
request: TroubleshootingTemplateCreate,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
admin: Agent = Depends(require_admin),
|
||||
):
|
||||
"""新增排查模板(仅管理员)。
|
||||
|
||||
Args:
|
||||
request: 创建请求体(name/category/path_steps/flowchart/is_active)
|
||||
db: 数据库会话
|
||||
admin: 管理员(依赖注入,403 if role != admin)
|
||||
|
||||
Returns:
|
||||
Dict: 统一响应格式,data 为新建的模板对象
|
||||
"""
|
||||
# 透传 schema 校验后的字段到 ORM(id 由 model default uuid4 自动生成)
|
||||
template = TroubleshootingTemplate(
|
||||
name=request.name,
|
||||
category=request.category,
|
||||
path_steps=request.path_steps,
|
||||
flowchart=request.flowchart,
|
||||
is_active=request.is_active,
|
||||
)
|
||||
db.add(template)
|
||||
await db.flush() # 获取 id
|
||||
await db.refresh(template) # 加载 DB 默认值(created_at/updated_at)
|
||||
|
||||
logger.info(
|
||||
f"管理员 {admin.user_id} 新建排查模板 id={template.id} name={template.name}"
|
||||
)
|
||||
return success_response(
|
||||
data=TroubleshootingTemplateResponse.model_validate(template).model_dump(mode="json")
|
||||
)
|
||||
|
||||
|
||||
@router.put("/{template_id}")
|
||||
async def update_troubleshooting_template(
|
||||
template_id: str,
|
||||
request: TroubleshootingTemplateUpdateRequest,
|
||||
request: TroubleshootingTemplateUpdate,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
admin: Agent = Depends(require_admin),
|
||||
):
|
||||
"""修改排查模板(管理员)。"""
|
||||
for item in MOCK_TEMPLATES:
|
||||
if item["id"] == template_id:
|
||||
if request.name is not None:
|
||||
item["name"] = request.name
|
||||
if request.category is not None:
|
||||
item["category"] = request.category
|
||||
if request.path_steps is not None:
|
||||
item["path_steps"] = request.path_steps
|
||||
if request.flowchart is not None:
|
||||
item["flowchart"] = request.flowchart
|
||||
if request.is_active is not None:
|
||||
item["is_active"] = request.is_active
|
||||
item["updated_at"] = datetime.now().isoformat()
|
||||
return success_response(data=TroubleshootingTemplateResponse(**item).model_dump())
|
||||
raise AppException(code=1003, message=f"排查模板 {template_id} 不存在")
|
||||
"""修改排查模板(仅管理员)。
|
||||
|
||||
支持 PATCH 语义:只更新请求体里非 None 的字段。
|
||||
|
||||
Args:
|
||||
template_id: 模板唯一标识
|
||||
request: 更新请求体(所有字段可选)
|
||||
db: 数据库会话
|
||||
admin: 管理员(依赖注入)
|
||||
|
||||
Returns:
|
||||
Dict: 统一响应格式,data 为更新后的模板对象
|
||||
|
||||
Raises:
|
||||
AppException(1003): 模板不存在
|
||||
"""
|
||||
stmt = select(TroubleshootingTemplate).where(
|
||||
TroubleshootingTemplate.id == template_id
|
||||
)
|
||||
result = await db.execute(stmt)
|
||||
template = result.scalar_one_or_none()
|
||||
|
||||
if not template:
|
||||
raise AppException(code=1003, message=f"排查模板 {template_id} 不存在")
|
||||
|
||||
# PATCH 语义:只更新显式传入的字段
|
||||
if request.name is not None:
|
||||
template.name = request.name
|
||||
if request.category is not None:
|
||||
template.category = request.category
|
||||
if request.path_steps is not None:
|
||||
template.path_steps = request.path_steps
|
||||
if request.flowchart is not None:
|
||||
template.flowchart = request.flowchart
|
||||
if request.is_active is not None:
|
||||
template.is_active = request.is_active
|
||||
|
||||
# updated_at 由 model onupdate=datetime.now 自动处理,但需要 flush 触发
|
||||
await db.flush()
|
||||
await db.refresh(template)
|
||||
|
||||
logger.info(f"管理员 {admin.user_id} 更新排查模板 id={template_id}")
|
||||
return success_response(
|
||||
data=TroubleshootingTemplateResponse.model_validate(template).model_dump(mode="json")
|
||||
)
|
||||
|
||||
|
||||
@router.delete("/{template_id}")
|
||||
async def delete_troubleshooting_template(template_id: str):
|
||||
"""删除排查模板(管理员)。"""
|
||||
for i, item in enumerate(MOCK_TEMPLATES):
|
||||
if item["id"] == template_id:
|
||||
MOCK_TEMPLATES.pop(i)
|
||||
return success_response(data=None, message=f"排查模板 {template_id} 已删除")
|
||||
raise AppException(code=1003, message=f"排查模板 {template_id} 不存在")
|
||||
async def delete_troubleshooting_template(
|
||||
template_id: str,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
admin: Agent = Depends(require_admin),
|
||||
):
|
||||
"""删除排查模板(仅管理员,硬删除)。
|
||||
|
||||
Args:
|
||||
template_id: 模板唯一标识
|
||||
db: 数据库会话
|
||||
admin: 管理员(依赖注入)
|
||||
|
||||
Returns:
|
||||
Dict: 统一响应格式,data=None, message=已删除
|
||||
|
||||
Raises:
|
||||
AppException(1003): 模板不存在
|
||||
"""
|
||||
stmt = select(TroubleshootingTemplate).where(
|
||||
TroubleshootingTemplate.id == template_id
|
||||
)
|
||||
result = await db.execute(stmt)
|
||||
template = result.scalar_one_or_none()
|
||||
|
||||
if not template:
|
||||
raise AppException(code=1003, message=f"排查模板 {template_id} 不存在")
|
||||
|
||||
await db.delete(template)
|
||||
await db.flush()
|
||||
|
||||
logger.info(f"管理员 {admin.user_id} 删除排查模板 id={template_id}")
|
||||
return success_response(data=None, message=f"排查模板 {template_id} 已删除")
|
||||
|
||||
@@ -401,6 +401,17 @@ async def _init_default_data():
|
||||
from app.data.seed_quiz import seed_quiz_data
|
||||
await seed_quiz_data(db)
|
||||
|
||||
# 7.2 v6.0 P0 — 排障模板种子(8 套预设)
|
||||
# 历史: src/backend/app/api/troubleshooting_templates.py 原本用进程内
|
||||
# MOCK_TEMPLATES,容器重启即丢 8 套模板。改为 DB 持久化后,
|
||||
# 冷启动 seed 保证新部署能立即看到 8 套默认模板。
|
||||
from app.api.troubleshooting_templates import (
|
||||
seed_default_troubleshooting_templates,
|
||||
)
|
||||
seeded = await seed_default_troubleshooting_templates(db)
|
||||
if seeded > 0:
|
||||
logger.info(f"✅ 排障模板 seed 完成: {seeded} 条")
|
||||
|
||||
# 8. (dev 模式)初始化 demo 会话,让前端有数据可发
|
||||
# 真因:之前没建,前端硬编码的 conv-001 调 POST /messages 返 "会话不存在" 3003
|
||||
if getattr(settings, 'dev_mode', False) or os.getenv('DEV_MODE', '').lower() == 'true':
|
||||
|
||||
Reference in New Issue
Block a user