diff --git a/config/prompts.yaml b/config/prompts.yaml index e33f364..1301ad9 100644 --- a/config/prompts.yaml +++ b/config/prompts.yaml @@ -307,5 +307,6 @@ content_prompts: 9. 一定要遵循知识库中提供的内容，严格遵守 10. 内容中不要提及招标单位的具体名称，如果需要提及，可用需求单位、业主单位等类似词语代替 11. 内容中不要出现专业软件的具体版本号，比如unity等 + 12. 如果内容需要，就生成markdown格式的表格，表格格式和内容由你来决定，一定要专业 请直接输出章节内容,不要包含章节标题。 diff --git a/src/bidmaster/nodes/content/save_to_word.py b/src/bidmaster/nodes/content/save_to_word.py index 6f0c0a6..1cff228 100644 --- a/src/bidmaster/nodes/content/save_to_word.py +++ b/src/bidmaster/nodes/content/save_to_word.py @@ -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 \ No newline at end of file + 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", + } \ No newline at end of file diff --git a/src/bidmaster/tools/word.py b/src/bidmaster/tools/word.py index 4cfea3d..95203d6 100644 --- a/src/bidmaster/tools/word.py +++ b/src/bidmaster/tools/word.py @@ -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): """配置标题的多级编号样式""" # 确保标题样式存在并配置编号格式 diff --git a/tests/unit/test_word.py b/tests/unit/test_word.py index d927ba5..5e24d4e 100644 --- a/tests/unit/test_word.py +++ b/tests/unit/test_word.py @@ -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) \ No newline at end of file + 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型设备" \ No newline at end of file