Compare commits

...

3 Commits

Author SHA1 Message Date
cxs
dad699a157 表格问题修复 2025-05-21 11:26:55 +08:00
cxs
dfa2b47b11 清洗后的图片问题 2025-05-21 10:57:19 +08:00
cxs
f9ab2ffce0 表格提取 2025-05-20 19:21:58 +08:00
7 changed files with 2516 additions and 26 deletions

136
README.md
View File

@@ -48,6 +48,129 @@ pip install -r requirements.txt
## 最近更新
### 2024年6月15日
- **移除文档中的图片标题**
- 移除了清洗后文档中的"图1"、"图2"等图片标题
- 保留图片在文档中的原始位置和显示
- 简化文档结构,使文档更加简洁
- 优化图片处理流程,确保只保留图片内容
- 保持文本输出中的图片引用标记,但不再显示编号
- 更接近于用户预期的输出效果
### 2024年6月14日
- **优化图片处理,保持原始位置且直接嵌入**
- 改进图片处理逻辑,保持图片在原始文档中的相对位置
- 不再生成外部images目录直接将图片嵌入到清洗后的文档中
- 智能识别原始文档中图片与文本的关联关系,确保图片插入的位置更合理
- 直接从内存中的图片数据创建图片对象,提高处理速度
- 简化图片说明caption格式不再显示文件名只保留编号信息
- 针对无法确定原始位置的图片,统一添加到文档末尾
- 图片处理过程更加稳定,避免因外部文件操作导致的错误
- 提升用户体验,文档外观更加接近原始文档
### 2024年6月13日
- **修复清洗后Word文档图片丢失问题**
- 增加了图片提取和保存功能确保清洗后的Word文档保留原始图片
- 使用多种方法提取文档中的图片支持多种图片格式PNG、JPG、GIF、BMP等
- 智能过滤无效和过小的图片,只保留有意义的内容
- 自动在段落间均匀分布图片,保持文档的可读性和美观性
- 为每张图片添加标题和编号,方便引用
- 所有图片保存在独立的images目录便于管理和查看
- 在文本输出中添加图片引用标记,保持文档内容的完整性
- 增强文档处理流程的稳定性,防止因图片处理错误导致的中断
### 2024年6月12日
- **TXT文件表格输出HTML标签**
- 改进表格处理逻辑使TXT文件中的表格也能以HTML标签形式输出
- 保持与Word文档中的表格输出格式一致提供结构化的表格内容
- 优化TXT文件处理流程正确保留HTML标签而不是转换为纯文本
- 确保表格的合并单元格属性和结构信息在TXT文件中也能被完整保留
- 保持HTML标签在TXT文件中的原始格式不进行换行符替换
- 改进文本合并逻辑智能区分普通文本和HTML标签内容
- 简化用户使用流程,无需额外操作即可获得格式一致的输出文件
- 提高系统处理效率和文档格式统一性
### 2024年6月11日
- **同时支持HTML标签显示和HTML文件生成**
- 优化表格处理功能,兼顾多种输出需求
- 移除Word文档中自动添加的表格标题保持文档结构简洁
- Word文档中直接以HTML标签形式显示所有表格方便查看表格结构
- 同时生成独立的HTML文件提供完整的表格视图支持交互和打印
- 在Word文档中添加蓝色超链接提示指引用户查看对应的HTML文件
- 改进HTML样式增强响应式布局和打印支持
- 优化表格HTML标签生成过程确保标签规范性和一致性
- 增强错误处理,即使某些表格转换失败也能保持系统稳定
- 提升整体文档处理流程的健壮性和用户体验
### 2024年6月10日
- **采用HTML标签形式输出表格**
- 改进表格处理机制直接输出HTML标签形式的表格而非创建Word表格
- 精确保留所有表格结构信息,包括表头、主体和合并单元格属性
- 自动为表格生成符合HTML规范的标签包括class和id属性
- 正确处理表格中的垂直和水平合并单元格添加rowspan和colspan属性
- 将表格标签以等宽字体显示,提高可读性和直观性
- 优化标签生成过程严格遵循HTML表格标准
- 自动区分表头和数据行使用正确的thead和tbody标签
- 简化表格处理流程,提高效率和准确性
### 2024年6月9日
- **改进Word文档表格显示方式**
- 修改表格处理机制直接在Word文档中显示表格不再需要外部HTML文件
- 准确复制原始表格的结构、内容和合并单元格信息
- 保留表格样式并自动设置表头格式
- 正确处理垂直和水平合并的单元格
- 改进表格位置控制,保持与原始文档的一致性
- 优化表格边框和样式,提供更专业的外观
- 简化处理流程,提高文档生成效率
- 修复合并单元格时的潜在错误
### 2024年6月8日
- **修复Word文档打开问题并改进表格处理**
- 解决了清洗后Word文档无法打开的关键问题
- 优化HTML表格生成方式确保文档处理的稳定性
- 在Word文档中添加醒目的HTML表格文件引用提示
- 保留表格的文本格式作为备用显示方式
- 改进错误处理,提供更详细的诊断信息
- 简化文档处理流程,提高代码可维护性
- 增强HTML表格文件的样式提供更好的打印支持
- 改进文档处理日志,便于追踪处理过程
### 2024年6月7日
- **表格直接HTML输出功能增强**
- 修改表格处理机制现在所有表格都将以HTML格式输出而非文本格式
- 彻底解决复杂表格的显示问题,包括多层表头和合并单元格
- 自动为每个表格生成独特的HTML标识符确保正确引用
- 提供更美观的表格样式,包括悬停效果和自适应宽度
- 改进表格边框和单元格间距,提升阅读体验
- 保留单元格格式化内容如换行符并在HTML中正确显示
- 针对打印场景优化表格样式,确保打印输出质量
- 技术说明由于Word文档格式限制HTML表格将保存在独立的HTML文件中
### 2024年6月6日
- **增强复杂表格识别与处理能力**
- 优化表格类型自动识别算法,通过多维度特征分析提高复杂表格的识别精度
- 增强表格结构分析能力,支持更精确地识别垂直和水平合并单元格
- 改进多级表头处理,提高复杂表头的识别和解析能力
- 引入表格宽高比分析,自动识别宽表格和复杂结构表格
- 新增单元格数一致性检查机制,提高对不规则表格的处理能力
- 优化垂直合并单元格的内容填充算法,改进空单元格的值传播机制
- 添加更详细的表格处理日志,便于诊断和调试复杂表格处理问题
- 完善异常处理,提高处理复杂表格时的稳定性和鲁棒性
### 2024年6月5日
- **模块化表格处理系统升级**
- 重构了表格处理架构,将不同类型的表格处理逻辑拆分为专门的处理器
- 实现了针对五种特殊表格类型的专用处理器:
- 多级表头表格处理器:处理具有复杂多级表头结构的表格
- 合并单元格密集型表格处理器:优化处理含有大量合并单元格的表格
- 带计算功能的表格处理器:识别并处理包含公式和计算的表格
- 嵌套表格处理器:处理表格内嵌套的子表格结构
- 跨页长表格处理器:正确识别和处理跨越多页的长表格
- 优化表格识别机制,智能匹配最适合的处理器
- 提高表格转文本的准确性和可读性
- 保持与原有系统的兼容性,同时提高处理复杂表格的能力
- 增强了系统对特殊表格结构的识别率和处理精度
### 2024年6月2日
- 改进Markdown表格转换功能
- 修复了合并单元格内容重复显示的问题
@@ -331,6 +454,19 @@ pip install -r requirements.txt
- 自动清理"表格无有效数据"等无效提示信息
- 优化文本拼接逻辑,确保输出格式的一致性
### 2024年6月20日
- **修复表格HTML生成错误**
- 解决了`_generate_table_html_tags`方法中`cell_map`变量未定义的错误
- 完全重写了表格HTML生成逻辑确保表格能正确转换为HTML格式
- 优化表格头部和主体分别处理的流程
- 添加了合并单元格的正确属性rowspan和colspan
- 完善了表格结构分析过程,准确标记垂直和水平合并的单元格
- 改进HTML标签属性生成逻辑确保输出符合HTML规范
- 增强了错误处理,提供具体的单元格位置信息方便调试
- 确保表格信息完整转换为HTML格式解决"表格的html没有正常输出"问题
### 2024年6月19日
## 安装说明
1. 克隆项目代码

