feat(scheduled-tasks): 新增 HTML 页面定时任务(生成/预览/收藏/技能)

后端:
- 新增 task_kind=html_page 任务类型,与普通定时任务并存,配置存 execution_context(无需迁移)
- 执行流水线:智能体检索数据 -> 选定模型渲染 HTML -> 清洗/校验 -> 失败自动修复 -> 完整性闸门(缺 doctype/html/head/body/title 或残留脚本即失败) -> 归档 index.html -> 落库
- 新增 scheduled_html_pages / html_page_favorites 表与 store;收藏去重(不可重复)、可删除、收藏时提取可复用风格
- HTML 预览接口(鉴权 + 公开)、收藏接口
- 渲染注入已启用的 HTML 技能(metadata.kind=html_page);全程使用用户所选模型,去除硬编码 GLM-5
- 定时任务跳过 TitleMiddleware,避免标题生成回退默认模型

前端:
- 创建/编辑弹窗增加任务类型与 HTML 配置(页面要求/限制/参考收藏页/模型下拉)
- 运行详情页与公开页 iframe 沙箱预览 HTML;收藏按钮(已收藏禁用)
- 参考收藏页可选可删

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
AdministratorandClaude Opus 4.8 committed 2026-06-02 23:46:20 +08:00
1 parent e1e9be8daa
commit d9afaca51f
22 files changed
+1996 -107

No files matched your search

