ocr_mkcontent.py 14 KB

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