feat(channel/render): 新增markdown渲染模块

实现了完整的markdown解析与渲染能力,包含IR中间节点定义、markdown解析器以及两种渲染风格(markdown格式与纯文本格式),支持常见markdown语法渲染
This commit is contained in:
Kris 2026-05-21 10:28:14 +08:00
parent dd1857221c
commit 8ee9c1425e
4 changed files with 849 additions and 0 deletions

View File

@ -0,0 +1,19 @@
from __future__ import annotations
from yuxi.channel.render.ir import Document
from yuxi.channel.render.parser import parse_markdown
from yuxi.channel.render.style_map import (
RenderStyleMap,
render,
render_md_table,
render_plain_table,
)
__all__ = [
"Document",
"RenderStyleMap",
"parse_markdown",
"render",
"render_md_table",
"render_plain_table",
]

View File

@ -0,0 +1,121 @@
from __future__ import annotations
from dataclasses import dataclass, field
class IrNode:
pass
@dataclass
class Text(IrNode):
content: str
@dataclass
class Bold(IrNode):
children: list[IrNode] = field(default_factory=list)
@dataclass
class Italic(IrNode):
children: list[IrNode] = field(default_factory=list)
@dataclass
class BoldItalic(IrNode):
children: list[IrNode] = field(default_factory=list)
@dataclass
class Strikethrough(IrNode):
children: list[IrNode] = field(default_factory=list)
@dataclass
class InlineCode(IrNode):
code: str
@dataclass
class CodeBlock(IrNode):
code: str
language: str | None = None
@dataclass
class Link(IrNode):
url: str
children: list[IrNode] = field(default_factory=list)
@dataclass
class Image(IrNode):
url: str
alt: str = ""
@dataclass
class Heading(IrNode):
level: int
children: list[IrNode] = field(default_factory=list)
@dataclass
class Paragraph(IrNode):
children: list[IrNode] = field(default_factory=list)
@dataclass
class BlockQuote(IrNode):
children: list[IrNode] = field(default_factory=list)
@dataclass
class ListItem(IrNode):
children: list[IrNode] = field(default_factory=list)
@dataclass
class UnorderedList(IrNode):
items: list[ListItem] = field(default_factory=list)
marker: str = "-"
@dataclass
class OrderedList(IrNode):
items: list[ListItem] = field(default_factory=list)
start: int = 1
@dataclass
class ThematicBreak(IrNode):
pass
@dataclass
class LineBreak(IrNode):
pass
@dataclass
class TableCell(IrNode):
children: list[IrNode] = field(default_factory=list)
align: str | None = None
@dataclass
class TableRow(IrNode):
cells: list[TableCell] = field(default_factory=list)
is_header: bool = False
@dataclass
class Table(IrNode):
header: TableRow | None = None
rows: list[TableRow] = field(default_factory=list)
@dataclass
class Document(IrNode):
children: list[IrNode] = field(default_factory=list)

View File

