""" 章节级JSON生成节点。 每个章节依据Markdown模板切片独立调用LLM，流式写入Raw文件， 完成后校验并落盘标准化JSON。该节点只负责“拿到合规章节”。 """ from __future__ import annotations import json from pathlib import Path import re from typing import Any, Dict, List, Tuple, Callable, Optional from loguru import logger from ..core import TemplateSection, ChapterStorage from ..ir import ALLOWED_BLOCK_TYPES, ALLOWED_INLINE_MARKS, IRValidator from ..prompts import ( SYSTEM_PROMPT_CHAPTER_JSON, SYSTEM_PROMPT_CHAPTER_JSON_REPAIR, build_chapter_repair_prompt, build_chapter_user_prompt, ) from .base_node import BaseNode try: from json_repair import repair_json as _json_repair_fn except ImportError: # pragma: no cover - optional dependency _json_repair_fn = None class ChapterJsonParseError(ValueError): """章节LLM输出无法解析为合法JSON时抛出的异常，附带原始文本方便排查。""" def __init__(self, message: str, raw_text: Optional[str] = None): """ 构造异常并附加原始输出，便于日志中定位。 Args: message: 人类可读的错误描述。 raw_text: 触发异常的完整LLM输出。 """ super().__init__(message) self.raw_text = raw_text class ChapterContentError(ValueError): """ 章节内容稀疏异常。 当LLM仅输出标题或正文不足以支撑一章时触发，驱动重试以保证报告质量。 """ class ChapterGenerationNode(BaseNode): """ 负责按章节调用LLM并校验JSON结构。 核心能力： - 构造章节级 payload 与提示词； - 以流式形式写入 raw 文件并透传 delta； - 尝试修复/解析LLM输出，并使用 IRValidator 校验； - 对block结构做容错修复，确保最终JSON可渲染。 """ _COLON_EQUALS_PATTERN = re.compile(r'(":\s*)=') _LINE_BREAK_SENTINEL = "__LINE_BREAK__" _INLINE_MARK_ALIASES = { "strong": "bold", "b": "bold", "em": "italic", "emphasis": "italic", "i": "italic", "u": "underline", "strike-through": "strike", "strikethrough": "strike", "s": "strike", "codeblock": "code", "monospace": "code", "hyperlink": "link", "url": "link", "colour": "color", "textcolor": "color", "bgcolor": "highlight", "background": "highlight", "highlightcolor": "highlight", "sub": "subscript", "sup": "superscript", } # 章节若仅包含标题或字符过少则视为失败，强制LLM重新生成 _MIN_NON_HEADING_BLOCKS = 2 _MIN_BODY_CHARACTERS = 400 _PARAGRAPH_FRAGMENT_MAX_CHARS = 80 _PARAGRAPH_FRAGMENT_NO_TERMINATOR_MAX_CHARS = 240 _TERMINATION_PUNCTUATION = set("。！？!?；;……") def __init__(self, llm_client, validator: IRValidator, storage: ChapterStorage): """ 记录LLM客户端/校验器/章节存储器，便于run方法调度。 Args: llm_client: 实际调用大模型的客户端 validator: IR结构校验器 storage: 负责章节流式落盘的存储器 """ super().__init__(llm_client, "ChapterGenerationNode") self.validator = validator self.storage = storage def run( self, section: TemplateSection, context: Dict[str, Any], run_dir: Path, stream_callback: Optional[Callable[[str, Dict[str, Any]], None]] = None, **kwargs, ) -> Dict[str, Any]: """ 针对单个章节调用LLM，校验/落盘章节JSON并返回结构化结果。 参数: section: 模板切片生成的章节对象，包含标题/顺序/slug。 context: Agent构造的共享上下文（主题、篇幅、布局等）。 run_dir: 章节存盘目录，由 `ChapterStorage.start_session` 返回。 stream_callback: 可选流式回调，将LLM delta 推送给前端。 **kwargs: 透传温度、top_p等采样参数。 返回: dict: 通过IR校验的章节JSON。 异常: ChapterJsonParseError: 多次尝试后仍无法解析合法JSON。 ChapterContentError: 正文密度不足或只有标题，需要触发重试。 """ chapter_meta = { "chapterId": section.chapter_id, "slug": section.slug, "title": section.title, "order": section.order, } chapter_dir = self.storage.begin_chapter(run_dir, chapter_meta) llm_payload = self._build_payload(section, context) user_message = build_chapter_user_prompt(llm_payload) raw_text = self._stream_llm( user_message, chapter_dir, stream_callback=stream_callback, section_meta=chapter_meta, **kwargs, ) chapter_json = self._parse_chapter(raw_text) # 自动补全关键字段后再校验 chapter_json.setdefault("chapterId", section.chapter_id) chapter_json.setdefault("anchor", section.slug) chapter_json.setdefault("title", section.title) chapter_json.setdefault("order", section.order) self._sanitize_chapter_blocks(chapter_json) valid, errors = self.validator.validate_chapter(chapter_json) if not valid and errors: repaired = self._attempt_llm_structural_repair( chapter_json, errors, raw_text=raw_text, ) if repaired: chapter_json = repaired chapter_json.setdefault("chapterId", section.chapter_id) chapter_json.setdefault("anchor", section.slug) chapter_json.setdefault("title", section.title) chapter_json.setdefault("order", section.order) self._sanitize_chapter_blocks(chapter_json) valid, errors = self.validator.validate_chapter(chapter_json) content_error: ChapterContentError | None = None if valid: try: self._ensure_content_density(chapter_json) except ChapterContentError as exc: content_error = exc error_messages: List[str] = [] if not valid and errors: error_messages.extend(errors) if content_error: error_messages.append(str(content_error)) self.storage.persist_chapter( run_dir, chapter_meta, chapter_json, errors=None if not error_messages else error_messages, ) if not valid: raise ValueError( f"{section.title} 章节JSON校验失败: {'; '.join(errors[:5])}" ) if content_error: raise content_error return chapter_json # ====== 内部方法 ====== def _build_payload(self, section: TemplateSection, context: Dict[str, Any]) -> Dict[str, Any]: """ 构造LLM输入payload。 参数: section: 当前要生成的章节，提供标题/编号/提纲。 context: 全局上下文字典，包含主题、三引擎报告、篇幅规划等。 返回: dict: 可以直接序列化进提示词的payload，兼顾章节信息与全局约束。 """ reports = context.get("reports", {}) # 章节篇幅规划（来自WordBudgetNode），用于指导字数与强调点 chapter_plan_map = context.get("chapter_directives", {}) chapter_plan = chapter_plan_map.get(section.chapter_id) if chapter_plan_map else {} payload = { "section": { "chapterId": section.chapter_id, "title": section.title, "slug": section.slug, "order": section.order, "number": section.number, "outline": section.outline, }, "globalContext": { "query": context.get("query"), "templateName": context.get("template_name"), "themeTokens": context.get("theme_tokens", {}), "styleDirectives": context.get("style_directives", {}), # layout里包含标题/目录/hero等信息，方便章节保持统一视觉调性 "layout": context.get("layout"), "templateOverview": context.get("template_overview", {}), }, "reports": { "query_engine": reports.get("query_engine", ""), "media_engine": reports.get("media_engine", ""), "insight_engine": reports.get("insight_engine", ""), }, "forumLogs": context.get("forum_logs", ""), "dataBundles": context.get("data_bundles", []), "constraints": { "language": "zh-CN", "maxTokens": context.get("max_tokens", 4096), "allowedBlocks": ALLOWED_BLOCK_TYPES, "styleHints": { "expectWidgets": True, "forceHeadingAnchors": True, "allowInlineMix": True, }, }, "chapterPlan": chapter_plan, "wordPlan": context.get("word_plan"), } if chapter_plan: constraints = payload["constraints"] if chapter_plan.get("targetWords"): constraints["wordTarget"] = chapter_plan["targetWords"] if chapter_plan.get("minWords"): constraints["minWords"] = chapter_plan["minWords"] if chapter_plan.get("maxWords"): constraints["maxWords"] = chapter_plan["maxWords"] if chapter_plan.get("emphasis"): constraints["emphasis"] = chapter_plan["emphasis"] if chapter_plan.get("sections"): constraints["sectionBudgets"] = chapter_plan["sections"] payload["globalContext"]["sectionBudgets"] = chapter_plan["sections"] return payload def _stream_llm( self, user_message: str, chapter_dir: Path, stream_callback: Optional[Callable[[str, Dict[str, Any]], None]] = None, section_meta: Optional[Dict[str, Any]] = None, **kwargs, ) -> str: """ 流式调用LLM并实时写入raw文件，同时通过回调将delta抛出。 参数: user_message: 拼装好的用户提示词。 chapter_dir: 章节的本地缓存目录，用于存放 stream.raw。 stream_callback: SSE流式推送的回调函数。 section_meta: 附带的章节ID/标题，用于回调payload。 **kwargs: 透传温度、top_p等参数。 返回: str: 将所有delta拼接后的原始文本。 """ chunks: List[str] = [] with self.storage.capture_stream(chapter_dir) as stream_fp: stream = self.llm_client.stream_invoke( SYSTEM_PROMPT_CHAPTER_JSON, user_message, temperature=kwargs.get("temperature", 0.2), top_p=kwargs.get("top_p", 0.95), ) for delta in stream: stream_fp.write(delta) chunks.append(delta) if stream_callback: meta = section_meta or {} try: stream_callback(delta, meta) except Exception as callback_error: # pragma: no cover - 仅记录，不阻断主流程 logger.warning(f"章节流式回调失败: {callback_error}") return "".join(chunks) def _parse_chapter(self, raw_text: str) -> Dict[str, Any]: """ 清洗LLM输出并解析JSON。 参数: raw_text: LLM原始输出（可能包含```包裹或额外说明）。 返回: dict: 章节JSON对象，至少包含 chapterId/title/blocks。 异常: ChapterJsonParseError: 多种修复策略仍无法解析合法JSON。 """ cleaned = raw_text.strip() if cleaned.startswith("```json"): cleaned = cleaned[7:] if cleaned.startswith("```"): cleaned = cleaned[3:] if cleaned.endswith("```"): cleaned = cleaned[:-3] cleaned = cleaned.strip() if not cleaned: raise ValueError("LLM返回空内容") candidate_payloads = [cleaned] repaired = self._repair_llm_json(cleaned) if repaired != cleaned: candidate_payloads.append(repaired) try: data = self._parse_with_candidates(candidate_payloads) except json.JSONDecodeError as exc: repaired_payload = self._attempt_json_repair(cleaned) if repaired_payload: candidate_payloads.append(repaired_payload) try: data = self._parse_with_candidates(candidate_payloads[-1:]) except json.JSONDecodeError as inner_exc: raise ChapterJsonParseError( f"章节JSON解析失败: {inner_exc}", raw_text=cleaned ) from inner_exc else: raise ChapterJsonParseError( f"章节JSON解析失败: {exc}", raw_text=cleaned ) from exc if "chapter" in data and isinstance(data["chapter"], dict): return data["chapter"] if isinstance(data, dict) and all( key in data for key in ("chapterId", "title", "blocks") ): return data if isinstance(data, list): for item in data: if isinstance(item, dict): if "chapter" in item and isinstance(item["chapter"], dict): return item["chapter"] if all(key in item for key in ("chapterId", "title", "blocks")): return item raise ValueError("章节JSON缺少chapter字段") def _repair_llm_json(self, text: str) -> str: """ 处理常见的LLM错误（如":=导致的非法JSON）。 参数: text: 原始章节JSON文本。 返回: str: 修复后的文本；若未做改动则返回原内容。 """ repaired = text mutated = False new_text = self._COLON_EQUALS_PATTERN.sub(r"\1", repaired) if new_text != repaired: logger.warning("检测到章节JSON中的\":=\"字符，已自动移除多余的'='号") repaired = new_text mutated = True repaired, escaped = self._escape_in_string_controls(repaired) if escaped: logger.warning("检测到章节JSON字符串中存在未转义的控制字符，已自动转换为转义序列") mutated = True repaired, balanced = self._balance_brackets(repaired) if balanced: logger.warning("检测到章节JSON括号不平衡，已自动补齐/剔除异常括号") mutated = True repaired, commas_fixed = self._fix_missing_commas(repaired) if commas_fixed: logger.warning("检测到章节JSON对象/数组之间缺少逗号，已自动补齐") mutated = True return repaired if mutated else text def _escape_in_string_controls(self, text: str) -> Tuple[str, bool]: """ 将字符串字面量中的裸换行/制表符/控制字符替换为JSON合法的转义序列。 """ if not text: return text, False result: List[str] = [] in_string = False escaped = False mutated = False control_map = {"\n": "\\n", "\r": "\\n", "\t": "\\t"} for ch in text: if escaped: result.append(ch) escaped = False continue if ch == "\\": result.append(ch) escaped = True continue if ch == '"': result.append(ch) in_string = not in_string continue if in_string and ch in control_map: result.append(control_map[ch]) mutated = True continue if in_string and ord(ch) < 0x20: result.append(f"\\u{ord(ch):04x}") mutated = True continue result.append(ch) return "".join(result), mutated def _fix_missing_commas(self, text: str) -> Tuple[str, bool]: """在对象/数组连续出现时自动补逗号""" if not text: return text, False chars: List[str] = [] mutated = False in_string = False escaped = False length = len(text) i = 0 while i < length: ch = text[i] chars.append(ch) if escaped: escaped = False i += 1 continue if ch == "\\": escaped = True i += 1 continue if ch == '"': in_string = not in_string i += 1 continue if not in_string and ch in "}]": j = i + 1 while j < length and text[j] in " \t\r\n": j += 1 if j < length: next_ch = text[j] if next_ch in "{[": chars.append(",") mutated = True i += 1 return "".join(chars), mutated def _balance_brackets(self, text: str) -> Tuple[str, bool]: """尝试修复因LLM多写/少写括号导致的不平衡结构""" if not text: return text, False result: List[str] = [] stack: List[str] = [] mutated = False in_string = False escaped = False opener_map = {"{": "}", "[": "]"} for ch in text: if escaped: result.append(ch) escaped = False continue if ch == "\\": result.append(ch) escaped = True continue if ch == '"': result.append(ch) in_string = not in_string continue if in_string: result.append(ch) continue if ch in "{[": stack.append(ch) result.append(ch) continue if ch in "}]": if stack and ((ch == "}" and stack[-1] == "{") or (ch == "]" and stack[-1] == "[")): stack.pop() result.append(ch) else: mutated = True continue result.append(ch) while stack: opener = stack.pop() result.append(opener_map[opener]) mutated = True return "".join(result), mutated def _attempt_json_repair(self, text: str) -> str | None: """使用可选的json_repair库进一步修复复杂语法错误""" if not _json_repair_fn: return None try: fixed = _json_repair_fn(text) except Exception as exc: # pragma: no cover - library failure logger.warning(f"json_repair 修复章节JSON失败: {exc}") return None if fixed == text: return None logger.warning("已使用json_repair自动修复章节JSON语法") return fixed def _attempt_llm_structural_repair( self, chapter: Dict[str, Any], validation_errors: List[str], raw_text: Optional[str] = None, ) -> Optional[Dict[str, Any]]: """将结构性错误的章节交给LLM兜底修复，保持Report Engine相同的API设置。""" if not validation_errors: return None payload = build_chapter_repair_prompt(chapter, validation_errors, raw_text) try: response = self.llm_client.invoke( SYSTEM_PROMPT_CHAPTER_JSON_REPAIR, payload, temperature=0.0, top_p=0.05, ) except Exception as exc: # pragma: no cover - 网络或API异常仅记录 logger.error(f"章节JSON LLM修复调用失败: {exc}") return None if not response: return None try: repaired = self._parse_chapter(response) except Exception as exc: logger.error(f"LLM修复后的章节JSON解析失败: {exc}") return None logger.warning("章节JSON经多次本地修复仍不合规，已成功启用LLM兜底修复") return repaired def _sanitize_chapter_blocks(self, chapter: Dict[str, Any]): """ 修正常见的结构性错误（例如list.items嵌套过深）。 参数: chapter: 章节JSON对象，会在原地被清理和规整。 """ def walk(blocks: List[Dict[str, Any]] | None): """递归检查并修复嵌套结构，保证每个block合法""" if not isinstance(blocks, list): return for block in blocks: if not isinstance(block, dict): continue self._ensure_block_type(block) self._sanitize_block_content(block) block_type = block.get("type") if block_type == "list": items = block.get("items") normalized = self._normalize_list_items(items) if normalized: block["items"] = normalized for entry in block.get("items", []): walk(entry) elif block_type in {"callout", "blockquote"}: walk(block.get("blocks")) elif block_type == "table": for row in block.get("rows", []): cells = row.get("cells") or [] for cell in cells: walk(cell.get("blocks")) elif block_type == "widget": self._normalize_widget_block(block) else: nested = block.get("blocks") if isinstance(nested, list): walk(nested) walk(chapter.get("blocks")) blocks = chapter.get("blocks") if isinstance(blocks, list): chapter["blocks"] = self._merge_fragment_sequences(blocks) def _ensure_content_density(self, chapter: Dict[str, Any]): """ 校验章节正文密度。 若blocks缺失、除标题外无有效区块，或正文字符数低于阈值， 则视为章节内容异常，触发ChapterContentError以便上游重试。 参数: chapter: 当前章节JSON。 异常: ChapterContentError: 当正文区块数量或字符数达不到下限时抛出。 """ blocks = chapter.get("blocks") if not isinstance(blocks, list) or not blocks: raise ChapterContentError("章节缺少正文区块，无法输出内容") non_heading_blocks = [ block for block in blocks if isinstance(block, dict) and block.get("type") not in {"heading", "divider", "toc"} ] body_characters = self._count_body_characters(blocks) if len(non_heading_blocks) < self._MIN_NON_HEADING_BLOCKS or body_characters < self._MIN_BODY_CHARACTERS: raise ChapterContentError( f"{chapter.get('title') or '该章节'} 正文不足：有效区块 {len(non_heading_blocks)} 个，估算字符数 {body_characters}" ) def _count_body_characters(self, blocks: Any) -> int: """ 递归统计正文字符数。 - 忽略heading/divider/widget等非正文类型； - 对paragraph/list/table/callout等结构抽取嵌套文本； - 仅用于粗粒度判断篇幅是否合理。 参数: blocks: 章节的 blocks 列表或子树。 返回: int: 估算的正文字符数量。 """ def walk(node: Any) -> int: """递归下钻block树并返回字符估算，跳过非正文类型""" if node is None: return 0 if isinstance(node, list): return sum(walk(item) for item in node) if isinstance(node, str): return len(node.strip()) if not isinstance(node, dict): return 0 block_type = node.get("type") if block_type in {"heading", "divider", "toc", "widget"}: return 0 if block_type == "paragraph": inlines = node.get("inlines") if isinstance(inlines, list): total = 0 for run in inlines: if isinstance(run, dict): text = run.get("text") if isinstance(text, str): total += len(text.strip()) return total text_value = node.get("text") if isinstance(text_value, str): return len(text_value.strip()) return len(self._extract_block_text(node).strip()) if block_type == "list": total = 0 for item in node.get("items", []): total += walk(item) return total if block_type in {"blockquote", "callout"}: return walk(node.get("blocks")) if block_type == "table": total = 0 for row in node.get("rows", []): cells = row.get("cells") or [] for cell in cells: total += walk(cell.get("blocks")) return total nested = node.get("blocks") if isinstance(nested, list): return walk(nested) return len(self._extract_block_text(node).strip()) return walk(blocks) def _sanitize_block_content(self, block: Dict[str, Any]): """根据类型做精细化修复，例如清理paragraph内的非法inline mark""" block_type = block.get("type") if block_type == "paragraph": self._normalize_paragraph_block(block) elif block_type == "table": self._sanitize_table_block(block) def _sanitize_table_block(self, block: Dict[str, Any]): """保证表格的rows/cells结构合法且每个单元格包含至少一个block""" rows = self._normalize_table_rows(block.get("rows")) block["rows"] = rows def _normalize_table_rows(self, rows: Any) -> List[Dict[str, Any]]: """确保rows始终是由row对象组成的列表""" if rows is None: rows_iterable: List[Any] = [] elif isinstance(rows, list): rows_iterable = rows else: rows_iterable = [rows] normalized_rows: List[Dict[str, Any]] = [] for row in rows_iterable: sanitized_row = self._normalize_table_row(row) if sanitized_row: normalized_rows.append(sanitized_row) if not normalized_rows: normalized_rows.append({"cells": [self._build_default_table_cell()]}) return normalized_rows def _normalize_table_row(self, row: Any) -> Dict[str, Any] | None: """将各种行表达统一成{'cells': [...]}结构""" if row is None: return None if isinstance(row, dict): result = dict(row) cells_value = result.get("cells") else: result = {} cells_value = row cells = self._normalize_table_cells(cells_value) if not cells: cells = [self._build_default_table_cell()] result["cells"] = cells return result def _normalize_table_cells(self, cells: Any) -> List[Dict[str, Any]]: """清洗单元格，保证每个cell下都有非空blocks""" if cells is None: cell_entries: List[Any] = [] elif isinstance(cells, list): cell_entries = cells else: cell_entries = [cells] normalized_cells: List[Dict[str, Any]] = [] for cell in cell_entries: sanitized = self._normalize_table_cell(cell) if sanitized: normalized_cells.append(sanitized) return normalized_cells def _normalize_table_cell(self, cell: Any) -> Dict[str, Any] | None: """把各种单元格写法规整为schema认可的形式""" if cell is None: return {"blocks": [self._as_paragraph_block("")]} if isinstance(cell, dict): normalized = dict(cell) blocks = self._coerce_cell_blocks(normalized.get("blocks"), normalized) elif isinstance(cell, list): normalized = {} blocks = self._coerce_cell_blocks(cell, None) elif isinstance(cell, (str, int, float)): normalized = {} blocks = [self._as_paragraph_block(str(cell))] else: normalized = {} blocks = [self._as_paragraph_block(str(cell))] normalized["blocks"] = blocks or [self._as_paragraph_block("")] return normalized def _coerce_cell_blocks( self, blocks: Any, source: Dict[str, Any] | None ) -> List[Dict[str, Any]]: """将cell.blocks字段强制转换为合法的block数组""" if isinstance(blocks, list): entries = blocks elif blocks is None: entries = [] else: entries = [blocks] normalized_blocks: List[Dict[str, Any]] = [] for entry in entries: if isinstance(entry, dict): normalized_blocks.append(entry) elif isinstance(entry, list): normalized_blocks.extend(self._coerce_cell_blocks(entry, None)) elif isinstance(entry, (str, int, float)): normalized_blocks.append(self._as_paragraph_block(str(entry))) elif entry is None: continue else: normalized_blocks.append(self._as_paragraph_block(str(entry))) if normalized_blocks: return normalized_blocks text_hint = "" if isinstance(source, dict): text_hint = self._extract_block_text(source).strip() return [self._as_paragraph_block(text_hint or "--")] def _build_default_table_cell(self) -> Dict[str, Any]: """生成一个最小可渲染的空白单元格""" return {"blocks": [self._as_paragraph_block("--")]} def _normalize_paragraph_block(self, block: Dict[str, Any]): """将paragraph的inlines统一规整，剔除非法marks""" inlines = block.get("inlines") normalized_runs: List[Dict[str, Any]] = [] if isinstance(inlines, list) and inlines: for run in inlines: normalized_runs.extend(self._coerce_inline_run(run)) else: normalized_runs = [self._as_inline_run(self._extract_block_text(block))] if not normalized_runs: normalized_runs = [self._as_inline_run("")] block["inlines"] = self._strip_inline_artifacts(normalized_runs) def _strip_inline_artifacts(self, inlines: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """移除被LLM误写入的JSON哨兵文本，防止渲染出`{\"type\": \"\"}`等垃圾字符""" cleaned: List[Dict[str, Any]] = [] for run in inlines or []: if not isinstance(run, dict): continue text = run.get("text") if isinstance(text, str): stripped = text.strip() if stripped.startswith("{") and stripped.endswith("}"): try: payload = json.loads(stripped) except json.JSONDecodeError: payload = None if isinstance(payload, dict) and set(payload.keys()).issubset({"type", "value"}): continue cleaned.append(run) return cleaned or [self._as_inline_run("")] def _merge_fragment_sequences(self, blocks: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """合并被LLM拆成多段的句子片段，避免HTML出现大量孤立

