138 lines
5.2 KiB
Python
138 lines
5.2 KiB
Python
"""HTML 模板管理 API(模板管理页)。
|
|
|
|
后端存储 :mod:`deerflow.runtime.scheduler.template_store`(文件型,
|
|
``{base_dir}/html-templates/<id>/``),启动时把随包内置模板 seed 进去(缺失才复制)。
|
|
读对所有登录用户开放(定时任务表单下拉也要用),写(创建/编辑/上传覆盖/删除/重置)
|
|
仅管理员——与敏感词管理同一鉴权口径。旧端点 ``GET /api/scheduled-tasks/html-templates``
|
|
保持不变(内部同样读存储),本路由是它的管理超集。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, HTTPException, Request
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app.gateway.catalog_cache import cached_catalog, catalog_key, invalidate_catalog
|
|
from app.gateway.deps import get_optional_user_from_request
|
|
from deerflow.runtime.scheduler import template_store
|
|
|
|
router = APIRouter(prefix="/api/html-templates", tags=["html-templates"])
|
|
|
|
|
|
async def _require_admin(request: Request) -> None:
|
|
"""写操作仅管理员;关闭鉴权时(request 无用户)放行,与 scheduled_tasks 口径一致。"""
|
|
user = await get_optional_user_from_request(request)
|
|
if user is None:
|
|
return
|
|
if getattr(user, "system_role", None) != "admin":
|
|
raise HTTPException(status_code=403, detail="仅管理员可管理 HTML 模板")
|
|
|
|
|
|
def _not_found(template_id: str) -> HTTPException:
|
|
return HTTPException(status_code=404, detail=f"模板不存在：{template_id}")
|
|
|
|
|
|
class TemplateCreateRequest(BaseModel):
|
|
template_id: str = Field(..., description="模板 id（字母/数字/./-/_，创建后不可改）")
|
|
name: str = Field(..., description="显示名称，如 no.28")
|
|
description: str = ""
|
|
html: str = Field(..., description="完整 HTML 模板正文")
|
|
|
|
|
|
class TemplateUpdateRequest(BaseModel):
|
|
name: str | None = None
|
|
description: str | None = None
|
|
html: str | None = Field(default=None, description="新的模板正文（上传覆盖走这里）")
|
|
|
|
|
|
@router.get("")
|
|
async def list_html_templates(request: Request) -> dict:
|
|
"""模板清单(含 builtin/convertible/size/时间),供管理页与定时任务表单共用。"""
|
|
async def load() -> dict:
|
|
return {"templates": await asyncio.to_thread(template_store.list_templates)}
|
|
|
|
return await cached_catalog(request, "html_templates", "admin-list", load)
|
|
|
|
|
|
@router.get("/{template_id}")
|
|
async def get_html_template(template_id: str, request: Request) -> dict[str, Any]:
|
|
async def load() -> dict[str, Any]:
|
|
meta = await asyncio.to_thread(template_store.get_template_meta, template_id)
|
|
if meta is None:
|
|
raise _not_found(template_id)
|
|
return meta
|
|
|
|
return await cached_catalog(request, "html_templates", catalog_key("meta", template_id), load)
|
|
|
|
|
|
@router.get("/{template_id}/content")
|
|
async def get_html_template_content(template_id: str, request: Request) -> dict:
|
|
async def load() -> dict:
|
|
try:
|
|
html = await asyncio.to_thread(template_store.read_template_html, template_id)
|
|
except template_store.TemplateStoreError:
|
|
raise _not_found(template_id)
|
|
return {"template_id": template_id, "html": html}
|
|
|
|
return await cached_catalog(request, "html_templates", catalog_key("content", template_id), load)
|
|
|
|
|
|
@router.post("")
|
|
async def create_html_template(body: TemplateCreateRequest, request: Request) -> dict[str, Any]:
|
|
await _require_admin(request)
|
|
try:
|
|
meta = template_store.create_template(
|
|
template_id=body.template_id,
|
|
name=body.name,
|
|
description=body.description,
|
|
html=body.html,
|
|
)
|
|
except template_store.TemplateStoreError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
await invalidate_catalog(request, "html_templates")
|
|
return meta
|
|
|
|
|
|
@router.put("/{template_id}")
|
|
async def update_html_template(
|
|
template_id: str, body: TemplateUpdateRequest, request: Request
|
|
) -> dict[str, Any]:
|
|
await _require_admin(request)
|
|
try:
|
|
meta = template_store.update_template(
|
|
template_id,
|
|
name=body.name,
|
|
description=body.description,
|
|
html=body.html,
|
|
)
|
|
except template_store.TemplateStoreError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
await invalidate_catalog(request, "html_templates")
|
|
return meta
|
|
|
|
|
|
@router.delete("/{template_id}")
|
|
async def delete_html_template(template_id: str, request: Request) -> dict:
|
|
await _require_admin(request)
|
|
try:
|
|
template_store.delete_template(template_id)
|
|
except template_store.TemplateStoreError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
await invalidate_catalog(request, "html_templates")
|
|
return {"ok": True, "template_id": template_id}
|
|
|
|
|
|
@router.post("/{template_id}/reset")
|
|
async def reset_html_template(template_id: str, request: Request) -> dict[str, Any]:
|
|
"""把内置模板正文重置回随包原版(自建模板无原版,400)。"""
|
|
await _require_admin(request)
|
|
try:
|
|
meta = template_store.reset_template(template_id)
|
|
except template_store.TemplateStoreError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
await invalidate_catalog(request, "html_templates")
|
|
return meta
|