@ -0,0 +1,432 @@
from __future__ import annotations
import logging
import re
from yuxi.channel.render.ir import (
BlockQuote,
Bold,
BoldItalic,
CodeBlock,
Document,
Heading,
Image,
InlineCode,
IrNode,
Italic,
LineBreak,
Link,
ListItem,
OrderedList,
Paragraph,
Strikethrough,
Table,
TableCell,
TableRow,
Text,
ThematicBreak,
UnorderedList,
)
logger = logging.getLogger(__name__)
MAX_INPUT_LENGTH = 500_000
_CODE_FENCE_RE = re.compile(r"^( {0,3})(`{3,}|~{3,})\s*(\S*)\s*$")
_HEADING_RE = re.compile(r"^(#{1,6})\s+(.+)$")
_UNORDERED_LIST_RE = re.compile(r"^(\s*)([-*+])\s+(.+)$")
_ORDERED_LIST_RE = re.compile(r"^(\s*)(\d+)\.\s+(.+)$")
_BLOCKQUOTE_RE = re.compile(r"^( {0,3})>\s?(.*)$")
_HR_RE = re.compile(r"^( {0,3})([-*_])\s*\2\s*\2[ \2]*$")
_TABLE_ROW_RE = re.compile(r"^\|(.+)\|$")
_TABLE_SEP_RE = re.compile(r"^\|[\s\|:-]+\|$")
def _strip_trailing_whitespace(lines: list[str]) -> list[str]:
while lines and lines[-1] == "":
lines.pop()
while lines and lines[0] == "":
lines.pop(0)
return lines
_INLINE_RE = re.compile(r"(~~|\*\*\*|\*\*|__|\*|_|`|\!\[|\[|\n)")
def parse_inline(text: str) -> list[IrNode]:
return _parse_inline_range(text, 0, len(text))
def _parse_inline_range(text: str, start: int, end: int) -> list[IrNode]:
nodes: list[IrNode] = []
i = start
n = end
while i < n:
m = _INLINE_RE.search(text, i, end)
if m is None:
if i < n:
nodes.append(Text(content=text[i:end]))
break
match_start = m.start()
if match_start > i:
nodes.append(Text(content=text[i:match_start]))
i = match_start
marker = m.group(1)
if marker == "~~":
if i + 3 < n:
closing = text.find("~~", i + 2, end)
if closing != -1:
nodes.append(Strikethrough(children=_parse_inline_range(text, i + 2, closing)))
i = closing + 2
continue
nodes.append(Text(content="~~"))
i += 2
continue
if marker == "***":
closing = text.find("***", i + 3, end)
if closing != -1:
nodes.append(BoldItalic(children=_parse_inline_range(text, i + 3, closing)))
i = closing + 3
continue
nodes.append(Text(content="***"))
i += 3
continue
if marker == "**":
if i + 3 < n:
closing = text.find("**", i + 2, end)
if closing != -1:
nodes.append(Bold(children=_parse_inline_range(text, i + 2, closing)))
i = closing + 2
continue
nodes.append(Text(content="**"))
i += 2
continue
if marker == "__":
if i + 3 < n:
closing = text.find("__", i + 2, end)
if closing != -1:
nodes.append(Bold(children=_parse_inline_range(text, i + 2, closing)))
i = closing + 2
continue
nodes.append(Text(content="__"))
i += 2
continue
if marker == "*":
if i + 2 < n:
closing = text.find("*", i + 1, end)
if closing != -1:
nodes.append(Italic(children=_parse_inline_range(text, i + 1, closing)))
i = closing + 1
continue
nodes.append(Text(content="*"))
i += 1
continue
if marker == "_":
if i + 2 < n and not (i > 0 and text[i - 1].isalnum()):
closing = text.find("_", i + 1, end)
if closing != -1 and (closing + 1 >= n or not text[closing + 1].isalnum()):
nodes.append(Italic(children=_parse_inline_range(text, i + 1, closing)))
i = closing + 1
continue
nodes.append(Text(content="_"))
i += 1
continue
if marker == "`":
closing = text.find("`", i + 1, end)
if closing != -1:
nodes.append(InlineCode(code=text[i + 1 : closing]))
i = closing + 1
continue
nodes.append(Text(content="`"))
i += 1
continue
if marker == "![":
close_bracket = text.find("]", i + 2, end)
if close_bracket != -1 and close_bracket + 1 < n and text[close_bracket + 1] == "(":
close_paren = text.find(")", close_bracket + 2, end)
if close_paren != -1:
alt = text[i + 2 : close_bracket]
url = text[close_bracket + 2 : close_paren]
nodes.append(Image(url=url, alt=alt))
i = close_paren + 1
continue
nodes.append(Text(content="!["))
i += 2
continue
if marker == "[":
close_bracket = text.find("]", i + 1, end)
if close_bracket != -1 and close_bracket + 1 < n and text[close_bracket + 1] == "(":
close_paren = text.find(")", close_bracket + 2, end)
if close_paren != -1:
link_url = text[close_bracket + 2 : close_paren]
nodes.append(Link(url=link_url, children=_parse_inline_range(text, i + 1, close_bracket)))
i = close_paren + 1
continue
nodes.append(Text(content="["))
i += 1
continue
if marker == "\n":
nodes.append(LineBreak())
i += 1
continue
return _merge_adjacent_text(nodes)
def _merge_adjacent_text(nodes: list[IrNode]) -> list[IrNode]:
merged: list[IrNode] = []
buf: list[str] = []
for node in nodes:
if isinstance(node, Text):
buf.append(node.content)
else:
if buf:
merged.append(Text(content="".join(buf)))
buf.clear()
merged.append(node)
if buf:
merged.append(Text(content="".join(buf)))
return merged
def _count_leading_spaces(line: str) -> int:
return len(line) - len(line.lstrip(" "))
def _split_table_cells(line: str) -> list[str]:
content = line.strip()
if content.startswith("|"):
content = content[1:]
if content.endswith("|"):
content = content[:-1]
return [c.strip() for c in content.split("|")]
def _parse_table_aligns(sep_line: str) -> list[str | None]:
cells = _split_table_cells(sep_line)
aligns: list[str | None] = []
for cell in cells:
c = cell.strip()
if c.startswith(":") and c.endswith(":"):
aligns.append("center")
elif c.endswith(":"):
aligns.append("right")
elif c.startswith(":"):
aligns.append("left")
else:
aligns.append(None)
return aligns
def _build_table_cells(cell_texts: list[str], aligns: list[str | None]) -> list[TableCell]:
max_len = max(len(cell_texts), len(aligns))
while len(aligns) < max_len:
aligns.append(None)
return [TableCell(children=parse_inline(cell_texts[i]), align=aligns[i]) for i in range(len(cell_texts))]
def _parse_list_items(
lines: list[str], start_idx: int, item_re: re.Pattern, group_idx: int
) -> tuple[list[ListItem], int]:
items: list[ListItem] = []
i = start_idx
n = len(lines)
base_indent: int | None = None
while i < n:
cl = lines[i]
cm = item_re.match(cl)
if cm:
indent = len(cm.group(1))
if base_indent is None:
base_indent = indent
elif indent < base_indent:
break
content = cm.group(group_idx)
children: list[IrNode] = parse_inline(content)
i += 1
sub_items: list[IrNode] = []
while i < n:
nl = lines[i]
if nl.strip() == "":
i += 1
continue
next_indent = _count_leading_spaces(nl)
if next_indent > indent and next_indent >= (indent + 2):
sub_item_re = re.match(r"^(\s*)([-*+])\s+(.+)$", nl)
sub_ordered_re = re.match(r"^(\s*)(\d+)\.\s+(.+)$", nl)
if sub_item_re:
sub_list_items, sub_i = _parse_list_items(lines, i, re.compile(r"^(\s*)([-*+])\s+(.+)$"), 3)
sub_ul = UnorderedList(items=sub_list_items, marker=sub_item_re.group(2))
sub_items.append(sub_ul)
i = sub_i
elif sub_ordered_re:
sub_list_items, sub_i = _parse_list_items(lines, i, re.compile(r"^(\s*)(\d+)\.\s+(.+)$"), 3)
sub_ol = OrderedList(items=sub_list_items, start=int(sub_ordered_re.group(2)))
sub_items.append(sub_ol)
i = sub_i
elif next_indent > base_indent:
para_lines: list[str] = []
while i < n and (lines[i].strip() == "" or _count_leading_spaces(lines[i]) >= indent + 2):
if lines[i].strip() == "":
if para_lines:
break
i += 1
continue
if _HEADING_RE.match(lines[i].strip()) or _CODE_FENCE_RE.match(lines[i]):
break
para_lines.append(lines[i].strip())
i += 1
if para_lines:
sub_items.append(Paragraph(children=parse_inline(" ".join(para_lines))))
else:
break
else:
break
items.append(ListItem(children=children + sub_items))
elif cl.strip() == "":
i += 1
if i < n and not item_re.match(lines[i]):
peek_indent = _count_leading_spaces(lines[i]) if lines[i].strip() else 999
if peek_indent <= (base_indent or 0):
break
else:
break
return items, i
def _parse_block_lines(lines: list[str]) -> list[IrNode]:
nodes: list[IrNode] = []
i = 0
n = len(lines)
while i < n:
line = lines[i]
fence_m = _CODE_FENCE_RE.match(line)
if fence_m:
fence_char = fence_m.group(2)[0]
lang = fence_m.group(3) or None
code_lines: list[str] = []
i += 1
while i < n:
fl = lines[i]
fm_close = _CODE_FENCE_RE.match(fl)
if fm_close and fm_close.group(2)[0] == fence_char:
i += 1
break
code_lines.append(fl)
i += 1
nodes.append(CodeBlock(code="\n".join(code_lines), language=lang))
continue
if line.strip() == "":
i += 1
continue
hr_m = _HR_RE.match(line)
if hr_m:
nodes.append(ThematicBreak())
i += 1
continue
heading_m = _HEADING_RE.match(line)
if heading_m:
level = len(heading_m.group(1))
content = heading_m.group(2)
nodes.append(Heading(level=level, children=parse_inline(content)))
i += 1
continue
bq_m = _BLOCKQUOTE_RE.match(line)
if bq_m:
bq_lines: list[str] = []
while i < n:
bl = lines[i]
bq_match = _BLOCKQUOTE_RE.match(bl)
if bq_match:
bq_lines.append(bq_match.group(2))
i += 1
elif bl.strip() == "":
bq_lines.append("")
i += 1
else:
break
inner = _parse_block_lines(bq_lines)
nodes.append(BlockQuote(children=inner))
continue
ul_m = _UNORDERED_LIST_RE.match(line)
if ul_m:
marker = ul_m.group(2)
items, i = _parse_list_items(lines, i, _UNORDERED_LIST_RE, 3)
nodes.append(UnorderedList(items=items, marker=marker))
continue
ol_m = _ORDERED_LIST_RE.match(line)
if ol_m:
start_num = int(ol_m.group(2))
items, i = _parse_list_items(lines, i, _ORDERED_LIST_RE, 3)
nodes.append(OrderedList(items=items, start=start_num))
continue
table_line = line.strip()
if _TABLE_ROW_RE.match(table_line):
if i + 1 < n and _TABLE_SEP_RE.match(lines[i + 1].strip()):
aligns = _parse_table_aligns(lines[i + 1])
header_cells = _build_table_cells(_split_table_cells(table_line), aligns)
header = TableRow(cells=header_cells, is_header=True)
i += 2
body_rows: list[TableRow] = []
while i < n and _TABLE_ROW_RE.match(lines[i].strip()):
row_cells = _build_table_cells(_split_table_cells(lines[i].strip()), aligns)
body_rows.append(TableRow(cells=row_cells))
i += 1
nodes.append(Table(header=header, rows=body_rows))
continue
para_lines: list[str] = []
while i < n and lines[i].strip() != "":
pl = lines[i]
if _HEADING_RE.match(pl) or _CODE_FENCE_RE.match(pl) or _HR_RE.match(pl):
break
if _BLOCKQUOTE_RE.match(pl):
break
if _UNORDERED_LIST_RE.match(pl) or _ORDERED_LIST_RE.match(pl):
break
if _TABLE_SEP_RE.match(pl.strip()):
break
para_lines.append(pl)
i += 1
para_text = " ".join(para_lines)
nodes.append(Paragraph(children=parse_inline(para_text)))
return nodes
def parse_markdown(text: str) -> Document:
if not text:
return Document(children=[])
if len(text) > MAX_INPUT_LENGTH:
logger.warning("Input text exceeds max length %d, truncating", MAX_INPUT_LENGTH)
text = text[:MAX_INPUT_LENGTH]
lines = text.split("\n")
lines = _strip_trailing_whitespace(lines)
children = _parse_block_lines(lines)
return Document(children=children)

