utils.py 3.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120
  1. """
  2. utils.py - 工具函数模块
  3. 提供日志配置、文本分块和表格转换等通用工具函数。
  4. """
  5. import logging
  6. import sys
  7. logger = logging.getLogger(__name__)
  8. def setup_logging(verbose: bool = False):
  9. """配置日志格式。
  10. - verbose=True: DEBUG级别,显示文件名行号
  11. - verbose=False: INFO级别,简洁格式
  12. Args:
  13. verbose: 是否开启详细日志模式
  14. """
  15. level = logging.DEBUG if verbose else logging.INFO
  16. fmt = (
  17. "%(asctime)s [%(levelname)s] %(name)s (%(filename)s:%(lineno)d): %(message)s"
  18. if verbose
  19. else "[%(levelname)s] %(message)s"
  20. )
  21. logging.basicConfig(
  22. level=level,
  23. format=fmt,
  24. stream=sys.stderr,
  25. force=True,
  26. )
  27. def chunk_text(text: str, max_chars: int = 4000) -> list[str]:
  28. """将长文本按最大字符数切分,尽量在段落边界处切分。
  29. Args:
  30. text: 待切分的文本
  31. max_chars: 每个块的最大字符数(默认 4000)
  32. Returns:
  33. 切分后的文本块列表
  34. """
  35. if not text:
  36. logger.debug("chunk_text 收到空文本,返回空列表")
  37. return []
  38. if max_chars <= 0:
  39. logger.warning("max_chars=%d 无效,使用默认值 4000", max_chars)
  40. max_chars = 4000
  41. if len(text) <= max_chars:
  42. logger.debug("文本长度 %d <= %d,无需切分", len(text), max_chars)
  43. return [text]
  44. chunks: list[str] = []
  45. start = 0
  46. while start < len(text):
  47. # 如果剩余文本不足 max_chars,直接取剩余全部
  48. if start + max_chars >= len(text):
  49. chunks.append(text[start:])
  50. break
  51. # 在当前块末尾附近寻找段落边界(换行符)
  52. end = start + max_chars
  53. # 在 [start+max_chars//2, start+max_chars] 范围内向前找最后一个换行符
  54. search_start = max(start + max_chars // 2, start)
  55. boundary = text.rfind("\n", search_start, end)
  56. if boundary <= start:
  57. # 找不到合适换行符,在最大范围内向后找
  58. boundary = text.find("\n", end)
  59. if boundary == -1 or boundary >= start + int(max_chars * 1.5):
  60. # 仍然找不到,直接在 max_chars 处切分
  61. boundary = end
  62. chunk = text[start:boundary].strip()
  63. if chunk:
  64. chunks.append(chunk)
  65. start = boundary + 1 # 跳过换行符
  66. logger.debug("文本已切分为 %d 个块", len(chunks))
  67. return chunks
  68. def table_to_markdown(table_info) -> str:
  69. """将 pdf_table_to_docx.table_parser.TableInfo 转换为 Markdown 表格格式。
  70. 用于将表格内容序列化为文本,方便传给 AI。
  71. Args:
  72. table_info: 来自 pdf_table_to_docx 的 TableInfo 对象
  73. Returns:
  74. Markdown 格式的表格字符串
  75. """
  76. # 输入校验
  77. if table_info is None:
  78. logger.warning("table_to_markdown 收到 None 输入")
  79. return "(空表格)"
  80. if not hasattr(table_info, "cells"):
  81. logger.warning("table_to_markdown 收到的对象没有 cells 属性")
  82. return "(空表格)"
  83. rows = table_info.cells # list[list[CellInfo]]
  84. if not rows:
  85. logger.debug("table_to_markdown: 空表格(无行)")
  86. return "(空表格)"
  87. lines: list[str] = []
  88. for row_idx, row in enumerate(rows):
  89. # 提取该行每个单元格的文本,去除首尾空白
  90. cell_texts = [cell.text.strip() if hasattr(cell, "text") else str(cell).strip() for cell in row]
  91. # 用 | 分隔单元格
  92. lines.append("| " + " | ".join(cell_texts) + " |")
  93. # 在第一行之后添加分隔行(表头分隔符)
  94. if row_idx == 0:
  95. lines.append("|" + "|".join("---" for _ in row) + "|")
  96. return "\n".join(lines)