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)