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)}" + + 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)}" + + 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..f073904 --- /dev/null +++ b/backend/app/export/exporters/pdf.py @@ -0,0 +1,375 @@ +"""PdfExporter:Document AST → PDF(reportlab platypus)。 + +v1 为文本优先:标题/段落/行内强调与链接/列表/引用/表格/代码块/数学文本均可导出; +function_plot 与 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..b47aa14 --- /dev/null +++ b/backend/app/plot/render.py @@ -0,0 +1,334 @@ +"""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}$") + + +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 _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]] = [] + 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 = [] + 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 = [] + continue + points.append((px, py)) + if points: + segments.append(points) + return segments + + +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..03a6819 --- /dev/null +++ b/backend/app/plot/render_reportlab.py @@ -0,0 +1,116 @@ +"""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: + # 竖向标签:rotate(90) 在 y-up 坐标下等价于 SVG 的 rotate(-90) + label = Group() + label.add( + String( + 16, geo.height / 2, geo.ylabel, + fontName=_FONT, fontSize=_LABEL_FONT_SIZE, fillColor=_LABEL_COLOR, textAnchor="middle", + ) + ) + label.rotate(90, 16, geo.height / 2) + 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('![替代文本](https://a.b/img.png "标题")') + 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,