View File

@@ -1,24 +0,0 @@
1111
【图片识别文本】
“ 完 善 了 异 常 处 理 , 防 止 惑 时 目 录 券 除 失 败 导 致 程 序 崖
4. 更 新 README.md
* 在 暨 近 更 新 部 分 记 录 了 临 时 文 件 处 理 机 制 的 改 进
。 添 加 了 Excel 文 件 句 柄 管 理
使 用 说 明
这 东 改 进 不 需 要 您 做 任 何 额 外 操 作 , 系 统 会 自 动 -
1. 在 处 理 Excel 文 件 时 正 球 关 闭 文 件 句 柄
2 当 尝 试 删 除 文 件 通 刨 “ 文 件 被 占 用 “ 错 误 时 , 自 动 等 待
并 重 试
3 即 使 无 法 券 除 临 时 文 件 , 也 不 影 响 处 #
如 果 仍 然 通 到 惧 时 文 件 问 题 , 系 统 会 在 下 次 启 动 时 自 动
清 理 所 有 临 时 文 件 , 不 会 影 响 系 统 功 能 。
以 上 优 化 星 觞 失 了 临 时 文 伟 删 除 问 题 , 又 保 持 了 系 统 的
稳 定 性 , 让 您 能 雪 顺 畅 地 处 理 Bxcel 文 件 。

