import os import re import uuid import logging import tempfile from datetime import datetime from docx import Document from docx.shared import Pt, RGBColor, Inches from docx.enum.text import WD_ALIGN_PARAGRAPH from docx.oxml.ns import qn from lxml import etree logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) logger = logging.getLogger(__name__) _MML2OMML_XSLT = None _MML2OMML_LOADED = False def _load_mml2omml_xslt(): global _MML2OMML_XSLT, _MML2OMML_LOADED if _MML2OMML_LOADED: return _MML2OMML_XSLT _MML2OMML_LOADED = True _this_dir = os.path.dirname(os.path.abspath(__file__)) search_paths = [ os.path.join(_this_dir, "MML2OMML.XSL"), r"C:\Program Files\Microsoft Office\root\Office16\MML2OMML.XSL", r"C:\Program Files (x86)\Microsoft Office\root\Office16\MML2OMML.XSL", r"C:\Program Files\Microsoft Office\Office16\MML2OMML.XSL", r"C:\Program Files (x86)\Microsoft Office\Office16\MML2OMML.XSL", "/usr/share/mml2omml/MML2OMML.XSL", "/app/utils/MML2OMML.XSL", ] logger.info(f"[公式] 开始搜索 MML2OMML.XSL，共 {len(search_paths)} 个路径") for path in search_paths: exists = os.path.isfile(path) logger.info(f"[公式] 检查路径: {path} -> {'存在' if exists else '不存在'}") if exists: try: _MML2OMML_XSLT = etree.parse(path) logger.info(f"[公式] ✓ MML2OMML.XSL 加载成功: {path}") return _MML2OMML_XSLT except Exception as e: logger.error(f"[公式] ✗ MML2OMML.XSL 加载失败: {path}, 错误: {e}") logger.warning("[公式] ✗ 未找到 MML2OMML.XSL，公式将使用 matplotlib 图片回退方式渲染") return None class MarkdownWordGenerator: def __init__(self): try: self.doc = Document() self.cache_dir = os.path.join(tempfile.gettempdir(), "latex_formula_cache") os.makedirs(self.cache_dir, exist_ok=True) self.formula_cache = {} self._omml_available = _load_mml2omml_xslt() is not None logger.info(f"[公式] MarkdownWordGenerator 初始化完成, OMML可用: {self._omml_available}") except Exception as e: logger.error(f"初始化 MarkdownWordGenerator 失败: {e}") raise def set_font(self, run, size=11, bold=False): try: run.font.name = "微软雅黑" rPr = run._element.get_or_add_rPr() rFonts = rPr.find(qn('w:rFonts')) if rFonts is None: rFonts = run._element.makeelement(qn('w:rFonts'), {}) rPr.insert(0, rFonts) rFonts.set(qn('w:eastAsia'), "微软雅黑") run.font.size = Pt(size) run.font.bold = True if bold else False except Exception as e: logger.warning(f"设置字体时出错: {e}") def add_heading(self, text, level): try: p = self.doc.add_heading(level=level) size_map = {1: 18, 2: 16, 3: 14} font_size = size_map.get(level, 12) self._add_rich_text(p, text, default_size=font_size, default_bold=True) except Exception as e: logger.error(f"添加标题失败 '{text}': {e}") def add_paragraph(self, text): try: if not text.strip(): return p = self.doc.add_paragraph() indent_chars = 0 temp_text = text while temp_text.startswith('\u3000'): indent_chars += 1 temp_text = temp_text[1:] if indent_chars >= 2: p.paragraph_format.first_line_indent = Inches(0.5) text = temp_text elif indent_chars == 1: p.paragraph_format.first_line_indent = Inches(0.25) text = temp_text self._add_rich_text(p, text) except Exception as e: logger.error(f"添加段落失败: {e}") def _add_rich_text(self, paragraph, text, default_size=11, default_bold=False): if not text: return pattern = r'\*\*(.+?)\*\*' last_end = 0 for match in re.finditer(pattern, text): start, end = match.span() if start > last_end: plain = text[last_end:start] if plain: run = paragraph.add_run(plain) self.set_font(run, size=default_size, bold=default_bold) bold_content = match.group(1) if bold_content: run = paragraph.add_run(bold_content) self.set_font(run, size=default_size, bold=True) last_end = end if last_end < len(text): remaining = text[last_end:] if remaining: run = paragraph.add_run(remaining) self.set_font(run, size=default_size, bold=default_bold) @staticmethod def _preprocess_latex_for_omml(latex: str) -> str: replacements = { r'\leqslant': r'\leq', r'\geqslant': r'\geq', r'\nleqslant': r'\nleq', r'\ngeqslant': r'\ngeq', r'\eqslantless': r'\leq', r'\eqslantgtr': r'\geq', r'\varnothing': r'\emptyset', r'\dfrac': r'\frac', r'\tfrac': r'\frac', r'\boldsymbol': r'\mathbf', r'\bm': r'\mathbf', } for old, new in replacements.items(): latex = latex.replace(old, new) latex = re.sub(r'\\mathrm\s*\{([^}]*)\}', lambda m: r'\mathrm{' + m.group(1) + '}', latex) latex = re.sub(r'\\text\s*\{([^}]*)\}', lambda m: r'\mathrm{' + m.group(1) + '}', latex) latex = re.sub(r'\\textbf\s*\{([^}]*)\}', lambda m: r'\mathbf{' + m.group(1) + '}', latex) latex = re.sub(r'\\textit\s*\{([^}]*)\}', lambda m: r'\mathit{' + m.group(1) + '}', latex) return latex def _latex_to_omml(self, latex_str): logger.info(f"[公式] LaTeX->OMML 开始, 原始: '{latex_str}'") try: import latex2mathml.converter processed = self._preprocess_latex_for_omml(latex_str) logger.info(f"[公式] LaTeX 预处理后: '{processed}'") mathml = latex2mathml.converter.convert(processed) logger.info(f"[公式] MathML 生成成功, 长度: {len(mathml)}") mathml_tree = etree.fromstring(mathml.encode('utf-8')) xslt = _load_mml2omml_xslt() if xslt is None: logger.warning("[公式] XSLT 不可用，OMML 转换跳过") return None transform = etree.XSLT(xslt) omml_tree = transform(mathml_tree) omml_root = omml_tree.getroot() logger.info(f"[公式] ✓ OMML 转换成功: '{latex_str}'") return omml_root except Exception as e: logger.error(f"[公式] ✗ LaTeX->OMML 转换失败 '{latex_str}': {e}", exc_info=True) return None def _add_omml_to_paragraph(self, paragraph, latex_str): omml = self._latex_to_omml(latex_str) if omml is not None: paragraph._element.append(omml) return True return False def _latex_to_image(self, latex_str, inline=True): """matplotlib 回退方案：将公式渲染为 PNG 图片""" logger.info(f"[公式] LaTeX->图片 回退, 原始: '{latex_str}'") try: import matplotlib matplotlib.use('Agg') import matplotlib.pyplot as plt processed = self._preprocess_latex_for_omml(latex_str) logger.info(f"[公式] LaTeX 预处理后: '{processed}'") if latex_str in self.formula_cache: cached = self.formula_cache[latex_str] if os.path.isfile(cached): logger.info(f"[公式] 使用缓存: {cached}") return cached filename = os.path.join(self.cache_dir, f"{uuid.uuid4()}.png") fig_width = 6 if not inline else 4 fig_height = 1.2 if not inline else 0.8 font_size = 20 if not inline else 16 fig = plt.figure(figsize=(fig_width, fig_height)) fig.text(0.5, 0.5, f"${processed}$", fontsize=font_size, ha='center', va='center') plt.axis('off') fig.savefig(filename, dpi=300, bbox_inches="tight", pad_inches=0.05) plt.close(fig) self.formula_cache[latex_str] = filename logger.info(f"[公式] ✓ 图片渲染成功: {filename}") return filename except Exception as e: logger.error(f"[公式] ✗ LaTeX->图片 渲染失败 '{latex_str}': {e}", exc_info=True) try: plt.close('all') except Exception: pass return None def _insert_formula(self, paragraph, latex_str, inline=True): """统一公式插入方法：优先 OMML，回退 matplotlib 图片""" if self._omml_available: if self._add_omml_to_paragraph(paragraph, latex_str): return True logger.warning(f"[公式] OMML 失败，回退 matplotlib: '{latex_str}'") img_path = self._latex_to_image(latex_str, inline=inline) if img_path: try: width = Inches(0.7) if inline else Inches(2.5) run = paragraph.add_run() run.add_picture(img_path, width=width) return True except Exception as e: logger.error(f"[公式] ✗ 图片插入失败: {e}") run = paragraph.add_run(f" {latex_str} ") self.set_font(run, size=9) logger.warning(f"[公式] ✗ 所有方式失败，以纯文本插入: '{latex_str}'") return False def _add_rich_text_with_formulas(self, paragraph, text, default_size=11, default_bold=False): if not text: return has_formula = bool(re.search(r'\$[^$]+?\$', text)) has_image = bool(self._extract_image_urls(text)) if not has_formula and not has_image: self._add_rich_text(paragraph, text, default_size=default_size, default_bold=default_bold) return if has_image: img_urls = self._extract_image_urls(text) last_end = 0 for start, end, url in img_urls: before_text = text[last_end:start] if before_text and before_text.strip(): self._add_rich_text_with_formulas(paragraph, before_text, default_size=default_size, default_bold=default_bold) self.add_image(url, width_inches=4.5) last_end = end after_text = text[last_end:] if after_text and after_text.strip(): p = self.doc.add_paragraph() self._add_rich_text_with_formulas(p, after_text, default_size=default_size, default_bold=default_bold) return segments = re.split(r'(\$[^$]+?\$)', text) for segment in segments: if not segment: continue if segment.startswith('$') and segment.endswith('$') and len(segment) > 2: latex_content = segment[1:-1].strip() self._insert_formula(paragraph, latex_content, inline=True) else: self._add_rich_text(paragraph, segment, default_size=default_size, default_bold=default_bold) def add_list(self, text, numbered=False, number=None): try: text = text.strip() if not text: return if re.match(r'^[-_*]{2,}$', text): return if numbered and number is not None: p = self.doc.add_paragraph() p.paragraph_format.left_indent = Pt(18) p.paragraph_format.first_line_indent = Pt(-18) num_run = p.add_run(f"{number}. ") self.set_font(num_run, bold=False) self._add_rich_text_with_formulas(p, text) else: p = self.doc.add_paragraph(style="List Bullet") self._add_rich_text_with_formulas(p, text) except Exception as e: logger.error(f"添加列表失败: {e}") def add_sub_list(self, text): try: text = text.strip() if not text: return if re.match(r'^[-_*]{2,}$', text): return try: p = self.doc.add_paragraph(style="List Bullet 2") except KeyError: p = self.doc.add_paragraph(style="List Bullet") p.paragraph_format.left_indent = Pt(36) self._add_rich_text_with_formulas(p, text) except Exception as e: logger.error(f"添加子列表失败: {e}") def _convert_url_to_local_path(self, url: str) -> str: if not url: return None from urllib.parse import urlparse path = urlparse(url).path if path.startswith('/picture/'): docker_path = f"/app/picture/{path[len('/picture/'):]}" if os.path.isfile(docker_path): return docker_path try: project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) local_path = os.path.join(project_root, "picture", path[len('/picture/'):]) if os.path.isfile(local_path): return local_path except Exception: pass return None return None @staticmethod def _get_http_headers(url: str) -> dict: headers = {} if "/api/v1/knowledge/" in url: def add_header(key: str, value: object, fallback: str = None): value = str(value if value is not None else fallback or "") if not value: return try: value.encode("latin-1") except UnicodeEncodeError: logger.warning(f"跳过非 latin-1 请求头 {key}: {value}") return headers[key] = value try: from config import RAG_CONFIG add_header("x-user-id", RAG_CONFIG.get("x-user-id", "1"), "1") add_header("x-user-name", RAG_CONFIG.get("x-user-name", "testuser"), "testuser") add_header("x-role", RAG_CONFIG.get("x-role", "admin"), "admin") except Exception: headers["x-user-id"] = "1" headers["x-user-name"] = "testuser" headers["x-role"] = "admin" return headers def add_image(self, url_or_path: str, width_inches: float = 5.5): if not url_or_path: return try: import io image_stream = None if os.path.isfile(url_or_path): image_stream = url_or_path elif url_or_path.startswith("http://") or url_or_path.startswith("https://"): local_path = self._convert_url_to_local_path(url_or_path) if local_path and os.path.isfile(local_path): image_stream = local_path else: import requests headers = self._get_http_headers(url_or_path) resp = requests.get(url_or_path, timeout=15, headers=headers, proxies={"http": None, "https": None}) resp.raise_for_status() content_type = resp.headers.get("Content-Type", "") logger.info(f"HTTP 下载完成, Content-Type: {content_type}, 大小: {len(resp.content)} bytes") if not content_type.startswith("image/"): logger.error(f"响应不是图片类型: {content_type}") self.add_paragraph(f"[图片加载失败: {url_or_path}]") return image_stream = io.BytesIO(resp.content) else: logger.warning(f"图片路径无效: {url_or_path}") return from PIL import Image as PilImage if isinstance(image_stream, str): pil_img = PilImage.open(image_stream) else: pil_img = PilImage.open(image_stream) png_stream = io.BytesIO() pil_img.convert("RGBA" if pil_img.mode in ("RGBA", "P") else "RGB").save(png_stream, format="PNG") png_stream.seek(0) p = self.doc.add_paragraph() p.alignment = WD_ALIGN_PARAGRAPH.CENTER run = p.add_run() run.add_picture(png_stream, width=Inches(width_inches)) logger.info(f"图片插入成功: {url_or_path}") except Exception as e: logger.error(f"插入图片失败 '{url_or_path}': {e!r}", exc_info=True) self.add_paragraph(f"[图片加载失败: {url_or_path}]") def _extract_image_urls(self, line_content): """ 从一行文本中提取所有图片URL，支持多种格式： - ![alt](url) - ![alt]`url` - ! `url` 返回 [(start, end, url), ...] """ results = [] # 格式1: ![alt](url) for m in re.finditer(r'!\[([^\]]*)\]\(([^)]+)\)', line_content): url = m.group(2).strip() if url.startswith('`') and url.endswith('`'): url = url[1:-1] results.append((m.start(), m.end(), url)) # 格式2: ![alt]`url` (未被格式1匹配的) for m in re.finditer(r'!\[([^\]]*)\]\s*`([^`]+)`', line_content): url = m.group(2).strip() already = any(s <= m.start() and e >= m.end() for s, e, _ in results) if not already: results.append((m.start(), m.end(), url)) # 格式3: ! `url` (模型常见的破损格式) for m in re.finditer(r'!\s*`([^`]+\.(?:jpg|jpeg|png|gif|bmp|webp))`', line_content, re.IGNORECASE): url = m.group(1).strip() already = any(s <= m.start() and e >= m.end() for s, e, _ in results) if not already: results.append((m.start(), m.end(), url)) results.sort(key=lambda x: x[0]) return results def add_table(self, lines): try: rows = [] for line in lines: line = line.strip("|") cells = [c.strip() for c in line.split("|")] rows.append(cells) if not rows: return header = rows[0] data_start_index = 1 if len(rows) > 1 and all(c.startswith('-') or c == '' for c in rows[1]): data_start_index = 2 data = rows[data_start_index:] max_cols = len(header) table = self.doc.add_table(rows=len(data) + 1, cols=max_cols) table.style = "Table Grid" for i, cell in enumerate(header): if i < max_cols: table.rows[0].cells[i].text = cell for r, row in enumerate(data): for c, val in enumerate(row): if c < max_cols: table.rows[r + 1].cells[c].text = val logger.info("表格添加成功") except Exception as e: logger.error(f"添加表格失败: {e}") def _add_horizontal_rule(self): p = self.doc.add_paragraph() pPr = p._element.get_or_add_pPr() pBdr = p._element.makeelement(qn('w:pBdr'), {}) bottom = p._element.makeelement(qn('w:bottom'), { qn('w:val'): 'single', qn('w:sz'): '6', qn('w:space'): '1', qn('w:color'): 'CCCCCC', }) pBdr.append(bottom) pPr.append(pBdr) def parse(self, markdown): lines = markdown.split("\n") table_buffer = [] block_formula = [] in_block_formula = False i = 0 while i < len(lines): line = lines[i] line_stripped = line.rstrip() line_content = line_stripped.lstrip() # 1. 处理块级公式 if line_content.startswith("$$"): if not in_block_formula: remaining = line_content[2:] close_idx = remaining.find("$$") if close_idx >= 0: formula = remaining[:close_idx].strip() after_formula = remaining[close_idx + 2:].strip() p = self.doc.add_paragraph() p.alignment = WD_ALIGN_PARAGRAPH.CENTER self._insert_formula(p, formula, inline=False) if after_formula: self.add_paragraph(after_formula) else: in_block_formula = True block_formula = [] if remaining.strip(): block_formula.append(remaining.strip()) else: remaining = line_content[2:] in_block_formula = False formula = "".join(block_formula).strip() p = self.doc.add_paragraph() p.alignment = WD_ALIGN_PARAGRAPH.CENTER self._insert_formula(p, formula, inline=False) if remaining.strip(): self.add_paragraph(remaining.strip()) i += 1 continue if in_block_formula: block_formula.append(line) i += 1 continue # 2. 处理表格 if line_content.startswith("|") and line_content.endswith("|"): table_buffer.append(line_stripped) i += 1 continue else: if table_buffer: self.add_table(table_buffer) table_buffer = [] # 空行跳过 if not line_stripped: i += 1 continue # 3. 水平分隔线 if re.match(r'^\s*[-*_]{3,}\s*$', line_stripped): self._add_horizontal_rule() i += 1 continue # 4. 标题 heading_match = re.match(r'^(#{1,3})\s+(.*)', line_content) if heading_match: level = len(heading_match.group(1)) text = heading_match.group(2) self.add_heading(text, level) i += 1 continue # 5. 有序列表（带子列表支持） list_match_ordered = re.match(r'^(\d+)\.\s*(.*)', line_content) if list_match_ordered: number = int(list_match_ordered.group(1)) text = list_match_ordered.group(2) self.add_list(text, numbered=True, number=number) i += 1 while i < len(lines): next_line = lines[i] next_stripped = next_line.rstrip() sub_match = re.match(r'^[\s\t]+[-*]\s*(.*)', next_stripped) if sub_match: sub_text = sub_match.group(1) self.add_sub_list(sub_text) i += 1 elif (next_stripped.startswith(' ') or next_stripped.startswith('\t')): sub_content = next_stripped.strip() if sub_content and not re.match(r'^\d+\.\s', sub_content): self.add_sub_list(sub_content) i += 1 else: break else: break continue # 6. 无序列表 is_bold_line = line_content.startswith('**') if not is_bold_line: list_match_unordered = re.match(r'^\s*[-*]\s+(.*)', line_stripped) if not list_match_unordered: m = re.match(r'^\s*-([^-].*)$', line_stripped) if m and m.group(1).strip(): list_match_unordered = m if list_match_unordered: text = list_match_unordered.group(1).strip() if text: self.add_list(text, numbered=False) i += 1 continue # 7. 图片链接 - 统一提取所有格式的图片URL img_urls = self._extract_image_urls(line_content) if img_urls: has_only_img = (len(img_urls) == 1 and line_content.strip().startswith('!') and img_urls[0][0] == 0) if has_only_img: _, _, url = img_urls[0] self.add_image(url) else: p = self.doc.add_paragraph() last_end = 0 for start, end, url in img_urls: before_text = line_content[last_end:start].strip() if before_text: self._add_rich_text_with_formulas(p, before_text) self.add_image(url, width_inches=4.5) p = self.doc.add_paragraph() last_end = end after_text = line_content[last_end:].strip() if after_text: self._add_rich_text_with_formulas(p, after_text) i += 1 continue # 8. 普通链接 if line_content.startswith("["): match = re.match(r'\[(.*?)\]\((.*?)\)', line_content) if match: p = self.doc.add_paragraph() run = p.add_run(match.group(1)) run.font.color.rgb = RGBColor(0, 0, 255) run.font.underline = True i += 1 continue # 9. 普通段落 p = self.doc.add_paragraph() indent_chars = 0 temp_line = line_stripped while temp_line.startswith('\u3000'): indent_chars += 1 temp_line = temp_line[1:] if indent_chars >= 2: p.paragraph_format.first_line_indent = Inches(0.5) elif indent_chars == 1: p.paragraph_format.first_line_indent = Inches(0.25) self._add_rich_text_with_formulas(p, line_content) i += 1 if table_buffer: self.add_table(table_buffer) def generate(self, markdown, title=None, output=None, images=None): try: if title: logger.info(f"添加文档标题: {title}") p = self.doc.add_paragraph() run = p.add_run(title) self.set_font(run, 20, True) p.alignment = WD_ALIGN_PARAGRAPH.CENTER logger.info("开始解析 Markdown 内容") self.parse(markdown) if images: self._append_images_section(images) if not output: timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') output = f"report_{timestamp}.docx" logger.info(f"正在保存文件: {output}") self.doc.save(output) if os.path.exists(output): file_size = os.path.getsize(output) logger.info(f"文件保存成功: {output}, 大小: {file_size} bytes") return output else: logger.error(f"文件保存后未找到: {output}") raise FileNotFoundError(f"File {output} was not created.") except Exception as e: logger.error(f"生成报告过程中发生严重错误: {e}", exc_info=True) raise def _append_images_section(self, images): try: self.add_heading("附件图片", level=2) for idx, img in enumerate(images, 1): if isinstance(img, dict): url = img.get('url', '') summary = img.get('summary', '') else: url = str(img) summary = '' p = self.doc.add_paragraph() run = p.add_run(f"图片{idx}：") self.set_font(run, bold=True) if url: self.add_image(url, width_inches=4) if summary: p = self.doc.add_paragraph() run = p.add_run(f"说明：{summary}") self.set_font(run, size=10) self.doc.add_paragraph() logger.info(f"图片附件部分添加完成，共 {len(images)} 张图片") except Exception as e: logger.error(f"添加图片附件部分失败: {e}") def generate_report_word(markdown, title=None, output=None, images=None): logger.info("=== 开始生成报告 ===") try: generator = MarkdownWordGenerator() if images: logger.info(f"准备插入 {len(images)} 张图片") file_path = generator.generate(markdown, title=title, output=output, images=images) logger.info(f"=== 报告生成完成: {file_path} ===") return file_path except Exception as e: logger.critical(f"报告生成失败: {e}") raise if __name__ == "__main__": md = r""" ## **测试报告 - OMML公式和图片** ### **一、行内公式** 温度要求 $ \leqslant 40 ^ { \circ } \mathrm { C } $，不得超过。 公差范围 $ \geqslant 0.05 \ \mathrm { mm } $ 以内合格。 ### **二、列表中的公式** 1. 温度 $ \leqslant 40 ^ { \circ } \mathrm { C } $ 要求 2. 间隙 $ \leqslant 0.05 \ \mathrm { mm } $ 标准 - 清洁轴颈 $ \geqslant 0.05 \ \mathrm { mm } $ ### **四、图片测试** ![图片](http://192.168.0.46:11000/api/v1/knowledge/files/images/56b1c0a02ff03d9d202627e88603e6671d3f193511d98a9fab0daf8687e6a9fa.jpg) """ try: result_path = generate_report_word(md) print(f"报告已生成，路径: {result_path}") except Exception as e: print(f"程序执行出错: {e}") import traceback traceback.print_exc()