normalize_financial_numbers.py 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300
  1. import re
  2. import os
  3. from pathlib import Path
  4. def normalize_financial_numbers(text: str) -> str:
  5. """
  6. 标准化财务数字:将全角字符转换为半角字符
  7. Args:
  8. text: 原始文本
  9. Returns:
  10. 标准化后的文本
  11. """
  12. if not text:
  13. return text
  14. # 定义全角到半角的映射
  15. fullwidth_to_halfwidth = {
  16. '0': '0', '1': '1', '2': '2', '3': '3', '4': '4',
  17. '5': '5', '6': '6', '7': '7', '8': '8', '9': '9',
  18. ',': ',', # 全角逗号转半角逗号
  19. '。': '.', # 全角句号转半角句号
  20. '.': '.', # 全角句点转半角句点
  21. ':': ':', # 全角冒号转半角冒号
  22. ';': ';', # 全角分号转半角分号
  23. '(': '(', # 全角左括号转半角左括号
  24. ')': ')', # 全角右括号转半角右括号
  25. '-': '-', # 全角减号转半角减号
  26. '+': '+', # 全角加号转半角加号
  27. '%': '%', # 全角百分号转半角百分号
  28. }
  29. # 第一步:执行基础字符替换
  30. normalized_text = text
  31. for fullwidth, halfwidth in fullwidth_to_halfwidth.items():
  32. normalized_text = normalized_text.replace(fullwidth, halfwidth)
  33. # 第二步:处理数字序列中的空格和分隔符
  34. # 修改正则表达式以匹配完整的数字序列,包括空格
  35. # 匹配模式:数字 + (空格? + 逗号 + 空格? + 数字)* + (空格? + 小数点 + 数字+)?
  36. number_sequence_pattern = r'(\d+(?:\s*[,,]\s*\d+)*(?:\s*[。..]\s*\d+)?)'
  37. def normalize_number_sequence(match):
  38. sequence = match.group(1)
  39. # 处理千分位分隔符周围的空格
  40. # 将 "数字 + 空格 + 逗号 + 空格 + 数字" 标准化为 "数字,数字"
  41. sequence = re.sub(r'(\d)\s*[,,]\s*(\d)', r'\1,\2', sequence)
  42. # 处理小数点周围的空格
  43. # 将 "数字 + 空格 + 小数点 + 空格 + 数字" 标准化为 "数字.数字"
  44. sequence = re.sub(r'(\d)\s*[。..]\s*(\d)', r'\1.\2', sequence)
  45. return sequence
  46. normalized_text = re.sub(number_sequence_pattern, normalize_number_sequence, normalized_text)
  47. return normalized_text
  48. def normalize_markdown_table(markdown_content: str) -> str:
  49. """
  50. 专门处理Markdown表格中的数字标准化
  51. 注意:保留原始markdown中的换行符,只替换表格内的文本内容
  52. Args:
  53. markdown_content: Markdown内容
  54. Returns:
  55. 标准化后的Markdown内容
  56. """
  57. # 使用BeautifulSoup处理HTML表格
  58. from bs4 import BeautifulSoup, Tag
  59. import re
  60. # 使用正则表达式找到所有表格的位置,并保留其前后的内容
  61. # 匹配完整的HTML表格标签(包括嵌套)
  62. table_pattern = r'(<table[^>]*>.*?</table>)'
  63. def normalize_table_match(match):
  64. """处理单个表格匹配,保留原始格式"""
  65. table_html = match.group(1)
  66. original_table_html = table_html # 保存原始HTML用于比较
  67. # 解析表格HTML
  68. soup = BeautifulSoup(table_html, 'html.parser')
  69. tables = soup.find_all('table')
  70. # 记录所有需要替换的文本(原始文本 -> 标准化文本)
  71. replacements = []
  72. for table in tables:
  73. if isinstance(table, Tag):
  74. cells = table.find_all(['td', 'th'])
  75. for cell in cells:
  76. if isinstance(cell, Tag):
  77. # 获取单元格的纯文本内容
  78. original_text = cell.get_text()
  79. normalized_text = normalize_financial_numbers(original_text)
  80. # 如果内容发生了变化,记录替换
  81. if original_text != normalized_text:
  82. # 找到单元格中所有文本节点并替换
  83. from bs4.element import NavigableString
  84. for text_node in cell.find_all(string=True, recursive=True):
  85. if isinstance(text_node, NavigableString):
  86. text_str = str(text_node)
  87. if text_str.strip():
  88. normalized = normalize_financial_numbers(text_str.strip())
  89. if normalized != text_str.strip():
  90. # 保留原始文本节点的前后空白
  91. if text_str.strip() == text_str:
  92. # 纯文本节点,直接替换
  93. text_node.replace_with(normalized)
  94. else:
  95. # 有前后空白,需要保留
  96. leading_ws = text_str[:len(text_str) - len(text_str.lstrip())]
  97. trailing_ws = text_str[len(text_str.rstrip()):]
  98. text_node.replace_with(leading_ws + normalized + trailing_ws)
  99. # 获取修改后的HTML
  100. modified_html = str(soup)
  101. # 如果内容没有变化,返回原始HTML(保持原始格式)
  102. # 检查是否只是格式变化(换行、空格等)
  103. original_text_only = re.sub(r'\s+', '', original_table_html)
  104. modified_text_only = re.sub(r'\s+', '', modified_html)
  105. if original_text_only == modified_text_only:
  106. # 只有格式变化,返回原始HTML以保留换行符
  107. return original_table_html
  108. # 有实际内容变化,返回修改后的HTML
  109. return modified_html
  110. # 使用正则替换,只替换表格内容,保留其他部分(包括换行符)不变
  111. normalized_content = re.sub(table_pattern, normalize_table_match, markdown_content, flags=re.DOTALL)
  112. return normalized_content
  113. def normalize_json_table(json_content: str) -> str:
  114. """
  115. 专门处理JSON格式OCR结果中表格的数字标准化
  116. Args:
  117. json_content: JSON格式的OCR结果内容
  118. Returns:
  119. 标准化后的JSON内容
  120. """
  121. """
  122. json_content 示例:
  123. [
  124. {
  125. "category": "Table",
  126. "text": "<table>...</table>"
  127. },
  128. {
  129. "category": "Text",
  130. "text": "Some other text"
  131. }
  132. ]
  133. """
  134. import json
  135. try:
  136. # 解析JSON内容
  137. data = json.loads(json_content) if isinstance(json_content, str) else json_content
  138. # 确保data是列表格式
  139. if not isinstance(data, list):
  140. return json_content
  141. # 遍历所有OCR结果项
  142. for item in data:
  143. if not isinstance(item, dict):
  144. continue
  145. # 检查是否是表格类型
  146. if item.get('category') == 'Table' and 'text' in item:
  147. table_html = item['text']
  148. # 使用BeautifulSoup处理HTML表格
  149. from bs4 import BeautifulSoup, Tag
  150. soup = BeautifulSoup(table_html, 'html.parser')
  151. tables = soup.find_all('table')
  152. for table in tables:
  153. if isinstance(table, Tag):
  154. cells = table.find_all(['td', 'th'])
  155. for cell in cells:
  156. if isinstance(cell, Tag):
  157. original_text = cell.get_text()
  158. # 应用数字标准化
  159. normalized_text = normalize_financial_numbers(original_text)
  160. # 如果内容发生了变化,更新单元格内容
  161. if original_text != normalized_text:
  162. cell.string = normalized_text
  163. # 更新item中的表格内容
  164. item['text'] = str(soup)
  165. # 同时标准化普通文本中的数字(如果需要)
  166. # elif 'text' in item:
  167. # original_text = item['text']
  168. # normalized_text = normalize_financial_numbers(original_text)
  169. # if original_text != normalized_text:
  170. # item['text'] = normalized_text
  171. # 返回标准化后的JSON字符串
  172. return json.dumps(data, ensure_ascii=False, indent=2)
  173. except json.JSONDecodeError as e:
  174. print(f"⚠️ JSON解析失败: {e}")
  175. return json_content
  176. except Exception as e:
  177. print(f"⚠️ JSON表格标准化失败: {e}")
  178. return json_content
  179. def normalize_json_file(file_path: str, output_path: str | None = None) -> str:
  180. """
  181. 标准化JSON文件中的表格数字
  182. Args:
  183. file_path: 输入JSON文件路径
  184. output_path: 输出文件路径,如果为None则覆盖原文件
  185. Returns:
  186. 标准化后的JSON内容
  187. """
  188. input_file = Path(file_path)
  189. output_file = Path(output_path) if output_path else input_file
  190. if not input_file.exists():
  191. raise FileNotFoundError(f"找不到文件: {file_path}")
  192. # 读取原始JSON文件
  193. with open(input_file, 'r', encoding='utf-8') as f:
  194. original_content = f.read()
  195. print(f"🔧 正在标准化JSON文件: {input_file.name}")
  196. # 标准化内容
  197. normalized_content = normalize_json_table(original_content)
  198. # 保存标准化后的文件
  199. with open(output_file, 'w', encoding='utf-8') as f:
  200. f.write(normalized_content)
  201. # 统计变化
  202. changes = sum(1 for o, n in zip(original_content, normalized_content) if o != n)
  203. if changes > 0:
  204. print(f"✅ 标准化了 {changes} 个字符")
  205. # 如果输出路径不同,也保存原始版本
  206. if output_path and output_path != file_path:
  207. original_backup = Path(output_path).parent / f"{Path(output_path).stem}_original.json"
  208. with open(original_backup, 'w', encoding='utf-8') as f:
  209. f.write(original_content)
  210. print(f"📄 原始版本已保存到: {original_backup}")
  211. else:
  212. print("ℹ️ 无需标准化(已是标准格式)")
  213. print(f"📄 标准化结果已保存到: {output_file}")
  214. return normalized_content
  215. if __name__ == "__main__":
  216. # 简单测试
  217. test_strings = [
  218. "28, 239, 305.48",
  219. "2023年净利润为28,239,305.48元",
  220. "总资产为1,234,567.89元",
  221. "负债总额为500,000.00元",
  222. "收入增长了10.5%,达到1,200,000元",
  223. "费用为300,000元",
  224. "利润率为15.2%",
  225. "现金流量为-50,000元",
  226. "股东权益为2,500,000.00元",
  227. "每股收益为3.25元",
  228. "市盈率为20.5倍",
  229. "营业收入为750,000元",
  230. "净资产收益率为12.3%",
  231. "总负债为1,200,000元",
  232. "流动比率为1.5倍",
  233. "速动比率为1.2倍",
  234. "资产负债率为40%",
  235. "存货周转率为6次/年",
  236. "应收账款周转率为8次/年",
  237. "固定资产周转率为2次/年",
  238. "总资产周转率为1.2次/年",
  239. "经营活动产生的现金流量净额为200,000元"
  240. ]
  241. for s in test_strings:
  242. print("原始: ", s)
  243. print("标准化: ", normalize_financial_numbers(s))
  244. print("-" * 50)