wx-agent/utils/word_generator518.py
2026-06-30 13:47:51 +08:00

717 lines
26 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

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
def _load_mml2omml_xslt():
global _MML2OMML_XSLT
if _MML2OMML_XSLT is not None:
return _MML2OMML_XSLT
search_paths = [
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",
]
for path in search_paths:
if os.path.isfile(path):
_MML2OMML_XSLT = etree.parse(path)
logger.info(f"加载 MML2OMML.XSL 成功: {path}")
return _MML2OMML_XSLT
logger.warning("未找到 MML2OMML.XSL公式将使用图片回退方式渲染")
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("MarkdownWordGenerator 初始化成功")
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):
try:
import latex2mathml.converter
latex_str = self._preprocess_latex_for_omml(latex_str)
mathml = latex2mathml.converter.convert(latex_str)
mathml_tree = etree.fromstring(mathml.encode('utf-8'))
xslt = _load_mml2omml_xslt()
if xslt is None:
return None
transform = etree.XSLT(xslt)
omml_tree = transform(mathml_tree)
return omml_tree.getroot()
except Exception as e:
logger.error(f"LaTeX->OMML 转换失败 '{latex_str}': {e}")
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 _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()
if self._omml_available and self._add_omml_to_paragraph(paragraph, latex_content):
pass
else:
run = paragraph.add_run(segment)
self.set_font(run, size=default_size, bold=default_bold)
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:
try:
from config import RAG_CONFIG
headers["x-user-id"] = str(RAG_CONFIG.get("x-user-id", "1"))
headers["x-user-name"] = str(RAG_CONFIG.get("x-user-name", "testuser"))
headers["x-role"] = str(RAG_CONFIG.get("x-role", "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
if not (self._omml_available and self._add_omml_to_paragraph(p, formula)):
self.add_paragraph(f"[公式: {formula}]")
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
if not (self._omml_available and self._add_omml_to_paragraph(p, formula)):
self.add_paragraph(f"[公式: {formula}]")
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()