""" Markdown 生成器模块 提供 Markdown 输出功能: - 完整文档 Markdown 生成 - 按页 Markdown 生成 - MinerU union_make 集成 - 金额数字标准化(全角→半角) """ import sys from pathlib import Path from typing import Dict, Any, List, Tuple, Optional from loguru import logger # 导入 MinerU 组件 mineru_path = Path(__file__).parents[3] if str(mineru_path) not in sys.path: sys.path.insert(0, str(mineru_path)) try: from mineru.backend.vlm.vlm_middle_json_mkcontent import union_make as vlm_union_make from mineru.utils.enum_class import MakeMode MINERU_AVAILABLE = True except ImportError: MINERU_AVAILABLE = False vlm_union_make = None class MakeMode: MM_MD = 'mm_md' NLP_MD = 'nlp_md' # 导入数字标准化工具 from .normalize_financial_numbers import normalize_markdown_table class MarkdownGenerator: """Markdown 生成器类""" @staticmethod def save_markdown( results: Dict[str, Any], middle_json: Dict[str, Any], output_dir: Path, doc_name: str, use_mineru_union: bool = False, normalize_numbers: bool = True ) -> Tuple[Path, Optional[Path]]: """ 保存 Markdown 文件 默认使用自定义实现,确保所有元素类型(包括 table_caption 等)都被正确处理 可选使用 MinerU union_make(但它不处理 table_caption 等独立元素) Args: results: 处理结果 middle_json: middle.json 格式数据 output_dir: 输出目录 doc_name: 文档名称 use_mineru_union: 是否使用 MinerU union_make(默认 False) normalize_numbers: 是否标准化金额数字(全角→半角) Returns: (Markdown 文件路径, 原始文件路径 或 None) """ md_path = output_dir / f"{doc_name}.md" original_path = None if use_mineru_union and MINERU_AVAILABLE and vlm_union_make is not None: try: img_bucket_path = "images" markdown_content = vlm_union_make( middle_json['pdf_info'], MakeMode.MM_MD, img_bucket_path ) if markdown_content: if isinstance(markdown_content, list): markdown_content = '\n\n'.join(markdown_content) header = MarkdownGenerator._generate_header(results) markdown_content = header + str(markdown_content) # 金额数字标准化 if normalize_numbers: original_content = markdown_content markdown_content = normalize_markdown_table(markdown_content) if markdown_content != original_content: original_path = output_dir / f"{doc_name}_original.md" with open(original_path, 'w', encoding='utf-8') as f: f.write(original_content) logger.info(f"📝 Original Markdown saved: {original_path}") with open(md_path, 'w', encoding='utf-8') as f: f.write(markdown_content) logger.info(f"📝 Markdown saved (MinerU format): {md_path}") return md_path, original_path except Exception as e: logger.warning(f"MinerU union_make failed: {e}, falling back to custom implementation") # 使用自定义实现,确保所有元素类型都被处理 markdown_content = MarkdownGenerator._generate_full_markdown(results) # 金额数字标准化 if normalize_numbers: original_content = markdown_content markdown_content = normalize_markdown_table(markdown_content) if markdown_content != original_content: original_path = output_dir / f"{doc_name}_original.md" with open(original_path, 'w', encoding='utf-8') as f: f.write(original_content) logger.info(f"📝 Original Markdown saved: {original_path}") with open(md_path, 'w', encoding='utf-8') as f: f.write(markdown_content) logger.info(f"📝 Markdown saved (custom format): {md_path}") return md_path, original_path @staticmethod def save_page_markdowns( results: Dict[str, Any], output_dir: Path, doc_name: str, is_pdf: bool = True, normalize_numbers: bool = True ) -> List[str]: """ 按页保存 Markdown 文件 命名规则: - PDF输入: 文件名_page_001.md - 图片输入(单页): 文件名.md(跳过,因为已有完整版) Args: results: 处理结果 output_dir: 输出目录 doc_name: 文档名称 is_pdf: 是否为 PDF 输入 normalize_numbers: 是否标准化金额数字(全角→半角) Returns: 保存的 Markdown 文件路径列表 """ saved_paths = [] total_pages = len(results.get('pages', [])) # 单个图片输入时,跳过按页保存(因为已有完整版 doc_name.md) if not is_pdf and total_pages == 1: logger.debug("📝 Single image input, skipping page markdown (full version exists)") return saved_paths for page in results.get('pages', []): page_idx = page.get('page_idx', 0) # 根据输入类型决定命名 if is_pdf or total_pages > 1: page_name = f"{doc_name}_page_{page_idx + 1:03d}" else: page_name = doc_name # 生成单页 Markdown md_content = MarkdownGenerator._generate_page_markdown(page, doc_name, page_idx) # 金额数字标准化 if normalize_numbers: original_content = md_content md_content = normalize_markdown_table(md_content) if md_content != original_content: original_path = output_dir / f"{page_name}_original.md" with open(original_path, 'w', encoding='utf-8') as f: f.write(original_content) logger.debug(f"📝 Original page Markdown saved: {original_path}") # 保存 md_path = output_dir / f"{page_name}.md" with open(md_path, 'w', encoding='utf-8') as f: f.write(md_content) saved_paths.append(str(md_path)) logger.debug(f"📝 Page Markdown saved: {md_path}") if saved_paths: logger.info(f"📝 {len(saved_paths)} page Markdowns saved") return saved_paths @staticmethod def _generate_header(results: Dict[str, Any]) -> str: """生成 Markdown 文件头""" return f""" """ @staticmethod def _generate_full_markdown(results: Dict[str, Any]) -> str: """ 生成完整文档的 Markdown(自定义实现) 确保所有元素类型都被正确处理,包括 table_caption、table_footnote 等 Args: results: 处理结果 Returns: Markdown 内容字符串 """ md_lines = [ f"", "", ] for page in results.get('pages', []): # 按阅读顺序处理元素 for element in page.get('elements', []): elem_type = element.get('type', '') content = element.get('content', {}) if elem_type == 'title': text = content.get('text', '') if isinstance(content, dict) else str(content) level = element.get('level', 1) if text: md_lines.append(f"{'#' * min(level, 6)} {text}") md_lines.append("") elif elem_type in ['text', 'ocr_text', 'ref_text']: text = content.get('text', '') if isinstance(content, dict) else str(content) if text: md_lines.append(text) md_lines.append("") elif elem_type in ['table', 'table_body']: html = content.get('html', '') if html: md_lines.append(f"\n{html}\n") md_lines.append("") elif elem_type in ['image', 'image_body', 'figure']: img_filename = content.get('image_path', '') if img_filename: md_lines.append(f"![](images/{img_filename})") md_lines.append("") elif elem_type in ['interline_equation', 'inline_equation', 'equation']: latex = content.get('latex', '') if latex: md_lines.append(f"$$\n{latex}\n$$") md_lines.append("") elif elem_type in ['table_caption', 'table_footnote']: text = content.get('text', '') if isinstance(content, dict) else str(content) if text: if elem_type == 'table_caption': md_lines.append(f"**{text}**") else: md_lines.append(f"*{text}*") md_lines.append("") elif elem_type in ['image_caption', 'image_footnote']: text = content.get('text', '') if isinstance(content, dict) else str(content) if text: if elem_type == 'image_caption': md_lines.append(f"**{text}**") else: md_lines.append(f"*{text}*") md_lines.append("") return '\n'.join(md_lines) @staticmethod def _generate_fallback(results: Dict[str, Any]) -> str: """降级方案:自定义 Markdown 生成""" md_lines = [ f"", "", ] for page in results.get('pages', []): for element in page.get('elements', []): elem_type = element.get('type', '') content = element.get('content', {}) bbox = element.get('bbox', []) # 添加 bbox 注释 if bbox: md_lines.append(f"") if elem_type == 'title': text = content.get('text', '') if isinstance(content, dict) else str(content) level = element.get('level', 1) md_lines.append(f"{'#' * min(level, 6)} {text}") md_lines.append("") elif elem_type in ['text', 'ocr_text', 'ref_text']: text = content.get('text', '') if isinstance(content, dict) else str(content) if text: md_lines.append(text) md_lines.append("") elif elem_type in ['table', 'table_body']: # 表格标题 table_captions = content.get('table_caption', []) if isinstance(table_captions, str): table_captions = [table_captions] if table_captions else [] for caption in table_captions: md_lines.append(f"**{caption}**") html = content.get('html', '') if html: md_lines.append(f"\n{html}\n") md_lines.append("") elif elem_type in ['image', 'image_body', 'figure']: img_filename = content.get('image_path', '') if img_filename: md_lines.append(f"![](images/{img_filename})") md_lines.append("") elif elem_type in ['interline_equation', 'inline_equation', 'equation']: latex = content.get('latex', '') if latex: md_lines.append(f"$$\n{latex}\n$$") md_lines.append("") elif elem_type in ['table_caption', 'table_footnote']: text = content.get('text', '') if isinstance(content, dict) else str(content) if text: if elem_type == 'table_caption': md_lines.append(f"**{text}**") else: md_lines.append(f"*{text}*") md_lines.append("") elif elem_type in ['image_caption', 'image_footnote']: text = content.get('text', '') if isinstance(content, dict) else str(content) if text: if elem_type == 'image_caption': md_lines.append(f"**{text}**") else: md_lines.append(f"*{text}*") md_lines.append("") return '\n'.join(md_lines) @staticmethod def _generate_page_markdown( page: Dict[str, Any], doc_name: str, page_idx: int ) -> str: """ 生成单页的 Markdown 内容 Args: page: 页面数据 doc_name: 文档名称 page_idx: 页码索引 Returns: Markdown 内容字符串 """ md_lines = [ f"", "", ] for element in page.get('elements', []): elem_type = element.get('type', '') content = element.get('content', {}) bbox = element.get('bbox', []) reading_order = element.get('reading_order', 0) # 添加元素注释 md_lines.append(f"") if elem_type == 'title': text = content.get('text', '') if isinstance(content, dict) else str(content) level = element.get('level', 1) md_lines.append(f"{'#' * min(level, 6)} {text}") md_lines.append("") elif elem_type in ['text', 'ocr_text', 'ref_text']: text = content.get('text', '') if isinstance(content, dict) else str(content) if text: md_lines.append(text) md_lines.append("") elif elem_type in ['table', 'table_body']: table_captions = content.get('table_caption', []) if isinstance(table_captions, str): table_captions = [table_captions] if table_captions else [] for caption in table_captions: md_lines.append(f"**{caption}**") html = content.get('html', '') if html: md_lines.append(f"\n{html}\n") md_lines.append("") elif elem_type in ['image', 'image_body', 'figure']: img_filename = content.get('image_path', '') if img_filename: md_lines.append(f"![](images/{img_filename})") md_lines.append("") elif elem_type in ['interline_equation', 'inline_equation', 'equation']: latex = content.get('latex', '') if latex: md_lines.append(f"$$\n{latex}\n$$") md_lines.append("") elif elem_type in ['table_caption', 'table_footnote']: text = content.get('text', '') if isinstance(content, dict) else str(content) if text: # 表格标题加粗,表格脚注斜体 if elem_type == 'table_caption': md_lines.append(f"**{text}**") else: md_lines.append(f"*{text}*") md_lines.append("") elif elem_type in ['image_caption', 'image_footnote']: text = content.get('text', '') if isinstance(content, dict) else str(content) if text: # 图片标题加粗,图片脚注斜体 if elem_type == 'image_caption': md_lines.append(f"**{text}**") else: md_lines.append(f"*{text}*") md_lines.append("") elif elem_type == 'discarded': text = content.get('text', '') if isinstance(content, dict) else '' if text: md_lines.append(f"") md_lines.append("") # 处理丢弃元素 for element in page.get('discarded_blocks', []): content = element.get('content', {}) bbox = element.get('bbox', []) reading_order = element.get('reading_order', 0) original_category = element.get('original_category', 'unknown') md_lines.append(f"") text = content.get('text', '') if isinstance(content, dict) else '' if text: md_lines.append(f"") else: md_lines.append(f"") md_lines.append("") return '\n'.join(md_lines)