ocr_mkcontent.py 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301
  1. import re
  2. from loguru import logger
  3. from magic_pdf.config.make_content_config import DropMode, MakeMode
  4. from magic_pdf.config.ocr_content_type import BlockType, ContentType
  5. from magic_pdf.libs.commons import join_path
  6. from magic_pdf.libs.language import detect_lang
  7. from magic_pdf.libs.markdown_utils import ocr_escape_special_markdown_char
  8. from magic_pdf.para.para_split_v3 import ListLineTag
  9. def __is_hyphen_at_line_end(line):
  10. """Check if a line ends with one or more letters followed by a hyphen.
  11. Args:
  12. line (str): The line of text to check.
  13. Returns:
  14. bool: True if the line ends with one or more letters followed by a hyphen, False otherwise.
  15. """
  16. # Use regex to check if the line ends with one or more letters followed by a hyphen
  17. return bool(re.search(r'[A-Za-z]+-\s*$', line))
  18. def ocr_mk_mm_markdown_with_para_and_pagination(pdf_info_dict: list,
  19. img_buket_path):
  20. markdown_with_para_and_pagination = []
  21. page_no = 0
  22. for page_info in pdf_info_dict:
  23. paras_of_layout = page_info.get('para_blocks')
  24. if not paras_of_layout:
  25. markdown_with_para_and_pagination.append({
  26. 'page_no':
  27. page_no,
  28. 'md_content':
  29. '',
  30. })
  31. page_no += 1
  32. continue
  33. page_markdown = ocr_mk_markdown_with_para_core_v2(
  34. paras_of_layout, 'mm', img_buket_path)
  35. markdown_with_para_and_pagination.append({
  36. 'page_no':
  37. page_no,
  38. 'md_content':
  39. '\n\n'.join(page_markdown)
  40. })
  41. page_no += 1
  42. return markdown_with_para_and_pagination
  43. def ocr_mk_markdown_with_para_core_v2(paras_of_layout,
  44. mode,
  45. img_buket_path='',
  46. ):
  47. page_markdown = []
  48. for para_block in paras_of_layout:
  49. para_text = ''
  50. para_type = para_block['type']
  51. if para_type in [BlockType.Text, BlockType.List, BlockType.Index]:
  52. para_text = merge_para_with_text(para_block)
  53. elif para_type == BlockType.Title:
  54. para_text = f'# {merge_para_with_text(para_block)}'
  55. elif para_type == BlockType.InterlineEquation:
  56. para_text = merge_para_with_text(para_block)
  57. elif para_type == BlockType.Image:
  58. if mode == 'nlp':
  59. continue
  60. elif mode == 'mm':
  61. for block in para_block['blocks']: # 1st.拼image_body
  62. if block['type'] == BlockType.ImageBody:
  63. for line in block['lines']:
  64. for span in line['spans']:
  65. if span['type'] == ContentType.Image:
  66. if span.get('image_path', ''):
  67. para_text += f"\n![]({join_path(img_buket_path, span['image_path'])}) \n"
  68. for block in para_block['blocks']: # 2nd.拼image_caption
  69. if block['type'] == BlockType.ImageCaption:
  70. para_text += merge_para_with_text(block) + ' \n'
  71. for block in para_block['blocks']: # 3rd.拼image_footnote
  72. if block['type'] == BlockType.ImageFootnote:
  73. para_text += merge_para_with_text(block) + ' \n'
  74. elif para_type == BlockType.Table:
  75. if mode == 'nlp':
  76. continue
  77. elif mode == 'mm':
  78. for block in para_block['blocks']: # 1st.拼table_caption
  79. if block['type'] == BlockType.TableCaption:
  80. para_text += merge_para_with_text(block) + ' \n'
  81. for block in para_block['blocks']: # 2nd.拼table_body
  82. if block['type'] == BlockType.TableBody:
  83. for line in block['lines']:
  84. for span in line['spans']:
  85. if span['type'] == ContentType.Table:
  86. # if processed by table model
  87. if span.get('latex', ''):
  88. para_text += f"\n\n$\n {span['latex']}\n$\n\n"
  89. elif span.get('html', ''):
  90. para_text += f"\n\n{span['html']}\n\n"
  91. elif span.get('image_path', ''):
  92. para_text += f"\n![]({join_path(img_buket_path, span['image_path'])}) \n"
  93. for block in para_block['blocks']: # 3rd.拼table_footnote
  94. if block['type'] == BlockType.TableFootnote:
  95. para_text += merge_para_with_text(block) + ' \n'
  96. if para_text.strip() == '':
  97. continue
  98. else:
  99. page_markdown.append(para_text.strip() + ' ')
  100. return page_markdown
  101. def detect_language(text):
  102. en_pattern = r'[a-zA-Z]+'
  103. en_matches = re.findall(en_pattern, text)
  104. en_length = sum(len(match) for match in en_matches)
  105. if len(text) > 0:
  106. if en_length / len(text) >= 0.5:
  107. return 'en'
  108. else:
  109. return 'unknown'
  110. else:
  111. return 'empty'
  112. # 连写字符拆分
  113. def __replace_ligatures(text: str):
  114. text = re.sub(r'fi', 'fi', text) # 替换 fi 连写符
  115. text = re.sub(r'fl', 'fl', text) # 替换 fl 连写符
  116. text = re.sub(r'ff', 'ff', text) # 替换 ff 连写符
  117. text = re.sub(r'ffi', 'ffi', text) # 替换 ffi 连写符
  118. text = re.sub(r'ffl', 'ffl', text) # 替换 ffl 连写符
  119. return text
  120. def merge_para_with_text(para_block):
  121. block_text = ''
  122. for line in para_block['lines']:
  123. for span in line['spans']:
  124. if span['type'] in [ContentType.Text]:
  125. block_text += span['content']
  126. block_lang = detect_lang(block_text)
  127. para_text = ''
  128. for i, line in enumerate(para_block['lines']):
  129. if i >= 1 and line.get(ListLineTag.IS_LIST_START_LINE, False):
  130. para_text += ' \n'
  131. for j, span in enumerate(line['spans']):
  132. span_type = span['type']
  133. content = ''
  134. if span_type == ContentType.Text:
  135. content = ocr_escape_special_markdown_char(span['content'])
  136. elif span_type == ContentType.InlineEquation:
  137. content = f"${span['content']}$"
  138. elif span_type == ContentType.InterlineEquation:
  139. content = f"\n$$\n{span['content']}\n$$\n"
  140. content = content.strip()
  141. if content:
  142. langs = ['zh', 'ja', 'ko']
  143. # logger.info(f'block_lang: {block_lang}, content: {content}')
  144. if block_lang in langs: # 中文/日语/韩文语境下,换行不需要空格分隔
  145. if j == len(line['spans']) - 1:
  146. para_text += content
  147. else:
  148. para_text += f'{content} '
  149. else:
  150. if span_type in [ContentType.Text, ContentType.InlineEquation]:
  151. # 如果span是line的最后一个且末尾带有-连字符,那么末尾不应该加空格,同时应该把-删除
  152. if j == len(line['spans'])-1 and span_type == ContentType.Text and __is_hyphen_at_line_end(content):
  153. para_text += content[:-1]
  154. else: # 西方文本语境下 content间需要空格分隔
  155. para_text += f'{content} '
  156. elif span_type == ContentType.InterlineEquation:
  157. para_text += content
  158. else:
  159. continue
  160. # 连写字符拆分
  161. # para_text = __replace_ligatures(para_text)
  162. return para_text
  163. def para_to_standard_format_v2(para_block, img_buket_path, page_idx, drop_reason=None):
  164. para_type = para_block['type']
  165. para_content = {}
  166. if para_type in [BlockType.Text, BlockType.List, BlockType.Index]:
  167. para_content = {
  168. 'type': 'text',
  169. 'text': merge_para_with_text(para_block),
  170. }
  171. elif para_type == BlockType.Title:
  172. para_content = {
  173. 'type': 'text',
  174. 'text': merge_para_with_text(para_block),
  175. 'text_level': 1,
  176. }
  177. elif para_type == BlockType.InterlineEquation:
  178. para_content = {
  179. 'type': 'equation',
  180. 'text': merge_para_with_text(para_block),
  181. 'text_format': 'latex',
  182. }
  183. elif para_type == BlockType.Image:
  184. para_content = {'type': 'image', 'img_path': '', 'img_caption': [], 'img_footnote': []}
  185. for block in para_block['blocks']:
  186. if block['type'] == BlockType.ImageBody:
  187. for line in block['lines']:
  188. for span in line['spans']:
  189. if span['type'] == ContentType.Image:
  190. if span.get('image_path', ''):
  191. para_content['img_path'] = join_path(img_buket_path, span['image_path'])
  192. if block['type'] == BlockType.ImageCaption:
  193. para_content['img_caption'].append(merge_para_with_text(block))
  194. if block['type'] == BlockType.ImageFootnote:
  195. para_content['img_footnote'].append(merge_para_with_text(block))
  196. elif para_type == BlockType.Table:
  197. para_content = {'type': 'table', 'img_path': '', 'table_caption': [], 'table_footnote': []}
  198. for block in para_block['blocks']:
  199. if block['type'] == BlockType.TableBody:
  200. for line in block['lines']:
  201. for span in line['spans']:
  202. if span['type'] == ContentType.Table:
  203. if span.get('latex', ''):
  204. para_content['table_body'] = f"\n\n$\n {span['latex']}\n$\n\n"
  205. elif span.get('html', ''):
  206. para_content['table_body'] = f"\n\n{span['html']}\n\n"
  207. if span.get('image_path', ''):
  208. para_content['img_path'] = join_path(img_buket_path, span['image_path'])
  209. if block['type'] == BlockType.TableCaption:
  210. para_content['table_caption'].append(merge_para_with_text(block))
  211. if block['type'] == BlockType.TableFootnote:
  212. para_content['table_footnote'].append(merge_para_with_text(block))
  213. para_content['page_idx'] = page_idx
  214. if drop_reason is not None:
  215. para_content['drop_reason'] = drop_reason
  216. return para_content
  217. def union_make(pdf_info_dict: list,
  218. make_mode: str,
  219. drop_mode: str,
  220. img_buket_path: str = '',
  221. ):
  222. output_content = []
  223. for page_info in pdf_info_dict:
  224. drop_reason_flag = False
  225. drop_reason = None
  226. if page_info.get('need_drop', False):
  227. drop_reason = page_info.get('drop_reason')
  228. if drop_mode == DropMode.NONE:
  229. pass
  230. elif drop_mode == DropMode.NONE_WITH_REASON:
  231. drop_reason_flag = True
  232. elif drop_mode == DropMode.WHOLE_PDF:
  233. raise Exception((f'drop_mode is {DropMode.WHOLE_PDF} ,'
  234. f'drop_reason is {drop_reason}'))
  235. elif drop_mode == DropMode.SINGLE_PAGE:
  236. logger.warning((f'drop_mode is {DropMode.SINGLE_PAGE} ,'
  237. f'drop_reason is {drop_reason}'))
  238. continue
  239. else:
  240. raise Exception('drop_mode can not be null')
  241. paras_of_layout = page_info.get('para_blocks')
  242. page_idx = page_info.get('page_idx')
  243. if not paras_of_layout:
  244. continue
  245. if make_mode == MakeMode.MM_MD:
  246. page_markdown = ocr_mk_markdown_with_para_core_v2(
  247. paras_of_layout, 'mm', img_buket_path)
  248. output_content.extend(page_markdown)
  249. elif make_mode == MakeMode.NLP_MD:
  250. page_markdown = ocr_mk_markdown_with_para_core_v2(
  251. paras_of_layout, 'nlp')
  252. output_content.extend(page_markdown)
  253. elif make_mode == MakeMode.STANDARD_FORMAT:
  254. for para_block in paras_of_layout:
  255. if drop_reason_flag:
  256. para_content = para_to_standard_format_v2(
  257. para_block, img_buket_path, page_idx)
  258. else:
  259. para_content = para_to_standard_format_v2(
  260. para_block, img_buket_path, page_idx)
  261. output_content.append(para_content)
  262. if make_mode in [MakeMode.MM_MD, MakeMode.NLP_MD]:
  263. return '\n\n'.join(output_content)
  264. elif make_mode == MakeMode.STANDARD_FORMAT:
  265. return output_content