diff --git a/.gitignore b/.gitignore
index b89ebd6..952319d 100644
--- a/.gitignore
+++ b/.gitignore
@@ -18,6 +18,8 @@ backend/.env
# 运行期生成的 SQLite 索引(vault 下的 Markdown 测试数据需提交)
backend/data/*.db*
backend/data/credentials/
+# 运行期导出的 HTML/PDF/DOCX 产物(不提交)
+backend/data/exports/
backend/data/logs/
# 阶段验收笔记(验收用,不提交)
backend/data/vault/验收/
diff --git a/backend/app/config.py b/backend/app/config.py
index 00334c9..3d3d753 100644
--- a/backend/app/config.py
+++ b/backend/app/config.py
@@ -25,6 +25,7 @@ class Settings:
vault_path: Path
attachments_path: Path
benchmark_datasets_path: Path
+ exports_path: Path
@lru_cache
@@ -45,4 +46,5 @@ def get_settings() -> Settings:
benchmark_datasets_path=Path(
os.getenv("APP_BENCHMARK_DATASETS_PATH", str(data_dir / "benchmarks"))
),
+ exports_path=Path(os.getenv("APP_EXPORTS_PATH", str(data_dir / "exports"))),
)
diff --git a/backend/app/contracts.py b/backend/app/contracts.py
index 2cc66b1..154c062 100644
--- a/backend/app/contracts.py
+++ b/backend/app/contracts.py
@@ -2,7 +2,14 @@ from datetime import datetime
from enum import Enum
from typing import Annotated, Any, Literal
-from pydantic import BaseModel, ConfigDict, Field, SecretStr, field_validator, model_validator
+from pydantic import (
+ BaseModel,
+ ConfigDict,
+ Field,
+ SecretStr,
+ field_validator,
+ model_validator,
+)
from app.request_overrides import RequestOverride
@@ -1307,3 +1314,88 @@ class BenchmarkReport(Contract):
cases: list[RAGCaseResult] = Field(default_factory=list)
error: str | None = None
error_code: str | None = None
+
+
+# Export(多格式文档导出)
+class ExportStatus(str, Enum):
+ queued = "queued"
+ running = "running"
+ completed = "completed"
+ failed = "failed"
+ cancelled = "cancelled"
+
+
+class ExportFormat(str, Enum):
+ html = "html"
+ pdf = "pdf"
+ docx = "docx"
+
+
+class ExportSourceType(str, Enum):
+ note = "note"
+ markdown = "markdown"
+
+
+class ExportSource(Contract):
+ """导出源:note 引用已索引笔记,markdown 用于未保存预览(不持久化)。"""
+
+ type: ExportSourceType
+ note_id: str | None = None
+ markdown: str | None = None
+
+ @model_validator(mode="after")
+ def _validate_source(self) -> "ExportSource":
+ if self.type == ExportSourceType.note and not self.note_id:
+ raise ValueError("note source requires note_id")
+ if self.type == ExportSourceType.markdown and not self.markdown:
+ raise ValueError("markdown source requires markdown")
+ return self
+
+
+class ExportOptions(Contract):
+ theme_id: str = "light"
+ include_title: bool = True
+ include_metadata: bool = False
+ page_size: str = "A4"
+ code_theme: str = "github-light"
+
+
+class ExportRequest(Contract):
+ source: ExportSource
+ format: ExportFormat
+ options: ExportOptions = Field(default_factory=ExportOptions)
+
+
+class ExportProgress(Contract):
+ phase: str
+ current: int
+ total: int
+ percent: float | None = None
+ message: str | None = None
+
+
+class ExportFile(Contract):
+ file_name: str
+ mime_type: str
+ size: int
+ sha256: str
+ expires_at: datetime
+
+
+class ExportJob(Contract):
+ job_id: str
+ status: ExportStatus
+ format: ExportFormat
+ progress: ExportProgress | None = None
+ file: ExportFile | None = None
+ warnings: list[str] = Field(default_factory=list)
+ error: str | None = None
+ error_code: str | None = None
+ created_at: datetime
+ started_at: datetime | None = None
+ completed_at: datetime | None = None
+
+
+class ExportJobListResponse(Contract):
+ items: list[ExportJob] = Field(default_factory=list)
+ page: PageMeta = Field(default_factory=PageMeta)
diff --git a/backend/app/export/__init__.py b/backend/app/export/__init__.py
new file mode 100644
index 0000000..d5a6faa
--- /dev/null
+++ b/backend/app/export/__init__.py
@@ -0,0 +1,8 @@
+"""Export Service:多格式文档导出(首批 HTML)。
+
+模块划分:
+- document.py Document AST 内部协议 + DocumentExporter Protocol + ExportResult
+- markdown.py mistune → Document AST 解析
+- exporters/html.py HtmlExporter(Document AST → HTML5)
+- service.py 导出任务注册表、后台执行、取消与文件生命周期
+"""
diff --git a/backend/app/export/document.py b/backend/app/export/document.py
new file mode 100644
index 0000000..e09f5cb
--- /dev/null
+++ b/backend/app/export/document.py
@@ -0,0 +1,46 @@
+"""Document AST:导出器的内部中间表示(Internal Protocol,不放入 contracts.py)。
+
+契约 §10.3 规定节点用稳定判别字段 node_id / type / attributes / children / text,
+类型专有信息统一放 attributes(如 heading 的 level、link 的 href、image 的 src)。
+导出器据此递归渲染,对无法表示的节点记 warning,不静默丢弃。
+"""
+
+from __future__ import annotations
+
+from typing import Any, Protocol
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from app.contracts import ExportOptions
+
+
+class DocumentNode(BaseModel):
+ """递归文档节点;type 取契约 §10.3 首批 node type 之一。"""
+
+ model_config = ConfigDict(extra="forbid")
+
+ type: str
+ node_id: str
+ attributes: dict[str, Any] = Field(default_factory=dict)
+ children: list["DocumentNode"] = Field(default_factory=list)
+ text: str = ""
+
+
+class Document(DocumentNode):
+ """根节点,type 固定为 document。"""
+
+ type: str = "document"
+
+
+class DocumentExporter(Protocol):
+ """导出器协议(契约 §10.3):把 Document AST 渲染为指定格式的产物。"""
+
+ async def export(self, document: Document, options: ExportOptions) -> "ExportResult": ...
+
+
+class ExportResult(BaseModel):
+ model_config = ConfigDict(extra="forbid")
+
+ content: bytes
+ mime_type: str
+ warnings: list[str] = Field(default_factory=list)
diff --git a/backend/app/export/exporters/__init__.py b/backend/app/export/exporters/__init__.py
new file mode 100644
index 0000000..018a8ab
--- /dev/null
+++ b/backend/app/export/exporters/__init__.py
@@ -0,0 +1 @@
+"""Export 渲染器:Document AST → 具体格式产物。"""
diff --git a/backend/app/export/exporters/_common.py b/backend/app/export/exporters/_common.py
new file mode 100644
index 0000000..cbf41ed
--- /dev/null
+++ b/backend/app/export/exporters/_common.py
@@ -0,0 +1,79 @@
+"""导出器共享工具:URL 协议校验、函数图像预算与占位 warning 文案。
+
+html / pdf / docx 三个导出器共用同一套安全规则与函数图像资源预算,避免各写一份
+导致行为漂移。
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+from urllib.parse import urlparse
+
+# 链接/图片地址允许的协议;无 scheme 的相对地址视为安全,其余协议一律降级
+ALLOWED_URL_SCHEMES = frozenset({"http", "https", "mailto"})
+
+MERMAID_WARNING = "mermaid 需前端渲染,已保留为占位代码块"
+RAW_HTML_WARNING = "原始 HTML 已按纯文本转义保留"
+# DOCX 暂不支持静态渲染函数图像,统一回退源码占位
+PLOT_PLACEHOLDER_WARNING = "函数图像:该格式暂不支持静态渲染,已保留为源码占位"
+
+# 单篇文档允许的函数图像数量上限,超出部分回退占位,防止多图块并发采样耗尽内存/线程
+MAX_FUNCTION_PLOTS = 16
+# 单篇文档允许的函数图像累计 AST 节点预算,超出部分回退占位,防止组合复杂度(多图块
+# × 多表达式 × 深表达式)在采样求值时长时间占满 CPU
+MAX_TOTAL_PLOT_NODES = 8000
+
+
+class FunctionPlotBudget:
+ """函数图像文档级资源预算:数量上限 + 累计 AST 节点上限。
+
+ HTML 与 PDF 导出器在渲染每个 function-plot 图块前先问预算,超限即回退源码占位,
+ 不解析不采样,避免多图块组合复杂度耗尽内存/CPU。
+ """
+
+ def __init__(self, max_plots: int | None = None, max_total_nodes: int | None = None) -> None:
+ # 默认读模块常量(便于测试 monkeypatch 常量后重新生效)
+ self.max_plots = MAX_FUNCTION_PLOTS if max_plots is None else max_plots
+ self.max_total_nodes = MAX_TOTAL_PLOT_NODES if max_total_nodes is None else max_total_nodes
+ self.count = 0
+ self.total_nodes = 0
+
+ def check_count(self) -> str | None:
+ """图块数量 +1;超限返回 warning 文案,否则返回 None。"""
+ self.count += 1
+ if self.count > self.max_plots:
+ return f"函数图像:文档内函数图像数量超过上限 {self.max_plots},已回退为源码占位"
+ return None
+
+ def check_nodes(self, node_count: int) -> str | None:
+ """累计节点预算校验;超限返回 warning 文案(不累加),否则累加并返回 None。"""
+ if self.total_nodes + node_count > self.max_total_nodes:
+ return f"函数图像:文档内函数图像累计复杂度超过上限 {self.max_total_nodes} 节点,已回退为源码占位"
+ self.total_nodes += node_count
+ return None
+
+
+def format_plot_diagnostic(diag) -> str:
+ """把解析诊断格式化为面向用户的 warning 文案。"""
+ loc = f"(第 {diag.line} 行)" if diag.line else ""
+ return f"函数图像:{diag.message}{loc}"
+
+
+def safe_url(url: str) -> str | None:
+ """校验 URL 协议;安全返回原串,不安全返回 None。"""
+ url = url.strip()
+ if not url:
+ return None
+ scheme = urlparse(url).scheme.lower()
+ if scheme and scheme not in ALLOWED_URL_SCHEMES:
+ return None
+ return url
+
+
+def format_meta_value(value: object) -> str:
+ """把元数据值转成可读文本:datetime 转 ISO、列表用逗号连接。"""
+ if isinstance(value, datetime):
+ return value.isoformat()
+ if isinstance(value, list):
+ return ", ".join(str(item) for item in value)
+ return str(value)
diff --git a/backend/app/export/exporters/docx.py b/backend/app/export/exporters/docx.py
new file mode 100644
index 0000000..74c9d9c
--- /dev/null
+++ b/backend/app/export/exporters/docx.py
@@ -0,0 +1,347 @@
+"""DocxExporter:Document AST → DOCX(python-docx)。
+
+v1 为文本优先:标题/段落/行内强调与链接/列表/引用/表格/代码块/数学文本均可导出;
+function_plot 与 mermaid 保留源码占位并记 warning。中文字体通过 Normal 样式挂载
+w:eastAsia=宋体,保证 Word 打开时中文正常显示;bold/italic 由 Word 原生渲染。
+"""
+
+from __future__ import annotations
+
+from io import BytesIO
+
+from docx import Document as DocxDocument
+from docx.enum.text import WD_ALIGN_PARAGRAPH
+from docx.opc.constants import RELATIONSHIP_TYPE
+from docx.oxml import OxmlElement
+from docx.oxml.ns import qn
+from docx.shared import Inches, Mm, Pt, RGBColor
+
+from app.contracts import ExportOptions
+from app.export.document import Document, DocumentNode, ExportResult
+from app.export.exporters._common import (
+ MERMAID_WARNING,
+ PLOT_PLACEHOLDER_WARNING,
+ RAW_HTML_WARNING,
+ format_meta_value,
+ safe_url,
+)
+
+_MIME = "application/vnd.openxmlformats-officedocument.wordprocessingml.document"
+
+_HEADING_SIZES = {1: 20, 2: 16, 3: 14, 4: 12, 5: 11, 6: 10.5}
+
+
+def _plain_text(children: list[DocumentNode]) -> str:
+ """递归拼接行内节点的纯文本,供标题/链接文字等需要纯文本处使用。"""
+ parts: list[str] = []
+ for child in children:
+ if child.type == "text":
+ parts.append(child.text)
+ elif child.children:
+ parts.append(_plain_text(child.children))
+ elif child.text:
+ parts.append(child.text)
+ return "".join(parts)
+
+
+class DocxExporter:
+ """实现 DocumentExporter:递归渲染 Document AST 为 DOCX 字节流。"""
+
+ def render(self, document: Document, options: ExportOptions) -> ExportResult:
+ """同步渲染;CPU 密集,调用方应放入线程执行,避免阻塞事件循环。"""
+ self._doc = DocxDocument()
+ self._configure_normal_style()
+ self._configure_page(options)
+ warnings: list[str] = []
+
+ self._render_header(document, options, warnings)
+ self._render_children(document.children, warnings)
+
+ buf = BytesIO()
+ self._doc.save(buf)
+ return ExportResult(content=buf.getvalue(), mime_type=_MIME, warnings=warnings)
+
+ async def export(self, document: Document, options: ExportOptions) -> ExportResult:
+ """契约要求的 async 接口;渲染本身同步,直接转发到 render。"""
+ return self.render(document, options)
+
+ def _configure_normal_style(self) -> None:
+ """Normal 样式挂载 CJK 字体;拉丁用 Calibri,中文用宋体。"""
+ style = self._doc.styles["Normal"]
+ style.font.name = "Calibri"
+ style.font.size = Pt(11)
+ rfonts = style.element.get_or_add_rPr().get_or_add_rFonts()
+ rfonts.set(qn("w:eastAsia"), "宋体")
+
+ def _configure_page(self, options: ExportOptions) -> None:
+ section = self._doc.sections[0]
+ size = (options.page_size or "A4").lower()
+ if size == "a4":
+ section.page_width = Mm(210)
+ section.page_height = Mm(297)
+ elif size == "letter":
+ section.page_width = Inches(8.5)
+ section.page_height = Inches(11)
+
+ # --- 文档头部 ---
+ def _render_header(self, document: Document, options: ExportOptions, warnings: list[str]) -> None:
+ title = str(document.attributes.get("title") or "")
+ if options.include_title and title:
+ p = self._doc.add_paragraph()
+ run = p.add_run(title)
+ run.bold = True
+ run.font.size = Pt(22)
+ p.paragraph_format.space_after = Pt(12)
+ if options.include_metadata:
+ metadata = document.attributes.get("metadata")
+ if metadata:
+ for key, value in metadata.items():
+ p = self._doc.add_paragraph()
+ run = p.add_run(f"{key}: {format_meta_value(value)}")
+ run.font.size = Pt(9)
+ run.font.color.rgb = RGBColor(0x57, 0x60, 0x6A)
+
+ # --- 块级 ---
+ def _render_children(self, children: list[DocumentNode], warnings: list[str]) -> None:
+ for child in children:
+ self._render_block(child, warnings)
+
+ def _render_block(self, node: DocumentNode, warnings: list[str]) -> None:
+ handler = getattr(self, f"_block_{node.type}", None)
+ if handler is not None:
+ handler(node, warnings)
+ else:
+ warnings.append(f"无法表示的节点类型已跳过:{node.type}")
+
+ def _block_heading(self, node: DocumentNode, warnings: list[str]) -> None:
+ level = max(1, min(6, int(node.attributes.get("level", 1))))
+ p = self._doc.add_paragraph()
+ run = p.add_run(_plain_text(node.children))
+ run.bold = True
+ run.font.size = Pt(_HEADING_SIZES[level])
+ p.paragraph_format.space_before = Pt(14 if level <= 2 else 10)
+ p.paragraph_format.space_after = Pt(6)
+
+ def _block_paragraph(self, node: DocumentNode, warnings: list[str]) -> None:
+ p = self._doc.add_paragraph()
+ self._render_inline(p, node.children, warnings)
+
+ def _block_blockquote(self, node: DocumentNode, warnings: list[str]) -> None:
+ # 引用块的直接子节点是块级节点(paragraph/list 等),不能交给行内渲染器,
+ # 否则正文会被当作「无法表示的行内节点」丢弃;逐个渲染并继承引用缩进/颜色。
+ for child in node.children:
+ if child.type == "paragraph":
+ p = self._doc.add_paragraph()
+ self._render_inline(p, child.children, warnings)
+ p.paragraph_format.left_indent = Pt(16)
+ for run in p.runs:
+ run.font.color.rgb = RGBColor(0x57, 0x60, 0x6A)
+ elif child.type == "list":
+ self._block_list(child, warnings, level=1, color=RGBColor(0x57, 0x60, 0x6A))
+ else:
+ self._render_block(child, warnings)
+
+ def _block_list(
+ self,
+ node: DocumentNode,
+ warnings: list[str],
+ level: int = 0,
+ color: RGBColor | None = None,
+ ) -> None:
+ ordered = bool(node.attributes.get("ordered"))
+ for index, item in enumerate(node.children, start=1):
+ self._block_list_item(item, warnings, ordered, index, level, color)
+
+ def _block_list_item(
+ self,
+ item: DocumentNode,
+ warnings: list[str],
+ ordered: bool,
+ index: int,
+ level: int,
+ color: RGBColor | None = None,
+ ) -> None:
+ if item.attributes.get("task"):
+ marker = "☑ " if item.attributes.get("checked") else "☐ "
+ else:
+ marker = f"{index}. " if ordered else "• "
+ indent = Pt(18 + 18 * level)
+ first = True
+ for child in item.children:
+ if child.type == "list":
+ self._block_list(child, warnings, level + 1, color)
+ continue
+ p = self._doc.add_paragraph()
+ p.paragraph_format.left_indent = indent
+ if first:
+ self._add_run(p, marker)
+ first = False
+ if child.type == "paragraph":
+ # 块级容器:展开其行内子节点
+ self._render_inline(p, child.children, warnings)
+ else:
+ # 直接行内节点(text/strong/emphasis/link/codespan 等):走行内渲染保留
+ # 语义(加粗/斜体/超链接),不能只渲染其 children 而丢掉格式。
+ self._render_inline_node(p, child, warnings)
+ if color is not None:
+ for run in p.runs:
+ run.font.color.rgb = color
+
+ def _block_table(self, node: DocumentNode, warnings: list[str]) -> None:
+ rows = node.children
+ ncols = max((len(r.children) for r in rows), default=0)
+ if not rows or ncols == 0:
+ return
+ table = self._doc.add_table(rows=len(rows), cols=ncols)
+ table.style = "Table Grid"
+ for ri, row in enumerate(rows):
+ head = bool(row.attributes.get("head"))
+ for ci in range(ncols):
+ cell = table.cell(ri, ci)
+ p = cell.paragraphs[0]
+ if ci < len(row.children):
+ self._render_inline(p, row.children[ci].children, warnings, bold=head)
+
+ def _block_code_block(self, node: DocumentNode, warnings: list[str]) -> None:
+ lines = node.text.split("\n")
+ p = self._doc.add_paragraph()
+ self._shade_paragraph(p)
+ p.paragraph_format.left_indent = Pt(8)
+ p.paragraph_format.right_indent = Pt(8)
+ p.paragraph_format.space_before = Pt(6)
+ p.paragraph_format.space_after = Pt(8)
+ for i, line in enumerate(lines):
+ run = p.add_run(line)
+ run.font.name = "Consolas"
+ run.font.size = Pt(10)
+ if i < len(lines) - 1:
+ run.add_break()
+
+ def _block_thematic_break(self, node: DocumentNode, warnings: list[str]) -> None:
+ p = self._doc.add_paragraph()
+ pPr = p._p.get_or_add_pPr()
+ pBdr = OxmlElement("w:pBdr")
+ bottom = OxmlElement("w:bottom")
+ bottom.set(qn("w:val"), "single")
+ bottom.set(qn("w:sz"), "6")
+ bottom.set(qn("w:space"), "1")
+ bottom.set(qn("w:color"), "D0D7DE")
+ pBdr.append(bottom)
+ pPr.append(pBdr)
+
+ def _block_mermaid(self, node: DocumentNode, warnings: list[str]) -> None:
+ warnings.append(MERMAID_WARNING)
+ self._block_code_block(node, warnings)
+
+ def _block_function_plot(self, node: DocumentNode, warnings: list[str]) -> None:
+ warnings.append(PLOT_PLACEHOLDER_WARNING)
+ self._block_code_block(node, warnings)
+
+ def _block_math_block(self, node: DocumentNode, warnings: list[str]) -> None:
+ p = self._doc.add_paragraph()
+ p.alignment = WD_ALIGN_PARAGRAPH.CENTER
+ p.add_run(f"$${node.text}$$")
+
+ def _block_html_block(self, node: DocumentNode, warnings: list[str]) -> None:
+ # 原始 HTML 不可信,按纯文本保留正文
+ warnings.append(RAW_HTML_WARNING)
+ self._doc.add_paragraph(node.text)
+
+ # --- 行内(写入 run) ---
+ def _render_inline(
+ self,
+ paragraph,
+ children: list[DocumentNode],
+ warnings: list[str],
+ bold: bool = False,
+ italic: bool = False,
+ ) -> None:
+ for child in children:
+ self._render_inline_node(paragraph, child, warnings, bold, italic)
+
+ def _render_inline_node(
+ self,
+ paragraph,
+ node: DocumentNode,
+ warnings: list[str],
+ bold: bool = False,
+ italic: bool = False,
+ ) -> None:
+ t = node.type
+ if t == "text":
+ self._add_run(paragraph, node.text, bold=bold, italic=italic)
+ elif t == "strong":
+ self._render_inline(paragraph, node.children, warnings, bold=True, italic=italic)
+ elif t == "emphasis":
+ self._render_inline(paragraph, node.children, warnings, bold=bold, italic=True)
+ elif t == "codespan":
+ self._add_run(paragraph, node.text, code=True)
+ elif t == "link":
+ inner = _plain_text(node.children)
+ href = str(node.attributes.get("href") or "")
+ safe_href = safe_url(href)
+ if safe_href is None:
+ warnings.append(f"链接协议不安全,已降级为纯文本:{href!r}")
+ self._render_inline(paragraph, node.children, warnings, bold, italic)
+ else:
+ self._add_hyperlink(paragraph, safe_href, inner)
+ elif t == "image":
+ src = str(node.attributes.get("src") or "")
+ alt = str(node.attributes.get("alt") or "")
+ if safe_url(src) is None:
+ warnings.append(f"图片地址不安全,已跳过:{src!r}")
+ else:
+ warnings.append("图片未内嵌到 DOCX,已用替代文本表示")
+ if alt:
+ self._add_run(paragraph, alt)
+ elif t == "math_inline":
+ self._add_run(paragraph, f"\\({node.text}\\)")
+ elif t == "linebreak":
+ self._add_run(paragraph, "").add_break()
+ else:
+ warnings.append(f"无法表示的行内节点已跳过:{t}")
+
+ def _add_run(self, paragraph, text: str, bold: bool = False, italic: bool = False, code: bool = False):
+ run = paragraph.add_run(text)
+ run.bold = bold
+ run.italic = italic
+ if code:
+ run.font.name = "Consolas"
+ run.font.size = Pt(10)
+ return run
+
+ def _add_hyperlink(self, paragraph, url: str, text: str) -> None:
+ """写入可点击的超链接 run(python-docx 无公开 API,需手写 w:hyperlink)。"""
+ part = paragraph.part
+ r_id = part.relate_to(url, RELATIONSHIP_TYPE.HYPERLINK, is_external=True)
+ hyperlink = OxmlElement("w:hyperlink")
+ hyperlink.set(qn("r:id"), r_id)
+ run = OxmlElement("w:r")
+ rPr = OxmlElement("w:rPr")
+ rFonts = OxmlElement("w:rFonts")
+ rFonts.set(qn("w:ascii"), "Calibri")
+ rFonts.set(qn("w:hAnsi"), "Calibri")
+ rFonts.set(qn("w:eastAsia"), "宋体")
+ rPr.append(rFonts)
+ color = OxmlElement("w:color")
+ color.set(qn("w:val"), "0969DA")
+ rPr.append(color)
+ u = OxmlElement("w:u")
+ u.set(qn("w:val"), "single")
+ rPr.append(u)
+ run.append(rPr)
+ t = OxmlElement("w:t")
+ t.text = text
+ t.set(qn("xml:space"), "preserve")
+ run.append(t)
+ hyperlink.append(run)
+ paragraph._p.append(hyperlink)
+
+ def _shade_paragraph(self, paragraph, fill: str = "F2F2F2") -> None:
+ """给段落加浅灰底纹,用于代码块占位。"""
+ pPr = paragraph._p.get_or_add_pPr()
+ shd = OxmlElement("w:shd")
+ shd.set(qn("w:val"), "clear")
+ shd.set(qn("w:color"), "auto")
+ shd.set(qn("w:fill"), fill)
+ pPr.append(shd)
diff --git a/backend/app/export/exporters/html.py b/backend/app/export/exporters/html.py
new file mode 100644
index 0000000..54af119
--- /dev/null
+++ b/backend/app/export/exporters/html.py
@@ -0,0 +1,284 @@
+"""HtmlExporter:Document AST → 完整 HTML5 文档(内嵌基础 CSS)。
+
+mermaid 等无法静态表达的节点渲染为占位代码块并记 warning,不静默丢失;function_plot
+解析为静态 SVG 内嵌(解析失败回退占位并转诊断);严重内容缺失由 service 层以
+EXPORT_UNSUPPORTED_CONTENT 判定,本层只负责逐节点渲染。
+"""
+
+from __future__ import annotations
+
+import html
+from datetime import datetime
+from urllib.parse import urlparse
+
+from app.contracts import ExportOptions
+from app.export.document import Document, DocumentNode, ExportResult
+from app.export.exporters._common import FunctionPlotBudget, format_plot_diagnostic
+from app.plot.renderer import FunctionPlotStaticRenderer, StaticRenderRequest
+
+_MERMAID_WARNING = "mermaid 需前端渲染,已保留为占位代码块"
+_RAW_HTML_WARNING = "原始 HTML 已按纯文本转义保留"
+
+# 链接/图片地址允许的协议;无 scheme 的相对地址视为安全,其余协议一律降级
+_ALLOWED_URL_SCHEMES = frozenset({"http", "https", "mailto"})
+
+
+def _safe_url(url: str) -> str | None:
+ """校验 URL 协议;安全返回原串,不安全返回 None。"""
+ url = url.strip()
+ if not url:
+ return None
+ scheme = urlparse(url).scheme.lower()
+ if scheme and scheme not in _ALLOWED_URL_SCHEMES:
+ return None
+ return url
+
+_BASE_CSS = """
+body { margin: 0; background: #f6f7f9; color: #1f2328; font: 15px/1.7 -apple-system, 'Segoe UI', 'Microsoft YaHei', sans-serif; }
+article { max-width: 860px; margin: 0 auto; padding: 40px 48px; background: #fff; }
+article.theme-dark { background: #0d1117; color: #c9d1d9; }
+h1, h2, h3, h4, h5, h6 { line-height: 1.3; margin: 1.4em 0 0.6em; }
+h1.title { margin-top: 0; }
+p { margin: 0.6em 0; }
+a { color: #0969da; }
+code { font-family: 'JetBrains Mono', Consolas, monospace; font-size: 0.9em; background: #f0f1f3; padding: 0.15em 0.35em; border-radius: 3px; }
+pre { background: #f6f8fa; padding: 14px 16px; border-radius: 6px; overflow-x: auto; }
+pre.code-theme-github-dark { background: #0d1117; color: #c9d1d9; }
+pre code { background: none; padding: 0; }
+pre.mermaid, pre.function-plot { border: 1px dashed #d0d7de; }
+figure.function-plot { margin: 1em 0; text-align: center; }
+figure.function-plot svg { max-width: 100%; height: auto; }
+blockquote { margin: 0.8em 0; padding: 0.2em 1em; border-left: 4px solid #d0d7de; color: #57606a; }
+img { max-width: 100%; }
+table { border-collapse: collapse; margin: 0.8em 0; }
+th, td { border: 1px solid #d0d7de; padding: 6px 12px; }
+th { background: #f6f8fa; }
+dl.metadata { font-size: 0.85em; color: #57606a; border-top: 1px solid #eaeef2; border-bottom: 1px solid #eaeef2; padding: 0.6em 0; }
+dl.metadata dt { display: inline; font-weight: 600; margin-right: 0.4em; }
+dl.metadata dd { display: inline; margin: 0 1.2em 0 0; }
+.math, .math-block { overflow-x: auto; padding: 0.4em 0; }
+.task-list-item { list-style: none; }
+.task-list-item input { margin-right: 0.4em; }
+hr { border: none; border-top: 1px solid #d0d7de; margin: 1.4em 0; }
+""".strip()
+
+
+class HtmlExporter:
+ """实现 DocumentExporter:递归渲染 Document AST 为完整 HTML5 文档。"""
+
+ def render(self, document: Document, options: ExportOptions) -> ExportResult:
+ """同步渲染;CPU 密集,调用方应放入线程执行,避免阻塞事件循环。"""
+ self._options = options
+ self._plot_budget = FunctionPlotBudget()
+ self._plot_renderer = FunctionPlotStaticRenderer()
+ warnings: list[str] = []
+ body = self._render_children(document.children, warnings)
+ content = self._assemble(document, options, body, warnings)
+ return ExportResult(
+ content=content.encode("utf-8"), mime_type="text/html", warnings=warnings
+ )
+
+ async def export(self, document: Document, options: ExportOptions) -> ExportResult:
+ """契约要求的 async 接口;渲染本身同步,直接转发到 render。"""
+ return self.render(document, options)
+
+ def _assemble(
+ self, document: Document, options: ExportOptions, body: str, warnings: list[str]
+ ) -> str:
+ title = str(document.attributes.get("title") or "")
+ parts = [
+ "",
+ '',
+ "
",
+ ' ',
+ ' ',
+ ]
+ if title:
+ parts.append(f"{html.escape(title)} ")
+ parts.append(f"")
+ parts.append("")
+ parts.append("")
+ parts.append(f'')
+ if options.include_title and title:
+ parts.append(f'{html.escape(title)} ')
+ if options.include_metadata:
+ metadata = document.attributes.get("metadata")
+ if metadata:
+ parts.append(self._render_metadata(metadata))
+ parts.append(body)
+ parts.append(" ")
+ parts.append("")
+ parts.append("")
+ return "\n".join(parts) + "\n"
+
+ def _render_metadata(self, metadata: dict) -> str:
+ entries = ["']
+ for key, value in metadata.items():
+ entries.append(f"{html.escape(str(key))} ")
+ entries.append(f"{html.escape(self._fmt_meta_value(value))} ")
+ entries.append(" ")
+ return "".join(entries)
+
+ @staticmethod
+ def _fmt_meta_value(value: object) -> str:
+ if isinstance(value, datetime):
+ return value.isoformat()
+ if isinstance(value, list):
+ return ", ".join(str(item) for item in value)
+ return str(value)
+
+ def _render_children(self, children: list[DocumentNode], warnings: list[str]) -> str:
+ return "".join(self._render_node(child, warnings) for child in children)
+
+ def _render_node(self, node: DocumentNode, warnings: list[str]) -> str:
+ handler = getattr(self, f"_render_{node.type}", None)
+ if handler is not None:
+ return handler(node, warnings)
+ warnings.append(f"无法表示的节点类型已跳过:{node.type}")
+ return ""
+
+ # --- 块级 ---
+ def _render_heading(self, node: DocumentNode, warnings: list[str]) -> str:
+ level = max(1, min(6, int(node.attributes.get("level", 1))))
+ return f"{self._render_children(node.children, warnings)} "
+
+ def _render_paragraph(self, node: DocumentNode, warnings: list[str]) -> str:
+ return f"{self._render_children(node.children, warnings)}
"
+
+ def _render_blockquote(self, node: DocumentNode, warnings: list[str]) -> str:
+ return f"{self._render_children(node.children, warnings)} "
+
+ def _render_list(self, node: DocumentNode, warnings: list[str]) -> str:
+ tag = "ol" if node.attributes.get("ordered") else "ul"
+ return f"<{tag}>{self._render_children(node.children, warnings)}{tag}>"
+
+ def _render_list_item(self, node: DocumentNode, warnings: list[str]) -> str:
+ inner = self._render_children(node.children, warnings)
+ if node.attributes.get("task"):
+ checked = " checked" if node.attributes.get("checked") else ""
+ return (
+ ''
+ f' {inner} '
+ )
+ return f"{inner} "
+
+ def _render_table(self, node: DocumentNode, warnings: list[str]) -> str:
+ rows = node.children
+ head_rows = [r for r in rows if r.attributes.get("head")]
+ body_rows = [r for r in rows if not r.attributes.get("head")]
+ parts = [""]
+ if head_rows:
+ parts.append("")
+ parts.extend(self._render_node(r, warnings) for r in head_rows)
+ parts.append(" ")
+ if body_rows:
+ parts.append("")
+ parts.extend(self._render_node(r, warnings) for r in body_rows)
+ parts.append(" ")
+ parts.append("
")
+ return "".join(parts)
+
+ def _render_table_row(self, node: DocumentNode, warnings: list[str]) -> str:
+ return f"{self._render_children(node.children, warnings)} "
+
+ def _render_table_cell(self, node: DocumentNode, warnings: list[str]) -> str:
+ tag = "th" if node.attributes.get("head") else "td"
+ return f"<{tag}>{self._render_children(node.children, warnings)}{tag}>"
+
+ def _render_code_block(self, node: DocumentNode, warnings: list[str]) -> str:
+ lang = str(node.attributes.get("language") or "")
+ code = html.escape(node.text)
+ lang_cls = f' class="language-{html.escape(lang)}"' if lang else ""
+ theme = html.escape(self._options.code_theme)
+ return f'{code} '
+
+ def _render_thematic_break(self, node: DocumentNode, warnings: list[str]) -> str:
+ return " "
+
+ def _render_mermaid(self, node: DocumentNode, warnings: list[str]) -> str:
+ warnings.append(_MERMAID_WARNING)
+ return f'{html.escape(node.text)} '
+
+ def _render_function_plot(self, node: DocumentNode, warnings: list[str]) -> str:
+ # 文档级数量上限:超出部分直接回退占位,不解析不采样,防止海量图像耗尽资源
+ over = self._plot_budget.check_count()
+ if over is not None:
+ warnings.append(over)
+ return f'{html.escape(node.text)} '
+ # 解析与渲染共同纳入局部异常回退:单个图像失败只回退占位 + warning,
+ # 绝不阻断整篇导出(含复杂表达式触发的 RecursionError 等异常)。
+ try:
+ request = StaticRenderRequest(
+ kind="function_plot", source=node.text, theme=self._options.theme_id
+ )
+ parsed = self._plot_renderer.parse(request)
+ for diag in parsed.diagnostics:
+ warnings.append(format_plot_diagnostic(diag))
+ if parsed.plot is None:
+ return f'{html.escape(node.text)} '
+ # 文档级累计复杂度预算:超出后回退占位,不再采样求值
+ over = self._plot_budget.check_nodes(parsed.plot.node_count)
+ if over is not None:
+ warnings.append(over)
+ return f'{html.escape(node.text)} '
+ rendered = self._plot_renderer.render_plot(parsed.plot)
+ except Exception as exc:
+ warnings.append(f"函数图像:解析或渲染失败,已回退占位({exc})")
+ return f'{html.escape(node.text)} '
+ warnings.extend(rendered.warnings)
+ return f'{rendered.content} '
+
+ def _render_math_block(self, node: DocumentNode, warnings: list[str]) -> str:
+ return f'$${html.escape(node.text)}$$
'
+
+ def _render_html_block(self, node: DocumentNode, warnings: list[str]) -> str:
+ # 原始 HTML 不可信,转义为纯文本展示,保证正文不丢且无注入风险
+ warnings.append(_RAW_HTML_WARNING)
+ return f'{html.escape(node.text)}
'
+
+ # --- 行内 ---
+ def _render_text(self, node: DocumentNode, warnings: list[str]) -> str:
+ return html.escape(node.text)
+
+ def _render_emphasis(self, node: DocumentNode, warnings: list[str]) -> str:
+ return f"{self._render_children(node.children, warnings)} "
+
+ def _render_strong(self, node: DocumentNode, warnings: list[str]) -> str:
+ return f"{self._render_children(node.children, warnings)} "
+
+ def _render_link(self, node: DocumentNode, warnings: list[str]) -> str:
+ inner = self._render_children(node.children, warnings)
+ href = str(node.attributes.get("href") or "")
+ safe_href = _safe_url(href)
+ if safe_href is None:
+ # 危险协议(如 javascript:)降级为纯文本,不输出可点击链接
+ warnings.append(f"链接协议不安全,已降级为纯文本:{href!r}")
+ return inner
+ title = str(node.attributes.get("title") or "")
+ attrs = [f'href="{html.escape(safe_href)}"']
+ if title:
+ attrs.append(f'title="{html.escape(title)}"')
+ return f"{inner} "
+
+ def _render_codespan(self, node: DocumentNode, warnings: list[str]) -> str:
+ return f"{html.escape(node.text)}"
+
+ def _render_image(self, node: DocumentNode, warnings: list[str]) -> str:
+ src = str(node.attributes.get("src") or "")
+ alt = str(node.attributes.get("alt") or "")
+ safe_src = _safe_url(src)
+ if safe_src is None:
+ # 危险协议(如 data:/javascript:)跳过图片,仅输出 alt 文本
+ warnings.append(f"图片地址不安全,已跳过:{src!r}")
+ return html.escape(alt) if alt else ""
+ title = str(node.attributes.get("title") or "")
+ attrs = [f'src="{html.escape(safe_src)}"', f'alt="{html.escape(alt)}"']
+ if title:
+ attrs.append(f'title="{html.escape(title)}"')
+ return f" "
+
+ def _render_math_inline(self, node: DocumentNode, warnings: list[str]) -> str:
+ return f"\\({html.escape(node.text)}\\)"
+
+ def _render_linebreak(self, node: DocumentNode, warnings: list[str]) -> str:
+ return " "
diff --git a/backend/app/export/exporters/pdf.py b/backend/app/export/exporters/pdf.py
new file mode 100644
index 0000000..e15f742
--- /dev/null
+++ b/backend/app/export/exporters/pdf.py
@@ -0,0 +1,375 @@
+"""PdfExporter:Document AST → PDF(reportlab platypus)。
+
+v1 为文本优先:标题/段落/行内强调与链接/列表/引用/表格/代码块/数学文本均可导出;
+function_plot 内嵌为矢量图(reportlab Drawing),mermaid 保留源码占位并记 warning。
+中文字体用 reportlab 内置 STSong-Light CID 字体,避免外部字体依赖。CID 字体无独立
+bold/italic 字重,故行内强调退化为普通文本(内容不丢、样式简化),标题靠字号区分层级。
+"""
+
+from __future__ import annotations
+
+import html as _html
+from io import BytesIO
+
+from reportlab.lib.enums import TA_CENTER
+from reportlab.lib.pagesizes import A4, letter
+from reportlab.lib.styles import ParagraphStyle
+from reportlab.lib.units import mm
+from reportlab.pdfbase import pdfmetrics
+from reportlab.pdfbase.cidfonts import UnicodeCIDFont
+from reportlab.platypus import (
+ Paragraph,
+ Preformatted,
+ SimpleDocTemplate,
+ Spacer,
+ Table,
+ TableStyle,
+)
+from reportlab.platypus.flowables import HRFlowable
+
+from app.contracts import ExportOptions
+from app.export.document import Document, DocumentNode, ExportResult
+from app.export.exporters._common import (
+ MERMAID_WARNING,
+ RAW_HTML_WARNING,
+ FunctionPlotBudget,
+ format_meta_value,
+ format_plot_diagnostic,
+ safe_url,
+)
+from app.plot.render_reportlab import render_drawing
+from app.plot.renderer import FunctionPlotStaticRenderer, StaticRenderRequest
+
+_FONT = "STSong-Light"
+pdfmetrics.registerFont(UnicodeCIDFont(_FONT))
+
+_MIME = "application/pdf"
+
+_PAGE_SIZES = {"a4": A4, "letter": letter}
+
+# 标题字号随层级递减;标题不依赖粗体(CID 无粗体字重),靠字号拉开层级
+_HEADING_SIZES = {1: 20, 2: 16, 3: 14, 4: 12, 5: 11, 6: 10.5}
+# 引用块文字颜色,与 HtmlExporter 的引用灰一致
+_QUOTE_COLOR = "#57606a"
+
+
+def _make_styles() -> dict[str, ParagraphStyle]:
+ body = ParagraphStyle(
+ "pdf-body",
+ fontName=_FONT,
+ fontSize=10.5,
+ leading=16,
+ spaceAfter=6,
+ )
+ title = ParagraphStyle("pdf-title", parent=body, fontSize=22, leading=28, spaceAfter=12)
+ quote = ParagraphStyle(
+ "pdf-quote",
+ parent=body,
+ leftIndent=14,
+ textColor="#57606a",
+ spaceBefore=4,
+ spaceAfter=6,
+ )
+ code = ParagraphStyle(
+ "pdf-code",
+ parent=body,
+ fontSize=9,
+ leading=12,
+ leftIndent=6,
+ rightIndent=6,
+ backColor="#f6f8fa",
+ borderColor="#d0d7de",
+ borderWidth=0.5,
+ borderPadding=6,
+ spaceBefore=4,
+ spaceAfter=8,
+ )
+ math = ParagraphStyle("pdf-math", parent=body, alignment=TA_CENTER, spaceBefore=6)
+ cell = ParagraphStyle("pdf-cell", parent=body, fontSize=10, leading=14, spaceAfter=0)
+ cell_head = ParagraphStyle(
+ "pdf-cell-head", parent=cell, textColor="#1f2328", fontSize=10
+ )
+ meta = ParagraphStyle("pdf-meta", parent=body, fontSize=8.5, leading=13, textColor="#57606a")
+ styles: dict[str, ParagraphStyle] = {
+ "body": body,
+ "title": title,
+ "quote": quote,
+ "code": code,
+ "math": math,
+ "cell": cell,
+ "cell_head": cell_head,
+ "meta": meta,
+ }
+ for level, size in _HEADING_SIZES.items():
+ styles[f"h{level}"] = ParagraphStyle(
+ f"pdf-h{level}",
+ parent=body,
+ fontSize=size,
+ leading=size * 1.4,
+ spaceBefore=14 if level <= 2 else 10,
+ spaceAfter=6,
+ )
+ return styles
+
+
+class PdfExporter:
+ """实现 DocumentExporter:递归渲染 Document AST 为 PDF 字节流。"""
+
+ def render(self, document: Document, options: ExportOptions) -> ExportResult:
+ """同步渲染;CPU 密集,调用方应放入线程执行,避免阻塞事件循环。"""
+ self._styles = _make_styles()
+ warnings: list[str] = []
+
+ page = _PAGE_SIZES.get((options.page_size or "A4").lower(), A4)
+ self._options = options
+ self._plot_budget = FunctionPlotBudget()
+ self._plot_renderer = FunctionPlotStaticRenderer()
+ # 内容区宽度(左右各 20mm 边距),供函数图像缩放适配页面
+ self._plot_width = page[0] - 40 * mm
+ buf = BytesIO()
+ doc = SimpleDocTemplate(
+ buf,
+ pagesize=page,
+ leftMargin=20 * mm,
+ rightMargin=20 * mm,
+ topMargin=18 * mm,
+ bottomMargin=18 * mm,
+ title=str(document.attributes.get("title") or "") or None,
+ )
+
+ story: list = []
+ self._render_header(document, options, story)
+ self._render_children(document.children, story, warnings)
+
+ doc.build(story)
+ return ExportResult(content=buf.getvalue(), mime_type=_MIME, warnings=warnings)
+
+ async def export(self, document: Document, options: ExportOptions) -> ExportResult:
+ """契约要求的 async 接口;渲染本身同步,直接转发到 render。"""
+ return self.render(document, options)
+
+ # --- 文档头部 ---
+ def _render_header(self, document: Document, options: ExportOptions, story: list) -> None:
+ title = str(document.attributes.get("title") or "")
+ if options.include_title and title:
+ story.append(Paragraph(_html.escape(title), self._styles["title"]))
+ if options.include_metadata:
+ metadata = document.attributes.get("metadata")
+ if metadata:
+ for key, value in metadata.items():
+ text = f"{_html.escape(str(key))}: {_html.escape(format_meta_value(value))}"
+ story.append(Paragraph(text, self._styles["meta"]))
+
+ # --- 块级 ---
+ def _render_children(self, children: list[DocumentNode], story: list, warnings: list[str]) -> None:
+ for child in children:
+ self._render_block(child, story, warnings)
+
+ def _render_block(self, node: DocumentNode, story: list, warnings: list[str]) -> None:
+ handler = getattr(self, f"_block_{node.type}", None)
+ if handler is not None:
+ handler(node, story, warnings)
+ else:
+ warnings.append(f"无法表示的节点类型已跳过:{node.type}")
+
+ def _block_heading(self, node: DocumentNode, story: list, warnings: list[str]) -> None:
+ level = max(1, min(6, int(node.attributes.get("level", 1))))
+ inline = self._render_inline(node.children, warnings)
+ story.append(Paragraph(inline, self._styles[f"h{level}"]))
+
+ def _block_paragraph(self, node: DocumentNode, story: list, warnings: list[str]) -> None:
+ story.append(Paragraph(self._render_inline(node.children, warnings), self._styles["body"]))
+
+ def _block_blockquote(self, node: DocumentNode, story: list, warnings: list[str]) -> None:
+ # 引用块的直接子节点是块级节点(paragraph/list 等),不能交给行内渲染器,
+ # 否则正文会被当作「无法表示的行内节点」丢弃;逐个渲染并继承引用缩进/颜色。
+ for child in node.children:
+ if child.type == "paragraph":
+ story.append(
+ Paragraph(self._render_inline(child.children, warnings), self._styles["quote"])
+ )
+ elif child.type == "list":
+ self._block_list(child, story, warnings, indent=14, color=_QUOTE_COLOR)
+ else:
+ self._render_block(child, story, warnings)
+
+ def _block_list(
+ self,
+ node: DocumentNode,
+ story: list,
+ warnings: list[str],
+ indent: int = 14,
+ color: str | None = None,
+ ) -> None:
+ ordered = bool(node.attributes.get("ordered"))
+ for index, item in enumerate(node.children, start=1):
+ self._block_list_item(item, story, warnings, ordered, index, indent, color)
+
+ def _block_list_item(
+ self,
+ item: DocumentNode,
+ story: list,
+ warnings: list[str],
+ ordered: bool,
+ index: int,
+ indent: int,
+ color: str | None = None,
+ ) -> None:
+ if item.attributes.get("task"):
+ marker = "☑ " if item.attributes.get("checked") else "☐ "
+ else:
+ marker = f"{index}. " if ordered else "• "
+ style_kwargs: dict = dict(
+ parent=self._styles["body"],
+ leftIndent=indent,
+ firstLineIndent=-7,
+ spaceAfter=2,
+ )
+ if color:
+ style_kwargs["textColor"] = color
+ style = ParagraphStyle(f"pdf-li-{indent}-{color or 'normal'}", **style_kwargs)
+ # 按 AST 顺序逐段输出:正文暂存为行内标记文本,遇到嵌套列表先 flush 再递归、
+ # 之后继续后续正文,保持「父段—子列表—后续段」的原始顺序(而不是把所有正文
+ # 都挤到子列表之前)。直接行内节点(text/strong/link 等)走 _render_inline_node,
+ # 保留加粗/链接等语义,不能只渲染其 children 而丢掉格式。
+ parts: list[str] = []
+ first = True
+
+ def flush() -> None:
+ nonlocal first
+ text = " ".join(parts)
+ if first:
+ text = marker + text
+ first = False
+ if text:
+ story.append(Paragraph(text, style))
+ parts.clear()
+
+ for child in item.children:
+ if child.type == "list":
+ flush()
+ self._block_list(child, story, warnings, indent + 14, color)
+ elif child.type == "paragraph":
+ parts.append(self._render_inline(child.children, warnings))
+ else:
+ parts.append(self._render_inline_node(child, warnings))
+ flush()
+
+ def _block_table(self, node: DocumentNode, story: list, warnings: list[str]) -> None:
+ rows = node.children
+ if not rows:
+ return
+ data: list[list[Paragraph]] = []
+ head_row_count = 0
+ for row in rows:
+ head = bool(row.attributes.get("head"))
+ if head:
+ head_row_count += 1
+ cells = [
+ Paragraph(
+ self._render_inline(cell.children, warnings),
+ self._styles["cell_head" if cell.attributes.get("head") else "cell"],
+ )
+ for cell in row.children
+ ]
+ data.append(cells)
+ table = Table(data, repeatRows=head_row_count)
+ commands = [
+ ("GRID", (0, 0), (-1, -1), 0.5, "#d0d7de"),
+ ("VALIGN", (0, 0), (-1, -1), "TOP"),
+ ("LEFTPADDING", (0, 0), (-1, -1), 6),
+ ("RIGHTPADDING", (0, 0), (-1, -1), 6),
+ ("TOPPADDING", (0, 0), (-1, -1), 4),
+ ("BOTTOMPADDING", (0, 0), (-1, -1), 4),
+ ]
+ if head_row_count:
+ commands.append(("BACKGROUND", (0, 0), (-1, head_row_count - 1), "#f6f8fa"))
+ table.setStyle(TableStyle(commands))
+ story.append(table)
+
+ def _block_code_block(self, node: DocumentNode, story: list, warnings: list[str]) -> None:
+ story.append(Preformatted(node.text, self._styles["code"]))
+
+ def _block_thematic_break(self, node: DocumentNode, story: list, warnings: list[str]) -> None:
+ story.append(Spacer(1, 4))
+ story.append(HRFlowable(width="100%", color="#d0d7de", thickness=0.5))
+ story.append(Spacer(1, 6))
+
+ def _block_mermaid(self, node: DocumentNode, story: list, warnings: list[str]) -> None:
+ warnings.append(MERMAID_WARNING)
+ story.append(Preformatted(node.text, self._styles["code"]))
+
+ def _block_function_plot(self, node: DocumentNode, story: list, warnings: list[str]) -> None:
+ # 文档级数量上限:超出部分直接回退占位,不解析不采样,防止海量图像耗尽资源
+ over = self._plot_budget.check_count()
+ if over is not None:
+ warnings.append(over)
+ story.append(Preformatted(node.text, self._styles["code"]))
+ return
+ # 解析与渲染共同纳入局部异常回退:单个图像失败只回退占位 + warning,
+ # 绝不阻断整篇导出(含复杂表达式触发的 RecursionError 等异常)。
+ try:
+ request = StaticRenderRequest(
+ kind="function_plot", source=node.text, theme=self._options.theme_id
+ )
+ parsed = self._plot_renderer.parse(request)
+ for diag in parsed.diagnostics:
+ warnings.append(format_plot_diagnostic(diag))
+ if parsed.plot is None:
+ story.append(Preformatted(node.text, self._styles["code"]))
+ return
+ # 文档级累计复杂度预算:超出后回退占位,不再采样求值
+ over = self._plot_budget.check_nodes(parsed.plot.node_count)
+ if over is not None:
+ warnings.append(over)
+ story.append(Preformatted(node.text, self._styles["code"]))
+ return
+ # Drawing 本身即 Flowable,缩放后追加到 story,与 HTML 视觉一致
+ drawing = render_drawing(parsed.plot, width=self._plot_width)
+ story.append(drawing)
+ except Exception as exc:
+ warnings.append(f"函数图像:解析或渲染失败,已回退占位({exc})")
+ story.append(Preformatted(node.text, self._styles["code"]))
+
+ def _block_math_block(self, node: DocumentNode, story: list, warnings: list[str]) -> None:
+ story.append(Paragraph(f"$${_html.escape(node.text)}$$", self._styles["math"]))
+
+ def _block_html_block(self, node: DocumentNode, story: list, warnings: list[str]) -> None:
+ # 原始 HTML 不可信,按纯文本保留正文
+ warnings.append(RAW_HTML_WARNING)
+ story.append(Paragraph(_html.escape(node.text), self._styles["body"]))
+
+ # --- 行内(产出 reportlab Paragraph 标记文本) ---
+ def _render_inline(self, children: list[DocumentNode], warnings: list[str]) -> str:
+ return "".join(self._render_inline_node(child, warnings) for child in children)
+
+ def _render_inline_node(self, node: DocumentNode, warnings: list[str]) -> str:
+ t = node.type
+ if t == "text":
+ return _html.escape(node.text)
+ if t in ("strong", "emphasis"):
+ return self._render_inline(node.children, warnings)
+ if t == "codespan":
+ return f'{_html.escape(node.text)} '
+ if t == "link":
+ inner = self._render_inline(node.children, warnings)
+ href = str(node.attributes.get("href") or "")
+ safe_href = safe_url(href)
+ if safe_href is None:
+ warnings.append(f"链接协议不安全,已降级为纯文本:{href!r}")
+ return inner
+ return f'{inner} '
+ if t == "image":
+ src = str(node.attributes.get("src") or "")
+ alt = str(node.attributes.get("alt") or "")
+ if safe_url(src) is None:
+ warnings.append(f"图片地址不安全,已跳过:{src!r}")
+ else:
+ warnings.append("图片未内嵌到 PDF,已用替代文本表示")
+ return _html.escape(alt) if alt else ""
+ if t == "math_inline":
+ return f"\\({_html.escape(node.text)}\\)"
+ if t == "linebreak":
+ return " "
+ warnings.append(f"无法表示的行内节点已跳过:{t}")
+ return ""
diff --git a/backend/app/export/markdown.py b/backend/app/export/markdown.py
new file mode 100644
index 0000000..bf9934d
--- /dev/null
+++ b/backend/app/export/markdown.py
@@ -0,0 +1,229 @@
+"""Markdown → Document AST:用 mistune 的 ast renderer 产出通用 token,再映射为内部节点。
+
+选用 mistune 内置 'ast' renderer 而非自写 BaseRenderer,是因为 mistune 的行内渲染按
+字符串拼接、无法承载结构化子节点;ast renderer 直接给出带 children/attrs/raw 的 token
+树,映射层只做 token → DocumentNode 的搬运,不掺入任何 HTML。
+"""
+
+from __future__ import annotations
+
+import mistune
+
+from app.export.document import Document, DocumentNode
+
+_PLUGINS = ["table", "math", "url", "task_lists"]
+
+# fenced code 语言分流:命中则转为专用节点,其余按普通代码块
+_MERMAID_LANG = "mermaid"
+_FUNCTION_PLOT_LANGS = {"function-plot", "function_plot", "functionplot"}
+
+
+def parse_document(markdown: str) -> Document:
+ """把 Markdown 文本解析为 Document AST 根节点。"""
+ renderer = mistune.create_markdown(renderer="ast", plugins=_PLUGINS)
+ tokens = renderer(markdown)
+ mapper = _AstMapper()
+ return Document(node_id=mapper.next_id(), children=mapper.map_blocks(tokens))
+
+
+class _AstMapper:
+ """token 树 → DocumentNode 树的映射器;node_id 按遍历顺序递增,无需跨请求稳定。"""
+
+ def __init__(self) -> None:
+ self._seq = 0
+
+ def next_id(self) -> str:
+ self._seq += 1
+ return f"node_{self._seq:03d}"
+
+ def map_blocks(self, tokens: list[dict]) -> list[DocumentNode]:
+ nodes: list[DocumentNode] = []
+ for token in tokens:
+ node = self.map_block(token)
+ if node is not None:
+ nodes.append(node)
+ return nodes
+
+ def map_block(self, token: dict) -> DocumentNode | None:
+ kind = token["type"]
+ if kind == "heading":
+ return DocumentNode(
+ type="heading",
+ node_id=self.next_id(),
+ attributes={"level": token["attrs"]["level"]},
+ children=self.map_inline(token.get("children", [])),
+ )
+ if kind in ("paragraph", "block_text"):
+ # block_text 是列表项内的段落块,仍按 paragraph 表达,由 list_item 包裹
+ return DocumentNode(
+ type="paragraph",
+ node_id=self.next_id(),
+ children=self.map_inline(token.get("children", [])),
+ )
+ if kind == "list":
+ return DocumentNode(
+ type="list",
+ node_id=self.next_id(),
+ attributes={"ordered": bool(token.get("attrs", {}).get("ordered"))},
+ children=[self.map_list_item(child) for child in token.get("children", [])],
+ )
+ if kind == "block_code":
+ return self._map_code(token)
+ if kind == "block_quote":
+ return DocumentNode(
+ type="blockquote",
+ node_id=self.next_id(),
+ children=self.map_blocks(token.get("children", [])),
+ )
+ if kind == "table":
+ return self._map_table(token)
+ if kind == "block_math":
+ return DocumentNode(
+ type="math_block", node_id=self.next_id(), text=token.get("raw", "")
+ )
+ if kind == "thematic_break":
+ return DocumentNode(type="thematic_break", node_id=self.next_id())
+ if kind == "blank_line":
+ return None
+ if kind == "block_html":
+ # 原始 HTML 块降级为纯文本节点,由 HtmlExporter 转义并记 warning,避免静默丢失正文
+ return DocumentNode(
+ type="html_block", node_id=self.next_id(), text=token.get("raw", "")
+ )
+ # 未知块级 token 保守保留原文;映射为带 text 子节点的 paragraph,避免被渲染层丢弃
+ raw = token.get("raw", "")
+ if raw:
+ return DocumentNode(
+ type="paragraph",
+ node_id=self.next_id(),
+ children=[DocumentNode(type="text", node_id=self.next_id(), text=raw)],
+ )
+ return None
+
+ def map_list_item(self, token: dict) -> DocumentNode:
+ """列表项:block_text 展平为行内子节点,嵌套 list 保留为子节点。"""
+ attributes: dict = {}
+ if token["type"] == "task_list_item":
+ attributes = {"task": True, "checked": bool(token.get("attrs", {}).get("checked"))}
+ children: list[DocumentNode] = []
+ for child in token.get("children", []):
+ if child["type"] == "block_text":
+ children.extend(self.map_inline(child.get("children", [])))
+ elif child["type"] == "list":
+ children.append(self.map_block(child))
+ else:
+ node = self.map_block(child)
+ if node is not None:
+ children.append(node)
+ return DocumentNode(
+ type="list_item", node_id=self.next_id(), attributes=attributes, children=children
+ )
+
+ def map_inline(self, tokens: list[dict]) -> list[DocumentNode]:
+ nodes: list[DocumentNode] = []
+ for token in tokens:
+ node = self.map_inline_token(token)
+ if node is not None:
+ nodes.append(node)
+ return nodes
+
+ def map_inline_token(self, token: dict) -> DocumentNode | None:
+ kind = token["type"]
+ if kind == "text":
+ return DocumentNode(type="text", node_id=self.next_id(), text=token.get("raw", ""))
+ if kind == "strong":
+ return DocumentNode(
+ type="strong", node_id=self.next_id(),
+ children=self.map_inline(token.get("children", [])),
+ )
+ if kind == "emphasis":
+ return DocumentNode(
+ type="emphasis", node_id=self.next_id(),
+ children=self.map_inline(token.get("children", [])),
+ )
+ if kind == "link":
+ attrs = token.get("attrs", {})
+ attributes = {"href": attrs.get("url", "")}
+ if attrs.get("title"):
+ attributes["title"] = attrs["title"]
+ return DocumentNode(
+ type="link", node_id=self.next_id(), attributes=attributes,
+ children=self.map_inline(token.get("children", [])),
+ )
+ if kind == "codespan":
+ return DocumentNode(type="codespan", node_id=self.next_id(), text=token.get("raw", ""))
+ if kind == "image":
+ # mistune 图片 token:src 在 attrs.url,alt 来自 children 的文本,title 在 attrs.title
+ attrs = token.get("attrs", {})
+ alt = "".join(
+ child.get("raw", "")
+ for child in token.get("children", [])
+ if child.get("type") == "text"
+ )
+ attributes = {"src": attrs.get("url", "")}
+ if alt:
+ attributes["alt"] = alt
+ if attrs.get("title"):
+ attributes["title"] = attrs["title"]
+ return DocumentNode(type="image", node_id=self.next_id(), attributes=attributes)
+ if kind == "inline_math":
+ return DocumentNode(
+ type="math_inline", node_id=self.next_id(), text=token.get("raw", "")
+ )
+ if kind == "softbreak":
+ # HTML 中换行会折叠为空白,软换行按空格表达
+ return DocumentNode(type="text", node_id=self.next_id(), text=" ")
+ if kind == "linebreak":
+ return DocumentNode(type="linebreak", node_id=self.next_id())
+ # 未知行内 token 保守保留原文
+ raw = token.get("raw", "")
+ if raw:
+ return DocumentNode(type="text", node_id=self.next_id(), text=raw)
+ return None
+
+ def _map_code(self, token: dict) -> DocumentNode:
+ info = (token.get("attrs", {}).get("info") or "").strip()
+ lang = info.split()[0].lower() if info else ""
+ code = token.get("raw", "").rstrip("\n")
+ if lang == _MERMAID_LANG:
+ return DocumentNode(type="mermaid", node_id=self.next_id(), text=code)
+ if lang in _FUNCTION_PLOT_LANGS:
+ return DocumentNode(type="function_plot", node_id=self.next_id(), text=code)
+ attributes = {"language": lang} if lang else {}
+ return DocumentNode(
+ type="code_block", node_id=self.next_id(), attributes=attributes, text=code
+ )
+
+ def _map_table(self, token: dict) -> DocumentNode:
+ rows: list[DocumentNode] = []
+ for child in token.get("children", []):
+ if child["type"] == "table_head":
+ rows.append(self._map_table_row(child, head=True))
+ elif child["type"] == "table_body":
+ for row in child.get("children", []):
+ if row["type"] == "table_row":
+ rows.append(self._map_table_row(row, head=False))
+ elif child["type"] == "table_row":
+ rows.append(self._map_table_row(child, head=False))
+ return DocumentNode(type="table", node_id=self.next_id(), children=rows)
+
+ def _map_table_row(self, token: dict, *, head: bool) -> DocumentNode:
+ cells: list[DocumentNode] = []
+ for cell in token.get("children", []):
+ if cell["type"] != "table_cell":
+ continue
+ attrs = cell.get("attrs", {})
+ cell_attributes = {"head": bool(attrs.get("head", head))}
+ if attrs.get("align"):
+ cell_attributes["align"] = attrs["align"]
+ cells.append(
+ DocumentNode(
+ type="table_cell",
+ node_id=self.next_id(),
+ attributes=cell_attributes,
+ children=self.map_inline(cell.get("children", [])),
+ )
+ )
+ return DocumentNode(
+ type="table_row", node_id=self.next_id(), attributes={"head": head}, children=cells
+ )
diff --git a/backend/app/export/service.py b/backend/app/export/service.py
new file mode 100644
index 0000000..10666eb
--- /dev/null
+++ b/backend/app/export/service.py
@@ -0,0 +1,391 @@
+"""Export 服务:任务注册表、后台渲染、取消与产物生命周期。
+
+与 Benchmark 一致采用「创建即返回 queued、后台 Task 异步执行」的内存模型:任务与产物
+暂存内存与 exports 目录,不持久化到 SQLite。导出是单阶段渲染,无 SSE 事件流,取消主要
+在渲染前/后让出执行权的边界生效;产物带 24h 过期时间,过期后不可下载。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import logging
+import re
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from uuid import uuid4
+
+from app.config import get_settings
+from app.contracts import (
+ ExportFile,
+ ExportFormat,
+ ExportJob,
+ ExportOptions,
+ ExportProgress,
+ ExportRequest,
+ ExportSource,
+ ExportSourceType,
+ ExportStatus,
+)
+from app.errors import ApiError
+from app.export.document import Document, ExportResult
+from app.export.exporters.docx import DocxExporter
+from app.export.exporters.html import HtmlExporter
+from app.export.exporters.pdf import PdfExporter
+from app.export.markdown import parse_document
+from app.services import note_service
+
+logger = logging.getLogger(__name__)
+
+_jobs: dict[str, ExportJob] = {}
+_tasks: dict[str, asyncio.Task] = {}
+_cancel_flags: dict[str, asyncio.Event] = {}
+MAX_JOBS = 100
+# 输入源(note / markdown)统一大小上限,防止未保存预览或超长笔记塞爆内存/产物
+MAX_MARKDOWN_CHARS = 200_000
+# 最终导出产物大小上限,防止超大 HTML 耗尽内存/磁盘
+MAX_EXPORT_BYTES = 20 * 1024 * 1024 # 20 MB
+# 并发渲染上限:解析/渲染是 CPU 密集的同步工作,限制同时执行的任务数,
+# 防止大量任务同时占满工作线程与内存
+MAX_CONCURRENT_RENDERS = 2
+_render_slots = asyncio.Semaphore(MAX_CONCURRENT_RENDERS)
+# 产物有效期
+FILE_TTL = timedelta(hours=24)
+
+_INVALID_FILE_CHARS = re.compile(r'[\\/:*?"<>|]')
+
+# 格式 → 导出器;新增格式只需在此登记,路由与任务模型无需改动
+_EXPORTERS: dict[ExportFormat, type] = {
+ ExportFormat.html: HtmlExporter,
+ ExportFormat.pdf: PdfExporter,
+ ExportFormat.docx: DocxExporter,
+}
+
+# 格式 → 文件扩展名(用于落盘文件名与产物清理)
+_EXTENSIONS: dict[ExportFormat, str] = {
+ ExportFormat.html: ".html",
+ ExportFormat.pdf: ".pdf",
+ ExportFormat.docx: ".docx",
+}
+
+
+def _extension_for(format: ExportFormat) -> str:
+ return _EXTENSIONS[format]
+
+
+class ExportCancelled(Exception):
+ """导出在渲染前被取消时抛出,用于标记 cancelled。"""
+
+
+class ExportTooLarge(Exception):
+ """导出产物超过大小上限时抛出,用于标记 failed 并携带专用错误码。"""
+
+
+def _now() -> datetime:
+ return datetime.now(timezone.utc)
+
+
+def _safe_download_name(title: str) -> str:
+ """清洗标题得到安全的下载文件名;空标题回退到 export。"""
+ name = _INVALID_FILE_CHARS.sub("_", title).strip() or "export"
+ return name[:80]
+
+
+def _export_path(job_id: str, ext: str) -> Path:
+ return get_settings().exports_path / f"{job_id}{ext}"
+
+
+def _delete_file(job_id: str, ext: str) -> None:
+ """删除导出产物文件;文件不存在时忽略。"""
+ try:
+ _export_path(job_id, ext).unlink(missing_ok=True)
+ except OSError:
+ logger.warning("Failed to delete export file: %s", job_id)
+
+
+def cleanup_orphan_files() -> int:
+ """清理 exports 目录下无对应内存任务的孤立产物(服务重启后调用)。"""
+ exports_dir = get_settings().exports_path
+ if not exports_dir.is_dir():
+ return 0
+ removed = 0
+ for ext in _EXTENSIONS.values():
+ for path in exports_dir.glob(f"*{ext}"):
+ if path.stem not in _jobs:
+ try:
+ path.unlink()
+ removed += 1
+ except OSError:
+ logger.warning("Failed to delete orphan export file: %s", path)
+ return removed
+
+
+def _render_document(document: Document, options: ExportOptions, format: ExportFormat) -> ExportResult:
+ """按 format 分发到对应导出器;每次新建实例避免跨线程复用。"""
+ exporter_cls = _EXPORTERS[format]
+ return exporter_cls().render(document, options)
+
+
+def _forget(job_id: str) -> None:
+ job = _jobs.get(job_id)
+ ext = _extension_for(job.format) if job is not None else ".html"
+ _jobs.pop(job_id, None)
+ _tasks.pop(job_id, None)
+ _cancel_flags.pop(job_id, None)
+ _delete_file(job_id, ext)
+
+
+def _evict_terminal() -> bool:
+ """超过容量时淘汰最旧的终态任务;全为活动任务无法淘汰时返回 False。"""
+ terminal = (ExportStatus.completed, ExportStatus.failed, ExportStatus.cancelled)
+ while len(_jobs) >= MAX_JOBS:
+ victim = next((jid for jid, job in _jobs.items() if job.status in terminal), None)
+ if victim is None:
+ return False
+ _forget(victim)
+ return True
+
+
+async def _resolve_source(source: ExportSource) -> tuple[str, str, dict | None]:
+ """把导出源解析为 (markdown, title, metadata);metadata 仅 note 源提供。"""
+ if source.type == ExportSourceType.note:
+ note = await note_service.get_note(source.note_id)
+ if note is None:
+ raise ApiError(
+ 404,
+ "EXPORT_SOURCE_NOT_FOUND",
+ "note not found",
+ {"note_id": source.note_id},
+ )
+ if len(note.markdown) > MAX_MARKDOWN_CHARS:
+ raise ApiError(
+ 400,
+ "EXPORT_OPTIONS_INVALID",
+ f"note source exceeds {MAX_MARKDOWN_CHARS} characters",
+ {"size": len(note.markdown), "limit": MAX_MARKDOWN_CHARS},
+ )
+ metadata = {
+ "file_path": note.file_path,
+ "tags": note.tags,
+ "created_at": note.created_at,
+ "updated_at": note.updated_at,
+ }
+ return note.markdown, note.title, metadata
+
+ markdown = source.markdown or ""
+ if not markdown.strip():
+ raise ApiError(400, "EXPORT_OPTIONS_INVALID", "markdown source must not be empty")
+ if len(markdown) > MAX_MARKDOWN_CHARS:
+ raise ApiError(
+ 400,
+ "EXPORT_OPTIONS_INVALID",
+ f"markdown source exceeds {MAX_MARKDOWN_CHARS} characters",
+ {"size": len(markdown), "limit": MAX_MARKDOWN_CHARS},
+ )
+ return markdown, "", None
+
+
+async def create_export(request: ExportRequest) -> ExportJob:
+ """创建导出任务,立即返回 queued 的 ExportJob,由后台 Task 渲染。"""
+ markdown, title, metadata = await _resolve_source(request.source)
+
+ if not _evict_terminal():
+ raise ApiError(
+ 429,
+ "EXPORT_CAPACITY_EXCEEDED",
+ "Export capacity exceeded; wait for active jobs to finish.",
+ {},
+ )
+
+ job_id = "export_" + uuid4().hex[:12]
+ job = ExportJob(
+ job_id=job_id,
+ status=ExportStatus.queued,
+ format=request.format,
+ created_at=_now(),
+ )
+ _jobs[job_id] = job
+ _cancel_flags[job_id] = asyncio.Event()
+ _tasks[job_id] = asyncio.create_task(
+ _execute(job_id, request.format, markdown, title, metadata, request.options)
+ )
+ return job
+
+
+async def _acquire_render_slot(cancel_event: asyncio.Event) -> bool:
+ """等待渲染槽位,同时响应取消:拿到槽位返回 True,被取消返回 False。
+
+ 等待期间任务保持 queued;取消即时生效,不必等前面的渲染完成。
+ """
+ while True:
+ if cancel_event.is_set():
+ return False
+ acquire = asyncio.create_task(_render_slots.acquire())
+ cancel_wait = asyncio.create_task(cancel_event.wait())
+ done, pending = await asyncio.wait(
+ (acquire, cancel_wait), return_when=asyncio.FIRST_COMPLETED
+ )
+ if acquire in done:
+ # 拿到槽位;收掉仍在等待取消标志的任务(不释放刚拿到的槽位)
+ for task in pending:
+ task.cancel()
+ await asyncio.gather(*pending, return_exceptions=True)
+ return True
+ # 取消先到:取消尚未完成的 acquire(Semaphore.acquire 取消不会递减计数)
+ acquire.cancel()
+ cancel_wait.cancel()
+ await asyncio.gather(acquire, cancel_wait, return_exceptions=True)
+ return False
+
+
+async def _execute(
+ job_id: str,
+ format: ExportFormat,
+ markdown: str,
+ title: str,
+ metadata: dict | None,
+ options: ExportOptions,
+) -> None:
+ """后台渲染:排队 → 解析 → 导出 → 写文件 → 挂载产物元信息。"""
+ cancel_event = _cancel_flags[job_id]
+ acquired = False
+ try:
+ # 并发渲染限额:解析/渲染是 CPU 密集的同步工作,用信号量限制同时执行的任务数。
+ # 等待槽位期间保持 queued 并同时监听取消,取消即时生效,不必等前面的渲染完成。
+ if not await _acquire_render_slot(cancel_event):
+ raise ExportCancelled()
+ acquired = True
+
+ # 拿到槽位后才进入 running
+ _jobs[job_id] = _jobs[job_id].model_copy(
+ update={
+ "status": ExportStatus.running,
+ "started_at": _now(),
+ "progress": ExportProgress(phase="rendering", current=0, total=1, percent=0.0),
+ }
+ )
+ # 让出一次,使「创建后立即取消」的 queued 任务能及时进入 cancelled
+ await asyncio.sleep(0)
+ if cancel_event.is_set():
+ raise ExportCancelled()
+
+ # 解析与渲染都是 CPU 密集的同步工作,放入线程执行避免阻塞事件循环,
+ # 使运行中的取消能在渲染边界生效;写文件前再次检查取消。
+ document = await asyncio.to_thread(parse_document, markdown)
+ document.attributes["title"] = title
+ if metadata:
+ document.attributes["metadata"] = metadata
+
+ result = await asyncio.to_thread(_render_document, document, options, format)
+ if cancel_event.is_set():
+ raise ExportCancelled()
+ if len(result.content) > MAX_EXPORT_BYTES:
+ raise ExportTooLarge()
+
+ ext = _extension_for(format)
+ out_dir = get_settings().exports_path
+ out_dir.mkdir(parents=True, exist_ok=True)
+ path = _export_path(job_id, ext)
+ path.write_bytes(result.content)
+
+ completed_at = _now()
+ _jobs[job_id] = _jobs[job_id].model_copy(
+ update={
+ "status": ExportStatus.completed,
+ "progress": ExportProgress(
+ phase="completed", current=1, total=1, percent=1.0
+ ),
+ "file": ExportFile(
+ file_name=f"{_safe_download_name(title)}{ext}",
+ mime_type=result.mime_type,
+ size=len(result.content),
+ sha256=hashlib.sha256(result.content).hexdigest(),
+ expires_at=completed_at + FILE_TTL,
+ ),
+ "warnings": result.warnings,
+ "completed_at": completed_at,
+ }
+ )
+ except ExportCancelled:
+ _jobs[job_id] = _jobs[job_id].model_copy(
+ update={
+ "status": ExportStatus.cancelled,
+ "completed_at": _now(),
+ }
+ )
+ except ExportTooLarge:
+ _jobs[job_id] = _jobs[job_id].model_copy(
+ update={
+ "status": ExportStatus.failed,
+ "error": "Export output exceeds size limit.",
+ "error_code": "EXPORT_OUTPUT_TOO_LARGE",
+ "completed_at": _now(),
+ }
+ )
+ except Exception as exc: # 渲染失败不拖垮服务,只记日志与项目错误码
+ logger.exception("Export failed: job_id=%s", job_id)
+ _jobs[job_id] = _jobs[job_id].model_copy(
+ update={
+ "status": ExportStatus.failed,
+ "error": "Export render failed.",
+ "error_code": "EXPORT_RENDER_FAILED",
+ "completed_at": _now(),
+ }
+ )
+ finally:
+ if acquired:
+ _render_slots.release()
+ _cancel_flags.pop(job_id, None)
+
+
+def list_exports(
+ status: ExportStatus | None = None,
+ format: ExportFormat | None = None,
+ limit: int = 50,
+ offset: int = 0,
+) -> tuple[list[ExportJob], int]:
+ jobs = list(_jobs.values())
+ if status is not None:
+ jobs = [j for j in jobs if j.status == status]
+ if format is not None:
+ jobs = [j for j in jobs if j.format == format]
+ jobs.sort(key=lambda j: j.created_at, reverse=True)
+ total = len(jobs)
+ return jobs[offset : offset + limit], total
+
+
+def get_export(job_id: str) -> ExportJob | None:
+ return _jobs.get(job_id)
+
+
+def cancel_export(job_id: str) -> ExportJob | None:
+ """取消导出:仅 queued/running 可取消,后台 Task 在让出边界标记 cancelled。"""
+ job = _jobs.get(job_id)
+ if job is None:
+ return None
+ if job.status in (ExportStatus.queued, ExportStatus.running):
+ _cancel_flags[job_id].set()
+ return job
+
+
+def get_export_file(job_id: str) -> Path:
+ """返回可下载产物的存储路径;未完成返回 404、过期返回 410。"""
+ job = _jobs.get(job_id)
+ if job is None:
+ raise ApiError(404, "EXPORT_JOB_NOT_FOUND", "export job not found", {"job_id": job_id})
+ if job.status != ExportStatus.completed or job.file is None:
+ raise ApiError(
+ 404, "EXPORT_JOB_NOT_FOUND", "export file not ready", {"job_id": job_id}
+ )
+ if job.file.expires_at <= _now():
+ _forget(job_id) # 过期即清理内存记录与产物文件
+ raise ApiError(410, "EXPORT_FILE_EXPIRED", "export file has expired", {"job_id": job_id})
+ return _export_path(job_id, _extension_for(job.format))
+
+
+async def wait_for_export(job_id: str) -> ExportJob | None:
+ """等待后台任务结束(测试/轮询用);无任务时直接返回当前状态。"""
+ task = _tasks.get(job_id)
+ if task is not None:
+ await task
+ return _jobs.get(job_id)
diff --git a/backend/app/main.py b/backend/app/main.py
index c47593e..c9e0651 100644
--- a/backend/app/main.py
+++ b/backend/app/main.py
@@ -11,6 +11,7 @@ from starlette.exceptions import HTTPException as StarletteHttpException
from app.config import get_settings
from app.container import container
from app.errors import ApiError, api_error_handler, http_error_handler, validation_error_handler
+from app.export import service as export_service
from app.routes import router as api_router
from app.media_routes import router as media_router
from app.local_model_routes import router as local_model_router
@@ -27,6 +28,8 @@ settings = get_settings()
async def lifespan(_: FastAPI):
install_logging()
log_event('system', 'service.started')
+ # 重启后内存注册表为空,清理上一次运行遗留的导出产物,避免磁盘垃圾堆积。
+ export_service.cleanup_orphan_files()
from app.services import transcription_service
transcription_service.recover_interrupted()
try:
@@ -41,6 +44,7 @@ async def lifespan(_: FastAPI):
from app.local_models import manager
for _, key in list(manager._downloads):
await manager.cancel_download(key)
+ # 第三方 MCP Server 必须跟随 AI Core 退出,不能遗留孤儿进程。
container.plugins.shutdown()
container.mcp_servers.shutdown()
log_event('system', 'service.stopped')
diff --git a/backend/app/plot/__init__.py b/backend/app/plot/__init__.py
new file mode 100644
index 0000000..2bf1044
--- /dev/null
+++ b/backend/app/plot/__init__.py
@@ -0,0 +1,7 @@
+"""Function Plot:函数图像的白名单表达式解析与静态 SVG 渲染。
+
+模块划分:
+- model.py FunctionPlot 等内部数据模型(不进 contracts.py,同 Document AST)
+- parser.py function-plot 源码与表达式解析(ast 白名单,绝不 eval/exec)
+- render.py 把 FunctionPlot 渲染为内嵌 SVG(纯几何 + ,无脚本)
+"""
diff --git a/backend/app/plot/model.py b/backend/app/plot/model.py
new file mode 100644
index 0000000..f11273e
--- /dev/null
+++ b/backend/app/plot/model.py
@@ -0,0 +1,57 @@
+"""Function Plot 内部数据模型。
+
+契约 §12.2 的 FunctionPlot 结构与 §10.4 的 StaticRenderResult 只在导出链路的后端内部
+流转,不进入 HTTP 契约,因此与 Document AST 一样放在独立包内,不进 contracts.py。
+"""
+
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+
+class FunctionPlotExpression(BaseModel):
+ """单条函数表达式;expression 为数学表达式文本(不含 ``y =`` 前缀)。"""
+
+ expression: str
+ label: str | None = None
+ color: str | None = None
+
+
+class PlotAxes(BaseModel):
+ xlabel: str | None = None
+ ylabel: str | None = None
+ grid: bool = True
+
+
+class FunctionPlot(BaseModel):
+ version: int = 1
+ expressions: list[FunctionPlotExpression]
+ domain: tuple[float, float] = (-10.0, 10.0)
+ range: tuple[float, float] | None = None
+ axes: PlotAxes = Field(default_factory=PlotAxes)
+ # 该块所有表达式 AST 节点数之和,供导出器做文档级累计复杂度预算
+ node_count: int = 0
+
+
+class PlotDiagnostic(BaseModel):
+ severity: Literal["warning", "error"]
+ code: str
+ message: str
+ line: int | None = None
+
+
+class FunctionPlotParseResult(BaseModel):
+ """解析结果:任一表达式 error 时 plot 为 None(整块回退占位),仅 warning 时 plot 有效。"""
+
+ plot: FunctionPlot | None = None
+ diagnostics: list[PlotDiagnostic] = Field(default_factory=list)
+
+
+class StaticRenderResult(BaseModel):
+ content: str
+ mime_type: str = "image/svg+xml"
+ width: int
+ height: int
+ warnings: list[str] = Field(default_factory=list)
diff --git a/backend/app/plot/parser.py b/backend/app/plot/parser.py
new file mode 100644
index 0000000..4dab231
--- /dev/null
+++ b/backend/app/plot/parser.py
@@ -0,0 +1,412 @@
+"""Function Plot 表达式解析:白名单数学语法,绝不执行 eval / 函数构造器 / 属性访问。
+
+安全模型:先用 ``ast.parse(mode='eval')`` 把表达式变成纯 AST(这一步不执行任何代码),
+再逐节点白名单校验(只允许数字、变量 ``x``、常量 ``pi/e``、白名单函数调用与四则/幂
+运算),最后用递归解释器直接计算数值——全程不 ``compile``/``exec`` 字符串。
+"""
+
+from __future__ import annotations
+
+import ast
+import math
+import re
+from typing import NoReturn
+
+from app.plot.model import (
+ FunctionPlot,
+ FunctionPlotExpression,
+ FunctionPlotParseResult,
+ PlotAxes,
+ PlotDiagnostic,
+)
+
+# 白名单函数(ln 是 log 的别名);abs 用内置函数,其余映射到 math
+_FUNCTION_IMPL: dict[str, object] = {
+ "sin": math.sin,
+ "cos": math.cos,
+ "tan": math.tan,
+ "asin": math.asin,
+ "acos": math.acos,
+ "atan": math.atan,
+ "sinh": math.sinh,
+ "cosh": math.cosh,
+ "tanh": math.tanh,
+ "exp": math.exp,
+ "log": math.log,
+ "ln": math.log,
+ "log10": math.log10,
+ "log2": math.log2,
+ "sqrt": math.sqrt,
+ "abs": abs,
+}
+_FUNCTIONS = frozenset(_FUNCTION_IMPL)
+_CONSTANTS: dict[str, float] = {"pi": math.pi, "e": math.e}
+
+_ALLOWED_BINOPS = (ast.Add, ast.Sub, ast.Mult, ast.Div, ast.Pow)
+_ALLOWED_UNARY = (ast.UAdd, ast.USub)
+_DIRECTIVE_KEYS = frozenset({"domain", "range", "xlabel", "ylabel", "grid"})
+_NUMBER_RE = re.compile(r"^(\d+\.?\d*|\.\d+)([eE][+-]?\d+)?$")
+
+# 表达式复杂度上限:深层嵌套或海量节点在递归校验/求值时会触发 RecursionError,
+# 用白名单校验提前拦截,保证失败走正常诊断路径而不是异常逃逸出导出链路。
+_MAX_AST_DEPTH = 200
+_MAX_AST_NODES = 1000
+# 单块 function-plot 允许的表达式数量上限,防止海量表达式导致超大 SVG 与海量采样求值
+_MAX_EXPRESSIONS = 16
+
+
+class PlotParseError(Exception):
+ """表达式解析/校验失败,携带可定位诊断。"""
+
+ def __init__(self, diagnostic: PlotDiagnostic) -> None:
+ super().__init__(diagnostic.message)
+ self.diagnostic = diagnostic
+
+
+def _unsafe(message: str) -> NoReturn:
+ raise PlotParseError(
+ PlotDiagnostic(severity="error", code="FUNCTION_PLOT_EXPRESSION_UNSAFE", message=message)
+ )
+
+
+def _is_number(tok: str) -> bool:
+ return bool(_NUMBER_RE.match(tok))
+
+
+def _tokenize(s: str) -> list[str]:
+ """把预处理后的表达式切成数字/标识符/运算符/括号 token。"""
+ tokens: list[str] = []
+ i = 0
+ n = len(s)
+ while i < n:
+ ch = s[i]
+ if ch.isspace():
+ i += 1
+ continue
+ if ch.isdigit() or ch == ".":
+ j = i
+ while j < n and (s[j].isdigit() or s[j] == "."):
+ j += 1
+ # 科学计数法:数字后紧跟 e/E[+-]数字 视为同一数字
+ if j < n and s[j] in "eE":
+ k = j + 1
+ if k < n and s[k] in "+-":
+ k += 1
+ if k < n and s[k].isdigit():
+ while k < n and s[k].isdigit():
+ k += 1
+ j = k
+ tokens.append(s[i:j])
+ i = j
+ continue
+ if ch.isalpha() or ch == "_":
+ j = i
+ while j < n and (s[j].isalnum() or s[j] == "_"):
+ j += 1
+ tokens.append(s[i:j])
+ i = j
+ continue
+ if ch == "*" and i + 1 < n and s[i + 1] == "*":
+ tokens.append("**")
+ i += 2
+ continue
+ tokens.append(ch)
+ i += 1
+ return tokens
+
+
+def _is_value_end(tok: str) -> bool:
+ """该 token 之后允许补乘号(数字/右括号/变量 x/常量)。"""
+ return tok == ")" or _is_number(tok) or tok == "x" or tok in _CONSTANTS
+
+
+def _is_value_start(tok: str) -> bool:
+ """该 token 可作为乘号右侧起点(左括号/数字/任意标识符,含函数名)。"""
+ return tok == "(" or _is_number(tok) or (tok and (tok[0].isalpha() or tok[0] == "_"))
+
+
+def _insert_implicit_multiplication(s: str) -> str:
+ """补隐式乘法:2x、2(x+1)、(x+1)(x-1)、x sin(x) 等;函数名后的 ``(`` 是调用不补。"""
+ tokens = _tokenize(s)
+ out: list[str] = []
+ prev: str | None = None
+ for tok in tokens:
+ if prev is not None and _is_value_end(prev) and _is_value_start(tok):
+ out.append("*")
+ out.append(tok)
+ prev = tok
+ return "".join(out)
+
+
+def _preprocess(expr: str) -> str:
+ """``^`` 视为幂,补隐式乘法后再交给 ast.parse。"""
+ return _insert_implicit_multiplication(expr.replace("^", "**"))
+
+
+def _check_node(node: ast.AST, depth: int = 0, counter: list[int] | None = None) -> None:
+ """白名单校验:任何越界节点都抛 FUNCTION_PLOT_EXPRESSION_UNSAFE。
+
+ 同时限制 AST 深度与节点总数,避免超长/超深表达式在递归校验或求值时触发
+ RecursionError 而绕过解析失败路径。
+ """
+ if counter is None:
+ counter = [0]
+ if depth > _MAX_AST_DEPTH:
+ _unsafe(f"表达式嵌套过深(超过 {_MAX_AST_DEPTH} 层)")
+ counter[0] += 1
+ if counter[0] > _MAX_AST_NODES:
+ _unsafe(f"表达式过于复杂(节点数超过 {_MAX_AST_NODES})")
+ if isinstance(node, ast.Constant):
+ if isinstance(node.value, bool) or not isinstance(node.value, (int, float)):
+ _unsafe(f"不支持的常量 {node.value!r}")
+ return
+ if isinstance(node, ast.Name):
+ if node.id == "x" or node.id in _CONSTANTS:
+ return
+ _unsafe(f"未知标识符 {node.id!r}")
+ if isinstance(node, ast.BinOp):
+ if not isinstance(node.op, _ALLOWED_BINOPS):
+ _unsafe(f"不支持的运算符 {type(node.op).__name__}")
+ _check_node(node.left, depth + 1, counter)
+ _check_node(node.right, depth + 1, counter)
+ return
+ if isinstance(node, ast.UnaryOp):
+ if not isinstance(node.op, _ALLOWED_UNARY):
+ _unsafe(f"不支持的运算符 {type(node.op).__name__}")
+ _check_node(node.operand, depth + 1, counter)
+ return
+ if isinstance(node, ast.Call):
+ if not isinstance(node.func, ast.Name) or node.func.id not in _FUNCTIONS:
+ _unsafe(f"不支持的函数调用 {ast.dump(node.func)!r}")
+ if node.keywords:
+ _unsafe("函数调用不支持关键字参数")
+ # 白名单内所有函数均恰取 1 个参数,提前校验避免求值期 TypeError
+ if len(node.args) != 1:
+ _unsafe(f"{node.func.id} 需要 1 个参数,实际 {len(node.args)} 个")
+ for arg in node.args:
+ _check_node(arg, depth + 1, counter)
+ return
+ _unsafe(f"不支持的语法 {type(node).__name__}")
+
+
+def parse_expression(expr: str) -> ast.Expression:
+ """把数学表达式解析为已通过白名单校验的 AST(可直接交给 evaluate)。"""
+ preprocessed = _preprocess(expr)
+ try:
+ tree = ast.parse(preprocessed, mode="eval")
+ except SyntaxError as exc:
+ raise PlotParseError(
+ PlotDiagnostic(
+ severity="error",
+ code="FUNCTION_PLOT_PARSE_FAILED",
+ message=f"表达式语法错误:{exc.msg}",
+ )
+ ) from exc
+ except RecursionError as exc:
+ # 极深嵌套可能在 ast.parse 阶段就触发 RecursionError,转为可定位诊断
+ raise PlotParseError(
+ PlotDiagnostic(
+ severity="error",
+ code="FUNCTION_PLOT_PARSE_FAILED",
+ message="表达式嵌套过深,无法解析",
+ )
+ ) from exc
+ _check_node(tree.body)
+ return tree
+
+
+def _count_nodes(node: ast.AST) -> int:
+ """统计已通过校验的表达式 AST 节点数,供文档级累计复杂度预算使用。"""
+ counter = [0]
+ _check_node(node, counter=counter)
+ return counter[0]
+
+
+def evaluate(expr_ast: ast.Expression, x: float) -> float:
+ """递归解释已校验 AST 得到数值,全程不编译/执行代码。"""
+ return _eval_node(expr_ast.body, x)
+
+
+def _eval_node(node: ast.AST, x: float) -> float:
+ if isinstance(node, ast.Constant):
+ return float(node.value)
+ if isinstance(node, ast.Name):
+ return x if node.id == "x" else _CONSTANTS[node.id]
+ if isinstance(node, ast.BinOp):
+ left = _eval_node(node.left, x)
+ right = _eval_node(node.right, x)
+ if isinstance(node.op, ast.Add):
+ return left + right
+ if isinstance(node.op, ast.Sub):
+ return left - right
+ if isinstance(node.op, ast.Mult):
+ return left * right
+ if isinstance(node.op, ast.Div):
+ return left / right
+ # 负数底 + 非整数指数会得到复数,数学绘图不支持,抛 ValueError 让采样点作为断点处理
+ if left < 0 and not right.is_integer():
+ raise ValueError("negative base with fractional exponent")
+ return left**right
+ if isinstance(node, ast.UnaryOp):
+ value = _eval_node(node.operand, x)
+ return -value if isinstance(node.op, ast.USub) else value
+ if isinstance(node, ast.Call):
+ args = [_eval_node(arg, x) for arg in node.args]
+ return _FUNCTION_IMPL[node.func.id](*args) # type: ignore[operator]
+ raise ValueError("unreachable node")
+
+
+def _strip_comment(line: str) -> str:
+ return line.split("#", 1)[0].strip()
+
+
+def _parse_pair(value: str) -> tuple[float, float]:
+ """解析 ``min, max`` / ``min max`` 数值对。"""
+ parts = [p for p in re.split(r"[,,\s]+", value.strip()) if p]
+ if len(parts) != 2:
+ raise ValueError("需要两个数值")
+ return float(parts[0]), float(parts[1])
+
+
+def _parse_directive(line: str) -> tuple[str, str] | None:
+ """指令行形如 ``key: value``(表达式不含冒号,冒号是可靠判别)。"""
+ if ":" not in line or "=" in line:
+ return None
+ key, _, value = line.partition(":")
+ key = key.strip().lower()
+ if not key or " " in key:
+ return None
+ return key, value.strip()
+
+
+def parse_source(source: str) -> FunctionPlotParseResult:
+ """把 function-plot fenced block 源码解析为 FunctionPlot + 诊断。"""
+ diagnostics: list[PlotDiagnostic] = []
+ expressions: list[FunctionPlotExpression] = []
+ domain: tuple[float, float] = (-10.0, 10.0)
+ range_: tuple[float, float] | None = None
+ xlabel: str | None = None
+ ylabel: str | None = None
+ grid: bool = True
+ has_error = False
+ total_nodes = 0
+
+ for lineno, raw_line in enumerate(source.splitlines(), start=1):
+ line = raw_line.strip()
+ if not line or line.startswith("#"):
+ continue
+
+ directive = _parse_directive(line)
+ if directive is not None:
+ key, value = directive
+ if key == "domain":
+ try:
+ domain = _parse_pair(value)
+ except ValueError:
+ diagnostics.append(
+ PlotDiagnostic(
+ severity="warning",
+ code="FUNCTION_PLOT_PARSE_FAILED",
+ message=f"domain 需要两个数值,已忽略:{value!r}",
+ line=lineno,
+ )
+ )
+ elif key == "range":
+ try:
+ range_ = _parse_pair(value)
+ except ValueError:
+ diagnostics.append(
+ PlotDiagnostic(
+ severity="warning",
+ code="FUNCTION_PLOT_PARSE_FAILED",
+ message=f"range 需要两个数值,已忽略:{value!r}",
+ line=lineno,
+ )
+ )
+ elif key == "xlabel":
+ xlabel = value or None
+ elif key == "ylabel":
+ ylabel = value or None
+ elif key == "grid":
+ grid = value.lower() in ("true", "1", "yes", "on")
+ else:
+ diagnostics.append(
+ PlotDiagnostic(
+ severity="warning",
+ code="FUNCTION_PLOT_PARSE_FAILED",
+ message=f"未知指令 {key!r} 已忽略",
+ line=lineno,
+ )
+ )
+ continue
+
+ # 表达式行:y = 或裸
+ expr_text = _strip_comment(line)
+ if not expr_text:
+ continue
+ if "=" in expr_text:
+ lhs, _, rhs = expr_text.partition("=")
+ if lhs.strip().lower() not in ("y", ""):
+ diagnostics.append(
+ PlotDiagnostic(
+ severity="error",
+ code="FUNCTION_PLOT_PARSE_FAILED",
+ message="表达式应形如 'y = '",
+ line=lineno,
+ )
+ )
+ has_error = True
+ continue
+ expr_text = rhs.strip()
+ if not expr_text:
+ diagnostics.append(
+ PlotDiagnostic(
+ severity="error",
+ code="FUNCTION_PLOT_PARSE_FAILED",
+ message="表达式为空",
+ line=lineno,
+ )
+ )
+ has_error = True
+ continue
+
+ try:
+ tree = parse_expression(expr_text)
+ except PlotParseError as exc:
+ exc.diagnostic.line = lineno
+ diagnostics.append(exc.diagnostic)
+ has_error = True
+ continue
+ total_nodes += _count_nodes(tree.body)
+ expressions.append(FunctionPlotExpression(expression=expr_text))
+ # 表达式数量超限:整块回退并提前终止,避免对海量表达式做采样求值
+ if len(expressions) > _MAX_EXPRESSIONS:
+ diagnostics.append(
+ PlotDiagnostic(
+ severity="error",
+ code="FUNCTION_PLOT_TOO_MANY_EXPRESSIONS",
+ message=f"表达式数量超过上限 {_MAX_EXPRESSIONS},已回退为源码占位",
+ )
+ )
+ return FunctionPlotParseResult(plot=None, diagnostics=diagnostics)
+
+ if has_error:
+ return FunctionPlotParseResult(plot=None, diagnostics=diagnostics)
+ if not expressions:
+ diagnostics.append(
+ PlotDiagnostic(
+ severity="error",
+ code="FUNCTION_PLOT_PARSE_FAILED",
+ message="没有找到任何函数表达式",
+ )
+ )
+ return FunctionPlotParseResult(plot=None, diagnostics=diagnostics)
+
+ plot = FunctionPlot(
+ expressions=expressions,
+ domain=domain,
+ range=range_,
+ axes=PlotAxes(xlabel=xlabel, ylabel=ylabel, grid=grid),
+ node_count=total_nodes,
+ )
+ return FunctionPlotParseResult(plot=plot, diagnostics=diagnostics)
diff --git a/backend/app/plot/render.py b/backend/app/plot/render.py
new file mode 100644
index 0000000..efd4d34
--- /dev/null
+++ b/backend/app/plot/render.py
@@ -0,0 +1,434 @@
+"""Function Plot → 静态 SVG 渲染 + 共享几何计算。
+
+只输出纯几何与 的 SVG(无 script/foreignObject/内联事件),可安全内嵌 HTML。
+所有文本与颜色都经过转义/校验,不把用户输入直接拼进标记。
+
+几何计算(范围解析、采样、刻度、非有限点分段)统一收敛到 ``compute_geometry``,
+返回像素坐标的 ``PlotGeometry``;``render_svg`` 只做 SVG 序列化,reportlab 后端
+(``render_reportlab.py``)消费同一份几何,保证 PDF 与 SVG 视觉一致。
+"""
+
+from __future__ import annotations
+
+import html
+import math
+import re
+from dataclasses import dataclass
+
+from app.plot.model import FunctionPlot, StaticRenderResult
+from app.plot.parser import PlotParseError, evaluate, parse_expression
+
+_WIDTH = 640
+_HEIGHT = 480
+_MARGIN = 52 # 四周留白,放轴刻度与标签
+_SAMPLES = 400
+_PALETTE = ["#0969da", "#d1242f", "#1a7f37", "#8250df", "#bf8700", "#e36209"]
+_COLOR_RE = re.compile(r"^#[0-9a-fA-F]{3,8}$")
+# 绘图矩形(像素,SVG y-down):曲线与坐标轴所在区域,坐标轴/网格均在此范围内
+_PLOT_X0 = _MARGIN
+_PLOT_Y0 = _MARGIN
+_PLOT_X1 = _WIDTH - _MARGIN
+_PLOT_Y1 = _HEIGHT - _MARGIN
+
+
+def _safe_color(color: str | None, fallback: str) -> str:
+ return color.strip() if color and _COLOR_RE.match(color.strip()) else fallback
+
+
+def _valid_span(lo: float, hi: float) -> bool:
+ """范围跨度有效:端点有限、跨度有限且大于零。
+
+ 端点相减可能溢出为 ``inf``(如 ``-1e308`` 到 ``1e308``),需单独校验跨度,
+ 否则后续坐标换算会生成含 ``nan`` 的 SVG。
+ """
+ span = hi - lo
+ return math.isfinite(lo) and math.isfinite(hi) and math.isfinite(span) and span > 0
+
+
+def _fmt_num(v: float) -> str:
+ if v == 0:
+ return "0"
+ if abs(v) >= 1e6 or abs(v) < 1e-6:
+ return f"{v:.2e}"
+ return f"{v:.6g}"
+
+
+def _nice_step(span: float, target_ticks: int = 6) -> float:
+ raw = abs(span) / target_ticks
+ if not math.isfinite(raw) or raw <= 0:
+ return 1.0 # 兜底步长,避免 span 为 0/inf 时产生非法刻度
+ mag = 10 ** math.floor(math.log10(raw))
+ for m in (1, 2, 5, 10):
+ if raw <= m * mag:
+ return m * mag
+ return 10 * mag
+
+
+def _ticks(lo: float, hi: float, step: float) -> list[float]:
+ # 防御:非法步长直接返回空,避免除零
+ if not math.isfinite(step) or step <= 0:
+ return []
+ first = math.ceil(lo / step) * step
+ values: list[float] = []
+ v = first
+ # 有上限的整数索引推进 + 步长推进校验,防止浮点精度导致 v+step==v 的死循环
+ for _ in range(1000):
+ if v > hi + step * 1e-9:
+ break
+ values.append(v)
+ nxt = v + step
+ if nxt <= v:
+ break # 步长小于当前数值的浮点精度,已无法推进
+ v = nxt
+ return values
+
+
+def _compute_range(
+ fns: list[tuple[object, object]],
+ xmin: float,
+ xmax: float,
+) -> tuple[float, float]:
+ """采样确定 y 范围;取有限样本的 min/max 加 5% 余量。"""
+ ys: list[float] = []
+ for _expr, tree in fns:
+ for i in range(_SAMPLES + 1):
+ x = xmin + (xmax - xmin) * i / _SAMPLES
+ try:
+ y = evaluate(tree, x) # type: ignore[arg-type]
+ except (ValueError, ZeroDivisionError, OverflowError, TypeError):
+ continue
+ # 复数等非实数结果直接跳过,不参与范围统计
+ if isinstance(y, (int, float)) and math.isfinite(y):
+ ys.append(y)
+
+ if not ys:
+ return -10.0, 10.0
+ lo, hi = min(ys), max(ys)
+ if lo == hi:
+ lo -= 1.0
+ hi += 1.0
+ pad = (hi - lo) * 0.05
+ return lo - pad, hi + pad
+
+
+def _sx(x: float, xmin: float, xmax: float) -> float:
+ """数据 x → 像素 x(SVG y-down 约定,原点左上)。"""
+ return _MARGIN + (x - xmin) / (xmax - xmin) * (_WIDTH - 2 * _MARGIN)
+
+
+def _sy(y: float, ymin: float, ymax: float) -> float:
+ """数据 y → 像素 y(SVG y-down 约定,原点左上)。"""
+ return _HEIGHT - _MARGIN - (y - ymin) / (ymax - ymin) * (_HEIGHT - 2 * _MARGIN)
+
+
+@dataclass
+class PlotGeometry:
+ """已解析的几何:范围、轴位置、刻度、曲线像素点段、标签与 warnings。
+
+ 像素坐标统一为 SVG y-down 约定;reportlab 后端(y-up)自行翻转 y。
+ """
+
+ width: int
+ height: int
+ xmin: float
+ xmax: float
+ ymin: float
+ ymax: float
+ x_axis_y: float # 数据空间里 x 轴所在 y(过原点则 0,否则贴边)
+ y_axis_x: float # 数据空间里 y 轴所在 x(过原点则 0,否则贴边)
+ xticks: list[float]
+ yticks: list[float]
+ polylines: list[list[list[tuple[float, float]]]] # 按表达式分组:段 → 像素点
+ colors: list[str] # 与 polylines 对齐
+ xlabel: str | None
+ ylabel: str | None
+ grid: bool
+ warnings: list[str]
+
+
+def _clip_segment(
+ p0: tuple[float, float],
+ p1: tuple[float, float],
+ x0: float,
+ y0: float,
+ x1: float,
+ y1: float,
+) -> tuple[tuple[float, float], tuple[float, float]] | None:
+ """Liang-Barsky:把线段裁剪到轴对齐矩形 [x0,x1]×[y0,y1],完全在外返回 None。"""
+ dx = p1[0] - p0[0]
+ dy = p1[1] - p0[1]
+ p = (-dx, dx, -dy, dy)
+ q = (p0[0] - x0, x1 - p0[0], p0[1] - y0, y1 - p0[1])
+ u1, u2 = 0.0, 1.0
+ for pk, qk in zip(p, q):
+ if pk == 0:
+ if qk < 0:
+ return None
+ else:
+ r = qk / pk
+ if pk < 0:
+ if r > u2:
+ return None
+ if r > u1:
+ u1 = r
+ else:
+ if r < u1:
+ return None
+ if r < u2:
+ u2 = r
+ if u1 > u2:
+ return None
+ return (p0[0] + u1 * dx, p0[1] + u1 * dy), (p0[0] + u2 * dx, p0[1] + u2 * dy)
+
+
+def _points_close(
+ a: tuple[float, float], b: tuple[float, float], eps: float = 1e-9
+) -> bool:
+ return abs(a[0] - b[0]) < eps and abs(a[1] - b[1]) < eps
+
+
+def _clip_polyline(
+ points: list[tuple[float, float]],
+ x0: float,
+ y0: float,
+ x1: float,
+ y1: float,
+) -> list[list[tuple[float, float]]]:
+ """把折线裁剪到矩形,返回若干连续子段;相邻点不衔接处自动断段。"""
+ if not points:
+ return []
+ segments: list[list[tuple[float, float]]] = []
+ current: list[tuple[float, float]] = []
+ for i in range(len(points) - 1):
+ clipped = _clip_segment(points[i], points[i + 1], x0, y0, x1, y1)
+ if clipped is None:
+ if current:
+ segments.append(current)
+ current = []
+ continue
+ a, b = clipped
+ # 共享点被裁剪修改(折线短暂越界后折返)时,a 与上一段末点不衔接,需断段
+ if current and not _points_close(a, current[-1]):
+ segments.append(current)
+ current = []
+ if not current:
+ current.append(a)
+ current.append(b)
+ if current:
+ segments.append(current)
+ return segments
+
+
+def _sample_segments(
+ tree: object,
+ xmin: float,
+ xmax: float,
+ ymin: float,
+ ymax: float,
+) -> list[list[tuple[float, float]]]:
+ """采样并映射为像素点段,再裁剪到绘图矩形。
+
+ 两处断段:非有限点处(画穿渐近线);相邻有限采样点横跨可见范围上下两侧时
+ (渐近点恰好落在两个采样点之间,否则会被裁剪成贯穿绘图区的伪竖线)。
+ """
+ segments: list[list[tuple[float, float]]] = []
+ points: list[tuple[float, float]] = []
+ prev_y: float | None = None
+ for i in range(_SAMPLES + 1):
+ x = xmin + (xmax - xmin) * i / _SAMPLES
+ try:
+ y = evaluate(tree, x) # type: ignore[arg-type]
+ except (ValueError, ZeroDivisionError, OverflowError, TypeError):
+ y = math.nan
+ if not isinstance(y, (int, float)) or not math.isfinite(y):
+ if points:
+ segments.append(points)
+ points = []
+ prev_y = None
+ continue
+ px = _sx(x, xmin, xmax)
+ py = _sy(y, ymin, ymax)
+ # 映射后的坐标必须有限:显式 range 下极端 y 值可能让像素坐标溢出为 inf
+ if not (math.isfinite(px) and math.isfinite(py)):
+ if points:
+ segments.append(points)
+ points = []
+ prev_y = None
+ continue
+ # 渐近线检测:相邻有限采样点分居可见范围上下两侧(一个 < ymin、一个 > ymax),
+ # 说明两者之间夹着竖直渐近线,断段避免被 Liang-Barsky 裁剪成贯穿绘图区的伪竖线
+ if prev_y is not None and (
+ (prev_y < ymin and y > ymax) or (prev_y > ymax and y < ymin)
+ ):
+ if points:
+ segments.append(points)
+ points = []
+ points.append((px, py))
+ prev_y = y
+ if points:
+ segments.append(points)
+
+ # 裁剪到绘图矩形:reportlab 无 SVG viewport 那样的自动裁剪,超出显式 range 的
+ # 曲线会覆盖页面其他内容,故在共享几何层统一裁剪(SVG 也一并收敛到绘图区)。
+ clipped: list[list[tuple[float, float]]] = []
+ for seg in segments:
+ clipped.extend(_clip_polyline(seg, _PLOT_X0, _PLOT_Y0, _PLOT_X1, _PLOT_Y1))
+ return clipped
+
+
+def compute_geometry(plot: FunctionPlot) -> PlotGeometry:
+ """解析并计算几何,供 SVG 与 reportlab 后端复用。"""
+ warnings: list[str] = []
+ xmin, xmax = plot.domain
+ if not _valid_span(xmin, xmax):
+ warnings.append("domain 无效,回退到 [-10, 10]")
+ xmin, xmax = -10.0, 10.0
+
+ # 重新解析并编译表达式(parse_source 已校验,这里异常只在模型被绕过时触发)
+ fns: list[tuple[object, object]] = []
+ for expr in plot.expressions:
+ try:
+ tree = parse_expression(expr.expression)
+ except PlotParseError as exc:
+ warnings.append(f"表达式无法渲染,已跳过:{expr.expression}({exc.diagnostic.message})")
+ continue
+ fns.append((expr, tree))
+
+ # 纵轴范围:显式 range 有效则用之;无效(退化/非有限/跨度溢出)丢弃并自动采样重算
+ if plot.range is not None:
+ lo, hi = float(plot.range[0]), float(plot.range[1])
+ if _valid_span(lo, hi):
+ ymin, ymax = lo, hi
+ else:
+ warnings.append("range 无效,改用自动范围")
+ ymin, ymax = _compute_range(fns, xmin, xmax)
+ else:
+ ymin, ymax = _compute_range(fns, xmin, xmax)
+
+ # 最终防线:自动范围在极端样本下也可能溢出,坐标映射前必须保证跨度有限且大于零
+ if not _valid_span(ymin, ymax):
+ warnings.append("y 范围跨度无法表示,回退到 [-10, 10]")
+ ymin, ymax = -10.0, 10.0
+
+ x_axis_y = 0.0 if ymin <= 0 <= ymax else ymin
+ y_axis_x = 0.0 if xmin <= 0 <= xmax else xmin
+ xticks = _ticks(xmin, xmax, _nice_step(xmax - xmin))
+ yticks = _ticks(ymin, ymax, _nice_step(ymax - ymin))
+
+ polylines: list[list[list[tuple[float, float]]]] = []
+ colors: list[str] = []
+ for i, (expr, tree) in enumerate(fns):
+ color = _safe_color(expr.color, _PALETTE[i % len(_PALETTE)])
+ colors.append(color)
+ polylines.append(_sample_segments(tree, xmin, xmax, ymin, ymax))
+
+ return PlotGeometry(
+ width=_WIDTH,
+ height=_HEIGHT,
+ xmin=xmin,
+ xmax=xmax,
+ ymin=ymin,
+ ymax=ymax,
+ x_axis_y=x_axis_y,
+ y_axis_x=y_axis_x,
+ xticks=xticks,
+ yticks=yticks,
+ polylines=polylines,
+ colors=colors,
+ xlabel=plot.axes.xlabel,
+ ylabel=plot.axes.ylabel,
+ grid=plot.axes.grid,
+ warnings=warnings,
+ )
+
+
+# --- SVG 序列化(与 compute_geometry 共用,保证字节级稳定) ---
+def _grid_svg(geo: PlotGeometry) -> str:
+ sx = lambda x: _sx(x, geo.xmin, geo.xmax)
+ sy = lambda y: _sy(y, geo.ymin, geo.ymax)
+ parts: list[str] = []
+ for x in geo.xticks:
+ parts.append(
+ f' '
+ )
+ for y in geo.yticks:
+ parts.append(
+ f' '
+ )
+ return "".join(parts)
+
+
+def _axes_svg(geo: PlotGeometry) -> str:
+ sx = lambda x: _sx(x, geo.xmin, geo.xmax)
+ sy = lambda y: _sy(y, geo.ymin, geo.ymax)
+ parts: list[str] = []
+ # 坐标轴:过原点则画在原点,否则贴边,保证始终有参照系
+ parts.append(
+ f' '
+ )
+ parts.append(
+ f' '
+ )
+ # x 轴刻度数字(画在轴下方)
+ for x in geo.xticks:
+ parts.append(
+ f'{html.escape(_fmt_num(x))} '
+ )
+ # y 轴刻度数字(画在轴左侧)
+ for y in geo.yticks:
+ parts.append(
+ f'{html.escape(_fmt_num(y))} '
+ )
+ return "".join(parts)
+
+
+def _polylines_svg(geo: PlotGeometry) -> str:
+ parts: list[str] = []
+ for segments, color in zip(geo.polylines, geo.colors):
+ for seg in segments:
+ points = " ".join(f"{px:.2f},{py:.2f}" for px, py in seg)
+ parts.append(f' ')
+ return "".join(parts)
+
+
+def _labels_svg(geo: PlotGeometry) -> str:
+ parts: list[str] = []
+ if geo.xlabel:
+ parts.append(
+ f'{html.escape(geo.xlabel)} '
+ )
+ if geo.ylabel:
+ parts.append(
+ f''
+ f'{html.escape(geo.ylabel)} '
+ )
+ return "".join(parts)
+
+
+def render_svg(plot: FunctionPlot) -> StaticRenderResult:
+ """把已解析的 FunctionPlot 渲染为内嵌 SVG。"""
+ geo = compute_geometry(plot)
+ parts: list[str] = [
+ f''
+ ]
+ if geo.grid:
+ parts.append(_grid_svg(geo))
+ parts.append(_axes_svg(geo))
+ parts.append(_polylines_svg(geo))
+ parts.append(_labels_svg(geo))
+ parts.append(" ")
+
+ return StaticRenderResult(
+ content="".join(parts),
+ width=geo.width,
+ height=geo.height,
+ warnings=geo.warnings,
+ )
diff --git a/backend/app/plot/render_reportlab.py b/backend/app/plot/render_reportlab.py
new file mode 100644
index 0000000..3bd5452
--- /dev/null
+++ b/backend/app/plot/render_reportlab.py
@@ -0,0 +1,120 @@
+"""Function Plot → reportlab 矢量 Drawing(供 PDF 内嵌)。
+
+消费 ``render.compute_geometry`` 的共享几何,产出 ``reportlab.graphics.shapes.Drawing``:
+网格/坐标轴用 ``Line``、曲线用 ``PolyLine``、刻度数字与轴标签用 ``String``。
+reportlab 原点在左下(y-up),与 SVG 的 y-down 相反,故对几何里的像素 y 统一翻转;
+轴标签(ylabel)用 ``Group.rotate`` 旋转为竖向文本。中文字体复用内置 STSong-Light,
+guarded 注册避免与 pdf.py 重复注册。
+"""
+
+from __future__ import annotations
+
+from reportlab.graphics.shapes import Drawing, Group, Line, PolyLine, String
+from reportlab.lib.colors import HexColor
+from reportlab.pdfbase import pdfmetrics
+from reportlab.pdfbase.cidfonts import UnicodeCIDFont
+
+from app.plot.model import FunctionPlot
+from app.plot.render import PlotGeometry, _fmt_num, _sx, _sy, compute_geometry
+
+_FONT = "STSong-Light"
+if _FONT not in pdfmetrics.getRegisteredFontNames():
+ pdfmetrics.registerFont(UnicodeCIDFont(_FONT))
+
+_GRID_COLOR = HexColor("#eaeef2")
+_AXIS_COLOR = HexColor("#57606a")
+_LABEL_COLOR = HexColor("#1f2328")
+_TICK_FONT_SIZE = 10
+_LABEL_FONT_SIZE = 12
+
+
+def _build_drawing(geo: PlotGeometry) -> Drawing:
+ """由共享几何构建矢量 Drawing(坐标翻转后仍沿用 SVG 的像素布局)。"""
+ drawing = Drawing(geo.width, geo.height)
+
+ # SVG y-down → reportlab y-up:翻转像素 y
+ def sx(x: float) -> float:
+ return _sx(x, geo.xmin, geo.xmax)
+
+ def sy(y: float) -> float:
+ return geo.height - _sy(y, geo.ymin, geo.ymax)
+
+ # 网格
+ if geo.grid:
+ for x in geo.xticks:
+ drawing.add(
+ Line(sx(x), sy(geo.ymin), sx(x), sy(geo.ymax), strokeColor=_GRID_COLOR, strokeWidth=0.5)
+ )
+ for y in geo.yticks:
+ drawing.add(
+ Line(sx(geo.xmin), sy(y), sx(geo.xmax), sy(y), strokeColor=_GRID_COLOR, strokeWidth=0.5)
+ )
+
+ # 坐标轴(过原点画在原点,否则贴边,与 SVG 一致)
+ drawing.add(
+ Line(sx(geo.xmin), sy(geo.x_axis_y), sx(geo.xmax), sy(geo.x_axis_y), strokeColor=_AXIS_COLOR, strokeWidth=0.7)
+ )
+ drawing.add(
+ Line(sx(geo.y_axis_x), sy(geo.ymin), sx(geo.y_axis_x), sy(geo.ymax), strokeColor=_AXIS_COLOR, strokeWidth=0.7)
+ )
+
+ # 刻度数字(x 轴下方、y 轴左侧)
+ for x in geo.xticks:
+ drawing.add(
+ String(
+ sx(x), sy(geo.x_axis_y) - 14, _fmt_num(x),
+ fontName=_FONT, fontSize=_TICK_FONT_SIZE, fillColor=_AXIS_COLOR, textAnchor="middle",
+ )
+ )
+ for y in geo.yticks:
+ drawing.add(
+ String(
+ sx(geo.y_axis_x) - 6, sy(y) - 3, _fmt_num(y),
+ fontName=_FONT, fontSize=_TICK_FONT_SIZE, fillColor=_AXIS_COLOR, textAnchor="end",
+ )
+ )
+
+ # 曲线(非有限点处已由几何断成多段)
+ for segments, color in zip(geo.polylines, geo.colors):
+ for seg in segments:
+ flipped = [(px, geo.height - py) for px, py in seg]
+ drawing.add(PolyLine(flipped, strokeColor=HexColor(color), strokeWidth=1.4))
+
+ # 轴标签
+ if geo.xlabel:
+ drawing.add(
+ String(
+ geo.width / 2, 10, geo.xlabel,
+ fontName=_FONT, fontSize=_LABEL_FONT_SIZE, fillColor=_LABEL_COLOR, textAnchor="middle",
+ )
+ )
+ if geo.ylabel:
+ # 竖向标签:Group.rotate(90) 在 y-up 坐标下等价于 SVG 的 rotate(-90)。
+ # 文本放在组内局部坐标 (0,0),先平移后旋转得到 T·R(先绕原点旋转、再平移到
+ # 目标位置),避免用绝对坐标定位又用相同坐标当旋转中心造成的重复变换,
+ # 后者会把标签甩到画布之外(负 x 区域)。
+ label = Group()
+ label.add(
+ String(
+ 0, 0, geo.ylabel,
+ fontName=_FONT, fontSize=_LABEL_FONT_SIZE, fillColor=_LABEL_COLOR, textAnchor="middle",
+ )
+ )
+ label.translate(16, geo.height / 2)
+ label.rotate(90)
+ drawing.add(label)
+
+ return drawing
+
+
+def render_drawing(plot: FunctionPlot, width: float | None = None) -> Drawing:
+ """把已解析的 FunctionPlot 渲染为 reportlab Drawing(可直接追加到 platypus story)。
+
+ ``width`` 为目标输出宽度(点),用于把 640px 的几何缩放到页面内容宽;省略则按
+ 原始尺寸输出。缩放只影响 PDF 渲染,不改动共享几何。
+ """
+ geo = compute_geometry(plot)
+ drawing = _build_drawing(geo)
+ if width is not None and width > 0:
+ drawing.renderScale = min(1.0, width / geo.width)
+ return drawing
diff --git a/backend/app/plot/renderer.py b/backend/app/plot/renderer.py
new file mode 100644
index 0000000..78ee049
--- /dev/null
+++ b/backend/app/plot/renderer.py
@@ -0,0 +1,66 @@
+"""StaticRenderer 内部契约(契约 §10.4)。
+
+把「静态可视化」抽象为统一请求/协议:导出器只面向 StaticRenderer,不再直接调用
+``render_svg`` 等具体实现。后端当前仅能静态渲染函数图像;Mermaid 后端无渲染能力,
+返回占位结果交前端渲染。
+"""
+
+from __future__ import annotations
+
+from typing import Literal, Protocol
+
+from pydantic import BaseModel, Field
+
+from app.plot.model import FunctionPlot, FunctionPlotParseResult, StaticRenderResult
+from app.plot.parser import parse_source
+from app.plot.render import render_svg
+
+
+class StaticRenderRequest(BaseModel):
+ """一次静态渲染请求;source_hash 供缓存/去重,theme 供主题化渲染。"""
+
+ kind: Literal["function_plot", "mermaid"]
+ source: str
+ source_hash: str = ""
+ theme: str | None = None
+ width: int | None = None
+ height: int | None = None
+
+
+class StaticRenderer(Protocol):
+ """静态渲染器协议:请求 → 渲染结果(content 为可直接内嵌的标记)。"""
+
+ def render(self, request: StaticRenderRequest) -> StaticRenderResult: ...
+
+
+class FunctionPlotStaticRenderer:
+ """函数图像渲染器:parse_source 解析 → render_svg 输出内嵌 SVG。
+
+ ``parse`` 与 ``render_plot`` 拆开,供导出器在渲染前先拿 node_count 做文档级
+ 累计复杂度预算、并消费解析诊断。
+ """
+
+ def parse(self, request: StaticRenderRequest) -> FunctionPlotParseResult:
+ return parse_source(request.source)
+
+ def render(self, request: StaticRenderRequest) -> StaticRenderResult:
+ parsed = self.parse(request)
+ if parsed.plot is None:
+ raise ValueError("function-plot source has no valid plot")
+ return self.render_plot(parsed.plot)
+
+ def render_plot(self, plot: FunctionPlot) -> StaticRenderResult:
+ return render_svg(plot)
+
+
+class MermaidStaticRenderer:
+ """Mermaid 后端无渲染能力:返回空占位结果,交前端渲染。"""
+
+ def render(self, request: StaticRenderRequest) -> StaticRenderResult:
+ return StaticRenderResult(
+ content="",
+ mime_type="text/plain",
+ width=0,
+ height=0,
+ warnings=["mermaid 需前端渲染,已保留为占位代码块"],
+ )
diff --git a/backend/app/routes.py b/backend/app/routes.py
index 8ae1e73..e2a5e7c 100644
--- a/backend/app/routes.py
+++ b/backend/app/routes.py
@@ -6,7 +6,7 @@ from datetime import datetime, timezone
from uuid import uuid4
from fastapi import APIRouter, Header, Query, Request
-from fastapi.responses import StreamingResponse
+from fastapi.responses import FileResponse, StreamingResponse
from app.agent import AgentCapacityError, AgentRunNotFoundError
from app.container import container
@@ -57,6 +57,11 @@ from app.contracts import (
ModelRoutingResponse,
SpeakerMatchRequest,
SpeakerMatchResult,
+ ExportFormat,
+ ExportJob,
+ ExportJobListResponse,
+ ExportRequest,
+ ExportStatus,
Note,
NoteCreateRequest,
NoteListResponse,
@@ -105,9 +110,11 @@ from app.contracts import (
from app.agent import AgentCapacityError, AgentRunNotFoundError
from app.benchmarks import datasets as benchmark_datasets
from app.benchmarks import service as benchmark_service
+from app.config import get_settings
from app.container import container
from app.services.persona_settings import PersonaSettings, load_persona, save_persona
from app.errors import ApiError
+from app.export import service as export_service
from app.extensions import ExtensionError
from app.extensions.mcp_registry import McpRegistryError
from app.providers.base import ProviderError
@@ -1507,6 +1514,81 @@ async def get_benchmark_report(run_id: str) -> BenchmarkReport:
return report
+@router.post(
+ "/exports",
+ response_model=ExportJob,
+ status_code=202,
+ tags=["Export"],
+)
+async def create_export(request: ExportRequest) -> ExportJob:
+ return await export_service.create_export(request)
+
+
+@router.get(
+ "/exports",
+ response_model=ExportJobListResponse,
+ tags=["Export"],
+)
+async def list_exports(
+ status: ExportStatus | None = Query(default=None),
+ format: ExportFormat | None = Query(default=None),
+ limit: int = Query(default=50, ge=1, le=200),
+ offset: int = Query(default=0, ge=0),
+) -> ExportJobListResponse:
+ items, total = export_service.list_exports(
+ status=status, format=format, limit=limit, offset=offset
+ )
+ return ExportJobListResponse(
+ items=items, page=PageMeta(total=total, limit=limit, offset=offset)
+ )
+
+
+@router.get(
+ "/exports/{job_id}",
+ response_model=ExportJob,
+ tags=["Export"],
+)
+async def get_export(job_id: str) -> ExportJob:
+ job = export_service.get_export(job_id)
+ if job is None:
+ raise ApiError(
+ 404, "EXPORT_JOB_NOT_FOUND", "export job not found", {"job_id": job_id}
+ )
+ return job
+
+
+@router.get(
+ "/exports/{job_id}/file",
+ tags=["Export"],
+)
+async def get_export_file(job_id: str) -> FileResponse:
+ path = export_service.get_export_file(job_id) # 未完成/过期分别抛 404/410
+ job = export_service.get_export(job_id)
+ if job is None or job.file is None:
+ raise ApiError(
+ 404, "EXPORT_JOB_NOT_FOUND", "export file not ready", {"job_id": job_id}
+ )
+ return FileResponse(
+ path=path,
+ media_type=job.file.mime_type,
+ filename=job.file.file_name,
+ )
+
+
+@router.post(
+ "/exports/{job_id}/cancel",
+ response_model=OperationResponse,
+ tags=["Export"],
+)
+async def cancel_export(job_id: str) -> OperationResponse:
+ job = export_service.cancel_export(job_id)
+ if job is None:
+ raise ApiError(
+ 404, "EXPORT_JOB_NOT_FOUND", "export job not found", {"job_id": job_id}
+ )
+ return OperationResponse(
+ status="accepted", resource_id=job_id, message="Export cancellation accepted."
+ )
@router.get("/settings/persona", response_model=PersonaSettings, tags=["Settings"])
diff --git a/backend/pyproject.toml b/backend/pyproject.toml
index 6c3d4c0..c5e9baf 100644
--- a/backend/pyproject.toml
+++ b/backend/pyproject.toml
@@ -9,8 +9,11 @@ dependencies = [
"fastapi>=0.116,<1.0",
"httpx>=0.28,<1.0",
"jsonschema>=4.25,<5.0",
+ "mistune>=3.0,<4.0",
+ "python-docx>=1.1,<2.0",
"pyyaml>=6.0,<7.0",
"referencing>=0.36,<1.0",
+ "reportlab>=4.0,<5.0",
"sqlite-vec>=0.1.9",
"uvicorn[standard]>=0.35,<1.0",
]
diff --git a/backend/tests/test_export.py b/backend/tests/test_export.py
new file mode 100644
index 0000000..cef356a
--- /dev/null
+++ b/backend/tests/test_export.py
@@ -0,0 +1,793 @@
+"""Export Service 的单元与端到端测试。
+
+沿用 conftest 隔离机制:APP_DATA_DIR / DB / Vault / exports 目录都落在临时目录,
+不读写真实数据。导出采用「创建即 queued + 后台 Task 执行」的异步模型,测试在同一
+事件循环内创建并等待后台任务结束,得到终态 ExportJob 后再断言。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import base64
+import re
+import zlib
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from pydantic import ValidationError
+
+from app.config import get_settings
+from app.contracts import (
+ ExportFormat,
+ ExportJob,
+ ExportOptions,
+ ExportRequest,
+ ExportSource,
+ ExportSourceType,
+ ExportStatus,
+)
+from app.errors import ApiError
+from app.export import service as export_service
+from app.export.exporters.html import HtmlExporter
+from app.export.markdown import parse_document
+
+MD = """# 进程调度
+
+一些 **加粗** 和 *斜体*,[链接](https://a.b) 与 `code`。
+
+- 项目一
+- 项目二
+
+```python
+print(1)
+```
+
+```mermaid
+graph LR
+```
+
+```function_plot
+y = x
+```
+
+| a | b |
+|---|---|
+| 1 | 2 |
+
+行内 $x^2$ 与块级
+$$
+y = mx + b
+$$
+"""
+
+
+@pytest.fixture(autouse=True)
+def _reset_export_state():
+ """清空内存注册表,避免跨用例的任务/取消标志互相污染。
+
+ 每个用例经 `asyncio.run()` 使用独立事件循环,模块级 Semaphore 会绑定到首个
+ 循环,跨用例复用会触发「bound to a different event loop」;此处每例重建槽位。
+ """
+ export_service._jobs.clear()
+ export_service._tasks.clear()
+ export_service._cancel_flags.clear()
+ export_service._render_slots = asyncio.Semaphore(export_service.MAX_CONCURRENT_RENDERS)
+ yield
+ export_service._jobs.clear()
+ export_service._tasks.clear()
+ export_service._cancel_flags.clear()
+
+
+def _create_and_wait(request: ExportRequest) -> object:
+ """创建导出并在同一事件循环内等待后台任务结束,返回终态 ExportJob。"""
+
+ async def _execute():
+ job = await export_service.create_export(request)
+ return await export_service.wait_for_export(job.job_id)
+
+ return asyncio.run(_execute())
+
+
+# --------------------------------------------------------------------------- #
+# markdown → Document AST
+# --------------------------------------------------------------------------- #
+def _types(nodes) -> list[str]:
+ return [n.type for n in nodes]
+
+
+def test_parse_document_heading_and_inline() -> None:
+ doc = parse_document("# 标题\n\n一段 **加粗** 和 [链接](https://a.b)。")
+
+ assert doc.type == "document"
+ heading = doc.children[0]
+ assert heading.type == "heading"
+ assert heading.attributes["level"] == 1
+
+ para = doc.children[1]
+ assert para.type == "paragraph"
+ kinds = _types(para.children)
+ assert "text" in kinds
+ assert "strong" in kinds
+ assert "link" in kinds
+
+ link = next(c for c in para.children if c.type == "link")
+ assert link.attributes["href"] == "https://a.b"
+
+
+def test_parse_document_list_and_code_fencing() -> None:
+ doc = parse_document("- a\n- b\n\n```mermaid\ngraph LR\n```\n\n```function_plot\ny=x\n```\n\n```python\nx\n```")
+
+ kinds = [c.type for c in doc.children]
+ assert kinds[0] == "list"
+ assert kinds[1] == "mermaid"
+ assert kinds[2] == "function_plot"
+ assert kinds[3] == "code_block"
+
+ code = doc.children[3]
+ assert code.attributes["language"] == "python"
+ assert code.text == "x"
+
+
+def test_parse_document_table_and_math() -> None:
+ doc = parse_document("| a | b |\n|---|---|\n| 1 | 2 |\n\n$x^2$\n\n$$\ny=mx\n$$")
+
+ table = doc.children[0]
+ assert table.type == "table"
+ assert table.children[0].type == "table_row"
+ assert table.children[0].children[0].attributes["head"] is True
+
+ # 表格后是「行内数学所在段落」与「块级数学」
+ kinds = [c.type for c in doc.children[1:]]
+ assert "paragraph" in kinds
+ assert "math_block" in kinds
+
+
+def test_parse_document_image_maps_src_alt_title() -> None:
+ doc = parse_document('')
+ img = doc.children[0].children[0]
+ assert img.type == "image"
+ assert img.attributes["src"] == "https://a.b/img.png"
+ assert img.attributes["alt"] == "替代文本"
+ assert img.attributes["title"] == "标题"
+
+
+def test_parse_document_function_plot_dash_alias() -> None:
+ doc = parse_document("```function-plot\ny = x^2\n```")
+ assert doc.children[0].type == "function_plot"
+ assert doc.children[0].text == "y = x^2"
+
+
+# --------------------------------------------------------------------------- #
+# HtmlExporter
+# --------------------------------------------------------------------------- #
+async def _render(markdown: str, *, title: str = "") -> str:
+ doc = parse_document(markdown)
+ doc.attributes["title"] = title
+ result = await HtmlExporter().export(doc, ExportOptions())
+ return result.content.decode("utf-8")
+
+
+def test_html_exporter_renders_basic_nodes_and_escapes() -> None:
+ html = asyncio.run(_render("# 标题\n\n**加粗** [链接](https://a.b) 与 原始 。"))
+
+ assert "标题 " in html
+ assert "加粗 " in html
+ assert '链接 ' in html
+ # 原始 HTML 必须被转义,不能注入文档
+ assert "<b>原始</b>" in html
+ assert "原始 " not in html
+
+
+def test_html_exporter_marks_mermaid_and_function_plot() -> None:
+ result = asyncio.run(HtmlExporter().export(parse_document("```mermaid\ngraph LR\n```"), ExportOptions()))
+
+ html = result.content.decode("utf-8")
+ assert 'graph LR ' in html
+ assert any("mermaid" in w for w in result.warnings)
+
+
+def test_html_exporter_rejects_unsafe_link_protocol() -> None:
+ result = asyncio.run(
+ HtmlExporter().export(parse_document("[点我](javascript:alert(1))"), ExportOptions())
+ )
+ html = result.content.decode("utf-8")
+ assert "javascript:" not in html
+ assert "点我" in html
+ assert any("不安全" in w for w in result.warnings)
+
+
+def test_html_exporter_rejects_unsafe_image_protocol() -> None:
+ result = asyncio.run(
+ HtmlExporter().export(parse_document("![alt](data:text/html,