report_utils.py 10 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319
  1. # -*- coding: utf-8 -*-
  2. """
  3. 审查报告生成工具模块
  4. 功能:
  5. - 将 Markdown 格式的审查报告转换为 Word (.docx) 文档
  6. - 使用 pypandoc + 自定义样式模板(data/template.docx)
  7. - 生成静态文件下载 URL
  8. 工具列表:
  9. - convert_markdown_to_docx: Markdown → Word 文档转换
  10. - get_docx_download_url: 生成 Word 文档的下载 URL
  11. """
  12. import os
  13. import re
  14. import uuid
  15. from pathlib import Path
  16. # ============================================================
  17. # 配置常量
  18. # ============================================================
  19. # 项目根目录
  20. PROJECT_ROOT = Path(__file__).parent.parent
  21. # 样式模板路径
  22. TEMPLATE_FILENAME = "template.docx"
  23. TEMPLATE_PATH = str(PROJECT_ROOT / "data" / TEMPLATE_FILENAME)
  24. # 报告输出目录(StaticFiles 挂载的静态目录)
  25. REPORT_OUTPUT_DIR = PROJECT_ROOT / "static" / "reports"
  26. # 服务器地址(读取 .env,无配置则用 localhost)
  27. def _get_server_host() -> str:
  28. """从 .env 读取 SERVER_HOST,默认 localhost:8000。"""
  29. from dotenv import dotenv_values
  30. cfg = dotenv_values(str(PROJECT_ROOT / ".env"))
  31. host = cfg.get("HOST", "0.0.0.0")
  32. port = cfg.get("PORT", "8000")
  33. # 对外暴露用 localhost 或配置的域名
  34. return cfg.get("SERVER_HOST", f"http://{host}:{port}")
  35. SERVER_HOST = _get_server_host()
  36. # ============================================================
  37. # Markdown → Word 转换
  38. # ============================================================
  39. def _normalize_markdown(md_content: str) -> str:
  40. """规范化 Markdown 内容,修复常见的 LLM 输出格式问题。
  41. 修复以下问题:
  42. 1. 标题行首有空格 → 去除(保证 ## 顶格)
  43. 2. `##` 与标题文字之间缺少空格 → 补空格
  44. 3. 标题前缺少空行 → 补空行
  45. 4. 以 `**##` 或 `**###` 开头的伪标题 → 转为正常标题
  46. 5. 全角 `#` 替换为半角 `#`
  47. 注意:不修改表格内部内容,避免破坏正确的 markdown 表格格式。
  48. Args:
  49. md_content: 原始 Markdown 内容
  50. Returns:
  51. str: 规范化后的 Markdown 内容
  52. """
  53. import re
  54. lines = md_content.split("\n")
  55. result = []
  56. in_table = False # 跟踪是否在表格内
  57. for i, line in enumerate(lines):
  58. # 检测表格边界
  59. stripped_line = line.strip()
  60. if stripped_line.startswith("|") and stripped_line.endswith("|"):
  61. in_table = True
  62. elif stripped_line.startswith("|") and "---" in stripped_line:
  63. in_table = True # 表格分隔行
  64. elif in_table and not stripped_line.startswith("|"):
  65. in_table = False
  66. # 1. 全角 # 替换为半角
  67. line = line.replace("\uff03", "#")
  68. # 表格内的行:跳过标题规范化处理,原样保留
  69. if in_table:
  70. result.append(line)
  71. continue
  72. # 2. 标题行首去空格 + 补空格:检测 ## 开头的行
  73. stripped = line.lstrip()
  74. # 匹配: 行首(可能有空格)的 # 标记,后面可选空格 + 文字
  75. if re.match(r"^#{1,6}(\s|$)", stripped) or re.match(r"^#{1,6}[^\s#]", stripped):
  76. # 确保 # 后面有空格(如 "##标题" → "## 标题")
  77. fixed = re.sub(r"^(#{1,6})\s*", r"\1 ", stripped)
  78. fixed = fixed.rstrip()
  79. # 确保标题前有空行(避免标题粘在上文后)
  80. if result and result[-1].strip() != "":
  81. result.append("")
  82. result.append(fixed)
  83. continue
  84. # 3. 检测 **## 标题** 这类伪标题 → 转为 ## 标题
  85. bold_header = re.match(r"^\*\*(#{1,6})\s+(.+?)\*\*\s*$", line)
  86. if bold_header:
  87. fixed = f"{bold_header.group(1)} {bold_header.group(2).rstrip('*').strip()}"
  88. if result and result[-1].strip() != "":
  89. result.append("")
  90. result.append(fixed)
  91. continue
  92. # 4. 检测 **### 标题** 变体
  93. bold_header2 = re.match(r"^\*\*(#{1,6})\s*(.+?)\*\*\s*$", line)
  94. if bold_header2:
  95. fixed = f"{bold_header2.group(1)} {bold_header2.group(2).rstrip('*').strip()}"
  96. if result and result[-1].strip() != "":
  97. result.append("")
  98. result.append(fixed)
  99. continue
  100. result.append(line)
  101. return "\n".join(result)
  102. def convert_markdown_to_docx(md_content: str, save_path: str) -> None:
  103. """将 Markdown 内容转换为 Word (.docx) 文档。
  104. 使用 pypandoc 进行转换,应用 data/template.docx 作为样式模板。
  105. 首次使用时会自动下载 pandoc。
  106. Args:
  107. md_content: Markdown 格式的报告内容
  108. save_path: 输出 .docx 文件的完整路径
  109. Raises:
  110. FileNotFoundError: 样式模板不存在
  111. RuntimeError: pandoc 转换失败
  112. """
  113. try:
  114. import pypandoc
  115. except ImportError:
  116. raise ImportError(
  117. "pypandoc 未安装。请执行: pip install pypandoc"
  118. )
  119. # 自动检测并下载 pandoc
  120. try:
  121. pypandoc.get_pandoc_path()
  122. except OSError:
  123. print("[报告] 未检测到 pandoc,正在自动下载...")
  124. pypandoc.download_pandoc()
  125. # 清理内容
  126. report_content = md_content.strip()
  127. if not report_content:
  128. raise ValueError("报告内容为空,无法生成文档")
  129. # 规范化 Markdown 格式(修复 LLM 输出中常见的排版问题)
  130. # report_content = _normalize_markdown(report_content)
  131. print("[报告] Markdown 规范化完成")
  132. # 检查模板
  133. if not os.path.exists(TEMPLATE_PATH):
  134. raise FileNotFoundError(
  135. f"样式模板不存在!路径:{TEMPLATE_PATH}\n"
  136. f"请将模板文件放置到 data/template.docx"
  137. )
  138. # 确保输出目录存在
  139. os.makedirs(os.path.dirname(save_path), exist_ok=True)
  140. # pandoc 转换参数
  141. extra_args = [
  142. "--standalone",
  143. f"--reference-doc={TEMPLATE_PATH}",
  144. "--from=gfm", # GitHub Flavored Markdown,表格支持更好
  145. "--wrap=none", # 禁止自动换行,保持表格单元格完整
  146. "--columns=120", # 宽列宽,避免窄列强制断行
  147. ]
  148. try:
  149. pypandoc.convert_text(
  150. source=report_content,
  151. format="md",
  152. to="docx",
  153. outputfile=save_path,
  154. extra_args=extra_args,
  155. encoding="utf-8",
  156. )
  157. print(f"[报告] ✅ 转换成功!文件路径:{save_path}")
  158. except Exception as e:
  159. raise RuntimeError(f"pandoc 转换失败: {e}")
  160. # ============================================================
  161. # 下载 URL 生成
  162. # ============================================================
  163. def get_docx_download_url(word_filename: str) -> str:
  164. """生成 Word 文档的下载 URL。
  165. Args:
  166. word_filename: Word 文档文件名(不含路径),如 "report_abc123.docx"
  167. Returns:
  168. str: 完整下载 URL,如 "http://localhost:8000/static/reports/report_abc123.docx"
  169. """
  170. return f"{SERVER_HOST}/static/reports/{word_filename}"
  171. def generate_report_filename(title: str = None) -> str:
  172. """生成审查报告文件名。
  173. Args:
  174. title: 报告标题(如煤矿名称),为空时使用默认名称
  175. Returns:
  176. str: 文件名,如 "XX煤矿配风计划审查报告_20260713_150132.docx"
  177. """
  178. from datetime import datetime
  179. date_str = datetime.now().strftime("%Y%m%d_%H%M%S")
  180. if title:
  181. # 清理标题中的非法文件名字符
  182. safe_title = re.sub(r'[\\/:*?"<>|]', '', title).strip()
  183. if safe_title:
  184. return f"{safe_title}配风计划审查报告_{date_str}.docx"
  185. return f"配风计划审查报告_{date_str}.docx"
  186. def extract_mine_name(text: str) -> str | None:
  187. """从 PDF 文本中提取煤矿/公司名称。
  188. 在文本前 5000 字中查找含"煤矿/煤业/矿业/能源/公司"的行,
  189. 提取其中最可能的企业名称。
  190. Args:
  191. text: PDF 提取的文本内容
  192. Returns:
  193. str | None: 提取到的名称,未找到返回 None
  194. """
  195. import re
  196. head = text[:5000] if len(text) > 5000 else text
  197. # 先找含关键字的行(每行独立匹配,避免跨行粘连)
  198. lines = head.split('\n')
  199. candidates = []
  200. for line in lines:
  201. line = line.strip()
  202. if not line or len(line) < 4:
  203. continue
  204. # 匹配模式:任意前缀 + 煤矿/煤业/矿业 + 可选后缀
  205. m = re.search(
  206. r'([\u4e00-\u9fa5\w()()\u3000]+?'
  207. r'(?:煤矿|煤业|矿业|能源)'
  208. r'(?:有限责任公司|股份有限公司|集团有限公司|有限公司|集团公司|集团|公司)?)',
  209. line
  210. )
  211. if m:
  212. name = m.group(1).strip()
  213. if len(name) >= 4:
  214. candidates.append(name)
  215. # 匹配 XX矿(非矿山/矿区/矿井/矿务局/矿长)
  216. m2 = re.search(
  217. r'([\u4e00-\u9fa5\w()()\u3000]{4,20}矿)'
  218. r'(?!山|区|井|业|务|长|灯|车|石|泥|泉水)',
  219. line
  220. )
  221. if m2:
  222. name = m2.group(1).strip()
  223. if len(name) >= 4 and name not in candidates:
  224. candidates.append(name)
  225. if not candidates:
  226. return None
  227. # 返回最长匹配(通常最完整)
  228. candidates.sort(key=len, reverse=True)
  229. best = candidates[0]
  230. # 清理末尾多余字符
  231. best = re.sub(r'[,。;:、!?\s]+$', '', best)
  232. # 如果以"公司"结尾且前面有"有限/责任/集团",确保完整性
  233. # 如果名称太长(>30字),截取到最后一个关键字处
  234. if len(best) > 40:
  235. # 尝试截断到 煤矿/煤业/矿业/公司 处
  236. trunc = re.match(r'(.{1,30}(?:煤矿|煤业|矿业|集团|公司))', best)
  237. if trunc:
  238. best = trunc.group(1)
  239. return best if len(best) >= 4 else None
  240. def save_review_report(md_content: str, title: str = None) -> tuple[str, str]:
  241. """保存审查报告为 Word 文档,返回 (文件路径, 下载URL)。
  242. 一站式函数:生成文件名 → 转换 docx → 返回路径和下载链接。
  243. Args:
  244. md_content: Markdown 格式的审查报告
  245. title: 报告标题(煤矿名称),用于生成文件名
  246. Returns:
  247. tuple[str, str]: (本地文件路径, 下载URL)
  248. """
  249. filename = generate_report_filename(title)
  250. save_path = str(REPORT_OUTPUT_DIR / filename)
  251. convert_markdown_to_docx(md_content, save_path)
  252. download_url = get_docx_download_url(filename)
  253. return save_path, download_url