feat: auto table rendering for markdown outputs
Co-authored-by: factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com>
This commit is contained in:
parent
f0c29304c7
commit
cdc9b1d757
@ -307,5 +307,6 @@ content_prompts:
|
||||
9. 一定要遵循知识库中提供的内容，严格遵守
|
||||
10. 内容中不要提及招标单位的具体名称，如果需要提及，可用需求单位、业主单位等类似词语代替
|
||||
11. 内容中不要出现专业软件的具体版本号，比如unity等
|
||||
12. 如果内容需要，就生成markdown格式的表格，表格格式和内容由你来决定，一定要专业
|
||||
|
||||
请直接输出章节内容,不要包含章节标题。
|
||||
|
||||
@ -4,8 +4,9 @@
|
||||
"""
|
||||
|
||||
import logging
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from ..base import BaseNode, NodeContext
|
||||
from ...utils.validation import ContentValidationError, ContentValidator
|
||||
@ -91,7 +92,7 @@ class SaveToWordNode(BaseNode):
|
||||
return state
|
||||
|
||||
def _fill_word_document(
|
||||
self, word_file: str, chapter: Dict[str, Any], content: str
|
||||
self, word_file: str, chapter: Dict[str, Any], content: Any
|
||||
) -> None:
|
||||
"""填充Word文档
|
||||
|
||||
@ -114,11 +115,99 @@ class SaveToWordNode(BaseNode):
|
||||
# 如果没有占位符，使用章节ID构建默认占位符
|
||||
placeholder = f"{{{{{chapter['id']}_content}}}}"
|
||||
|
||||
structured_content = self._maybe_structure_tables(content)
|
||||
|
||||
try:
|
||||
# 填充章节内容
|
||||
word_processor.fill_chapter_content(word_path, chapter, content)
|
||||
word_processor.fill_chapter_content(word_path, chapter, structured_content)
|
||||
logger.info(f"成功填充占位符: {placeholder}")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"填充Word文档失败: {e}", exc_info=True)
|
||||
raise ValueError(f"章节 {chapter['id']} Word文档填充失败") from e
|
||||
raise ValueError(f"章节 {chapter['id']} Word文档填充失败") from e
|
||||
|
||||
def _maybe_structure_tables(self, content: Any) -> Any:
|
||||
if not isinstance(content, str):
|
||||
return content
|
||||
|
||||
lines = content.splitlines()
|
||||
blocks: List[Dict[str, Any]] = []
|
||||
buffer: List[str] = []
|
||||
i = 0
|
||||
table_found = False
|
||||
|
||||
while i < len(lines):
|
||||
line = lines[i].rstrip()
|
||||
stripped = line.strip()
|
||||
|
||||
if self._is_table_header(stripped) and i + 1 < len(lines) and self._is_table_separator(lines[i + 1]):
|
||||
if buffer and any(s.strip() for s in buffer):
|
||||
blocks.append({"type": "paragraph", "text": "\n".join(buffer).strip()})
|
||||
buffer = []
|
||||
|
||||
table_lines = [stripped, lines[i + 1].strip()]
|
||||
i += 2
|
||||
while i < len(lines) and self._is_table_row(lines[i]):
|
||||
table_lines.append(lines[i].strip())
|
||||
i += 1
|
||||
|
||||
table_block = self._parse_table_block(table_lines)
|
||||
if table_block:
|
||||
blocks.append(table_block)
|
||||
table_found = True
|
||||
continue
|
||||
|
||||
buffer.append(line)
|
||||
i += 1
|
||||
|
||||
if buffer and any(s.strip() for s in buffer):
|
||||
blocks.append({"type": "paragraph", "text": "\n".join(buffer).strip()})
|
||||
|
||||
if table_found:
|
||||
return {"blocks": blocks}
|
||||
return content
|
||||
|
||||
@staticmethod
|
||||
def _is_table_header(line: str) -> bool:
|
||||
return line.count("|") >= 2 and not re.fullmatch(r"\s*\|?[:\-\s|]+\|?\s*", line)
|
||||
|
||||
@staticmethod
|
||||
def _is_table_separator(line: str) -> bool:
|
||||
return bool(re.fullmatch(r"\s*\|?\s*:?-{3,}:?\s*(\|\s*:?-{3,}:?\s*)+\|?\s*", line))
|
||||
|
||||
@staticmethod
|
||||
def _is_table_row(line: str) -> bool:
|
||||
return line.strip().count("|") >= 2 and not SaveToWordNode._is_table_separator(line)
|
||||
|
||||
@staticmethod
|
||||
def _split_table_line(line: str) -> List[str]:
|
||||
trimmed = line.strip()
|
||||
if trimmed.startswith("|"):
|
||||
trimmed = trimmed[1:]
|
||||
if trimmed.endswith("|"):
|
||||
trimmed = trimmed[:-1]
|
||||
return [cell.strip() for cell in trimmed.split("|")]
|
||||
|
||||
def _parse_table_block(self, table_lines: List[str]) -> Optional[Dict[str, Any]]:
|
||||
if len(table_lines) < 2:
|
||||
return None
|
||||
|
||||
header_cells = self._split_table_line(table_lines[0])
|
||||
if not any(cell for cell in header_cells):
|
||||
return None
|
||||
|
||||
data_lines = table_lines[2:] if len(table_lines) >= 2 else []
|
||||
rows = []
|
||||
for line in data_lines:
|
||||
if not line.strip():
|
||||
continue
|
||||
row_cells = self._split_table_line(line)
|
||||
if len(row_cells) != len(header_cells):
|
||||
continue
|
||||
rows.append(row_cells)
|
||||
|
||||
return {
|
||||
"type": "table",
|
||||
"headers": header_cells,
|
||||
"rows": rows,
|
||||
"style": "Table Grid",
|
||||
}
|
||||
@ -3,6 +3,7 @@
|
||||
根据三级标题结构生成专业的标书Word模板。
|
||||
"""
|
||||
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
import unicodedata
|
||||
@ -11,9 +12,11 @@ from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
|
||||
from docx import Document
|
||||
from docx.enum.table import WD_TABLE_ALIGNMENT
|
||||
from docx.shared import Pt, Cm
|
||||
from docx.enum.text import WD_ALIGN_PARAGRAPH, WD_LINE_SPACING
|
||||
from docx.oxml import OxmlElement
|
||||
from docx.oxml.ns import qn
|
||||
from docx.text.paragraph import Paragraph
|
||||
|
||||
from .parser import DocumentChapter, ChapterGuidance
|
||||
@ -204,7 +207,7 @@ class WordProcessor:
|
||||
|
||||
self.fill_chapter_content(doc_path, chapter_meta, content)
|
||||
|
||||
def fill_chapter_content(self, doc_path: str | Path, chapter_meta: Dict[str, Any], content: str) -> None:
|
||||
def fill_chapter_content(self, doc_path: str | Path, chapter_meta: Dict[str, Any], content: Any) -> None:
|
||||
"""填充指定章节的内容到Word文档"""
|
||||
|
||||
if not Path(doc_path).exists():
|
||||
@ -231,7 +234,7 @@ class WordProcessor:
|
||||
chapter_level = self._get_chapter_level_from_heading(target_paragraph)
|
||||
chapter_level = chapter_level or DEFAULT_CHAPTER_LEVEL
|
||||
|
||||
parsed_paragraphs = self._parse_markdown_to_paragraphs(content, chapter_level)
|
||||
parsed_paragraphs = self._parse_content_to_paragraph_structs(content, chapter_level)
|
||||
|
||||
if matched_strategy == "any_heading_fallback":
|
||||
notice_text = (
|
||||
@ -272,16 +275,92 @@ class WordProcessor:
|
||||
"成功填充内容到文档: %s", chapter_meta.get("placeholder", chapter_meta.get("id"))
|
||||
)
|
||||
|
||||
def _parse_content_to_paragraph_structs(self, content: Any, chapter_level: int) -> list:
|
||||
structured_blocks = self._load_structured_blocks(content)
|
||||
if structured_blocks is not None:
|
||||
normalized = self._normalize_blocks_to_paragraph_structs(structured_blocks)
|
||||
if normalized:
|
||||
return normalized
|
||||
|
||||
content_str = content if isinstance(content, str) else json.dumps(content, ensure_ascii=False)
|
||||
return self._parse_markdown_to_paragraphs(content_str or "", chapter_level)
|
||||
|
||||
def _load_structured_blocks(self, content: Any) -> Optional[list]:
|
||||
if isinstance(content, (dict, list)):
|
||||
candidate = content
|
||||
elif isinstance(content, str):
|
||||
stripped = content.strip()
|
||||
if not stripped.startswith('{') and not stripped.startswith('['):
|
||||
return None
|
||||
try:
|
||||
candidate = json.loads(stripped)
|
||||
except Exception:
|
||||
return None
|
||||
else:
|
||||
return None
|
||||
|
||||
if isinstance(candidate, dict) and isinstance(candidate.get('blocks'), list):
|
||||
return candidate['blocks']
|
||||
if isinstance(candidate, list):
|
||||
return candidate
|
||||
return None
|
||||
|
||||
def _normalize_blocks_to_paragraph_structs(self, blocks: list) -> list:
|
||||
paragraphs: list = []
|
||||
|
||||
for block in blocks:
|
||||
if not isinstance(block, dict):
|
||||
continue
|
||||
block_type = (block.get('type') or '').lower()
|
||||
|
||||
if block_type in {'paragraph', 'text', 'p'}:
|
||||
text = str(block.get('text', '')).strip()
|
||||
if text:
|
||||
paragraphs.append({'type': 'paragraph', 'text': text})
|
||||
|
||||
elif block_type in {'bold', 'subheading', 'heading'}:
|
||||
text = str(block.get('text', '')).strip()
|
||||
if text:
|
||||
paragraphs.append({'type': 'bold_paragraph', 'text': text})
|
||||
|
||||
elif block_type in {'ordered_list', 'ol'}:
|
||||
items = block.get('items') or []
|
||||
for item in items:
|
||||
item_text = str(item).strip()
|
||||
if item_text:
|
||||
paragraphs.append({'type': 'ordered_list', 'text': item_text})
|
||||
|
||||
elif block_type in {'unordered_list', 'ul', 'bullet'}:
|
||||
items = block.get('items') or []
|
||||
for item in items:
|
||||
item_text = str(item).strip()
|
||||
if item_text:
|
||||
paragraphs.append({'type': 'unordered_list', 'text': item_text})
|
||||
|
||||
elif block_type == 'table':
|
||||
headers = block.get('headers') or []
|
||||
rows = block.get('rows') or []
|
||||
has_content = headers or rows
|
||||
if not has_content:
|
||||
continue
|
||||
paragraphs.append({
|
||||
'type': 'table',
|
||||
'title': block.get('title'),
|
||||
'headers': headers,
|
||||
'rows': rows,
|
||||
'style': block.get('style'),
|
||||
'column_widths_cm': block.get('column_widths_cm'),
|
||||
'alignment': block.get('alignment'),
|
||||
})
|
||||
|
||||
else:
|
||||
text = block.get('text') or block.get('value')
|
||||
if text:
|
||||
paragraphs.append({'type': 'paragraph', 'text': str(text)})
|
||||
|
||||
return paragraphs
|
||||
|
||||
def _parse_markdown_to_paragraphs(self, content: str, chapter_level: int) -> list:
|
||||
"""解析Markdown内容为段落结构列表
|
||||
|
||||
Args:
|
||||
content: Markdown格式的内容
|
||||
chapter_level: 当前章节层级（1/2/3）
|
||||
|
||||
Returns:
|
||||
段落结构列表，每项包含 {'type': 'heading'|'paragraph'|'list', 'text': str, 'style': str}
|
||||
"""
|
||||
paragraphs = []
|
||||
lines = content.split('\n')
|
||||
i = 0
|
||||
@ -293,7 +372,6 @@ class WordProcessor:
|
||||
i += 1
|
||||
continue
|
||||
|
||||
# Markdown标题：## 标题
|
||||
if MARKDOWN_HEADING_PATTERN.match(line):
|
||||
title_text = MARKDOWN_HEADING_PATTERN.sub('', line).strip()
|
||||
if not title_text:
|
||||
@ -304,7 +382,6 @@ class WordProcessor:
|
||||
'text': title_text,
|
||||
})
|
||||
|
||||
# 有序列表：1. / 1) / （1）
|
||||
elif ORDERED_LIST_PATTERN.match(line):
|
||||
text = ORDERED_LIST_PATTERN.sub('', line)
|
||||
paragraphs.append({
|
||||
@ -312,7 +389,6 @@ class WordProcessor:
|
||||
'text': text
|
||||
})
|
||||
|
||||
# 无序列表：- / * / •
|
||||
elif UNORDERED_LIST_PATTERN.match(line):
|
||||
text = UNORDERED_LIST_PATTERN.sub('', line)
|
||||
paragraphs.append({
|
||||
@ -320,11 +396,8 @@ class WordProcessor:
|
||||
'text': text
|
||||
})
|
||||
|
||||
# 普通段落
|
||||
else:
|
||||
# 处理行内格式：**粗体**
|
||||
text = BOLD_TEXT_PATTERN.sub(r'\1', line)
|
||||
# 处理行内格式：*斜体*（单星号，但避免误匹配列表）
|
||||
if not line.startswith('*'):
|
||||
text = ITALIC_TEXT_PATTERN.sub(r'\1', text)
|
||||
|
||||
@ -724,11 +797,98 @@ class WordProcessor:
|
||||
list_index = sum(1 for p in parsed_paragraphs[:i] if p['type'] == 'ordered_list') + 1
|
||||
para_struct['list_index'] = list_index
|
||||
|
||||
# 逐个插入段落
|
||||
for para_struct in parsed_paragraphs:
|
||||
if para_struct.get('type') == 'table':
|
||||
current_element = self._insert_table_after(doc, current_element, para_struct)
|
||||
continue
|
||||
|
||||
_, new_element = self._create_paragraph_element(doc, current_element, para_struct)
|
||||
current_element = new_element
|
||||
|
||||
def _insert_table_after(self, doc: Document, current_element, table_struct: dict):
|
||||
last_element = current_element
|
||||
|
||||
title = table_struct.get('title')
|
||||
if title:
|
||||
_, last_element = self._create_paragraph_element(doc, last_element, {
|
||||
'type': 'bold_paragraph',
|
||||
'text': str(title),
|
||||
})
|
||||
|
||||
headers = table_struct.get('headers') or []
|
||||
rows = table_struct.get('rows') or []
|
||||
col_count = max(len(headers), max((len(r) for r in rows), default=0)) or 1
|
||||
row_count = len(rows) + (1 if headers else 0)
|
||||
table = doc.add_table(rows=row_count or 1, cols=col_count)
|
||||
|
||||
style_name = table_struct.get('style') or 'Table Grid'
|
||||
try:
|
||||
table.style = style_name
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
current_row = 0
|
||||
if headers:
|
||||
for idx in range(col_count):
|
||||
cell = table.cell(0, idx)
|
||||
text = headers[idx] if idx < len(headers) else ''
|
||||
cell.text = str(text) if text is not None else ''
|
||||
if cell.paragraphs:
|
||||
for run in cell.paragraphs[0].runs:
|
||||
run.bold = True
|
||||
current_row = 1
|
||||
|
||||
for r_idx, row in enumerate(rows):
|
||||
cells = table.rows[current_row + r_idx].cells
|
||||
for c_idx in range(col_count):
|
||||
value = row[c_idx] if c_idx < len(row) else ''
|
||||
cells[c_idx].text = str(value) if value is not None else ''
|
||||
|
||||
col_widths = table_struct.get('column_widths_cm')
|
||||
if isinstance(col_widths, list) and len(col_widths) == col_count:
|
||||
for col_idx, width in enumerate(col_widths):
|
||||
try:
|
||||
width_value = float(width)
|
||||
except (TypeError, ValueError):
|
||||
continue
|
||||
for cell in table.columns[col_idx].cells:
|
||||
cell.width = Cm(width_value)
|
||||
|
||||
alignment = (table_struct.get('alignment') or '').lower()
|
||||
if alignment in {'center', 'centre'}:
|
||||
table.alignment = WD_TABLE_ALIGNMENT.CENTER
|
||||
elif alignment in {'right', 'end'}:
|
||||
table.alignment = WD_TABLE_ALIGNMENT.RIGHT
|
||||
else:
|
||||
table.alignment = WD_TABLE_ALIGNMENT.LEFT
|
||||
|
||||
self._apply_table_borders(table)
|
||||
|
||||
tbl_element = table._element
|
||||
last_element.addnext(tbl_element)
|
||||
return tbl_element
|
||||
|
||||
@staticmethod
|
||||
def _apply_table_borders(table) -> None:
|
||||
tbl_pr = table._element.tblPr
|
||||
if tbl_pr is None:
|
||||
tbl_pr = OxmlElement('w:tblPr')
|
||||
table._element.append(tbl_pr)
|
||||
borders = getattr(tbl_pr, 'tblBorders', None)
|
||||
if borders is None:
|
||||
borders = OxmlElement('w:tblBorders')
|
||||
tbl_pr.append(borders)
|
||||
|
||||
for border_name in ['top', 'left', 'bottom', 'right', 'insideH', 'insideV']:
|
||||
element = getattr(borders, border_name, None)
|
||||
if element is None:
|
||||
element = OxmlElement(f'w:{border_name}')
|
||||
borders.append(element)
|
||||
element.set(qn('w:val'), 'single')
|
||||
element.set(qn('w:sz'), '8')
|
||||
element.set(qn('w:space'), '0')
|
||||
element.set(qn('w:color'), '000000')
|
||||
|
||||
def _setup_numbering_styles(self, doc: Document):
|
||||
"""配置标题的多级编号样式"""
|
||||
# 确保标题样式存在并配置编号格式
|
||||
|
||||
@ -1,11 +1,13 @@
|
||||
"""Word处理器章节填充逻辑单元测试"""
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from tempfile import TemporaryDirectory
|
||||
|
||||
import pytest
|
||||
from docx import Document
|
||||
|
||||
from bidmaster.nodes.content.save_to_word import SaveToWordNode
|
||||
from bidmaster.tools.word import WordProcessor
|
||||
|
||||
|
||||
@ -194,4 +196,99 @@ def test_fill_matches_higher_level_heading(word_processor: WordProcessor) -> Non
|
||||
word_processor.fill_chapter_content(doc_path, chapter_meta, "深度内容")
|
||||
|
||||
doc = Document(doc_path)
|
||||
assert any("深度内容" in para.text for para in doc.paragraphs)
|
||||
assert any("深度内容" in para.text for para in doc.paragraphs)
|
||||
|
||||
|
||||
def test_save_node_auto_converts_markdown_table() -> None:
|
||||
with TemporaryDirectory() as tmp_dir:
|
||||
doc_path = Path(tmp_dir) / "table_from_md.docx"
|
||||
_create_document(doc_path, [("1 表格章节", 1)])
|
||||
|
||||
chapter_meta = {
|
||||
"id": "chapter_1",
|
||||
"title": "表格章节",
|
||||
"level": 1,
|
||||
"placeholder": "{{chapter_1_content}}",
|
||||
"normalized_title": WordProcessor()._normalize_heading_text("表格章节"),
|
||||
"heading_number": "1",
|
||||
"order_index": 1,
|
||||
}
|
||||
|
||||
content = """表格说明\n| 项目 | 数量 |\n| --- | --- |\n| A | 1 |\n| B | 2 |"""
|
||||
|
||||
node = SaveToWordNode()
|
||||
node._fill_word_document(str(doc_path), chapter_meta, content)
|
||||
|
||||
doc = Document(doc_path)
|
||||
assert len(doc.tables) == 1
|
||||
table = doc.tables[0]
|
||||
assert table.cell(0, 0).text.strip() == "项目"
|
||||
assert table.cell(1, 1).text.strip() == "1"
|
||||
|
||||
|
||||
def test_table_borders_are_solid() -> None:
|
||||
with TemporaryDirectory() as tmp_dir:
|
||||
doc_path = Path(tmp_dir) / "table_border.docx"
|
||||
_create_document(doc_path, [("1 边框", 1)])
|
||||
|
||||
chapter_meta = {
|
||||
"id": "chapter_1",
|
||||
"title": "边框",
|
||||
"level": 1,
|
||||
"placeholder": "{{chapter_1_content}}",
|
||||
"normalized_title": WordProcessor()._normalize_heading_text("边框"),
|
||||
"heading_number": "1",
|
||||
"order_index": 1,
|
||||
}
|
||||
|
||||
content = """| A | B |
|
||||
| --- | --- |
|
||||
| 1 | 2 |"""
|
||||
|
||||
node = SaveToWordNode()
|
||||
node._fill_word_document(str(doc_path), chapter_meta, content)
|
||||
|
||||
doc = Document(doc_path)
|
||||
table = doc.tables[0]
|
||||
xml = table._element.tblPr.xml if table._element.tblPr is not None else ""
|
||||
assert "tblBorders" in xml
|
||||
assert "w:val=\"single\"" in xml
|
||||
|
||||
|
||||
def test_fill_with_structured_table_block(word_processor: WordProcessor) -> None:
|
||||
with TemporaryDirectory() as tmp_dir:
|
||||
doc_path = Path(tmp_dir) / "table_block.docx"
|
||||
_create_document(doc_path, [("1 技术方案", 1)])
|
||||
|
||||
chapter_meta = {
|
||||
"id": "chapter_1",
|
||||
"title": "技术方案",
|
||||
"level": 1,
|
||||
"placeholder": "{{chapter_1_content}}",
|
||||
"normalized_title": word_processor._normalize_heading_text("技术方案"),
|
||||
"heading_number": "1",
|
||||
"order_index": 1,
|
||||
}
|
||||
|
||||
content = json.dumps({
|
||||
"blocks": [
|
||||
{"type": "paragraph", "text": "方案概述"},
|
||||
{
|
||||
"type": "table",
|
||||
"title": "设备清单",
|
||||
"headers": ["名称", "数量"],
|
||||
"rows": [["A型设备", "2"], ["B型设备", "4"]],
|
||||
"style": "Table Grid",
|
||||
"column_widths_cm": [6, 3],
|
||||
},
|
||||
]
|
||||
})
|
||||
|
||||
word_processor.fill_chapter_content(doc_path, chapter_meta, content)
|
||||
|
||||
doc = Document(doc_path)
|
||||
assert any("方案概述" in para.text for para in doc.paragraphs)
|
||||
assert len(doc.tables) == 1
|
||||
table = doc.tables[0]
|
||||
assert table.cell(0, 0).text == "名称"
|
||||
assert table.cell(1, 0).text == "A型设备"
|
||||
Loading…
Reference in New Issue
Block a user