386 lines
17 KiB
Python
386 lines
17 KiB
Python
"""HTML 模板管理 API(模板管理页)。
|
|
|
|
后端存储 :mod:`deerflow.runtime.scheduler.template_store`(文件型,
|
|
``{base_dir}/html-templates/<id>/``),启动时把随包内置模板 seed 进去(缺失才复制)。
|
|
读对所有登录用户开放(定时任务表单下拉也要用),写(创建/编辑/上传覆盖/删除/重置)
|
|
仅管理员——与敏感词管理同一鉴权口径。旧端点 ``GET /api/scheduled-tasks/html-templates``
|
|
保持不变(内部同样读存储),本路由是它的管理超集。
|
|
|
|
结构化模板(kind="structured",如 no.1)额外暴露默认数据的读写接口:
|
|
``GET /{id}/data`` 读、``PUT /{id}/data`` 整体替换(管理页「编辑数据」的持久化入口,
|
|
替换前经转换契约校验)。``GET /{id}/content`` 默认返回组合后的完整文档(预览用),
|
|
``?raw=1`` 返回骨架原文(编辑用)。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import re
|
|
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_fill, 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="新的模板正文（上传覆盖走这里）")
|
|
|
|
|
|
class TemplateDataUpdateRequest(BaseModel):
|
|
data: dict[str, Any] = Field(..., description="新的默认数据（整体替换 data.json，须通过契约校验）")
|
|
manifest: dict[str, Any] | None = Field(
|
|
default=None,
|
|
description="可选：新的 MANIFEST（整体替换 manifest.json，系统名称/导航页面/交互规则）",
|
|
)
|
|
|
|
|
|
class FieldVisibilityUpdateRequest(BaseModel):
|
|
hidden: list[str] = Field(default_factory=list, description="要隐藏的字段显示名列表（其余字段全部显示）")
|
|
extract_locked: list[str] | None = Field(
|
|
default=None,
|
|
description="可选：不接受抽取更新的字段名单（None=不动；给出则整体替换，空表=全部接受）",
|
|
)
|
|
|
|
|
|
class CustomModuleListUpdateRequest(BaseModel):
|
|
modules: list[dict[str, Any]] = Field(..., description="自定义模块全量列表（整体替换）")
|
|
|
|
|
|
class CustomModuleGenerateRequest(BaseModel):
|
|
page: str = Field(..., description="挂载的页面编号（manifest.页面 的 编号）")
|
|
title: str = Field(default="", description="模块标题（留空由模型起名）")
|
|
description: str = Field(..., min_length=4, description="用自然语言描述想新增的展示内容")
|
|
model_name: 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, raw: bool = False
|
|
) -> dict:
|
|
"""模板内容。默认返回可预览的完整文档(结构化模板 = 默认数据组合产物);
|
|
``?raw=1`` 返回正文原文(结构化模板 = 带 __PAGE_* 标记的骨架,编辑用)。"""
|
|
|
|
async def load() -> dict:
|
|
if raw:
|
|
try:
|
|
html = await asyncio.to_thread(template_store.read_template_html, template_id)
|
|
except template_store.TemplateStoreError:
|
|
raise _not_found(template_id)
|
|
else:
|
|
try:
|
|
html = await asyncio.to_thread(template_fill.read_template_text, template_id)
|
|
except template_fill.TemplateError:
|
|
raise _not_found(template_id)
|
|
return {"template_id": template_id, "html": html}
|
|
|
|
suffix = "raw" if raw else "doc"
|
|
return await cached_catalog(request, "html_templates", catalog_key("content", template_id, suffix), load)
|
|
|
|
|
|
@router.get("/{template_id}/data")
|
|
async def get_html_template_data(template_id: str, request: Request) -> dict[str, Any]:
|
|
"""结构化模板的默认数据(data.json) + MANIFEST + 各模块中文名(管理页「编辑数据」用)。"""
|
|
|
|
async def load() -> dict[str, Any]:
|
|
try:
|
|
data, manifest, modules = await asyncio.gather(
|
|
asyncio.to_thread(template_store.read_template_data, template_id),
|
|
asyncio.to_thread(template_store.read_template_manifest, template_id),
|
|
asyncio.to_thread(template_fill.structured_module_labels, template_id),
|
|
)
|
|
except (template_store.TemplateStoreError, template_fill.TemplateError) as exc:
|
|
status = 404 if "不存在" in str(exc) else 400
|
|
raise HTTPException(status_code=status, detail=str(exc))
|
|
return {
|
|
"template_id": template_id,
|
|
"data": data,
|
|
"manifest": manifest,
|
|
"modules": modules,
|
|
}
|
|
|
|
return await cached_catalog(request, "html_templates", catalog_key("data", template_id), load)
|
|
|
|
|
|
@router.put("/{template_id}/data")
|
|
async def update_html_template_data(
|
|
template_id: str, body: TemplateDataUpdateRequest, request: Request
|
|
) -> dict[str, Any]:
|
|
"""整体替换结构化模板的默认数据/MANIFEST 并持久化(仅管理员;先经契约校验)。"""
|
|
await _require_admin(request)
|
|
try:
|
|
meta = template_store.write_template_data(
|
|
template_id, body.data, manifest=body.manifest
|
|
)
|
|
except template_store.TemplateStoreError as exc:
|
|
status = 404 if "不存在" in str(exc) else 400
|
|
raise HTTPException(status_code=status, detail=str(exc))
|
|
await invalidate_catalog(request, "html_templates")
|
|
return meta
|
|
|
|
|
|
# ── 字段显示 / 自定义模块 ────────────────────────────────────────────────────
|
|
|
|
|
|
@router.get("/{template_id}/fields")
|
|
async def get_html_template_fields(template_id: str, request: Request) -> dict[str, Any]:
|
|
"""字段目录(fields.json,含中文说明)× 当前可见性 + 页面清单 + 自定义模块列表。"""
|
|
|
|
async def load() -> dict[str, Any]:
|
|
try:
|
|
catalog, manifest = await asyncio.gather(
|
|
asyncio.to_thread(template_store.read_template_fields, template_id),
|
|
asyncio.to_thread(template_store.read_template_manifest, template_id),
|
|
)
|
|
except template_store.TemplateStoreError as exc:
|
|
status = 404 if "不存在" in str(exc) else 400
|
|
raise HTTPException(status_code=status, detail=str(exc))
|
|
hidden = manifest.get("字段显示")
|
|
hidden = hidden if isinstance(hidden, dict) else {}
|
|
locks = manifest.get("字段抽取关闭")
|
|
locks = locks if isinstance(locks, dict) else {}
|
|
fields = [
|
|
{
|
|
"label": str(f.get("label") or ""),
|
|
"pages": [str(p) for p in (f.get("pages") or [])],
|
|
"description": str(f.get("description") or ""),
|
|
"visible": hidden.get(str(f.get("label"))) is not False,
|
|
# 是否接受定时任务抽取数据更新(默认 True);布局分区无独立数据键,恒为 True 且前端禁用开关。
|
|
"extract": locks.get(str(f.get("label"))) is not True,
|
|
"has_key": bool(f.get("keys")),
|
|
}
|
|
for f in catalog
|
|
]
|
|
pages = [
|
|
{"key": str(p.get("编号") or ""), "label": str(p.get("名称") or p.get("编号") or "")}
|
|
for p in manifest.get("页面") or []
|
|
if isinstance(p, dict)
|
|
]
|
|
custom = manifest.get("自定义模块")
|
|
return {
|
|
"template_id": template_id,
|
|
"fields": fields,
|
|
"pages": pages,
|
|
"custom_modules": custom if isinstance(custom, list) else [],
|
|
}
|
|
|
|
return await cached_catalog(request, "html_templates", catalog_key("fields", template_id), load)
|
|
|
|
|
|
@router.put("/{template_id}/fields")
|
|
async def update_html_template_fields(
|
|
template_id: str, body: FieldVisibilityUpdateRequest, request: Request
|
|
) -> dict[str, Any]:
|
|
"""整体替换字段显示/隐藏配置(可选一并替换抽取锁定名单;仅管理员)。"""
|
|
await _require_admin(request)
|
|
try:
|
|
meta = template_store.write_template_field_visibility(template_id, body.hidden)
|
|
if body.extract_locked is not None:
|
|
template_store.write_template_field_locks(template_id, body.extract_locked)
|
|
except template_store.TemplateStoreError as exc:
|
|
status = 404 if "不存在" in str(exc) else 400
|
|
raise HTTPException(status_code=status, detail=str(exc))
|
|
await invalidate_catalog(request, "html_templates")
|
|
return meta
|
|
|
|
|
|
@router.put("/{template_id}/custom-modules")
|
|
async def update_html_template_custom_modules(
|
|
template_id: str, body: CustomModuleListUpdateRequest, request: Request
|
|
) -> dict[str, Any]:
|
|
"""自定义模块列表整体替换(入库前逐条消毒;仅管理员)。"""
|
|
await _require_admin(request)
|
|
try:
|
|
meta = template_store.write_template_custom_modules(template_id, body.modules)
|
|
except template_store.TemplateStoreError as exc:
|
|
status = 404 if "不存在" in str(exc) else 400
|
|
raise HTTPException(status_code=status, detail=str(exc))
|
|
await invalidate_catalog(request, "html_templates")
|
|
return meta
|
|
|
|
|
|
_CUSTOM_MODULE_PROMPT = """你是资深前端设计师,为知识数据展示系统(红白主题)绘制一个自定义展示模块的 HTML 片段。
|
|
|
|
主题与可直接使用的样式(不要自造 CSS,尽量复用):
|
|
- 颜色: 背景 #f5f5f6、卡片白 #fff、主红 #c8102e、深红 #8f1024、浅红底 #fff1f3、正文 #262626、次要 #737373
|
|
- 布局类: panel(白卡容器) panel-title(红色竖条标题) grid2(两列) metrics(三列指标组)
|
|
- 组件类: metric metric-label metric-value metric-unit metric-note(指标卡) | chip chip-red chip-green chip-orange(标签)
|
|
- 组件类: text-list text-item(条目列表) | info-grid info-item(信息格) | bar-row bar(横条统计) | entry-grid entry-card(入口卡)
|
|
- 也可以用少量 inline style 微调(颜色只允许上述色板)
|
|
|
|
硬性约束:
|
|
1. 只输出一个最外层 <div> 包裹的 HTML 片段,不要 <html>/<head>/<body>/<script>/<style>/<iframe>/<form>
|
|
2. 禁止任何 on* 事件属性、javascript: 链接、外部资源引用(图片/字体/CDN)
|
|
3. 内容用简体中文;基于下面的描述合理组织版式;没有的数据写「待补充」,不要编造精确数字
|
|
4. **文字极简(最重要)**:整个模块总文字量 ≤120 字;任何一条文字 ≤30 字;严禁出现大段连续正文或 <p> 长段落
|
|
5. **呈现方式**:数字/统计一律用 metric 指标卡(大数字+短标签+一句 ≤20 字注释);分类/状态用 chip 标签;多条信息用 text-item 短条目(每条一行短句);条目间用并列结构,不要写成文章
|
|
6. 只输出片段本身,不要任何解释文字或 markdown 代码块
|
|
|
|
挂载页面:{page}
|
|
模块标题:{title}
|
|
模块描述:{description}"""
|
|
|
|
|
|
_THINK_BLOCK_RE = re.compile(r"<think>.*?</think>", re.IGNORECASE | re.DOTALL)
|
|
_THINK_TAG_RE = re.compile(r"</?think>", re.IGNORECASE)
|
|
_FENCE_HTML_RE = re.compile(r"```(?:html)?\s*(.*?)```", re.DOTALL | re.IGNORECASE)
|
|
|
|
|
|
def _extract_fragment(raw: str) -> str:
|
|
text = str(raw or "").strip()
|
|
text = _THINK_BLOCK_RE.sub("", text)
|
|
text = _THINK_TAG_RE.sub("", text).strip()
|
|
fence = _FENCE_HTML_RE.search(text)
|
|
if fence:
|
|
text = fence.group(1).strip()
|
|
body = re.search(r"<body[^>]*>(.*?)</body>", text, re.IGNORECASE | re.DOTALL)
|
|
if body:
|
|
text = body.group(1).strip()
|
|
# 思考文字混在片段前(无标签包裹)时,从第一个 "<div" 起截取。
|
|
div_at = text.find("<div")
|
|
if div_at > 0:
|
|
text = text[div_at:].strip()
|
|
return text
|
|
|
|
|
|
@router.post("/{template_id}/custom-modules/generate")
|
|
async def generate_html_template_custom_module(
|
|
template_id: str, body: CustomModuleGenerateRequest, request: Request
|
|
) -> dict[str, Any]:
|
|
"""按用户的自然语言描述调用模型生成自定义模块 HTML 片段(已消毒,未入库)。"""
|
|
await _require_admin(request)
|
|
try:
|
|
# 模板必须存在且页面编号合法,防止生成挂在不存在的页面上。
|
|
manifest = await asyncio.to_thread(template_store.read_template_manifest, template_id)
|
|
except template_store.TemplateStoreError as exc:
|
|
status = 404 if "不存在" in str(exc) else 400
|
|
raise HTTPException(status_code=status, detail=str(exc))
|
|
valid_pages = {
|
|
str(p.get("编号") or "") for p in manifest.get("页面") or [] if isinstance(p, dict)
|
|
}
|
|
if body.page not in valid_pages:
|
|
raise HTTPException(status_code=400, detail=f"未知页面编号：{body.page}")
|
|
|
|
from langchain_core.messages import HumanMessage
|
|
|
|
from deerflow.models import create_chat_model
|
|
|
|
prompt = _CUSTOM_MODULE_PROMPT.format(
|
|
page=body.page, title=body.title or "(由你起一个简短标题)", description=body.description
|
|
)
|
|
model = create_chat_model(name=body.model_name or None, thinking_enabled=False)
|
|
response = await model.ainvoke([HumanMessage(content=prompt)])
|
|
content = getattr(response, "content", response)
|
|
if isinstance(content, list):
|
|
content = "\n".join(
|
|
part.get("text", "") if isinstance(part, dict) else str(part) for part in content
|
|
)
|
|
fragment = template_store.sanitize_html_fragment(_extract_fragment(str(content)))
|
|
if "<div" not in fragment:
|
|
raise HTTPException(status_code=502, detail="模型未返回有效的 HTML 片段，请调整描述后重试")
|
|
return {"template_id": template_id, "page": body.page, "html": fragment}
|
|
|
|
|
|
@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
|