View File

@@ -1 +0,0 @@
1111 【图片识别文本】 “ 完 善 了 异 常 处 理 , 防 止 惑 时 目 录 券 除 失 败 导 致 程 序 崖 澎 澎 4. 更 新 README.md * 在 暨 近 更 新 部 分 记 录 了 临 时 文 件 处 理 机 制 的 改 进 。 添 加 了 Excel 文 件 句 柄 管 理 使 用 说 明 这 东 改 进 不 需 要 您 做 任 何 额 外 操 作 , 系 统 会 自 动 - 1. 在 处 理 Excel 文 件 时 正 球 关 闭 文 件 句 柄 2 当 尝 试 删 除 文 件 通 刨 “ 文 件 被 占 用 “ 错 误 时 , 自 动 等 待 并 重 试 3 即 使 无 法 券 除 临 时 文 件 , 也 不 影 响 处 # 如 果 仍 然 通 到 惧 时 文 件 问 题 , 系 统 会 在 下 次 启 动 时 自 动 清 理 所 有 临 时 文 件 , 不 会 影 响 系 统 功 能 。 以 上 优 化 星 觞 失 了 临 时 文 伟 删 除 问 题 , 又 保 持 了 系 统 的 稳 定 性 , 让 您 能 雪 顺 畅 地 处 理 Bxcel 文 件 。

File diff suppressed because one or more lines are too long

1539
table/table_cleaner.py Normal file

File diff suppressed because it is too large Load Diff

444
table/table_to_html.py Normal file
View File

