803 lines
30 KiB
Python
803 lines
30 KiB
Python
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`
|
||
- ! `url`
|
||
返回 [(start, end, url), ...]
|
||
"""
|
||
results = []
|
||
|
||
# 格式1: 
|
||
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 } $
|
||
|
||
|
||
### **四、图片测试**
|
||
|
||
|
||
"""
|
||
|
||
try:
|
||
result_path = generate_report_word(md)
|
||
print(f"报告已生成，路径: {result_path}")
|
||
except Exception as e:
|
||
print(f"程序执行出错: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|