""" if not isinstance(blocks, list): return blocks merged: List[Dict[str, Any]] = [] fragment_buffer: List[Dict[str, Any]] = [] def flush_buffer(): """将当前片段缓冲写入merged列表，必要时合并为单段paragraph""" nonlocal fragment_buffer if not fragment_buffer: return if len(fragment_buffer) == 1: merged.append(fragment_buffer[0]) else: merged.append(self._combine_paragraph_fragments(fragment_buffer)) fragment_buffer = [] for block in blocks: if self._is_paragraph_fragment(block): fragment_buffer.append(block) continue flush_buffer() merged.append(self._merge_nested_fragments(block)) flush_buffer() return merged def _merge_nested_fragments(self, block: Dict[str, Any]) -> Dict[str, Any]: """对嵌套结构（callout/list/table）递归处理片段合并""" block_type = block.get("type") if block_type in {"callout", "blockquote"}: nested = block.get("blocks") if isinstance(nested, list): block["blocks"] = self._merge_fragment_sequences(nested) elif block_type == "list": items = block.get("items") if isinstance(items, list): for entry in items: if isinstance(entry, list): merged_entry = self._merge_fragment_sequences(entry) entry[:] = merged_entry elif block_type == "table": for row in block.get("rows", []): cells = row.get("cells") or [] for cell in cells: nested_blocks = cell.get("blocks") if isinstance(nested_blocks, list): cell["blocks"] = self._merge_fragment_sequences(nested_blocks) return block def _combine_paragraph_fragments(self, fragments: List[Dict[str, Any]]) -> Dict[str, Any]: """将多个句子片段合并为单个paragraph block""" template = dict(fragments[0]) combined_inlines: List[Dict[str, Any]] = [] for fragment in fragments: runs = fragment.get("inlines") if isinstance(runs, list) and runs: combined_inlines.extend(runs) else: fallback_text = self._extract_block_text(fragment) combined_inlines.append(self._as_inline_run(fallback_text)) if not combined_inlines: combined_inlines.append(self._as_inline_run("")) template["inlines"] = combined_inlines return template def _is_paragraph_fragment(self, block: Dict[str, Any]) -> bool: """判断paragraph是否为被错误拆分的短片段""" if not isinstance(block, dict) or block.get("type") != "paragraph": return False inlines = block.get("inlines") text = "" has_marks = False if isinstance(inlines, list) and inlines: parts: List[str] = [] for run in inlines: if not isinstance(run, dict): continue parts.append(str(run.get("text") or "")) marks = run.get("marks") if isinstance(marks, list) and any(marks): has_marks = True text = "".join(parts) else: text = self._extract_block_text(block) stripped = (text or "").strip() if not stripped: return True if has_marks: return False if "\n" in stripped: return False short_limit = self._PARAGRAPH_FRAGMENT_MAX_CHARS long_limit = getattr( self, "_PARAGRAPH_FRAGMENT_NO_TERMINATOR_MAX_CHARS", short_limit * 3, ) if stripped[-1] in self._TERMINATION_PUNCTUATION: return len(stripped) <= short_limit if len(stripped) > long_limit: return False return True def _coerce_inline_run(self, run: Any) -> List[Dict[str, Any]]: """将任意inline写法规整为合法run""" if isinstance(run, dict): normalized_run = dict(run) text = normalized_run.get("text") if not isinstance(text, str): text = "" if text is None else str(text) marks = normalized_run.get("marks") sanitized_marks, extra_text = self._sanitize_inline_marks(marks) normalized_run["marks"] = sanitized_marks normalized_run["text"] = (text or "") + extra_text return [normalized_run] if isinstance(run, str): return [self._as_inline_run(run)] if isinstance(run, (int, float)): return [self._as_inline_run(str(run))] if isinstance(run, list): normalized: List[Dict[str, Any]] = [] for item in run: normalized.extend(self._coerce_inline_run(item)) return normalized return [self._as_inline_run("" if run is None else str(run))] def _sanitize_inline_marks(self, marks: Any) -> Tuple[List[Dict[str, Any]], str]: """过滤非法marks并将break类控制符转成文本""" text_suffix = "" if marks is None: return [], text_suffix mark_list = marks if isinstance(marks, list) else [marks] sanitized: List[Dict[str, Any]] = [] for mark in mark_list: normalized_mark, extra_text = self._normalize_inline_mark(mark) if normalized_mark: sanitized.append(normalized_mark) if extra_text: text_suffix += extra_text return sanitized, text_suffix def _normalize_inline_mark(self, mark: Any) -> Tuple[Dict[str, Any] | None, str]: """对单个mark做兼容映射，或者在必要时转换为文本""" if not isinstance(mark, dict): return None, "" canonical_type = self._canonical_inline_mark_type(mark.get("type")) if canonical_type == self._LINE_BREAK_SENTINEL: return None, "\n" if canonical_type in ALLOWED_INLINE_MARKS: normalized = dict(mark) normalized["type"] = canonical_type return normalized, "" return None, "" def _canonical_inline_mark_type(self, mark_type: Any) -> str | None: """将mark type映射为Schema所支持的取值""" if not isinstance(mark_type, str): return None normalized = mark_type.strip() if not normalized: return None lowered = normalized.lower() if lowered in {"break", "linebreak", "br"}: return self._LINE_BREAK_SENTINEL return self._INLINE_MARK_ALIASES.get(lowered, lowered) def _extract_block_text(self, block: Dict[str, Any]) -> str: """优先从text/content等字段提取fallback文本""" for key in ("text", "content", "value", "title"): value = block.get(key) if isinstance(value, str): return value if value is not None: return str(value) return "" def _normalize_list_items(self, items: Any) -> List[List[Dict[str, Any]]]: """确保list block的items为[[block, block], ...]结构""" if not isinstance(items, list): return [] normalized: List[List[Dict[str, Any]]] = [] for item in items: normalized.extend(self._coerce_list_item(item)) return [entry for entry in normalized if entry] def _coerce_list_item(self, item: Any) -> List[List[Dict[str, Any]]]: """将各种嵌套写法统一折算为区块数组""" result: List[List[Dict[str, Any]]] = [] if isinstance(item, dict): self._ensure_block_type(item) result.append([item]) return result if isinstance(item, list): dicts = [elem for elem in item if isinstance(elem, dict)] if dicts: for elem in dicts: self._ensure_block_type(elem) result.append(dicts) for elem in item: if isinstance(elem, list): result.extend(self._coerce_list_item(elem)) elif isinstance(elem, dict): continue elif isinstance(elem, str): result.append([self._as_paragraph_block(elem)]) elif isinstance(elem, (int, float)): result.append([self._as_paragraph_block(str(elem))]) elif isinstance(item, str): result.append([self._as_paragraph_block(item)]) elif isinstance(item, (int, float)): result.append([self._as_paragraph_block(str(item))]) return result def _normalize_widget_block(self, block: Dict[str, Any]): """确保widget具备顶层data或dataRef""" has_data = block.get("data") is not None or block.get("dataRef") is not None if has_data: return props = block.get("props") if isinstance(props, dict) and "data" in props: block["data"] = props.pop("data") return block["data"] = {"labels": [], "datasets": []} def _ensure_block_type(self, block: Dict[str, Any]): """若block缺少合法type，则降级为paragraph""" block_type = block.get("type") if isinstance(block_type, str) and block_type in ALLOWED_BLOCK_TYPES: return text = "" for key in ("text", "content", "title"): value = block.get(key) if isinstance(value, str) and value.strip(): text = value.strip() break if not text: try: text = json.dumps(block, ensure_ascii=False) except Exception: text = str(block) block.clear() block["type"] = "paragraph" block["inlines"] = [self._as_inline_run(text)] @staticmethod def _as_paragraph_block(text: str) -> Dict[str, Any]: """将字符串快速包装成paragraph block，方便统一处理""" return { "type": "paragraph", "inlines": [ChapterGenerationNode._as_inline_run(text)], } @staticmethod def _as_inline_run(text: str) -> Dict[str, Any]: """构造基础inline run，保证marks字段存在""" return {"text": text or "", "marks": []} @staticmethod def _parse_with_candidates(payloads: List[str]) -> Dict[str, Any]: """按顺序尝试多个payload，直到解析成功""" last_exc: json.JSONDecodeError | None = None for payload in payloads: try: return json.loads(payload) except json.JSONDecodeError as exc: last_exc = exc assert last_exc is not None raise last_exc __all__ = ["ChapterGenerationNode", "ChapterJsonParseError"]