@@ -0,0 +1,444 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import os
import docx
import re
from docx.table import Table, _Cell
from docx.oxml import parse_xml
from docx.oxml.ns import nsdecls
from typing import List, Dict, Tuple, Optional, Union
import uuid
from bs4 import BeautifulSoup
import html
class TableToHtml:
def __init__(self, debug: bool = False):
"""
初始化表格到HTML转换器
Args:
debug: 是否启用调试模式,输出更多日志信息
"""
self.debug = debug
# 为每个表格生成唯一ID
self.table_id = f"table_{uuid.uuid4().hex[:8]}"
def _log(self, message: str):
"""
输出调试日志
Args:
message: 日志消息
"""
if self.debug:
print(f"[TableToHtml] {message}")
def _get_vmerge_value(self, cell_element) -> Optional[str]:
"""
获取单元格的垂直合并属性
Args:
cell_element: 单元格元素
Returns:
str: 垂直合并属性值
"""
vmerge = cell_element.xpath('.//w:vMerge')
if vmerge:
return vmerge[0].get('{http://schemas.openxmlformats.org/wordprocessingml/2006/main}val', 'continue')
return None
def _get_gridspan_value(self, cell_element) -> int:
"""
获取单元格的水平合并数量
Args:
cell_element: 单元格元素
Returns:
int: 水平合并的列数
"""
try:
gridspan = cell_element.xpath('.//w:gridSpan')
if gridspan and gridspan[0].get('{http://schemas.openxmlformats.org/wordprocessingml/2006/main}val'):
return int(gridspan[0].get('{http://schemas.openxmlformats.org/wordprocessingml/2006/main}val'))
except (ValueError, TypeError, AttributeError) as e:
self._log(f"警告获取gridspan值时出错: {str(e)}")
return 1 # 默认返回1表示没有合并
def _get_cell_content(self, cell: _Cell) -> str:
"""
获取单元格的文本内容并处理HTML特殊字符
Args:
cell: docx表格单元格对象
Returns:
str: 处理后的HTML内容
"""
content = cell.text.strip()
# 转义HTML特殊字符
content = html.escape(content)
# 处理换行
content = content.replace('\n', '<br>')
return content
def _analyze_table_structure(self, table: Table) -> Dict:
"""
分析表格结构,包括合并单元格信息
Args:
table: docx表格对象
Returns:
Dict: 表格结构信息
"""
rows = len(table.rows)
cols = len(table.columns)
# 存储合并单元格信息
merged_cells = {}
# 存储垂直合并的源单元格
vmerge_sources = {}
# 分析合并单元格
for i in range(rows):
for j in range(cols):
try:
cell = table.cell(i, j)
# 检查垂直合并
if cell._element.tcPr is not None:
vmerge = cell._element.tcPr.xpath('.//w:vMerge')
if vmerge:
val = self._get_vmerge_value(cell._element)
if val == 'restart':
# 这是垂直合并的起始单元格
# 计算合并的行数
rowspan = 1
for k in range(i+1, rows):
next_cell = table.cell(k, j)
if self._get_vmerge_value(next_cell._element) == 'continue':
rowspan += 1
# 标记此单元格为被合并
merged_cells[(k, j)] = {'merged': True, 'source': (i, j)}
else:
break
# 记录合并信息
vmerge_sources[(i, j)] = {'rowspan': rowspan}
elif val == 'continue':
# 这是被合并的单元格,稍后处理
pass
# 检查水平合并
if cell._element.tcPr is not None:
gridspan = self._get_gridspan_value(cell._element)
if gridspan > 1:
# 记录colspan
merged_cells[(i, j)] = {'colspan': gridspan}
# 标记被合并的单元格
for k in range(1, gridspan):
if j + k < cols:
merged_cells[(i, j+k)] = {'merged': True, 'source': (i, j)}
except Exception as e:
self._log(f"警告:分析单元格 [{i},{j}] 时出错: {str(e)}")
continue
# 将垂直合并信息合并到主合并字典
for pos, info in vmerge_sources.items():
if pos in merged_cells:
merged_cells[pos].update(info)
else:
merged_cells[pos] = info
return {
'rows': rows,
'cols': cols,
'merged_cells': merged_cells
}
def _is_header_row(self, row_idx: int, table: Table, structure: Dict) -> bool:
"""
判断是否为表头行
Args:
row_idx: 行索引
table: 表格对象
structure: 表格结构信息
Returns:
bool: 是否为表头行
"""
# 简单策略:第一行通常是表头
if row_idx == 0:
return True
# 检查是否有垂直合并从第一行开始的单元格
for j in range(structure['cols']):
cell_pos = (row_idx, j)
if cell_pos in structure['merged_cells'] and 'merged' in structure['merged_cells'][cell_pos]:
source = structure['merged_cells'][cell_pos]['source']
if source[0] == 0: # 合并源是第一行
return True
return False
def _detect_table_headers(self, table: Table, structure: Dict) -> List[int]:
"""
检测表格表头行
Args:
table: 表格对象
structure: 表格结构信息
Returns:
List[int]: 表头行索引列表
"""
header_rows = []
rows = structure['rows']
# 检查前3行或所有行如果行数少于3
for i in range(min(3, rows)):
if self._is_header_row(i, table, structure):
header_rows.append(i)
# 如果没有检测到表头,默认第一行为表头
if not header_rows and rows > 0:
header_rows = [0]
self._log(f"检测到的表头行: {header_rows}")
return header_rows
def table_to_html(self, table: Table) -> str:
"""
将docx表格转换为HTML格式
Args:
table: docx表格对象
Returns:
str: HTML表格代码
"""
try:
# 分析表格结构
structure = self._analyze_table_structure(table)
rows = structure['rows']
cols = structure['cols']
merged_cells = structure['merged_cells']
self._log(f"表格结构: {rows}行 x {cols}列,合并单元格: {len(merged_cells)}")
# 检测表头
header_rows = self._detect_table_headers(table, structure)
# 构建HTML表格
soup = BeautifulSoup('<table></table>', 'html.parser')
table_tag = soup.table
table_tag['class'] = ['docx-table']
table_tag['id'] = self.table_id
# 添加表头部分thead
if header_rows:
thead = soup.new_tag('thead')
table_tag.append(thead)
for i in header_rows:
if i >= rows:
continue
tr = soup.new_tag('tr')
thead.append(tr)
j = 0
while j < cols:
cell_pos = (i, j)
# 检查是否被合并
if cell_pos in merged_cells and 'merged' in merged_cells[cell_pos]:
j += 1
continue
# 创建th元素
th = soup.new_tag('th')
# 处理合并
if cell_pos in merged_cells:
if 'rowspan' in merged_cells[cell_pos]:
th['rowspan'] = merged_cells[cell_pos]['rowspan']
if 'colspan' in merged_cells[cell_pos]:
th['colspan'] = merged_cells[cell_pos]['colspan']
j += merged_cells[cell_pos]['colspan'] - 1
# 设置单元格内容
cell = table.cell(i, j)
content = self._get_cell_content(cell)
th.string = content
tr.append(th)
j += 1
# 添加表格主体tbody
tbody = soup.new_tag('tbody')
table_tag.append(tbody)
# 计算数据行的起始索引
data_start = max(header_rows) + 1 if header_rows else 0
# 处理数据行
for i in range(data_start, rows):
tr = soup.new_tag('tr')
tbody.append(tr)
j = 0
while j < cols:
cell_pos = (i, j)
# 检查是否被合并
if cell_pos in merged_cells and 'merged' in merged_cells[cell_pos]:
j += 1
continue
# 创建td元素
td = soup.new_tag('td')
# 处理合并
if cell_pos in merged_cells:
if 'rowspan' in merged_cells[cell_pos]:
td['rowspan'] = merged_cells[cell_pos]['rowspan']
if 'colspan' in merged_cells[cell_pos]:
td['colspan'] = merged_cells[cell_pos]['colspan']
j += merged_cells[cell_pos]['colspan'] - 1
# 设置单元格内容
cell = table.cell(i, j)
content = self._get_cell_content(cell)
td.string = content
tr.append(td)
j += 1
# 添加基本的CSS样式
style = soup.new_tag('style')
style.string = f'''
#{self.table_id} {{
border-collapse: collapse;
width: 100%;
margin-bottom: 1em;
font-family: Arial, sans-serif;
}}
#{self.table_id} th, #{self.table_id} td {{
border: 1px solid #ddd;
padding: 8px;
text-align: left;
}}
#{self.table_id} th {{
background-color: #f2f2f2;
font-weight: bold;
}}
#{self.table_id} tr:nth-child(even) {{
background-color: #f9f9f9;
}}
#{self.table_id} tr:hover {{
background-color: #f5f5f5;
}}
'''
# 返回完整的HTML代码
html_code = str(style) + str(table_tag)
return html_code
except Exception as e:
self._log(f"转换表格到HTML时出错: {str(e)}")
import traceback
traceback.print_exc()
return f"<div class='error'>表格处理失败: {str(e)}</div>"
def process_document_tables(self, doc_path: str) -> List[str]:
"""
处理文档中的所有表格并转换为HTML
Args:
doc_path: 文档文件路径
Returns:
List[str]: HTML表格代码列表
"""
try:
# 打开文档
doc = docx.Document(doc_path)
html_tables = []
# 处理所有表格
for i, table in enumerate(doc.tables):
self._log(f"处理第 {i+1} 个表格")
self.table_id = f"table_{uuid.uuid4().hex[:8]}" # 为每个表格生成唯一ID
html_code = self.table_to_html(table)
html_tables.append(html_code)
return html_tables
except Exception as e:
self._log(f"处理文档表格时出错: {str(e)}")
import traceback
traceback.print_exc()
return [f"<div class='error'>文档处理失败: {str(e)}</div>"]
def convert_tables_to_html(doc_path: str, output_path: str = None, debug: bool = False):
"""
将文档中的表格转换为HTML并保存
Args:
doc_path: 文档文件路径
output_path: 输出HTML文件路径如果为None则使用原文件名+.html
debug: 是否启用调试模式
Returns:
str: 输出文件路径
"""
if output_path is None:
# 创建默认输出路径
base_name = os.path.splitext(doc_path)[0]
output_path = f"{base_name}_tables.html"
converter = TableToHtml(debug=debug)
html_tables = converter.process_document_tables(doc_path)
# 创建完整HTML文档
html_content = f'''<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<title>表格预览</title>
<meta name="viewport" content="width=device-width, initial-scale=1.0">
</head>
<body>
<h1>文档中的表格</h1>
{' '.join(html_tables)}
</body>
</html>'''
# 保存HTML文件
with open(output_path, 'w', encoding='utf-8') as f:
f.write(html_content)
if debug:
print(f"HTML文件已保存到: {output_path}")
return output_path
if __name__ == '__main__':
import argparse
parser = argparse.ArgumentParser(description='将Word文档中的表格转换为HTML')
parser.add_argument('input_file', help='输入文档文件路径')
parser.add_argument('-o', '--output', help='输出HTML文件路径', default=None)
parser.add_argument('-d', '--debug', action='store_true', help='启用调试模式')
args = parser.parse_args()
result_path = convert_tables_to_html(args.input_file, args.output, args.debug)
print(f"表格已转换为HTML文件路径: {result_path}")