+7 -1
View File
@@ -245,7 +245,9 @@ FastAPI application on port 8001 with health check at `GET /health`. Set `GATEWA
| **Thread Shares** (`/api/thread-shares`) | Per-conversation sharing with two modes (`thread_shares.mode`). **`import`**: `POST /` `{thread_id, mode}` - create (or reuse) a revocable share code; `GET /{code}/preview` + `POST /{code}/import` - copy that conversation into the caller's account (fresh `thread_id` via `app/gateway/thread_copy.py` + new `threads_meta` row + `user-data` copy; dedup on `metadata.shared_origin`). **`view`**: a public, read-only static page — see Public Shares below. Both modes: `GET /` - list my codes; `DELETE /{code}` - revoke. |
| **Public Shares** (`/api/public/shares`) | **Unauthenticated** (the `/api/public/` prefix is whitelisted in `auth_middleware._PUBLIC_PATH_PREFIXES`). `GET /{code}` - read-only conversation snapshot (title + serialized messages + artifacts) for a `view`-mode share; messages are read straight from the latest checkpoint. `GET /{code}/files/{path}` - serves a file referenced by that conversation (images, Markdown, text, PDF, …) so the shared page's links are openable; path resolution is confined to the owner's thread directory. Active content (`text/html`, `application/xhtml+xml`, `image/svg+xml`) is forced to download (`_public_share_disposition`); everything else is served inline. Missing/revoked/non-`view` codes all return 404. |
| **Roundtable Drafts** (`/api/roundtable-drafts`) | Per-user persistence for the roundtable-planning page (replaces the old browser `localStorage`). `GET /` - list drafts (lightweight meta); `POST /` - create; `GET/PUT/DELETE /{id}` - full read / partial update / delete. A draft = one full session (Step 1 intent + Step 2 roundtable), step snapshots stored as JSON in `PortableLongText`. Recommendation history bound to a draft: `GET /{id}/recommendations` - list newest-first; `POST /{id}/recommendations` - append one run (objective/status/rationale/picks/candidates). Backed by `deerflow.persistence.roundtable_drafts` (tables `roundtable_drafts`, `roundtable_recommend_history`); store wired in `deps.py` as `app.state.roundtable_draft_store`. |
| **Scheduled Tasks** (`/api/scheduled-tasks`) | Per-user cron-like agent tasks. `GET/POST /` - list/create owned tasks; `PATCH/DELETE /{id}` - update/delete (owner); `POST /{id}/{pause,resume,trigger}`; `GET /{id}/runs`, `GET /runs/{rid}`, run-file preview/edit/download; `GET /runs/{rid}/messages` - agent messages for the run's underlying agent run (used to inspect a stuck/running run; authorized via `_load_run`); `DELETE /runs/{rid}` - owner-scoped, with an admin fallback so admins can delete any user's run; publish/subscribe under `/{id}/{publish,unpublish,subscribe}` and `GET /published`. **Admin (cross-user):** `GET /admin/all` - every task; `PATCH/DELETE /admin/{id}` - edit/delete any task; `POST /admin/{id}/trigger`; `GET /admin/{id}/runs`. Admin routes require `system_role == "admin"`. |
| **Scheduled Tasks** (`/api/scheduled-tasks`) | Per-user cron-like agent tasks. `GET/POST /` - list/create owned tasks; `PATCH/DELETE /{id}` - update/delete (owner); `POST /{id}/{pause,resume,trigger}`; `GET /{id}/runs`, `GET /runs/{rid}`, run-file preview/edit/download; `GET /runs/{rid}/messages` - agent messages for the run's underlying agent run (used to inspect a stuck/running run; authorized via `_load_run`); `GET /runs/{rid}/html` - generated HTML document for an `html_page` run; `DELETE /runs/{rid}` - owner-scoped, with an admin fallback so admins can delete any user's run; publish/subscribe under `/{id}/{publish,unpublish,subscribe}` and `GET /published`. **Admin (cross-user):** `GET /admin/all` - every task; `PATCH/DELETE /admin/{id}` - edit/delete any task; `POST /admin/{id}/trigger`; `GET /admin/{id}/runs`. Admin routes require `system_role == "admin"`. |
| **HTML Page Favorites** (`/api/html-page-favorites`) | Bookmarked generated HTML pages, owner-scoped. `GET /` - list; `POST /` `{page_id, title?, description?, tags?}` - bookmark a `scheduled_html_pages` row (extracts a reusable style/layout summary via GLM5 at favorite time); a page can only be favorited once per user (`DuplicateFavoriteError` → HTTP 409). `GET/DELETE /{id}` - read/delete (the create-task reference picker deletes favorites here). A favorite can be selected as a style reference when creating a new `html_page` task. |
| **Public Scheduled Tasks** (`/api/public/scheduled-tasks`) | **Unauthenticated** mirror for embedding. Adds `GET /runs/{rid}/html` and `GET /{task_id}/latest-html` for the `html_page` preview; the run result keeps `result.html` (only file paths are redacted). |
Proxied through nginx: `/api/langgraph/*` → LangGraph, all other `/api/*` → Gateway.
@@ -257,6 +259,10 @@ Proxied through nginx: `/api/langgraph/*` → LangGraph, all other `/api/*` →
**Unique task names**: task names are unique per user (`uq_scheduled_tasks_user_name`). `create_task`/`update_task`/`admin_update_task` proactively check via `_assert_name_available()` and raise `DuplicateTaskNameError` on a collision; the router translates that into HTTP `409` instead of letting the raw DB `IntegrityError` surface as a `500`.
**HTML page tasks (`task_kind == "html_page"`)**: a second task kind that renders a full HTML document each run instead of a Markdown report. The kind + its config (`html_config`: `page_prompt`, `negative_prompt`, `reference_favorite_id`, `model_name`) live inside `execution_context` — no schema change to `scheduled_tasks`. Data gathering is delegated to the agent's own search skills (driven by `page_prompt`), so there is no explicit search-query/data-source field; the render model is user-selectable (`model_name`, default `zai-org/GLM-5-FP8`). `_run_task_attempt()` dispatches by kind; `_run_html_page_attempt()` runs the pipeline: gather real data via the lead agent (its search tools) → optionally load a reference favorite's reusable style/layout → render with the chosen model (thinking off) → fault-tolerant JSON parse → sanitize (BeautifulSoup, strips `<script>`/`<iframe>`/`on*`/`javascript:`) → structural validate → one GLM5 repair pass if invalid → **completeness gate**: if the HTML is still not a full valid document (missing doctype/html/head/body, empty body, or residual scripts) after the repair pass, raise `HtmlGenerationError` so the run is marked **failed** (a page only counts as successful when it is a complete, previewable HTML document) → best-effort Playwright screenshot (no-op when Playwright absent) → archive `index.html` → persist a `scheduled_html_pages` row → return `result` with `format: "html"` and the full HTML inline. Pure helpers live in `deerflow/runtime/scheduler/html_page.py`. The frontend previews the HTML in a sandboxed `iframe` (no `allow-scripts`) on the run-detail and public pages; a user can bookmark a page (style summary extracted via GLM5) and later pick it as a style reference. New tables `scheduled_html_pages` + `html_page_favorites` (`deerflow/persistence/html_pages/`, wired as `app.state.html_page_store`) are created by `Base.metadata.create_all` with no migration. Tests: `tests/test_html_page_tasks.py`.
**HTML skills (tools the render step uses).** Two offline skills under `skills/public/` shape the generated HTML and are toggled in the skills settings page (enabled by default): `html-page-builder` (a self-contained inline-CSS design system — colors/cards/tables/KPI/responsive) and `html-charts` (inline-SVG charts that render in the no-`allow-scripts` preview iframe; optional local ECharts documented in its `assets/README.md`). A skill opts into HTML injection via `metadata.kind: html_page` in its SKILL.md frontmatter; `_collect_html_skill_instructions()` reads the bodies of **enabled** such skills and `build_generation_prompt(skill_instructions=…)` injects them, so the (raw-model) render step honors the user's skill selection. The data-gathering step and the render/repair/style passes all use the task's selected `model_name` (no silent fallback to the default model — and there is **no hardcoded GLM-5**; an unset selection falls back to config `models[0]`). To stop auxiliary LLM calls from leaking onto the default model, **`TitleMiddleware` is skipped when `is_scheduled_run` is set** (`_build_middlewares` in `lead_agent/agent.py`) — scheduled/HTML runs use throwaway threads whose titles are never shown, and `config.title.model_name` is usually unset (→ would call `models[0]`, not the chosen model). The completeness gate (`validate_html`) also requires a non-empty `<title>` and forbids external stylesheets, on top of the doctype/html/head/body + non-empty-body + no-script checks.
### Custom Agent Persistence
Custom agent metadata lives in the `agents` ORM table (`deerflow.persistence.agents`). `agents.id` is the stable runtime id and the filesystem directory name under `.deer-flow/agents/{id}/`; `agents.name` is display-only, can contain Chinese characters, and is not unique. `user_id IS NULL` marks built-in agents, user-owned agents are visible only to their owner unless `published = true`, and update/delete operations require ownership. The runtime accepts `agent_id` in context/configurable, and non-default `assistant_id` values are treated as agent ids after visibility checks.
@@ -32,6 +32,7 @@ from app.gateway.routers import (
auth,
channels,
feedback,
html_page_favorites,
intent,
llm_metrics,
mcp,
@@ -870,6 +871,8 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
# surface results without a DeerFlow login.
app.include_router(public_scheduled_tasks.router)
app.include_router(html_page_favorites.router)
# System settings API is mounted at /api/system-settings
app.include_router(system_settings.router)
@@ -93,6 +93,10 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
app.state.scheduled_task_store = make_scheduled_task_store(sf)
from deerflow.persistence.html_pages import make_html_page_store
app.state.html_page_store = make_html_page_store(sf)
from deerflow.persistence.llm_metrics import make_llm_metrics_store
app.state.llm_metrics_store = make_llm_metrics_store(sf)
@@ -0,0 +1,93 @@
"""HTML page favorites API.
A favorite bookmarks a generated ``scheduled_html_pages`` row and stores a
reusable style/layout summary (extracted via GLM5 at favorite time) so a later
``html_page`` task can be generated "in the style of" the favorite.
All routes are owner-scoped; the underlying logic lives on
``ScheduledTaskService`` so GLM5 access and the page store stay in one place.
"""
from __future__ import annotations
from typing import Any
from fastapi import APIRouter, HTTPException, Request
from pydantic import BaseModel, Field
from app.gateway.deps import get_current_user
# Importing the ORM models here (at app-module load, before the lifespan runs
# ``Base.metadata.create_all``) registers ``scheduled_html_pages`` +
# ``html_page_favorites`` with the metadata so the tables are auto-created on
# startup. Without this the store's models are only imported later (inside
# ``langgraph_runtime``), after create_all has already run — yielding
# "no such table: html_page_favorites".
from deerflow.persistence.html_pages.model import ( # noqa: F401
HtmlPageFavoriteRow,
ScheduledHtmlPageRow,
)
from deerflow.runtime.scheduler import get_scheduled_task_service
from deerflow.runtime.scheduler.service import DuplicateFavoriteError
router = APIRouter(prefix="/api/html-page-favorites", tags=["html-page-favorites"])
class HtmlPageFavoriteCreateRequest(BaseModel):
page_id: str = Field(..., description="scheduled_html_pages.id to bookmark")
title: str | None = None
description: str | None = None
tags: list[str] | None = None
async def _user_id(request: Request) -> str:
user_id = await get_current_user(request)
if not user_id:
raise HTTPException(status_code=401, detail="Authentication required")
return user_id
@router.get("")
async def list_html_page_favorites(request: Request) -> list[dict[str, Any]]:
service = get_scheduled_task_service()
user_id = await _user_id(request)
return await service.list_html_page_favorites(user_id)
@router.post("")
async def create_html_page_favorite(body: HtmlPageFavoriteCreateRequest, request: Request) -> dict[str, Any]:
service = get_scheduled_task_service()
user_id = await _user_id(request)
try:
favorite = await service.create_html_page_favorite(
user_id,
body.page_id,
title=body.title,
description=body.description,
tags=body.tags,
)
except DuplicateFavoriteError:
raise HTTPException(status_code=409, detail="该页面已收藏,请勿重复收藏")
if favorite is None:
raise HTTPException(status_code=404, detail="页面不存在或无权收藏")
return favorite
@router.get("/{favorite_id}")
async def get_html_page_favorite(favorite_id: str, request: Request) -> dict[str, Any]:
service = get_scheduled_task_service()
user_id = await _user_id(request)
favorite = await service.get_html_page_favorite(user_id, favorite_id)
if favorite is None:
raise HTTPException(status_code=404, detail=f"Favorite {favorite_id} not found")
return favorite
@router.delete("/{favorite_id}")
async def delete_html_page_favorite(favorite_id: str, request: Request) -> dict[str, Any]:
service = get_scheduled_task_service()
user_id = await _user_id(request)
deleted = await service.delete_html_page_favorite(user_id, favorite_id)
if not deleted:
raise HTTPException(status_code=404, detail=f"Favorite {favorite_id} not found")
return {"success": True}
@@ -243,6 +243,39 @@ async def public_get_run(run_id: str) -> dict[str, Any]:
return _redact_run(run)
def _html_payload(run_id: str, page: dict[str, Any]) -> dict[str, Any]:
return {
"run_id": run_id,
"task_id": page.get("task_id"),
"title": page.get("title"),
"html": page.get("html_content"),
"style_summary": page.get("style_summary"),
"layout_summary": page.get("layout_summary"),
"page_id": page.get("id"),
"created_at": page.get("created_at"),
}
@router.get("/runs/{run_id}/html")
async def public_get_run_html(run_id: str) -> dict[str, Any]:
"""Return the generated HTML document for an ``html_page`` run (anonymous)."""
service = get_scheduled_task_service()
page = await service.get_html_page_for_run(run_id)
if page is None:
raise HTTPException(status_code=404, detail="该执行记录没有 HTML 页面")
return _html_payload(run_id, page)
@router.get("/{task_id}/latest-html")
async def public_get_latest_html(task_id: str) -> dict[str, Any]:
"""Return the most recent generated HTML page for a task (anonymous)."""
service = get_scheduled_task_service()
page = await service.get_latest_html_page_for_task(task_id)
if page is None:
raise HTTPException(status_code=404, detail="该任务暂无 HTML 页面")
return _html_payload(str(page.get("run_id") or ""), page)
@router.get("/runs/{run_id}/files/{file_index}")
async def public_get_run_file(run_id: str, file_index: int) -> dict[str, Any]:
"""Return a single result-file entry with its inline text content where
@@ -251,6 +251,29 @@ async def get_scheduled_task_run(run_id: str, request: Request) -> dict:
return run
@router.get("/runs/{run_id}/html")
async def get_scheduled_task_run_html(run_id: str, request: Request) -> dict:
"""Return the generated HTML document for an ``html_page`` run."""
service = get_scheduled_task_service()
user_id = await _user_id(request)
run = await _load_run(service, request, run_id, user_id)
if run is None:
raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
page = await service.get_html_page_for_run(run_id)
if page is None:
raise HTTPException(status_code=404, detail="该执行记录没有 HTML 页面")
return {
"run_id": run_id,
"task_id": page.get("task_id") or run.get("task_id"),
"title": page.get("title"),
"html": page.get("html_content"),
"style_summary": page.get("style_summary"),
"layout_summary": page.get("layout_summary"),
"page_id": page.get("id"),
"created_at": page.get("created_at"),
}
def _event_type_for_checkpoint_message(message: dict) -> str:
"""Map a serialized LangChain message to the same ``event_type`` strings
the run-event store uses, so the frontend's existing role labels keep
@@ -6,6 +6,8 @@
"code-retrieval": { "enabled": true },
"news-briefing": { "enabled": true },
"workspace-search": { "enabled": true },
"policy-knowledge": { "enabled": true }
"policy-knowledge": { "enabled": true },
"html-page-builder": { "enabled": true },
"html-charts": { "enabled": true }
}
}
@@ -280,8 +280,14 @@ def _build_middlewares(
if resolved_app_config.token_usage.enabled:
middlewares.append(TokenUsageMiddleware())
# Add TitleMiddleware
middlewares.append(TitleMiddleware(app_config=resolved_app_config))
# Add TitleMiddleware — but skip it for scheduled runs. Each scheduled run
# executes on a throwaway thread whose title is never shown to a user, and
# generating one would fire an extra LLM call on the *default* model
# (config.title.model_name is usually unset → models[0]), which is not the
# model the task selected. Skipping it keeps scheduled/HTML runs on exactly
# the chosen model.
if not cfg.get("is_scheduled_run", False):
middlewares.append(TitleMiddleware(app_config=resolved_app_config))
# Add MemoryMiddleware (after TitleMiddleware) — skip in bootstrap mode (agent creation flow)
memory_injection_enabled = cfg.get("memory_injection_enabled", True)
@@ -0,0 +1,23 @@
"""Generated HTML page + favorite persistence."""
from __future__ import annotations
from deerflow.persistence.html_pages.base import HtmlPageStore
from deerflow.persistence.html_pages.memory import MemoryHtmlPageStore
from deerflow.persistence.html_pages.model import HtmlPageFavoriteRow, ScheduledHtmlPageRow
from deerflow.persistence.html_pages.sql import HtmlPageRepository
__all__ = [
"HtmlPageStore",
"HtmlPageRepository",
"MemoryHtmlPageStore",
"HtmlPageFavoriteRow",
"ScheduledHtmlPageRow",
"make_html_page_store",
]
def make_html_page_store(session_factory):
if session_factory is None:
return MemoryHtmlPageStore()
return HtmlPageRepository(session_factory)
@@ -0,0 +1,53 @@
"""Abstract persistence API for generated HTML pages and favorites."""
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import Any
class HtmlPageStore(ABC):
# --- Generated pages ---
@abstractmethod
async def create_page(self, data: dict[str, Any]) -> dict[str, Any]:
raise NotImplementedError
@abstractmethod
async def get_page(self, page_id: str) -> dict[str, Any] | None:
"""Return a page by id without restricting by owner."""
raise NotImplementedError
@abstractmethod
async def get_page_by_run(self, run_id: str) -> dict[str, Any] | None:
"""Return the page produced by a given run, if any."""
raise NotImplementedError
@abstractmethod
async def latest_page_for_task(self, task_id: str) -> dict[str, Any] | None:
"""Return the most recently created page for a task, if any."""
raise NotImplementedError
# --- Favorites ---
@abstractmethod
async def create_favorite(self, data: dict[str, Any]) -> dict[str, Any]:
raise NotImplementedError
@abstractmethod
async def list_favorites(self, user_id: str, *, limit: int = 100) -> list[dict[str, Any]]:
raise NotImplementedError
@abstractmethod
async def get_favorite(self, favorite_id: str) -> dict[str, Any] | None:
"""Return a favorite by id without restricting by owner."""
raise NotImplementedError
@abstractmethod
async def get_favorite_by_page(self, user_id: str, page_id: str) -> dict[str, Any] | None:
"""Return the user's existing favorite for a page, if any (dedup check)."""
raise NotImplementedError
@abstractmethod
async def delete_favorite(self, favorite_id: str, user_id: str) -> bool:
raise NotImplementedError
@@ -0,0 +1,102 @@
"""In-memory HtmlPageStore for no-database / test deployments."""
from __future__ import annotations
import copy
from datetime import UTC, datetime
from typing import Any
from deerflow.persistence.html_pages.base import HtmlPageStore
def _now_iso() -> str:
return datetime.now(UTC).isoformat()
class MemoryHtmlPageStore(HtmlPageStore):
def __init__(self) -> None:
self._pages: dict[str, dict[str, Any]] = {}
self._favorites: dict[str, dict[str, Any]] = {}
# --- Generated pages ---
async def create_page(self, data: dict[str, Any]) -> dict[str, Any]:
row = {
"id": data["id"],
"task_id": data["task_id"],
"run_id": data["run_id"],
"user_id": data["user_id"],
"title": data.get("title") or "",
"html_content": data.get("html_content") or "",
"html_file_path": data.get("html_file_path"),
"data_snapshot": data.get("data_snapshot") or {},
"prompt_snapshot": data.get("prompt_snapshot") or {},
"reference_favorite_id": data.get("reference_favorite_id"),
"style_summary": data.get("style_summary"),
"layout_summary": data.get("layout_summary"),
"created_at": _now_iso(),
"updated_at": _now_iso(),
}
self._pages[row["id"]] = row
return copy.deepcopy(row)
async def get_page(self, page_id: str) -> dict[str, Any] | None:
row = self._pages.get(page_id)
return copy.deepcopy(row) if row is not None else None
async def get_page_by_run(self, run_id: str) -> dict[str, Any] | None:
matches = [r for r in self._pages.values() if r["run_id"] == run_id]
if not matches:
return None
matches.sort(key=lambda r: r["created_at"], reverse=True)
return copy.deepcopy(matches[0])
async def latest_page_for_task(self, task_id: str) -> dict[str, Any] | None:
matches = [r for r in self._pages.values() if r["task_id"] == task_id]
if not matches:
return None
matches.sort(key=lambda r: r["created_at"], reverse=True)
return copy.deepcopy(matches[0])
# --- Favorites ---
async def create_favorite(self, data: dict[str, Any]) -> dict[str, Any]:
row = {
"id": data["id"],
"user_id": data["user_id"],
"page_id": data["page_id"],
"title": data.get("title") or "",
"description": data.get("description"),
"tags": data.get("tags") or [],
"html_snapshot": data.get("html_snapshot") or "",
"style_summary": data.get("style_summary"),
"layout_summary": data.get("layout_summary"),
"content_structure_summary": data.get("content_structure_summary"),
"reuse_prompt": data.get("reuse_prompt"),
"created_at": _now_iso(),
"updated_at": _now_iso(),
}
self._favorites[row["id"]] = row
return copy.deepcopy(row)
async def list_favorites(self, user_id: str, *, limit: int = 100) -> list[dict[str, Any]]:
matches = [r for r in self._favorites.values() if r["user_id"] == user_id]
matches.sort(key=lambda r: r["created_at"], reverse=True)
return [copy.deepcopy(r) for r in matches[: max(1, limit)]]
async def get_favorite(self, favorite_id: str) -> dict[str, Any] | None:
row = self._favorites.get(favorite_id)
return copy.deepcopy(row) if row is not None else None
async def get_favorite_by_page(self, user_id: str, page_id: str) -> dict[str, Any] | None:
for row in self._favorites.values():
if row["user_id"] == user_id and row["page_id"] == page_id:
return copy.deepcopy(row)
return None
async def delete_favorite(self, favorite_id: str, user_id: str) -> bool:
row = self._favorites.get(favorite_id)
if row is None or row["user_id"] != user_id:
return False
del self._favorites[favorite_id]
return True
@@ -0,0 +1,65 @@
"""ORM models for generated HTML pages and user favorites.
These back the "HTML 页面定时任务" feature: a scheduled task of kind
``html_page`` renders a full HTML document each run. Every rendered page is
recorded in :class:`ScheduledHtmlPageRow`; a user may bookmark one into
:class:`HtmlPageFavoriteRow`, which also stores a reusable style/layout
summary so a later task can be generated "in the style of" the favorite.
Both tables are brand new, so ``Base.metadata.create_all`` creates them on the
next startup without touching any existing schema.
"""
from __future__ import annotations
from datetime import UTC, datetime
from sqlalchemy import String, Text
from sqlalchemy.orm import Mapped, mapped_column
from deerflow.persistence.base import Base
from deerflow.persistence.types import BeijingDateTime, PortableJSON, PortableLongText
class ScheduledHtmlPageRow(Base):
"""One HTML page produced by a single run of an ``html_page`` task."""
__tablename__ = "scheduled_html_pages"
id: Mapped[str] = mapped_column(String(64), primary_key=True)
task_id: Mapped[str] = mapped_column(String(64), index=True)
run_id: Mapped[str] = mapped_column(String(64), index=True)
user_id: Mapped[str] = mapped_column(String(64), index=True)
title: Mapped[str] = mapped_column(String(255))
# Full HTML document — can be large, so LONGTEXT on MySQL.
html_content: Mapped[str] = mapped_column(PortableLongText())
html_file_path: Mapped[str | None] = mapped_column(Text)
data_snapshot_json: Mapped[dict | None] = mapped_column(PortableJSON(), default=dict)
prompt_snapshot_json: Mapped[dict | None] = mapped_column(PortableJSON(), default=dict)
reference_favorite_id: Mapped[str | None] = mapped_column(String(64), index=True)
style_summary: Mapped[str | None] = mapped_column(Text)
layout_summary: Mapped[str | None] = mapped_column(Text)
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC))
updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
class HtmlPageFavoriteRow(Base):
"""A user-bookmarked HTML page, with a reusable style/layout summary."""
__tablename__ = "html_page_favorites"
id: Mapped[str] = mapped_column(String(64), primary_key=True)
user_id: Mapped[str] = mapped_column(String(64), index=True)
page_id: Mapped[str] = mapped_column(String(64), index=True)
title: Mapped[str] = mapped_column(String(255))
description: Mapped[str | None] = mapped_column(Text)
tags_json: Mapped[list | None] = mapped_column(PortableJSON(), default=list)
# Snapshot of the page HTML at favorite time, decoupled from the source run
# so deleting the run never breaks the favorite reference.
html_snapshot: Mapped[str] = mapped_column(PortableLongText())
style_summary: Mapped[str | None] = mapped_column(Text)
layout_summary: Mapped[str | None] = mapped_column(Text)
content_structure_summary: Mapped[str | None] = mapped_column(Text)
reuse_prompt: Mapped[str | None] = mapped_column(Text)
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC))
updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
@@ -0,0 +1,141 @@
"""SQLAlchemy repository for generated HTML pages and favorites."""
from __future__ import annotations
from datetime import datetime
from typing import Any
from sqlalchemy import delete, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from deerflow.persistence.html_pages.base import HtmlPageStore
from deerflow.persistence.html_pages.model import HtmlPageFavoriteRow, ScheduledHtmlPageRow
def _to_dict(row) -> dict[str, Any]:
data = row.to_dict()
for key in ("created_at", "updated_at"):
value = data.get(key)
if isinstance(value, datetime):
data[key] = value.isoformat()
# Normalize the JSON column names to API-friendly keys.
if "data_snapshot_json" in data:
data["data_snapshot"] = data.pop("data_snapshot_json") or {}
if "prompt_snapshot_json" in data:
data["prompt_snapshot"] = data.pop("prompt_snapshot_json") or {}
if "tags_json" in data:
data["tags"] = data.pop("tags_json") or []
return data
class HtmlPageRepository(HtmlPageStore):
def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
self._sf = session_factory
# --- Generated pages ---
async def create_page(self, data: dict[str, Any]) -> dict[str, Any]:
row = ScheduledHtmlPageRow(
id=data["id"],
task_id=data["task_id"],
run_id=data["run_id"],
user_id=data["user_id"],
title=data.get("title") or "",
html_content=data.get("html_content") or "",
html_file_path=data.get("html_file_path"),
data_snapshot_json=data.get("data_snapshot") or {},
prompt_snapshot_json=data.get("prompt_snapshot") or {},
reference_favorite_id=data.get("reference_favorite_id"),
style_summary=data.get("style_summary"),
layout_summary=data.get("layout_summary"),
)
async with self._sf() as session:
session.add(row)
await session.commit()
await session.refresh(row)
return _to_dict(row)
async def get_page(self, page_id: str) -> dict[str, Any] | None:
async with self._sf() as session:
row = await session.get(ScheduledHtmlPageRow, page_id)
return _to_dict(row) if row is not None else None
async def get_page_by_run(self, run_id: str) -> dict[str, Any] | None:
async with self._sf() as session:
result = await session.execute(
select(ScheduledHtmlPageRow)
.where(ScheduledHtmlPageRow.run_id == run_id)
.order_by(ScheduledHtmlPageRow.created_at.desc())
.limit(1)
)
row = result.scalar_one_or_none()
return _to_dict(row) if row is not None else None
async def latest_page_for_task(self, task_id: str) -> dict[str, Any] | None:
async with self._sf() as session:
result = await session.execute(
select(ScheduledHtmlPageRow)
.where(ScheduledHtmlPageRow.task_id == task_id)
.order_by(ScheduledHtmlPageRow.created_at.desc())
.limit(1)
)
row = result.scalar_one_or_none()
return _to_dict(row) if row is not None else None
# --- Favorites ---
async def create_favorite(self, data: dict[str, Any]) -> dict[str, Any]:
row = HtmlPageFavoriteRow(
id=data["id"],
user_id=data["user_id"],
page_id=data["page_id"],
title=data.get("title") or "",
description=data.get("description"),
tags_json=data.get("tags") or [],
html_snapshot=data.get("html_snapshot") or "",
style_summary=data.get("style_summary"),
layout_summary=data.get("layout_summary"),
content_structure_summary=data.get("content_structure_summary"),
reuse_prompt=data.get("reuse_prompt"),
)
async with self._sf() as session:
session.add(row)
await session.commit()
await session.refresh(row)
return _to_dict(row)
async def list_favorites(self, user_id: str, *, limit: int = 100) -> list[dict[str, Any]]:
async with self._sf() as session:
result = await session.execute(
select(HtmlPageFavoriteRow)
.where(HtmlPageFavoriteRow.user_id == user_id)
.order_by(HtmlPageFavoriteRow.created_at.desc())
.limit(max(1, limit))
)
return [_to_dict(row) for row in result.scalars()]
async def get_favorite(self, favorite_id: str) -> dict[str, Any] | None:
async with self._sf() as session:
row = await session.get(HtmlPageFavoriteRow, favorite_id)
return _to_dict(row) if row is not None else None
async def get_favorite_by_page(self, user_id: str, page_id: str) -> dict[str, Any] | None:
async with self._sf() as session:
result = await session.execute(
select(HtmlPageFavoriteRow)
.where(HtmlPageFavoriteRow.user_id == user_id, HtmlPageFavoriteRow.page_id == page_id)
.limit(1)
)
row = result.scalar_one_or_none()
return _to_dict(row) if row is not None else None
async def delete_favorite(self, favorite_id: str, user_id: str) -> bool:
async with self._sf() as session:
result = await session.execute(
delete(HtmlPageFavoriteRow).where(
HtmlPageFavoriteRow.id == favorite_id,
HtmlPageFavoriteRow.user_id == user_id,
)
)
await session.commit()
return (result.rowcount or 0) > 0
@@ -1,8 +1,16 @@
from .registry import get_scheduled_task_service, set_scheduled_task_service
from .service import DuplicateTaskNameError, MarkdownRequirementError, ScheduledTaskService
from .service import (
DuplicateFavoriteError,
DuplicateTaskNameError,
HtmlGenerationError,
MarkdownRequirementError,
ScheduledTaskService,
)
__all__ = [
"DuplicateFavoriteError",
"DuplicateTaskNameError",
"HtmlGenerationError",
"MarkdownRequirementError",
"ScheduledTaskService",
"get_scheduled_task_service",
@@ -0,0 +1,365 @@
"""HTML page generation helpers for ``html_page`` scheduled tasks.
Pure, dependency-light utilities the scheduler uses to turn a GLM5 response
into a safe, previewable HTML document:
- :func:`extract_json_object` — fault-tolerant JSON extraction from a model
reply that may be wrapped in prose or ```json fences.
- :func:`sanitize_html` — strip script-execution vectors (``<script>``,
``<iframe>``, ``on*`` handlers, ``javascript:`` URLs, …) using BeautifulSoup.
- :func:`validate_html` — cheap structural checks (doctype/html/head/body,
non-empty body, no residual scripts) that gate the optional repair pass.
- ``build_*_prompt`` — the GLM5 prompt templates (generation / repair / style
extraction).
- :func:`screenshot_html` — best-effort Playwright screenshot, a no-op when
Playwright is not installed (the offline image ships without a browser).
Kept free of any ``app.*`` import so it stays inside the harness boundary.
"""
from __future__ import annotations
import json
import logging
import re
from typing import Any
logger = logging.getLogger(__name__)
# Tags removed wholesale — they either execute code or load remote active
# content that the iframe sandbox alone shouldn't be relied on to contain.
_DANGEROUS_TAGS = frozenset(
{"script", "iframe", "object", "embed", "applet", "frame", "frameset", "base"}
)
# URL schemes that can execute script when navigated/loaded.
_DANGEROUS_URL_RE = re.compile(r"^\s*(javascript|vbscript|data:text/html)", re.IGNORECASE)
# Attributes carrying a URL we must scheme-check.
_URL_ATTRS = frozenset({"href", "src", "action", "formaction", "xlink:href", "background", "poster"})
def parse_frontmatter(text: str) -> tuple[dict[str, Any], str]:
"""Split a ``---`` YAML frontmatter block from the body of a SKILL.md.
Returns ``(frontmatter_dict, body)``. When there is no frontmatter the dict
is empty and the body is the whole text.
"""
if not text or not text.lstrip().startswith("---"):
return {}, text or ""
lines = text.splitlines()
# Skip leading blank lines before the opening fence.
start = 0
while start < len(lines) and not lines[start].strip():
start += 1
end = None
for i in range(start + 1, len(lines)):
if lines[i].strip() == "---":
end = i
break
if end is None:
return {}, text
fm_block = "\n".join(lines[start + 1 : end])
body = "\n".join(lines[end + 1 :])
try:
import yaml
data = yaml.safe_load(fm_block) or {}
if not isinstance(data, dict):
data = {}
except Exception:
data = {}
return data, body
def extract_json_object(text: str | None) -> dict[str, Any] | None:
"""Best-effort parse of a JSON object from a model reply.
Handles three common shapes: a bare JSON object, a ```json fenced block,
and a JSON object embedded in surrounding prose. Returns ``None`` when no
balanced object can be recovered.
"""
if not text or not isinstance(text, str):
return None
candidate = text.strip()
# Strip a single ```json ... ``` (or ``` ... ```) fence if present.
fence = re.search(r"```(?:json)?\s*(.*?)```", candidate, flags=re.DOTALL | re.IGNORECASE)
if fence:
candidate = fence.group(1).strip()
# Fast path: the whole thing is already valid JSON.
try:
parsed = json.loads(candidate)
return parsed if isinstance(parsed, dict) else None
except (ValueError, TypeError):
pass
# Fallback: scan for the first balanced ``{ ... }`` and parse that.
start = candidate.find("{")
if start == -1:
return None
depth = 0
in_string = False
escape = False
for index in range(start, len(candidate)):
char = candidate[index]
if in_string:
if escape:
escape = False
elif char == "\\":
escape = True
elif char == '"':
in_string = False
continue
if char == '"':
in_string = True
elif char == "{":
depth += 1
elif char == "}":
depth -= 1
if depth == 0:
blob = candidate[start : index + 1]
try:
parsed = json.loads(blob)
except (ValueError, TypeError):
return None
return parsed if isinstance(parsed, dict) else None
return None
def sanitize_html(html: str) -> tuple[str, list[str]]:
"""Remove script-execution vectors from ``html``.
Returns ``(cleaned_html, removed)`` where ``removed`` is a human-readable
report of what was stripped (used for logging / repair feedback). The
cleaner is defensive but not a full HTML validator — pair it with an
``iframe sandbox`` (no ``allow-scripts``) on the preview side.
"""
if not html or not isinstance(html, str):
return "", []
try:
from bs4 import BeautifulSoup
except Exception: # pragma: no cover - bs4 is a declared dependency
logger.warning("BeautifulSoup unavailable; returning HTML unsanitized")
return html, []
removed: list[str] = []
soup = BeautifulSoup(html, "html.parser")
# 1) Drop dangerous tags entirely.
for tag in soup.find_all(True):
name = (tag.name or "").lower()
if name in _DANGEROUS_TAGS:
removed.append(f"<{name}>")
tag.decompose()
continue
# External stylesheet / preload links pull remote resources.
if name == "link":
href = str(tag.get("href") or "")
rel = " ".join(tag.get("rel") or []) if isinstance(tag.get("rel"), list) else str(tag.get("rel") or "")
if re.match(r"^\s*(https?:)?//", href) and "stylesheet" in rel.lower():
removed.append("<link stylesheet>")
tag.decompose()
continue
# meta refresh can redirect to arbitrary URLs.
if name == "meta" and str(tag.get("http-equiv") or "").lower() == "refresh":
removed.append("<meta refresh>")
tag.decompose()
continue
# 2) Strip event handlers and dangerous URL schemes from surviving tags.
for tag in soup.find_all(True):
for attr in list(tag.attrs.keys()):
lowered = attr.lower()
if lowered.startswith("on"):
removed.append(f"{lowered}=")
del tag.attrs[attr]
continue
if lowered in _URL_ATTRS:
value = tag.attrs[attr]
value_str = value if isinstance(value, str) else " ".join(value)
if _DANGEROUS_URL_RE.match(value_str or ""):
removed.append(f"{lowered}={value_str[:24]}")
del tag.attrs[attr]
return str(soup), removed
def validate_html(html: str) -> list[str]:
"""Return a list of structural problems with ``html`` (empty == valid).
A page only counts as a complete document when it has the full skeleton
(doctype/html/head/body), a non-empty ``<title>``, visible body content
(no white-screen), no residual scripts, and no external stylesheet links.
"""
errors: list[str] = []
if not html or not isinstance(html, str):
return ["empty html"]
lowered = html.lower()
if "<!doctype html" not in lowered:
errors.append("缺少 <!doctype html> 声明")
if "<html" not in lowered:
errors.append("缺少 <html> 根元素")
if "<head" not in lowered:
errors.append("缺少 <head> 元素")
if "<body" not in lowered:
errors.append("缺少 <body> 元素")
if "<script" in lowered:
errors.append("仍包含 <script> 标签(不允许脚本)")
try:
from bs4 import BeautifulSoup
soup = BeautifulSoup(html, "html.parser")
# Title must exist and be non-empty.
title_el = soup.find("title")
if title_el is None or not title_el.get_text(strip=True):
errors.append("缺少非空 <title> 标题")
# External stylesheet links pull remote resources — not allowed offline.
for link in soup.find_all("link"):
href = str(link.get("href") or "")
rel = " ".join(link.get("rel") or []) if isinstance(link.get("rel"), list) else str(link.get("rel") or "")
if re.match(r"^\s*(https?:)?//", href) and "stylesheet" in rel.lower():
errors.append("包含外链样式表(不允许外部 CDN/资源)")
break
# Body must carry some visible content — guards against white-screen.
body = soup.body or soup
text = body.get_text(strip=True) if body else ""
has_media = bool(body.find(["img", "svg", "canvas", "table", "ul", "ol"])) if body else False
if len(text) < 5 and not has_media:
errors.append("页面正文为空,可能白屏")
except Exception: # pragma: no cover
pass
return errors
# ---------------------------------------------------------------------------
# Prompt templates
# ---------------------------------------------------------------------------
_GENERATION_TEMPLATE = """你是一个内网页面生成助手。请根据输入数据生成一个完整的 HTML 页面。
用户页面要求:
{page_prompt}
限制条件:
{negative_prompt}
本次数据:
{data}
参考页面风格:
{reference_style}
参考页面布局:
{reference_layout}
{skill_instructions}
要求:
1. 只输出 JSON,不要输出 Markdown,不要输出多余解释。
2. JSON 字段必须包含 title、html、style_summary、layout_summary。
3. html 必须是完整 HTML 文档,包含 <!doctype html>、html、head、body。
4. 所有 CSS 写在 <style> 标签内,不要引用外部 CDN 或外链样式表。
5. 不要包含 <script>、外链脚本、iframe、表单提交、on* 事件属性。
6. 参考页面只用于风格和布局,页面内容必须来自“本次数据”,不要复用参考页旧数据。
7. 页面使用中文,排版适合在浏览器中直接预览。
"""
_REPAIR_TEMPLATE = """你是一个 HTML 修复助手。下面的 HTML 存在问题,请修复后重新输出完整页面。
原始页面要求:
{requirements}
发现的问题:
{errors}
待修复 HTML:
{html}
要求:
1. 只输出 JSON,字段包含 fixed_html、changes(修复说明数组)。
2. fixed_html 必须是完整 HTML 文档,包含 <!doctype html>、html、head、body。
3. 移除所有 <script>、iframe、外链脚本、on* 事件属性、表单提交。
4. 保留原有内容与数据,只修复结构与安全问题。
"""
_STYLE_EXTRACT_TEMPLATE = """你是一个页面风格分析助手。请分析下面的 HTML 页面,提取可复用的风格与布局特征。
HTML 页面:
{html}
要求:
1. 只输出 JSON,字段包含 style_summary、layout_summary、content_structure_summary、reuse_prompt。
2. style_summary:颜色、字体、间距、圆角、阴影、视觉密度等视觉风格。
3. layout_summary:页面分区、卡片、表格、图表、导航等布局结构。
4. content_structure_summary:信息组织顺序与层级。
5. reuse_prompt:一段可直接拼接到下次生成 prompt 的中文风格说明,不要包含本页的具体数据。
"""
def _clip(value: Any, limit: int) -> str:
text = value if isinstance(value, str) else json.dumps(value, ensure_ascii=False, default=str)
text = text or ""
return text if len(text) <= limit else text[:limit] + "\n…(内容过长已截断)"
def build_generation_prompt(
*,
page_prompt: str,
negative_prompt: str | None,
data: Any,
reference_style: str | None,
reference_layout: str | None,
skill_instructions: str | None = None,
data_limit: int = 16000,
skill_limit: int = 16000,
) -> str:
skill_block = ""
if skill_instructions and skill_instructions.strip():
skill_block = "\n可用页面生成技能规范(请严格遵循):\n" + _clip(skill_instructions.strip(), skill_limit) + "\n"
return _GENERATION_TEMPLATE.format(
page_prompt=(page_prompt or "(未提供,使用通用看板风格)").strip(),
negative_prompt=(negative_prompt or "无").strip(),
data=_clip(data, data_limit),
reference_style=(reference_style or "无").strip(),
reference_layout=(reference_layout or "无").strip(),
skill_instructions=skill_block,
)
def build_repair_prompt(*, html: str, errors: list[str], requirements: str | None) -> str:
return _REPAIR_TEMPLATE.format(
requirements=(requirements or "无").strip(),
errors="\n".join(f"- {e}" for e in errors) or "- 未知问题",
html=_clip(html, 32000),
)
def build_style_extract_prompt(*, html: str) -> str:
return _STYLE_EXTRACT_TEMPLATE.format(html=_clip(html, 32000))
async def screenshot_html(html: str, output_path) -> str | None:
"""Best-effort PNG screenshot of ``html`` for white-screen / console QA.
Returns the screenshot path on success, or ``None`` when Playwright (or a
usable browser) is unavailable — the offline image ships without one, so
this must degrade silently rather than fail the run.
"""
try:
from playwright.async_api import async_playwright
except Exception:
return None
try:
async with async_playwright() as pw:
browser = await pw.chromium.launch()
try:
page = await browser.new_page()
await page.set_content(html, wait_until="networkidle")
await page.screenshot(path=str(output_path), full_page=True)
finally:
await browser.close()
return str(output_path)
except Exception:
logger.debug("Playwright screenshot failed; skipping QA screenshot", exc_info=True)
return None
@@ -22,6 +22,7 @@ from deerflow.config.agents_config import AGENT_ID_PATTERN
from deerflow.config.paths import get_paths
from deerflow.persistence.scheduled_tasks import ScheduledTaskStore
from deerflow.runtime import serialize_channel_values
from deerflow.runtime.scheduler import html_page as html_page_utils
from deerflow.runtime.scheduler.element import ElementDeliveryClient
from deerflow.runtime.user_context import reset_current_user, set_current_user
from deerflow.utils.time import now_iso
@@ -36,6 +37,11 @@ _RUN_ATTEMPT_TIMEOUT_SECONDS = 30 * 60 # 30 minutes
# Retries allowed after the first attempt before the run is marked "failed".
_MAX_RUN_RETRIES = 3
# Max HTML length (chars) inlined into the run's ``result_json``. Beyond this
# the inline copy is dropped to stay under MySQL's 64 KB ``TEXT`` cap; the full
# document is always retrievable from the page store (LONGTEXT) + index.html.
_MAX_INLINE_RESULT_HTML = 60000
# Minimum length (in characters) the generated Markdown file's content must
# reach when a task has ``require_markdown`` enabled. The model is NOT told about
# this threshold — it is enforced silently by the service.
@@ -58,6 +64,22 @@ class MarkdownRequirementError(Exception):
"""
class HtmlGenerationError(Exception):
"""Raised when an ``html_page`` run did not produce a complete, valid HTML
document (missing doctype/html/head/body, empty body, or residual scripts
after the sanitize + repair passes). The run is marked failed — a page only
counts as successful when it is a full, previewable HTML document."""
class DuplicateFavoriteError(Exception):
"""A user already bookmarked this page. Surfaced as HTTP 409 by the router
so the same page cannot be favorited twice."""
def __init__(self, page_id: str) -> None:
self.page_id = page_id
super().__init__(f"Page {page_id!r} is already favorited")
class DuplicateTaskNameError(Exception):
"""A user already owns a scheduled task with the requested name.
@@ -835,7 +857,7 @@ class ScheduledTaskService:
runtime_thread_id = str(uuid.uuid4())
try:
result = await asyncio.wait_for(
self._run_agent_attempt(task, runtime_thread_id=runtime_thread_id, scheduled_for=scheduled_for, run_id=run_id),
self._run_task_attempt(task, runtime_thread_id=runtime_thread_id, scheduled_for=scheduled_for, run_id=run_id),
timeout=_RUN_ATTEMPT_TIMEOUT_SECONDS,
)
except TimeoutError:
@@ -875,6 +897,333 @@ class ScheduledTaskService:
await self._deliver_element_result(task, result["content"])
return
async def _run_task_attempt(
self,
task: dict[str, Any],
*,
runtime_thread_id: str,
scheduled_for: datetime,
run_id: str,
) -> dict[str, Any]:
"""Dispatch a single execution attempt by task kind.
``html_page`` tasks render an HTML document via GLM5; everything else
runs the regular lead-agent attempt. The kind lives in
``execution_context.task_kind`` so no schema migration is needed.
"""
ctx = task.get("execution_context")
kind = ctx.get("task_kind") if isinstance(ctx, dict) else None
if kind == "html_page":
return await self._run_html_page_attempt(
task, runtime_thread_id=runtime_thread_id, scheduled_for=scheduled_for, run_id=run_id
)
return await self._run_agent_attempt(
task, runtime_thread_id=runtime_thread_id, scheduled_for=scheduled_for, run_id=run_id
)
@staticmethod
def _html_config(task: dict[str, Any]) -> dict[str, Any]:
ctx = task.get("execution_context")
cfg = ctx.get("html_config") if isinstance(ctx, dict) else None
return cfg if isinstance(cfg, dict) else {}
async def _invoke_chat_model(self, model_name: str | None, prompt: str) -> str:
"""Invoke a chat model once and return its text content.
Used for the GLM5 HTML generation / repair / style-extraction passes.
Thinking is disabled to keep structured-JSON output stable.
"""
from langchain_core.messages import HumanMessage
from deerflow.models import create_chat_model
# No hardcoded model: use exactly the selected model; only when none was
# selected do we fall back to the system default (config models[0]).
model = create_chat_model(name=model_name or None, thinking_enabled=False)
response = await model.ainvoke([HumanMessage(content=prompt)])
content = getattr(response, "content", response)
if isinstance(content, str):
return content
if isinstance(content, list):
parts = [part.get("text", "") if isinstance(part, dict) else str(part) for part in content]
return "\n".join(p for p in parts if p)
return str(content)
async def _gather_html_data(
self,
task: dict[str, Any],
*,
runtime_thread_id: str,
scheduled_for: datetime,
run_id: str,
) -> str:
"""Run the lead agent to collect source data for the HTML page.
Reuses the regular agent-run primitives (with the task's search query
folded into the prompt) and returns the agent's final text, which feeds
the GLM5 render step as ``本次数据``. Returns an empty string if the
agent produced nothing.
"""
cfg = self._html_config(task)
page_prompt = str(cfg.get("page_prompt") or "").strip()
gather_instruction = (
"请运用你自身的检索/搜索等技能,围绕下述页面需求获取最新、真实的资料,"
"并整理成结构化的中文要点(可用列表/小标题),供后续生成 HTML 页面使用,本步骤不要生成 HTML:\n"
)
parts = [gather_instruction]
if page_prompt:
parts.append(f"页面需求:{page_prompt}")
base_prompt = str(task.get("prompt") or "").strip()
if base_prompt:
parts.append(f"补充说明:{base_prompt}")
gather_task = {**task, "prompt": "\n".join(parts)}
# Never apply the markdown gate to the gather pass.
gather_ctx = dict(task.get("execution_context") or {})
gather_ctx.pop("require_markdown", None)
gather_ctx.pop("task_kind", None)
gather_ctx.pop("html_config", None)
# Use the task's selected model for the data-gathering agent too, so the
# whole HTML pipeline honors ``html_config.model_name`` instead of
# silently falling back to the default model for this step.
gather_model = str(cfg.get("model_name") or "").strip()
if gather_model:
gather_ctx["model_name"] = gather_model
gather_task["execution_context"] = gather_ctx
await self._ensure_thread_meta(task["user_id"], runtime_thread_id)
async with self._agent_run_lock:
record = await self._start_agent_run(gather_task, scheduled_for=scheduled_for, runtime_thread_id=runtime_thread_id)
await self.store.update_run(run_id, {"agent_run_id": record.run_id})
if record.task is not None:
await record.task
text = await self._extract_latest_ai_text(runtime_thread_id)
return text or ""
def _collect_html_skill_instructions(self) -> str:
"""Concatenate the bodies of enabled skills marked for HTML pages.
A skill opts in via ``metadata.kind: html_page`` in its SKILL.md
frontmatter. Only enabled skills (per the skills settings the user
toggles) are injected, so the render step honors the user's skill
selection. Best-effort — any failure yields an empty string.
"""
try:
from deerflow.skills import get_or_new_skill_storage
storage = get_or_new_skill_storage()
skills = storage.load_skills(enabled_only=True)
except Exception:
logger.debug("Failed to load skills for HTML render injection", exc_info=True)
return ""
blocks: list[str] = []
for skill in skills:
try:
text = Path(skill.skill_file).read_text(encoding="utf-8")
except Exception:
continue
frontmatter, body = html_page_utils.parse_frontmatter(text)
meta = frontmatter.get("metadata") if isinstance(frontmatter, dict) else None
if not (isinstance(meta, dict) and str(meta.get("kind")) == "html_page"):
continue
body = body.strip()
if body:
blocks.append(f"### 技能:{skill.name}\n{body}")
return "\n\n".join(blocks)
def _archive_html_index(self, run_id: str, html: str) -> Path | None:
"""Write the generated HTML to the run's archive dir as ``index.html``."""
try:
archive_dir = get_paths().base_dir / "scheduled-task-runs" / run_id / "files"
archive_dir.mkdir(parents=True, exist_ok=True)
target = archive_dir / "index.html"
target.write_text(html, encoding="utf-8")
return target
except Exception:
logger.warning("Failed to archive generated HTML for run %s", run_id, exc_info=True)
return None
async def _run_html_page_attempt(
self,
task: dict[str, Any],
*,
runtime_thread_id: str,
scheduled_for: datetime,
run_id: str,
) -> dict[str, Any]:
"""Generate an HTML page for an ``html_page`` task.
Pipeline: gather data via the lead agent → optionally load a reference
favorite's style/layout → GLM5 render → sanitize → validate → (one
GLM5 repair pass if invalid) → best-effort screenshot → archive
``index.html`` → persist a ``scheduled_html_pages`` row → build the run
result payload (``format: "html"`` with the full HTML inline).
"""
cfg = self._html_config(task)
# The user's selected model only — no hardcoded GLM-5. Empty → None →
# system default (config models[0]).
model_name = str(cfg.get("model_name") or "").strip() or None
page_prompt = str(cfg.get("page_prompt") or "").strip()
negative_prompt = str(cfg.get("negative_prompt") or "").strip()
reference_favorite_id = cfg.get("reference_favorite_id") or None
# 1) Gather source data (real, via the agent's search tools).
data_text = await self._gather_html_data(
task, runtime_thread_id=runtime_thread_id, scheduled_for=scheduled_for, run_id=run_id
)
# 2) Load reference style/layout from a favorite if one was selected.
reference_style = None
reference_layout = None
if reference_favorite_id:
store = self._html_store
if store is not None:
try:
favorite = await store.get_favorite(str(reference_favorite_id))
except Exception:
favorite = None
if favorite is not None:
reference_style = favorite.get("reuse_prompt") or favorite.get("style_summary")
reference_layout = favorite.get("layout_summary")
# 3) Render with the chosen model, injecting any enabled HTML skills.
skill_instructions = self._collect_html_skill_instructions()
gen_prompt = html_page_utils.build_generation_prompt(
page_prompt=page_prompt,
negative_prompt=negative_prompt,
data=data_text,
reference_style=reference_style,
reference_layout=reference_layout,
skill_instructions=skill_instructions,
)
raw = await self._invoke_chat_model(model_name, gen_prompt)
parsed = html_page_utils.extract_json_object(raw) or {}
title = str(parsed.get("title") or task.get("name") or "生成页面").strip()
html = parsed.get("html")
if not isinstance(html, str) or "<" not in html:
# GLM5 may have replied with bare HTML instead of JSON — accept it.
if isinstance(raw, str) and ("<html" in raw.lower() or "<!doctype" in raw.lower()):
html = raw
else:
raise ValueError("GLM5 未返回有效的 HTML(JSON 解析失败)")
style_summary = str(parsed.get("style_summary") or "").strip() or None
layout_summary = str(parsed.get("layout_summary") or "").strip() or None
# 4) Sanitize + validate, with one repair pass on failure.
html, _removed = html_page_utils.sanitize_html(html)
errors = html_page_utils.validate_html(html)
if errors:
logger.info("Generated HTML failed validation (%s); attempting repair", errors)
repair_prompt = html_page_utils.build_repair_prompt(
html=html, errors=errors, requirements=page_prompt or task.get("prompt")
)
try:
repaired_raw = await self._invoke_chat_model(model_name, repair_prompt)
repaired = html_page_utils.extract_json_object(repaired_raw) or {}
fixed = repaired.get("fixed_html")
if isinstance(fixed, str) and "<" in fixed:
fixed_clean, _ = html_page_utils.sanitize_html(fixed)
if not html_page_utils.validate_html(fixed_clean):
html = fixed_clean
except Exception:
logger.warning("HTML repair pass failed; keeping sanitized original", exc_info=True)
# 4b) Completeness gate — a page only counts as successful when it is a
# full, valid HTML document. If it still fails validation after the
# repair pass, fail the run (no half-baked page is archived/persisted).
final_errors = html_page_utils.validate_html(html)
if final_errors:
raise HtmlGenerationError(
"HTML 生成未通过完整性校验:" + ";".join(final_errors) + "。请调整页面要求或重试。"
)
# 5) Best-effort screenshot QA (no-op without Playwright).
screenshot_path = None
try:
shot_target = get_paths().base_dir / "scheduled-task-runs" / run_id / "files" / "preview.png"
shot_target.parent.mkdir(parents=True, exist_ok=True)
screenshot_path = await html_page_utils.screenshot_html(html, shot_target)
except Exception:
screenshot_path = None
# 6) Archive index.html.
archive_path = self._archive_html_index(run_id, html)
files: list[dict[str, Any]] = []
if archive_path is not None:
files.append(
{
"name": "index.html",
"path": str(archive_path),
"virtual_path": "/mnt/user-data/outputs/index.html",
"mime_type": "text/html",
"size": archive_path.stat().st_size if archive_path.exists() else None,
"kind": "html",
}
)
if screenshot_path:
shot = Path(screenshot_path)
files.append(
{
"name": "preview.png",
"path": str(shot),
"mime_type": "image/png",
"size": shot.stat().st_size if shot.exists() else None,
"kind": "artifact",
}
)
data_snapshot = {"data_text": data_text[:8000] if isinstance(data_text, str) else ""}
prompt_snapshot = {
"page_prompt": page_prompt,
"negative_prompt": negative_prompt,
"reference_favorite_id": reference_favorite_id,
"model_name": model_name,
}
# 7) Persist the page asset (best-effort — failure must not fail the run).
store = self._html_store
if store is not None:
try:
await store.create_page(
{
"id": str(uuid.uuid4()),
"task_id": task["task_id"],
"run_id": run_id,
"user_id": task["user_id"],
"title": title,
"html_content": html,
"html_file_path": str(archive_path) if archive_path else None,
"data_snapshot": data_snapshot,
"prompt_snapshot": prompt_snapshot,
"reference_favorite_id": reference_favorite_id,
"style_summary": style_summary,
"layout_summary": layout_summary,
}
)
except Exception:
logger.warning("Failed to persist scheduled_html_pages row for run %s", run_id, exc_info=True)
task_tz = ZoneInfo(task.get("timezone") or "UTC")
local_time = scheduled_for.astimezone(task_tz)
# ``scheduled_task_runs.result_json`` is TEXT (64 KB) on MySQL, so only
# inline the HTML when it comfortably fits. The full document always
# lives in ``scheduled_html_pages.html_content`` (LONGTEXT) + the
# archived ``index.html``; the ``/runs/{id}/html`` endpoints read from
# there, so previews work even when the inline copy is omitted here.
inline_html = html if len(html) <= _MAX_INLINE_RESULT_HTML else None
return {
"title": f"{title} - {local_time.strftime('%Y-%m-%d %H:%M')}",
"content": page_prompt or f"HTML 页面已生成:{title}",
"format": "html",
"html": inline_html,
"style_summary": style_summary,
"layout_summary": layout_summary,
"data_snapshot": data_snapshot,
"prompt_snapshot": prompt_snapshot,
"files": files,
"created_at": datetime.now(UTC).isoformat(),
}
async def _run_agent_attempt(
self,
task: dict[str, Any],
@@ -1275,3 +1624,132 @@ class ScheduledTaskService:
async def unsubscribe_task(self, user_id: str, task_id: str) -> bool:
return await self.store.delete_subscription(task_id, user_id)
# --- HTML pages / favorites ---
@property
def _html_store(self):
state = getattr(self.app, "state", None)
return getattr(state, "html_page_store", None)
async def get_html_page_for_run(self, run_id: str) -> dict[str, Any] | None:
"""Return the generated HTML page for a run (page-store row or run result)."""
store = self._html_store
if store is not None:
page = await store.get_page_by_run(run_id)
if page is not None:
return page
# Fallback to the inline HTML on the run result.
run = await self.store.get_run_any(run_id)
if run is None:
return None
result = run.get("result") if isinstance(run.get("result"), dict) else None
if not result or not isinstance(result.get("html"), str):
return None
return {
"run_id": run_id,
"task_id": run.get("task_id"),
"title": result.get("title"),
"html_content": result.get("html"),
"style_summary": result.get("style_summary"),
"layout_summary": result.get("layout_summary"),
"created_at": result.get("created_at"),
}
async def get_latest_html_page_for_task(self, task_id: str) -> dict[str, Any] | None:
store = self._html_store
if store is not None:
page = await store.latest_page_for_task(task_id)
if page is not None:
return page
# Fallback: newest run that carries inline HTML.
runs = await self.store.list_runs_any(task_id, limit=20)
for run in runs:
result = run.get("result") if isinstance(run.get("result"), dict) else None
if result and isinstance(result.get("html"), str):
return await self.get_html_page_for_run(run["id"])
return None
async def list_html_page_favorites(self, user_id: str) -> list[dict[str, Any]]:
store = self._html_store
if store is None:
return []
return await store.list_favorites(user_id)
async def get_html_page_favorite(self, user_id: str, favorite_id: str) -> dict[str, Any] | None:
store = self._html_store
if store is None:
return None
favorite = await store.get_favorite(favorite_id)
if favorite is None or favorite.get("user_id") != user_id:
return None
return favorite
async def delete_html_page_favorite(self, user_id: str, favorite_id: str) -> bool:
store = self._html_store
if store is None:
return False
return await store.delete_favorite(favorite_id, user_id)
async def create_html_page_favorite(
self,
user_id: str,
page_id: str,
*,
title: str | None = None,
description: str | None = None,
tags: list[str] | None = None,
) -> dict[str, Any] | None:
"""Bookmark a generated page, extracting a reusable style summary via GLM5.
Returns ``None`` when the page does not exist or is not owned by the
caller. Style extraction is best-effort — a failure still produces a
favorite (without the reuse hints) so the user's bookmark is never lost.
"""
store = self._html_store
if store is None:
return None
page = await store.get_page(page_id)
if page is None or page.get("user_id") != user_id:
return None
# A page can only be bookmarked once per user.
existing = await store.get_favorite_by_page(user_id, page_id)
if existing is not None:
raise DuplicateFavoriteError(page_id)
html = page.get("html_content") or ""
style_summary = page.get("style_summary")
layout_summary = page.get("layout_summary")
content_structure_summary = None
reuse_prompt = None
if html:
try:
# Reuse the model the page was generated with (recorded in the
# page's prompt snapshot) so favoriting doesn't fall back to the
# default model.
extract_model = (page.get("prompt_snapshot") or {}).get("model_name")
extract_prompt = html_page_utils.build_style_extract_prompt(html=html)
raw = await self._invoke_chat_model(extract_model, extract_prompt)
extracted = html_page_utils.extract_json_object(raw) or {}
style_summary = extracted.get("style_summary") or style_summary
layout_summary = extracted.get("layout_summary") or layout_summary
content_structure_summary = extracted.get("content_structure_summary")
reuse_prompt = extracted.get("reuse_prompt")
except Exception:
logger.warning("HTML style extraction failed for page %s; saving favorite without hints", page_id, exc_info=True)
return await store.create_favorite(
{
"id": str(uuid.uuid4()),
"user_id": user_id,
"page_id": page_id,
"title": (title or page.get("title") or "收藏页面").strip(),
"description": (description or None),
"tags": tags or [],
"html_snapshot": html,
"style_summary": style_summary,
"layout_summary": layout_summary,
"content_structure_summary": content_structure_summary,
"reuse_prompt": reuse_prompt,
}
)