View File

@ -0,0 +1,277 @@
from __future__ import annotations
import logging
from dataclasses import dataclass
from typing import Callable
from yuxi.channel.render.ir import (
BlockQuote,
Bold,
BoldItalic,
CodeBlock,
Document,
Heading,
Image,
InlineCode,
IrNode,
Italic,
LineBreak,
Link,
ListItem,
OrderedList,
Paragraph,
Strikethrough,
Table,
TableCell,
TableRow,
Text,
ThematicBreak,
UnorderedList,
)
logger = logging.getLogger(__name__)
_PIPE_ESCAPE_RE = str.maketrans({"|": "\\|"})
def render_md_table(
header_texts: list[str],
body_texts: list[list[str]],
aligns: list[str | None],
) -> str:
if not header_texts and not body_texts:
return ""
col_count = len(header_texts) or (len(body_texts[0]) if body_texts else 0)
if col_count == 0:
return ""
while len(header_texts) < col_count:
header_texts.append("")
while len(aligns) < col_count:
aligns.append(None)
align_map = {
"left": ":--",
"center": ":-:",
"right": "--:",
}
def _fmt_row(cells: list[str]) -> str:
parts = [cell.translate(_PIPE_ESCAPE_RE) for cell in cells[:col_count]]
while len(parts) < col_count:
parts.append("")
return "| " + " | ".join(parts) + " |"
def _fmt_sep() -> str:
parts = [align_map.get(a or "", "---") for a in aligns[:col_count]]
while len(parts) < col_count:
parts.append("---")
return "|" + "|".join(parts) + "|"
lines: list[str] = []
if header_texts:
lines.append(_fmt_row(header_texts))
lines.append(_fmt_sep())
for row in body_texts:
lines.append(_fmt_row(row))
return "\n".join(lines)
def _pad_cell(text: str, width: int, align: str | None) -> str:
if align == "right":
return text.rjust(width)
elif align == "center":
return text.center(width)
else:
return text.ljust(width)
def render_plain_table(
header_texts: list[str],
body_texts: list[list[str]],
aligns: list[str | None],
) -> str:
if not header_texts and not body_texts:
return ""
all_rows = ([header_texts] if header_texts else []) + body_texts
if not all_rows:
return ""
col_count = max(len(row) for row in all_rows)
col_widths = [0] * col_count
for row in all_rows:
for j, cell in enumerate(row[:col_count]):
col_widths[j] = max(col_widths[j], len(cell) + 2)
while len(aligns) < col_count:
aligns.append(None)
def _fmt_row(cells: list[str]) -> str:
padded = []
for j in range(col_count):
c = cells[j] if j < len(cells) else ""
padded.append(_pad_cell(c, col_widths[j], aligns[j]))
return "|" + "|".join(padded) + "|"
lines: list[str] = []
if header_texts:
lines.append(_fmt_row(header_texts))
sep_parts = ["-" * w for w in col_widths]
lines.append("|" + "|".join(sep_parts) + "|")
for row in body_texts:
lines.append(_fmt_row(row))
return "\n".join(lines)
@dataclass
class RenderStyleMap:
bold: Callable[[str], str]
italic: Callable[[str], str]
bold_italic: Callable[[str], str]
strikethrough: Callable[[str], str]
inline_code: Callable[[str], str]
code_block: Callable[[str, str | None], str]
link: Callable[[str, str], str]
image: Callable[[str, str], str]
heading: Callable[[int, str], str]
unordered_list_item: Callable[[str], str]
ordered_list_item: Callable[[str, int], str]
blockquote: Callable[[str], str]
table: Callable[[list[str], list[list[str]], list[str | None]], str] | None = None
thematic_break: str = ""
paragraph_separator: str = "\n\n"
@classmethod
def markdown(cls) -> RenderStyleMap:
return cls(
bold=lambda s: f"**{s}**",
italic=lambda s: f"*{s}*",
bold_italic=lambda s: f"***{s}***",
strikethrough=lambda s: f"~~{s}~~",
inline_code=lambda s: f"`{s}`",
code_block=lambda code, lang: f"```{lang or ''}\n{code}\n```",
link=lambda text, url: f"[{text}]({url})",
image=lambda alt, url: f"![{alt}]({url})",
heading=lambda level, text: f"{'#' * level} {text}",
unordered_list_item=lambda text: f"- {text}",
ordered_list_item=lambda text, idx: f"{idx}. {text}",
blockquote=lambda s: "\n".join(f"> {line}" for line in s.split("\n")),
table=render_md_table,
thematic_break="---",
)
@classmethod
def plain_text(cls) -> RenderStyleMap:
return cls(
bold=lambda s: s,
italic=lambda s: s,
bold_italic=lambda s: s,
strikethrough=lambda s: s,
inline_code=lambda s: s,
code_block=lambda code, lang: code,
link=lambda text, url: f"{text} ({url})" if text else url,
image=lambda alt, url: f"[{alt or 'Image'}]({url})",
heading=lambda level, text: text,
unordered_list_item=lambda text: f"- {text}",
ordered_list_item=lambda text, idx: f"{idx}. {text}",
blockquote=lambda s: s,
table=render_plain_table,
thematic_break="---",
)
def _render_inline(nodes: list[IrNode], style: RenderStyleMap) -> str:
parts: list[str] = []
for node in nodes:
if isinstance(node, Text):
parts.append(node.content)
elif isinstance(node, Bold):
parts.append(style.bold(_render_inline(node.children, style)))
elif isinstance(node, Italic):
parts.append(style.italic(_render_inline(node.children, style)))
elif isinstance(node, BoldItalic):
parts.append(style.bold_italic(_render_inline(node.children, style)))
elif isinstance(node, Strikethrough):
parts.append(style.strikethrough(_render_inline(node.children, style)))
elif isinstance(node, InlineCode):
parts.append(style.inline_code(node.code))
elif isinstance(node, Link):
text = _render_inline(node.children, style)
parts.append(style.link(text, node.url))
elif isinstance(node, Image):
parts.append(style.image(node.alt, node.url))
elif isinstance(node, LineBreak):
parts.append("\n")
return "".join(parts)
def render(doc: Document, style: RenderStyleMap) -> str:
parts: list[str] = []
for i, node in enumerate(doc.children):
if i > 0:
parts.append(style.paragraph_separator)
parts.append(_render_block(node, style))
return "".join(parts)
def _render_block(node: IrNode, style: RenderStyleMap) -> str:
if isinstance(node, Paragraph):
return _render_inline(node.children, style)
elif isinstance(node, Heading):
text = _render_inline(node.children, style)
return style.heading(node.level, text)
elif isinstance(node, CodeBlock):
return style.code_block(node.code, node.language)
elif isinstance(node, BlockQuote):
inner = _render_blocks(node.children, style)
return style.blockquote(inner)
elif isinstance(node, UnorderedList):
lines = []
for item in node.items:
text = _render_list_item(item, style)
lines.append(style.unordered_list_item(text))
return "\n".join(lines)
elif isinstance(node, OrderedList):
lines = []
for idx, item in enumerate(node.items, start=node.start):
text = _render_list_item(item, style)
lines.append(style.ordered_list_item(text, idx))
return "\n".join(lines)
elif isinstance(node, ThematicBreak):
return style.thematic_break
elif isinstance(node, Table):
if style.table is not None:
header_texts = [_render_inline(cell.children, style) for cell in (node.header.cells if node.header else [])]
aligns = [
cell.align for cell in (node.header.cells if node.header else node.rows[0].cells if node.rows else [])
]
body_texts = [[_render_inline(cell.children, style) for cell in row.cells] for row in node.rows]
return style.table(header_texts, body_texts, aligns)
return ""
else:
logger.warning("Unknown IR node type during render: %s", type(node).__name__)
return ""
def _render_list_item(item: ListItem, style: RenderStyleMap) -> str:
inline_parts: list[str] = []
block_parts: list[str] = []
for child in item.children:
if isinstance(child, (UnorderedList, OrderedList)):
block_parts.append(_render_block(child, style))
elif isinstance(child, Paragraph):
block_parts.append(_render_block(child, style))
else:
inline_parts.append(_render_inline([child], style))
result = "".join(inline_parts)
if block_parts:
indent = " "
result += "\n" + "\n".join(indent + line for bp in block_parts for line in bp.split("\n"))
return result
def _render_blocks(nodes: list[IrNode], style: RenderStyleMap) -> str:
parts: list[str] = []
for node in nodes:
parts.append(_render_block(node, style))
return "\n".join(parts)