396
table/text_splitter.py Normal file
View File

@@ -0,0 +1,396 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import re
import json
import argparse
def count_chinese_tokens(text):
"""
估算中文文本的token数量
1个汉字约等于1.5个token
1个英文单词约等于1个token
1个标点符号约等于1个token
"""
# 匹配中文字符
chinese_chars = len(re.findall(r'[\u4e00-\u9fff]', text))
# 匹配英文单词
english_words = len(re.findall(r'[a-zA-Z]+', text))
# 匹配标点符号
punctuations = len(re.findall(r'[^\w\s]', text))
# 计算总token数粗略估算
total_tokens = chinese_chars * 1.5 + english_words + punctuations
return int(total_tokens)
def process_table_content(table_content):
"""
处理表格内容,移除表格标记并进行智能分段
处理策略:
1. 清理无效内容
2. 智能分段
3. 保持语义完整性
4. 控制token长度
"""
# 移除表格标记和多余空白
content = re.sub(r'表格\s*\d+\s*(?:开始|结束)', '', table_content)
content = re.sub(r'\s+', ' ', content).strip()
# 分段处理
paragraphs = []
current_para = []
# 按句子分割
sentences = re.split(r'([。!?\n])', content)
for i in range(0, len(sentences), 2):
sentence = sentences[i].strip()
if not sentence:
continue
# 添加标点符号(如果存在)
if i + 1 < len(sentences):
sentence += sentences[i + 1]
# 检查是否是新段落的开始
if (re.match(r'^[的]', sentence) or # 以"的"开头
re.match(r'^[在]', sentence) or # 以"在"开头
re.match(r'^[\w()]+[:]', sentence)): # 以键值对形式开头
# 保存当前段落
if current_para:
full_para = ''.join(current_para).strip()
if full_para:
# 控制token长度
if count_chinese_tokens(full_para) > 512:
split_paras = split_long_paragraph(full_para)
paragraphs.extend(split_paras)
else:
paragraphs.append(full_para)
current_para = []
current_para.append(sentence)
# 处理最后一个段落
if current_para:
full_para = ''.join(current_para).strip()
if full_para:
if count_chinese_tokens(full_para) > 512:
split_paras = split_long_paragraph(full_para)
paragraphs.extend(split_paras)
else:
paragraphs.append(full_para)
return paragraphs
def split_long_paragraph(paragraph):
"""智能分割长段落,保持语义完整性"""
result = []
# 首先尝试按逗号分割
parts = re.split(r'([,。!?])', paragraph)
current_part = ""
current_tokens = 0
for i in range(0, len(parts), 2):
part = parts[i].strip()
if not part:
continue
# 添加标点符号(如果存在)
if i + 1 < len(parts):
part += parts[i + 1]
part_tokens = count_chinese_tokens(part)
if current_tokens + part_tokens > 512:
if current_part:
result.append(current_part)
current_part = part
current_tokens = part_tokens
else:
current_part += part
current_tokens += part_tokens
if current_part:
result.append(current_part)
return result
def format_group_to_text(group):
"""将分组数据格式化为易读的文本,采用通用的处理方式"""
if not group:
return ""
parts = []
# 通用处理:遍历所有键值对,构建文本
for key, value in group.items():
# 跳过空值
if not value:
continue
# 清理和格式化键名
clean_key = re.sub(r'[_\(\)]', ' ', key).strip()
# 清理值中的"表格无有效数据"字眼
if isinstance(value, str):
value = re.sub(r'[【\[]*表格无[有效]*数据[】\]]*', '', value)
if not value.strip(): # 如果清理后为空,则跳过
continue
# 构建文本片段
text = f"{clean_key}{value}"
parts.append(text)
# 使用逗号连接所有部分,并确保结果中没有"表格无有效数据"字眼
result = "".join(parts)
result = re.sub(r'[【\[]*表格无[有效]*数据[】\]]*', '', result)
return result.strip("") + "" if result.strip("") else ""
def split_long_text(text):
"""将长文本按token限制分割"""
if count_chinese_tokens(text) <= 512:
return [text]
result = []
parts = re.split(r'([。])', text)
current_part = ""
current_tokens = 0
for i in range(0, len(parts), 2):
sentence = parts[i]
if i + 1 < len(parts):
sentence += parts[i + 1] # 添加句号
sentence_tokens = count_chinese_tokens(sentence)
if current_tokens + sentence_tokens > 512:
if current_part:
result.append(current_part)
current_part = sentence
current_tokens = sentence_tokens
else:
current_part += sentence
current_tokens += sentence_tokens
if current_part:
result.append(current_part)
return result
def split_text_into_paragraphs(text):
"""
将连续文本智能分段
策略:
1. 基于标题和章节标记进行主要分段
2. 基于段落语义标记进行次要分段
3. 基于句子关联度进行内容分段
4. 基于token长度进行辅助分段确保每段不超过512个token
5. 保持段落的语义完整性
6. 智能处理表格内容
"""
# 清理文本中可能存在的多余空格和换行
text = re.sub(r'\s+', ' ', text).strip()
# 首先处理表格内容
table_pattern = re.compile(r'(表格\s*\d+\s*开始.*?表格\s*\d+\s*结束)', re.DOTALL)
parts = []
last_end = 0
for match in table_pattern.finditer(text):
# 添加表格前的文本
if match.start() > last_end:
parts.append(("text", text[last_end:match.start()]))
# 处理表格内容
table_content = match.group(1)
table_paragraphs = process_table_content(table_content)
for para in table_paragraphs:
# 确保表格段落没有冒号开头
para = re.sub(r'^[:]+\s*', '', para.strip())
if para: # 只添加非空段落
parts.append(("table", para))
last_end = match.end()
# 添加最后一个表格之后的文本
if last_end < len(text):
parts.append(("text", text[last_end:]))
# 如果没有找到表格,将整个文本作为一个文本部分
if not parts:
parts = [("text", text)]
# 主要分段标记(标题、章节等)
major_markers = [
r'^第[一二三四五六七八九十百千]+[章节篇]', # 中文数字章节
r'^第\d+[章节篇]', # 阿拉伯数字章节
r'^[一二三四五六七八九十][、.]', # 中文数字序号
r'^\d+[、.]', # 阿拉伯数字序号
r'^[(][一二三四五六七八九十][)]', # 带括号的中文数字
r'^[(]\d+[)]', # 带括号的阿拉伯数字
r'^[IVX]+[、.]', # 罗马数字序号
]
# 次要分段标记(语义转折等)
minor_markers = [
r'然而[,]',
r'但是[,]',
r'不过[,]',
r'相反[,]',
r'因此[,]',
r'所以[,]',
r'总的来说',
r'综上所述',
r'总而言之',
r'例如[,]',
r'比如[,]',
r'首先[,]',
r'其次[,]',
r'最后[,]',
r'另外[,]',
]
# 特殊段落标记
special_markers = [
r'^摘要',
r'^引言',
r'^前言',
r'^结论',
r'^致谢',
r'^参考文献',
r'^注释',
r'^附录',
]
# 合并所有标记模式
all_markers = major_markers + special_markers
marker_pattern = '|'.join(all_markers)
minor_marker_pattern = '|'.join(minor_markers)
# 按句子分割的分隔符
sentence_separators = r'([。!?\!\?])'
# 分段处理
paragraphs = []
for part_type, content in parts:
if part_type == "table":
# 表格内容已经过处理,直接添加
paragraphs.append(content)
continue
# 处理普通文本
current_para = ""
current_tokens = 0
# 按主要标记分段
text_parts = re.split(f'({marker_pattern})', content)
for i, part in enumerate(text_parts):
if not part.strip(): # 跳过空部分
continue
# 去除冒号开头
part = re.sub(r'^[:]+\s*', '', part.strip())
if not part: # 跳过清理后为空的部分
continue
if i % 2 == 1: # 是标记
if current_para:
paragraphs.append(current_para)
current_para = part
current_tokens = count_chinese_tokens(part)
else: # 是内容
sentences = re.split(sentence_separators, part)
for j, sentence in enumerate(sentences):
if not sentence.strip():
continue
# 去除句子开头的冒号
sentence = re.sub(r'^[:]+\s*', '', sentence.strip())
if not sentence:
continue
sentence_tokens = count_chinese_tokens(sentence)
# 检查是否有次要分段标记
has_minor_marker = bool(re.search(minor_marker_pattern, sentence))
if has_minor_marker and current_para:
paragraphs.append(current_para)
current_para = sentence
current_tokens = sentence_tokens
elif current_tokens + sentence_tokens > 512:
if current_para:
paragraphs.append(current_para)
current_para = sentence
current_tokens = sentence_tokens
else:
if current_para:
current_para += sentence
else:
current_para = sentence
current_tokens += sentence_tokens
if current_para:
paragraphs.append(current_para)
# 最后一次清理所有段落,确保没有冒号开头
cleaned_paragraphs = []
for para in paragraphs:
para = re.sub(r'^[:]+\s*', '', para.strip())
if para: # 只添加非空段落
cleaned_paragraphs.append(para)
return cleaned_paragraphs
def save_to_json(paragraphs, output_file):
"""将段落保存为JSON格式"""
data = {
"total_paragraphs": len(paragraphs),
"paragraphs": paragraphs
}
with open(output_file, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
print(f"成功将文本分成 {len(paragraphs)} 个段落并保存到 {output_file}")
def save_to_txt(paragraphs, output_file):
"""将段落保存为TXT格式每段用换行符分隔"""
with open(output_file, 'w', encoding='utf-8') as f:
for paragraph in paragraphs:
f.write(paragraph + '\n\n') # 使用两个换行符使段落分隔更清晰
print(f"成功将文本分成 {len(paragraphs)} 个段落并保存到 {output_file}")
def main():
parser = argparse.ArgumentParser(description="将连续文本智能分段并保存为TXT或JSON")
parser.add_argument("input_file", help="输入文件路径例如sample_continuous_text.txt")
parser.add_argument("--output", "-o", default="paragraphs.txt", help="输出文件路径,默认为当前目录下的 paragraphs.txt")
parser.add_argument("--format", "-f", choices=['txt', 'json'], default='txt', help="输出文件格式支持txt和json默认为txt")
args = parser.parse_args()
# 读取输入文件
try:
with open(args.input_file, 'r', encoding='utf-8') as f:
text = f.read()
except Exception as e:
print(f"读取文件出错: {e}")
return
# 分段
paragraphs = split_text_into_paragraphs(text)
# 根据指定格式保存
if args.format == 'json':
save_to_json(paragraphs, args.output)
else:
save_to_txt(paragraphs, args.output)
if __name__ == "__main__":
main()