fix: preserve exports on close and support themed PDF without export quotas

This commit is contained in:
2026-09-07 14:04:03 +08:00
parent 47c53b6f38
commit 9f097ea629
20 changed files with 485 additions and 123 deletions
+20 -2
View File
@@ -1426,7 +1426,18 @@ class ExportSource(Contract):
return self
class ExportPalette(Contract):
page: str = Field(pattern=r'^#[0-9a-fA-F]{6}$')
surface: str = Field(pattern=r'^#[0-9a-fA-F]{6}$')
text: str = Field(pattern=r'^#[0-9a-fA-F]{6}$')
muted: str = Field(pattern=r'^#[0-9a-fA-F]{6}$')
code: str = Field(pattern=r'^#[0-9a-fA-F]{6}$')
border: str = Field(pattern=r'^#[0-9a-fA-F]{6}$')
accent: str = Field(pattern=r'^#[0-9a-fA-F]{6}$')
class ExportOptions(Contract):
palette: ExportPalette | None = None
theme_id: str = "light"
include_title: bool = True
include_metadata: bool = False
@@ -1437,16 +1448,23 @@ class ExportOptions(Contract):
class ExportAsset(Contract):
kind: Literal['mermaid', 'math_block', 'math_inline', 'image']
source_hash: str = Field(pattern=r'^[a-f0-9]{64}$')
png_base64: str = Field(max_length=2800000)
png_base64: str
class ExportRequest(Contract):
assets: list[ExportAsset] = Field(default_factory=list, max_length=64)
assets: list[ExportAsset] = Field(default_factory=list)
title: str = Field(default="", max_length=200)
source: ExportSource
format: ExportFormat
options: ExportOptions = Field(default_factory=ExportOptions)
@model_validator(mode="after")
def _asset_limits(self) -> "ExportRequest":
if self.format != ExportFormat.pdf:
if len(self.assets) > 64 or any(len(asset.png_base64) > 2800000 for asset in self.assets):
raise ValueError("export asset count or size limit exceeded")
return self
class ExportProgress(Contract):
phase: str
+22 -19
View File
@@ -1,4 +1,4 @@
"""Bounded raster-only resource boundary. No URLs, XML or filesystem paths accepted."""
"""Raster-only resources; PDF bypasses export quotas but retains path/format validation."""
import base64
import hashlib
import threading
@@ -8,12 +8,14 @@ from app.errors import ApiError
_math_lock = threading.Lock()
def enrich_document(document, file_path=None):
"""Embed local vault images and bounded MathText. Unsupported TeX stays explicit."""
def enrich_document(document, file_path=None, unlimited=False, options=None):
"""Embed Vault images and MathText, with format-specific quotas and palette."""
from app.config import get_settings
from urllib.parse import unquote, urlsplit
vault = get_settings().vault_path.resolve()
base = (vault / (file_path or '')).parent if file_path else vault
from app.export.themes import pdf_palette
palette = pdf_palette(options, []) if unlimited and options else None
warnings = []
count = total = pixels = 0
def visit(node):
@@ -21,14 +23,14 @@ def enrich_document(document, file_path=None):
if node.type in {'image','math_block','math_inline'} or node.attributes.get('static_png'):
count += 1
try:
if count > 64: raise ValueError('resource count')
if not unlimited and count > 64: raise ValueError('resource count')
if node.attributes.get('static_png'):
raw = node.attributes['static_png']
elif node.type == 'image':
src = str(node.attributes.get('src',''))
if urlsplit(src).scheme or src.startswith('//'): raise ValueError('remote image')
path = (base / unquote(src)).resolve()
if not path.is_relative_to(vault) or path.suffix.lower() not in {'.png','.jpg','.jpeg','.webp'} or path.stat().st_size > 2_000_000:
if not path.is_relative_to(vault) or path.suffix.lower() not in {'.png','.jpg','.jpeg','.webp'} or (not unlimited and path.stat().st_size > 2_000_000):
raise ValueError('image path or budget')
raw = path.read_bytes()
else:
@@ -36,23 +38,24 @@ def enrich_document(document, file_path=None):
depth = 0
for char in source:
depth += (char == '{') - (char == '}')
if depth > 20: raise ValueError('math depth')
if len(source) > 512 or depth != 0: raise ValueError('math budget')
if not unlimited and depth > 20: raise ValueError('math depth')
if (not unlimited and len(source) > 512) or depth != 0: raise ValueError('math budget')
from matplotlib.mathtext import math_to_image
with _math_lock:
from matplotlib import rc_context
with _math_lock, rc_context({'savefig.transparent': bool(palette)}):
out = BytesIO()
math_to_image('$'+source+'$', out, dpi=180, format='png', color='black')
math_to_image('$'+source+'$', out, dpi=180, format='png', color=palette['text'] if palette else 'black')
raw = out.getvalue()
with Image.open(BytesIO(raw)) as image:
pixels += image.width * image.height
if pixels > 16_000_000: raise ValueError('document pixels')
if image.width * image.height > 4_000_000: raise ValueError('image dimensions')
if not unlimited and pixels > 16_000_000: raise ValueError('document pixels')
if not unlimited and image.width * image.height > 4_000_000: raise ValueError('image dimensions')
out = BytesIO()
# Flatten alpha on white for portable print/Word output.
rgba=image.convert('RGBA'); background=Image.new('RGBA',rgba.size,'white')
# Composite transparency over the PDF theme or the print/Word white surface.
rgba=image.convert('RGBA'); background=Image.new('RGBA',rgba.size,palette['surface'] if palette else 'white')
background.alpha_composite(rgba); background.convert('RGB').save(out,'PNG')
png=out.getvalue();total += len(png)
if total > 8_000_000: raise ValueError('resource bytes')
if not unlimited and total > 8_000_000: raise ValueError('resource bytes')
node.attributes['static_png']=png
except Exception:
node.attributes.pop('static_png', None)
@@ -66,26 +69,26 @@ def enrich_document(document, file_path=None):
def source_hash(source):
return hashlib.sha256(source.strip().encode()).hexdigest()
def validate_assets(assets):
def validate_assets(assets, unlimited=False):
result = {}
total = pixels = 0
for asset in assets:
try:
raw = base64.b64decode(asset.png_base64, validate=True)
total += len(raw)
if total > 8 * 1024 * 1024:
if not unlimited and total > 8 * 1024 * 1024:
raise ValueError('asset budget')
with Image.open(BytesIO(raw)) as image:
pixels += image.width * image.height
if pixels > 16_000_000: raise ValueError('document pixel budget')
if image.format != 'PNG' or image.width * image.height > 4_000_000:
if not unlimited and pixels > 16_000_000: raise ValueError('document pixel budget')
if image.format != 'PNG' or (not unlimited and image.width * image.height > 4_000_000):
raise ValueError('image budget')
image.load()
out = BytesIO()
rgba = image.convert('RGBA')
background = Image.new('RGBA', rgba.size, 'white')
background.alpha_composite(rgba)
background.convert('RGB').save(out, 'PNG')
(rgba if unlimited else background.convert('RGB')).save(out, 'PNG')
key = (asset.kind, asset.source_hash)
if key in result:
raise ValueError('duplicate asset')
+2 -3
View File
@@ -1,7 +1,6 @@
"""导出器共享工具:URL 协议校验、函数图像预算与占位 warning 文案。
html / pdf / docx 三个导出器共用同一套安全规则与函数图像资源预算,避免各写一份
导致行为漂移。
导出器共享 URL 规则;HTML / DOCX 使用文档资源预算,PDF 不使用这些预算。
"""
from __future__ import annotations
@@ -27,7 +26,7 @@ MAX_TOTAL_PLOT_NODES = 8000
class FunctionPlotBudget:
"""函数图像文档级资源预算:数量上限 + 累计 AST 节点上限。
HTML 与 PDF 导出器在渲染每个 function-plot 图块前先问预算,超限即回退源码占位,
HTML 与 DOCX 导出器在渲染每个 function-plot 图块前先问预算,超限即回退源码占位,
不解析不采样,避免多图块组合复杂度耗尽内存/CPU。
"""
+40 -39
View File
@@ -20,7 +20,7 @@ from reportlab.pdfbase.cidfonts import UnicodeCIDFont
from reportlab.platypus import (
Paragraph,
Indenter,
Preformatted,
XPreformatted,
SimpleDocTemplate,
Spacer,
Table,
@@ -29,12 +29,11 @@ from reportlab.platypus import (
from reportlab.platypus.flowables import HRFlowable
from app.contracts import ExportOptions
from app.export.themes import CALLOUTS, print_theme_warning
from app.export.themes import CALLOUTS, pdf_palette
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,
@@ -54,10 +53,11 @@ _HEADING_SIZES = {1: 20, 2: 16, 3: 14, 4: 12, 5: 11, 6: 10.5}
_QUOTE_COLOR = "#57606a"
def _make_styles() -> dict[str, ParagraphStyle]:
def _make_styles(palette) -> dict[str, ParagraphStyle]:
body = ParagraphStyle(
"pdf-body",
fontName=_FONT,
textColor=palette["text"],
fontSize=10.5,
leading=16,
spaceAfter=6,
@@ -67,7 +67,7 @@ def _make_styles() -> dict[str, ParagraphStyle]:
"pdf-quote",
parent=body,
leftIndent=14,
textColor="#57606a",
textColor=palette["muted"],
spaceBefore=4,
spaceAfter=6,
)
@@ -78,8 +78,8 @@ def _make_styles() -> dict[str, ParagraphStyle]:
leading=12,
leftIndent=6,
rightIndent=6,
backColor="#f6f8fa",
borderColor="#d0d7de",
backColor=palette["code"],
borderColor=palette["border"],
borderWidth=0.5,
borderPadding=6,
spaceBefore=4,
@@ -88,9 +88,9 @@ def _make_styles() -> dict[str, ParagraphStyle]:
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
"pdf-cell-head", parent=cell, textColor=palette["text"], fontSize=10
)
meta = ParagraphStyle("pdf-meta", parent=body, fontSize=8.5, leading=13, textColor="#57606a")
meta = ParagraphStyle("pdf-meta", parent=body, fontSize=8.5, leading=13, textColor=palette["muted"])
styles: dict[str, ParagraphStyle] = {
"body": body,
"title": title,
@@ -109,6 +109,7 @@ def _make_styles() -> dict[str, ParagraphStyle]:
leading=size * 1.4,
spaceBefore=14 if level <= 2 else 10,
spaceAfter=6,
keepWithNext=True,
)
return styles
@@ -118,17 +119,17 @@ class PdfExporter:
def render(self, document: Document, options: ExportOptions) -> ExportResult:
"""同步渲染;CPU 密集,调用方应放入线程执行,避免阻塞事件循环。"""
self._styles = _make_styles()
warnings: list[str] = []
print_theme_warning(options, warnings, "PDF")
self._palette = pdf_palette(options, warnings)
self._styles = _make_styles(self._palette)
if _FONT == "STSong-Light": warnings.append("PDF 使用 CID 字体,阅读器需提供中文字体;可配置 APP_EXPORT_FONT 嵌入 TrueType 字体")
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
self._plot_width = page[0] - 40 * mm - 12
self._plot_height = page[1] - 36 * mm - 12
buf = BytesIO()
doc = SimpleDocTemplate(
buf,
@@ -144,7 +145,14 @@ class PdfExporter:
self._render_header(document, options, story)
self._render_children(document.children, story, warnings)
doc.build(story)
def paint_page(canvas, template):
canvas.saveState()
canvas.setFillColor(self._palette['page'])
canvas.rect(0, 0, page[0], page[1], fill=1, stroke=0)
canvas.setFillColor(self._palette['surface'])
canvas.roundRect(12*mm, 10*mm, page[0]-24*mm, page[1]-20*mm, 5*mm, fill=1, stroke=0)
canvas.restoreState()
doc.build(story, onFirstPage=paint_page, onLaterPages=paint_page)
return ExportResult(content=buf.getvalue(), mime_type=_MIME, warnings=warnings)
async def export(self, document: Document, options: ExportOptions) -> ExportResult:
@@ -172,7 +180,7 @@ class PdfExporter:
if node.attributes.get('static_png'):
from reportlab.platypus import Image
image = Image(BytesIO(node.attributes['static_png']))
scale = min(1, self._plot_width / image.imageWidth, 600 / image.imageHeight)
scale = min(1, self._plot_width / image.imageWidth, self._plot_height / image.imageHeight)
image.drawWidth = image.imageWidth * scale
image.drawHeight = image.imageHeight * scale
story.append(image)
@@ -194,9 +202,13 @@ class PdfExporter:
def _block_callout(self, node, story, warnings):
kind = node.attributes['kind']
icon, color = CALLOUTS[kind]
from reportlab.lib.colors import HexColor
background = HexColor(self._palette['code'])
if .2126*background.red + .7152*background.green + .0722*background.blue < .5:
color = {'#0969da':'#a5d6ff','#7041a0':'#d2a8ff','#176f41':'#7ee787','#805400':'#f2cc60','#b42318':'#ffa198','#57606a':self._palette['muted']}[color]
title = self._render_inline(node.children[0].children,warnings)
style = ParagraphStyle('callout-'+kind,parent=self._styles['body'],textColor=color,
backColor='#f6f8fa',borderColor=color,borderWidth=1,borderPadding=6,spaceBefore=8,spaceAfter=8)
backColor=self._palette['code'],borderColor=color,borderWidth=1,borderPadding=6,spaceBefore=8,spaceAfter=8)
story.append(Paragraph(_html.escape(icon)+' '+title,style))
self._render_children(node.children[1:],story,warnings)
@@ -209,7 +221,7 @@ class PdfExporter:
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)
self._block_list(child, story, warnings, indent=14, color=self._palette['muted'])
else:
self._render_block(child, story, warnings)
@@ -301,7 +313,7 @@ class PdfExporter:
data.append(cells)
table = Table(data, repeatRows=head_row_count)
commands = [
("GRID", (0, 0), (-1, -1), 0.5, "#d0d7de"),
("GRID", (0, 0), (-1, -1), 0.5, self._palette["border"]),
("VALIGN", (0, 0), (-1, -1), "TOP"),
("LEFTPADDING", (0, 0), (-1, -1), 6),
("RIGHTPADDING", (0, 0), (-1, -1), 6),
@@ -309,53 +321,42 @@ class PdfExporter:
("BOTTOMPADDING", (0, 0), (-1, -1), 4),
]
if head_row_count:
commands.append(("BACKGROUND", (0, 0), (-1, head_row_count - 1), "#f6f8fa"))
commands.append(("BACKGROUND", (0, 0), (-1, head_row_count - 1), self._palette["code"]))
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"]))
story.append(XPreformatted(_html.escape(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(HRFlowable(width="100%", color=self._palette["border"], 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"]))
story.append(XPreformatted(_html.escape(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)
from app.plot.parser import parse_source
parsed = parse_source(request.source, unlimited=True)
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"]))
story.append(XPreformatted(_html.escape(node.text), self._styles["code"]))
return
# Drawing 本身即 Flowable,缩放后追加到 story,与 HTML 视觉一致
drawing = render_drawing(parsed.plot, width=self._plot_width)
drawing = render_drawing(parsed.plot, width=self._plot_width, palette=self._palette, unlimited=True, max_height=self._plot_height)
story.append(drawing)
except Exception as exc:
warnings.append(f"函数图像:解析或渲染失败,已回退占位({exc}")
story.append(Preformatted(node.text, self._styles["code"]))
story.append(XPreformatted(_html.escape(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"]))
@@ -393,7 +394,7 @@ class PdfExporter:
if safe_href is None:
warnings.append(f"链接协议不安全,已降级为纯文本:{href!r}")
return inner
return f'<a href="{_html.escape(safe_href)}">{inner}</a>'
return f'<a href="{_html.escape(safe_href)}" color="{self._palette["accent"]}">{inner}</a>'
if t == "image":
src = str(node.attributes.get("src") or "")
alt = str(node.attributes.get("alt") or "")
+7 -7
View File
@@ -146,7 +146,7 @@ def _evict_terminal() -> bool:
return True
async def _resolve_source(source: ExportSource) -> tuple[str, str, dict | None]:
async def _resolve_source(source: ExportSource, unlimited: bool = False) -> tuple[str, str, dict | None]:
"""把导出源解析为 (markdown, title, metadata)metadata 仅 note 源提供。"""
if source.type == ExportSourceType.note:
note = await note_service.get_note(source.note_id)
@@ -157,7 +157,7 @@ async def _resolve_source(source: ExportSource) -> tuple[str, str, dict | None]:
"note not found",
{"note_id": source.note_id},
)
if len(note.markdown) > MAX_MARKDOWN_CHARS:
if not unlimited and len(note.markdown) > MAX_MARKDOWN_CHARS:
raise ApiError(
400,
"EXPORT_OPTIONS_INVALID",
@@ -175,7 +175,7 @@ async def _resolve_source(source: ExportSource) -> tuple[str, str, dict | None]:
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:
if not unlimited and len(markdown) > MAX_MARKDOWN_CHARS:
raise ApiError(
400,
"EXPORT_OPTIONS_INVALID",
@@ -187,10 +187,10 @@ async def _resolve_source(source: ExportSource) -> tuple[str, str, dict | None]:
async def create_export(request: ExportRequest) -> ExportJob:
"""创建导出任务,立即返回 queued 的 ExportJob,由后台 Task 渲染。"""
markdown, title, metadata = await _resolve_source(request.source)
markdown, title, metadata = await _resolve_source(request.source, request.format == ExportFormat.pdf)
title = request.title or title
from app.export.assets import validate_assets
assets = await asyncio.to_thread(validate_assets, request.assets)
assets = await asyncio.to_thread(validate_assets, request.assets, request.format == ExportFormat.pdf)
if not _evict_terminal():
raise ApiError(
@@ -283,12 +283,12 @@ async def _execute(
document.attributes["metadata"] = metadata
from app.export.assets import enrich_document
resource_warnings = await asyncio.to_thread(enrich_document, document, (metadata or {}).get('file_path'))
resource_warnings = await asyncio.to_thread(enrich_document, document, (metadata or {}).get('file_path'), format == ExportFormat.pdf, options)
result = await asyncio.to_thread(_render_document, document, options, format)
result.warnings[:0] = resource_warnings
if cancel_event.is_set():
raise ExportCancelled()
if len(result.content) > MAX_EXPORT_BYTES:
if format != ExportFormat.pdf and len(result.content) > MAX_EXPORT_BYTES:
raise ExportTooLarge()
ext = _extension_for(format)
+10
View File
@@ -32,3 +32,13 @@ ALIASES = {'summary':'abstract','tldr':'abstract','hint':'tip',
'check':'success','done':'success','help':'question','faq':'question',
'caution':'warning','attention':'warning','fail':'failure','missing':'failure',
'error':'danger','cite':'quote'}
def pdf_palette(options, warnings):
if options.palette is not None:
return options.palette.model_dump()
theme_id = options.theme_id
if theme_id not in PALETTES:
warnings.append(f'PDF 不支持主题 {theme_id},已使用 light 导出配色')
theme_id = 'light'
return dict(zip(('page','surface','text','muted','code','border','accent'), PALETTES[theme_id]))
+12 -12
View File
@@ -143,7 +143,7 @@ def _preprocess(expr: str) -> str:
return _insert_implicit_multiplication(expr.replace("^", "**"))
def _check_node(node: ast.AST, depth: int = 0, counter: list[int] | None = None) -> None:
def _check_node(node: ast.AST, depth: int = 0, counter: list[int] | None = None, unlimited: bool = False) -> None:
"""白名单校验:任何越界节点都抛 FUNCTION_PLOT_EXPRESSION_UNSAFE。
同时限制 AST 深度与节点总数避免超长/超深表达式在递归校验或求值时触发
@@ -151,10 +151,10 @@ def _check_node(node: ast.AST, depth: int = 0, counter: list[int] | None = None)
"""
if counter is None:
counter = [0]
if depth > _MAX_AST_DEPTH:
if not unlimited and depth > _MAX_AST_DEPTH:
_unsafe(f"表达式嵌套过深(超过 {_MAX_AST_DEPTH} 层)")
counter[0] += 1
if counter[0] > _MAX_AST_NODES:
if not unlimited and 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)):
@@ -167,13 +167,13 @@ def _check_node(node: ast.AST, depth: int = 0, counter: list[int] | None = None)
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)
_check_node(node.left, depth + 1, counter, unlimited)
_check_node(node.right, depth + 1, counter, unlimited)
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)
_check_node(node.operand, depth + 1, counter, unlimited)
return
if isinstance(node, ast.Call):
if not isinstance(node.func, ast.Name) or node.func.id not in _FUNCTIONS:
@@ -184,12 +184,12 @@ def _check_node(node: ast.AST, depth: int = 0, counter: list[int] | None = None)
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)
_check_node(arg, depth + 1, counter, unlimited)
return
_unsafe(f"不支持的语法 {type(node).__name__}")
def parse_expression(expr: str) -> ast.Expression:
def parse_expression(expr: str, unlimited: bool = False) -> ast.Expression:
"""把数学表达式解析为已通过白名单校验的 AST(可直接交给 evaluate)。"""
preprocessed = _preprocess(expr)
try:
@@ -211,7 +211,7 @@ def parse_expression(expr: str) -> ast.Expression:
message="表达式嵌套过深,无法解析",
)
) from exc
_check_node(tree.body)
_check_node(tree.body, unlimited=unlimited)
return tree
@@ -279,7 +279,7 @@ def _parse_directive(line: str) -> tuple[str, str] | None:
return key, value.strip()
def parse_source(source: str) -> FunctionPlotParseResult:
def parse_source(source: str, unlimited: bool = False) -> FunctionPlotParseResult:
"""把 function-plot fenced block 源码解析为 FunctionPlot + 诊断。"""
diagnostics: list[PlotDiagnostic] = []
expressions: list[FunctionPlotExpression] = []
@@ -371,7 +371,7 @@ def parse_source(source: str) -> FunctionPlotParseResult:
continue
try:
tree = parse_expression(expr_text)
tree = parse_expression(expr_text, unlimited=unlimited)
except PlotParseError as exc:
exc.diagnostic.line = lineno
diagnostics.append(exc.diagnostic)
@@ -380,7 +380,7 @@ def parse_source(source: str) -> FunctionPlotParseResult:
total_nodes += _count_nodes(tree.body)
expressions.append(FunctionPlotExpression(expression=expr_text))
# 表达式数量超限:整块回退并提前终止,避免对海量表达式做采样求值
if len(expressions) > _MAX_EXPRESSIONS:
if not unlimited and len(expressions) > _MAX_EXPRESSIONS:
diagnostics.append(
PlotDiagnostic(
severity="error",
+2 -2
View File
@@ -339,7 +339,7 @@ def _sample_segments(
return clipped
def compute_geometry(plot: FunctionPlot) -> PlotGeometry:
def compute_geometry(plot: FunctionPlot, unlimited: bool = False) -> PlotGeometry:
"""解析并计算几何,供 SVG 与 reportlab 后端复用。"""
warnings: list[str] = []
xmin, xmax = plot.domain
@@ -351,7 +351,7 @@ def compute_geometry(plot: FunctionPlot) -> PlotGeometry:
fns: list[tuple[object, object]] = []
for expr in plot.expressions:
try:
tree = parse_expression(expr.expression)
tree = parse_expression(expr.expression, unlimited=unlimited)
except PlotParseError as exc:
warnings.append(f"表达式无法渲染,已跳过:{expr.expression}{exc.diagnostic.message}")
continue
+22 -13
View File
@@ -26,9 +26,12 @@ _TICK_FONT_SIZE = 10
_LABEL_FONT_SIZE = 12
def _build_drawing(geo: PlotGeometry) -> Drawing:
def _build_drawing(geo: PlotGeometry, palette=None) -> Drawing:
"""由共享几何构建矢量 Drawing(坐标翻转后仍沿用 SVG 的像素布局)。"""
drawing = Drawing(geo.width, geo.height)
grid_color = HexColor(palette['border']) if palette else _GRID_COLOR
axis_color = HexColor(palette['muted']) if palette else _AXIS_COLOR
label_color = HexColor(palette['text']) if palette else _LABEL_COLOR
# SVG y-down → reportlab y-up:翻转像素 y
def sx(x: float) -> float:
@@ -41,19 +44,19 @@ def _build_drawing(geo: PlotGeometry) -> Drawing:
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)
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)
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)
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)
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 轴左侧)
@@ -61,14 +64,14 @@ def _build_drawing(geo: PlotGeometry) -> Drawing:
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",
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",
fontName=_FONT, fontSize=_TICK_FONT_SIZE, fillColor=axis_color, textAnchor="end",
)
)
@@ -83,7 +86,7 @@ def _build_drawing(geo: PlotGeometry) -> Drawing:
drawing.add(
String(
geo.width / 2, 10, geo.xlabel,
fontName=_FONT, fontSize=_LABEL_FONT_SIZE, fillColor=_LABEL_COLOR, textAnchor="middle",
fontName=_FONT, fontSize=_LABEL_FONT_SIZE, fillColor=label_color, textAnchor="middle",
)
)
if geo.ylabel:
@@ -95,7 +98,7 @@ def _build_drawing(geo: PlotGeometry) -> Drawing:
label.add(
String(
0, 0, geo.ylabel,
fontName=_FONT, fontSize=_LABEL_FONT_SIZE, fillColor=_LABEL_COLOR, textAnchor="middle",
fontName=_FONT, fontSize=_LABEL_FONT_SIZE, fillColor=label_color, textAnchor="middle",
)
)
label.translate(16, geo.height / 2)
@@ -105,19 +108,25 @@ def _build_drawing(geo: PlotGeometry) -> Drawing:
return drawing
def render_drawing(plot: FunctionPlot, width: float | None = None) -> Drawing:
def render_drawing(plot: FunctionPlot, width: float | None = None, palette=None, unlimited=False, max_height=None) -> Drawing:
"""把已解析的 FunctionPlot 渲染为 reportlab Drawing(可直接追加到 platypus story)。
``width`` 为目标输出宽度用于把 640px 的几何缩放到页面内容宽省略则按
原始尺寸输出缩放只影响 PDF 渲染不改动共享几何
"""
geo = compute_geometry(plot)
drawing = _build_drawing(geo)
geo = compute_geometry(plot, unlimited=unlimited)
if palette:
from reportlab.lib.colors import HexColor as color
bg = color(palette['surface'])
if .2126*bg.red + .7152*bg.green + .0722*bg.blue < .5:
colors = ['#79c0ff','#ff9b9b','#7ee787','#d2a8ff','#f2cc60','#ffa657']
geo.colors = [value if plot.expressions[i].color else colors[i % len(colors)] for i,value in enumerate(geo.colors)]
drawing = _build_drawing(geo, palette)
legend_height = ((len(plot.expressions)+1)//2)*24
drawing.height += legend_height
for index, expression in enumerate(plot.expressions):
drawing.add(String(24+(index%2)*310,geo.height+legend_height-18-(index//2)*24,
expression.label or 'y = '+expression.expression,fontName=_FONT,fontSize=12,fillColor=HexColor(geo.colors[index])))
if width is not None and width > 0:
drawing.renderScale = min(1.0, width / geo.width)
drawing.renderScale = min(1.0, width / geo.width, max_height / drawing.height if max_height else 1.0)
return drawing
+6 -6
View File
@@ -334,17 +334,16 @@ def test_pdf_exporter_function_plot_fallback_on_error() -> None:
assert any("函数图像" in w for w in result.warnings)
def test_pdf_exporter_limits_function_plot_count() -> None:
def test_pdf_exporter_has_no_function_plot_count_quota() -> None:
from app.export.exporters.pdf import PdfExporter
blocks = "\n\n".join("```function-plot\ny = x\n```" for _ in range(20))
result = asyncio.run(PdfExporter().export(parse_document(blocks), ExportOptions()))
assert result.content[:4] == b"%PDF"
# 超出数量上限的图块回退占位并记 warning
assert any("数量超过上限" in w for w in result.warnings)
assert not any("函数图像" in w for w in result.warnings)
def test_pdf_exporter_limits_total_plot_nodes(monkeypatch) -> None:
def test_pdf_exporter_has_no_total_plot_node_quota(monkeypatch) -> None:
import app.export.exporters._common as common_mod
from app.export.exporters.pdf import PdfExporter
@@ -352,7 +351,7 @@ def test_pdf_exporter_limits_total_plot_nodes(monkeypatch) -> None:
md = "```function-plot\ny = x\n```\n\n```function-plot\ny = x + x + x + x\n```"
result = asyncio.run(PdfExporter().export(parse_document(md), ExportOptions()))
assert result.content[:4] == b"%PDF"
assert any("累计复杂度" in w for w in result.warnings)
assert not any("函数图像" in w for w in result.warnings)
def test_docx_exporter_embeds_plot_and_warns_missing_mermaid() -> None:
@@ -831,7 +830,8 @@ def test_callout_formats(name):
xml = z.read("word/document.xml").decode()
assert all(word in xml for word in ["Title", "Body", "item", "second", "w:shd"])
result = PdfExporter().render(doc, ExportOptions(theme_id="sepia"))
assert result.content.startswith(b"%PDF") and len(result.warnings) == 1
assert result.content.startswith(b"%PDF")
assert not any("浅色打印" in warning for warning in result.warnings)
@pytest.mark.parametrize("fold", ["", "+", "-"])
+96
View File
@@ -0,0 +1,96 @@
"""PDF theme and resource policy regressions; no real providers or user files."""
import asyncio
import base64
from io import BytesIO
import pytest
from PIL import Image
from pydantic import ValidationError
from app.contracts import ExportAsset, ExportOptions, ExportRequest
from app.export.assets import validate_assets, enrich_document, source_hash
from app.export.exporters.pdf import PdfExporter
from app.export.markdown import parse_document
from app.export.themes import PALETTES
from app.export import service
def png_asset(size=(40,30), source='graph LR; A-->B'):
out=BytesIO(); Image.new('RGBA',size,(0,0,0,0)).save(out,'PNG')
return ExportAsset(kind='mermaid',source_hash=source_hash(source),png_base64=base64.b64encode(out.getvalue()).decode())
@pytest.mark.parametrize('theme',list(PALETTES))
def test_pdf_theme_colors_are_written_on_every_page(theme):
import re, zlib
palette=PALETTES[theme]
doc=parse_document(('## Section\n\nText body\n\n> Quoted text\n\n```python\nprint(1)\n```\n\n')*30)
result=PdfExporter().render(doc,ExportOptions(theme_id=theme))
streams=[]
for match in re.finditer(rb'stream\r?\n(.*?)endstream',result.content,re.S):
try: streams.append(zlib.decompress(base64.a85decode(match[1].strip().removesuffix(b'~>'))))
except Exception: pass
from reportlab.lib.rl_accel import fp_str
command=(fp_str(*[int(palette[0][i:i+2],16)/255 for i in (1,3,5)])+' rg').encode()
pages=[s for s in streams if b'BT' in s and b'/F' in s]
assert len(pages)>1
assert all(command in s for s in pages)
assert not any('浅色打印' in w for w in result.warnings)
def test_pdf_accepts_asset_contract_beyond_previous_count_and_size():
assets=[png_asset(source=str(i)) for i in range(65)]
assets[0]=assets[0].model_copy(update={'png_base64':'A'*2800004})
values=dict(source={'type':'markdown','markdown':'content'},assets=assets)
ExportRequest(format='pdf',**values)
with pytest.raises(ValidationError): ExportRequest(format='html',**values)
with pytest.raises(ValidationError): ExportRequest(format='docx',**values)
def test_pdf_large_png_still_requires_valid_format():
asset=png_asset((2100,2000))
assert validate_assets([asset],unlimited=True)
with pytest.raises(Exception): validate_assets([asset])
with pytest.raises(Exception): validate_assets([asset.model_copy(update={'png_base64':'invalid'})],unlimited=True)
def test_pdf_embeds_more_than_64_resources_with_theme_background():
doc=parse_document(('```mermaid\ngraph LR; A-->B\n```\n\n')*65)
from app.export.assets import attach_assets
attach_assets(doc,validate_assets([png_asset()],unlimited=True))
assert not enrich_document(doc,unlimited=True,options=ExportOptions(theme_id='dark'))
assert all('static_png' in node.attributes for node in doc.children)
with Image.open(BytesIO(doc.children[-1].attributes['static_png'])) as image:
assert image.getpixel((0,0)) == (13,17,23)
assert PdfExporter().render(doc,ExportOptions(theme_id='dark')).content.startswith(b'%PDF')
def test_pdf_pipeline_ignores_source_and_output_quotas(monkeypatch):
monkeypatch.setattr(service,'MAX_MARKDOWN_CHARS',8)
monkeypatch.setattr(service,'MAX_EXPORT_BYTES',8)
async def run():
job=await service.create_export(ExportRequest(format='pdf',source={'type':'markdown','markdown':'Beyond the previous quota.'}))
done=await service.wait_for_export(job.job_id)
assert done.status.value=='completed'
assert service.get_export_file(job.job_id).stat().st_size>8
with pytest.raises(Exception):
await service.create_export(ExportRequest(format='html',source={'type':'markdown','markdown':'Beyond the previous quota.'}))
asyncio.run(run())
def test_pdf_accepts_more_than_16_curves_and_keeps_expression_safety():
doc=parse_document('```function-plot\n'+'\n'.join(f'y=x+{i}' for i in range(17))+'\n```')
result=PdfExporter().render(doc,ExportOptions(theme_id='dark'))
assert not any('函数图像' in w for w in result.warnings)
unsafe=PdfExporter().render(parse_document('```function-plot\ny=__import__("os")\n```'),ExportOptions())
assert any('函数图像' in w for w in unsafe.warnings)
def test_pdf_custom_palette_and_math_color():
palette=dict(zip(('page','surface','text','muted','code','border','accent'),PALETTES['midnight-purple']))
options=ExportOptions(theme_id='my-theme',palette=palette)
doc=parse_document('Formula $x^2$')
assert not enrich_document(doc,unlimited=True,options=options)
math=next(n for n in doc.children[0].children if n.type=='math_inline')
with Image.open(BytesIO(math.attributes['static_png'])) as image:
assert image.getpixel((0,0))==(25,19,34)
assert not any('主题' in w for w in PdfExporter().render(doc,options).warnings)
with pytest.raises(ValidationError): ExportOptions(palette={**palette,'text':'url(file:///private)'})