"""
文本处理器模块 - 重构版
实现全局编号系统,确保段落精确对应 + 清理HTML标签
"""
import re
from bs4 import BeautifulSoup
from typing import List, Dict
from loguru import logger
class TextProcessor:
"""文本处理器 - 简化版,专注核心功能"""
def __init__(self, config: Dict):
"""
初始化文本处理器
Args:
config: 配置字典
"""
self.config = config
# 不再使用最小长度限制,只要有内容就提取
self.chunk_size = config['translation']['chunk_size']
self._global_id_counter = 0
logger.info(f"文本处理器初始化: chunk_size={self.chunk_size}, 无最小长度限制")
@staticmethod
def get_valid_text_elements(soup) -> List:
"""
获取有效的文本元素列表,自动过滤嵌套容器
(静态方法,供Builder共用,确保遍历顺序一致)
Args:
soup: BeautifulSoup对象
Returns:
过滤后的元素列表
"""
# 定义关注的标签
tags = ['p', 'div', 'h1', 'h2', 'h3', 'h4', 'h5', 'h6', 'blockquote', 'li', 'td']
# 1. 获取所有候选元素
all_candidates = soup.find_all(tags)
# 2. 转换为集合以提高查找速度
candidate_set = set(all_candidates)
final_elements = []
for element in all_candidates:
# 3. 检查当前元素是否包含其他候选元素
# 如果包含,说明它是父容器,应该跳过,让子元素去被处理
has_candidate_children = False
# 只查找直接子级或后代中的候选标签
descendants = element.find_all(tags)
for child in descendants:
if child in candidate_set:
has_candidate_children = True
break
if has_candidate_children:
# 这是一个容器元素,跳过
continue
final_elements.append(element)
return final_elements
def extract_paragraphs_with_global_id(self, html_content: str, source_file: str = "") -> List[Dict]:
"""
提取段落并分配全局唯一ID
Args:
html_content: HTML内容
source_file: 来源文件名(用于调试)
Returns:
带全局ID的段落列表
"""
try:
soup = BeautifulSoup(html_content, 'html.parser')
paragraphs = []
# 移除不需要的元素
for element in soup(['script', 'style', 'meta', 'link']):
element.decompose()
# 获取有效的文本元素 (使用统一的过滤逻辑)
text_elements = self.get_valid_text_elements(soup)
position = 0
for element in text_elements:
# 清理文本:移除上标、下标等
clean_text = self._clean_element_text(element)
# 过滤逻辑:
# 1. 如果是导航元素,跳过
if self._is_navigation_element(element):
continue
# 2. 内容检查:只要不是空字符串,就保留
if not clean_text:
continue
# 分配全局唯一ID
global_id = self._generate_global_id()
paragraphs.append({
'global_id': global_id,
'text': clean_text,
'html_element': str(element),
'source_file': source_file,
'position': position, # 在文件中的位置(重要!)
'tag': element.name,
'length': len(clean_text)
})
position += 1
logger.info(f"从 {source_file} 提取了 {len(paragraphs)} 个段落")
return paragraphs
except Exception as e:
logger.error(f"提取段落失败: {e}")
return []
@staticmethod
def clean_element_text(element) -> str:
"""
清理元素文本:移除上标、下标、脚注等 (静态方法,供Builder共用)
Args:
element: HTML元素
Returns:
清理后的文本
"""
# 复制元素,避免修改原始DOM
element_copy = element.__copy__()
# 移除上标和下标(通常是脚注引用)
for tag in element_copy.find_all(['sup', 'sub']):
tag.decompose()
# 移除带有特定class的span/a标签 (脚注常见写法)
footnote_patterns = re.compile(r'footnote|endnote|reference|note|super|sub', re.I)
for tag in element_copy.find_all(['a', 'span', 'div'], class_=footnote_patterns):
tag.decompose()
# 移除仅包含数字或中括号数字的小型文本节点 (针对单纯文本形式的脚注 [1] 或 1)
for tag in element_copy.find_all('span'):
text = tag.get_text().strip()
# 匹配 [1], (1), 1, 12
if re.match(r'^(\[\d+\]|\(\d+\)|\d+)$', text):
tag.decompose()
# 获取清理后的文本
text = element_copy.get_text().strip()
# 额外的正则清理:移除正文末尾残留的引用标记,如 "text.[1]" 或 "text.1"
text = re.sub(r'(\.|。|,|,)\s*(\[\d+\]|\d+)(?=\s|$)', r'\1', text)
# 清理多余的空白
text = re.sub(r'\s+', ' ', text)
return text
@staticmethod
def is_navigation_element(element) -> bool:
"""
判断是否是导航元素 (静态方法,供Builder共用)
Args:
element: HTML元素
Returns:
是否是导航元素
"""
# 检查class属性
classes = element.get('class', [])
nav_classes = ['nav', 'navigation', 'toc', 'menu', 'header', 'footer', 'page-number']
# 处理 class 可能是列表或字符串的情况
if isinstance(classes, list):
class_str = ' '.join(classes).lower()
else:
class_str = str(classes).lower()
if any(nav_class in class_str for nav_class in nav_classes):
return True
# 检查父元素
parent = element.parent
if parent:
parent_classes = parent.get('class', [])
if isinstance(parent_classes, list):
parent_class_str = ' '.join(parent_classes).lower()
else:
parent_class_str = str(parent_classes).lower()
if any(nav_class in parent_class_str for nav_class in nav_classes):
return True
return False
def _clean_element_text(self, element) -> str:
"""兼容旧调用的包装器"""
return self.clean_element_text(element)
def _is_navigation_element(self, element) -> bool:
"""兼容旧调用的包装器"""
return self.is_navigation_element(element)
def _generate_global_id(self) -> str:
"""
生成全局唯一ID
Returns:
全局ID字符串,格式:p_0001
"""
self._global_id_counter += 1
return f"p_{self._global_id_counter:04d}"
def create_chunks_by_size(self, paragraphs: List[Dict]) -> List[List[Dict]]:
"""
按字符数创建chunks,不切断段落,不考虑章节边界
Args:
paragraphs: 带全局ID的段落列表
Returns:
分块的段落列表
"""
if not paragraphs:
return []
chunks = []
current_chunk = []
current_size = 0
for paragraph in paragraphs:
para_length = paragraph['length']
# 如果当前chunk加上这个段落不超过限制,就加入
if current_size + para_length <= self.chunk_size:
current_chunk.append(paragraph)
current_size += para_length
else:
# 保存当前chunk(如果有内容)
if current_chunk:
chunks.append(current_chunk)
# 开始新chunk
current_chunk = [paragraph]
current_size = para_length
# 保存最后一个chunk
if current_chunk:
chunks.append(current_chunk)
# 统计信息
total_chars = sum(p['length'] for p in paragraphs)
avg_chunk_size = total_chars / len(chunks) if chunks else 0
logger.info(f"创建了 {len(chunks)} 个chunk,"
f"总段落数: {len(paragraphs)}, "
f"平均chunk大小: {avg_chunk_size:.0f} 字符")
# 显示chunk分布
for i, chunk in enumerate(chunks, 1):
chunk_size = sum(p['length'] for p in chunk)
logger.debug(f" Chunk {i}: {len(chunk)} 段落, {chunk_size} 字符, "
f"ID范围: {chunk[0]['global_id']} - {chunk[-1]['global_id']}")
return chunks
def validate_translation(self, original: str, translation: str) -> bool:
"""
验证翻译质量
Args:
original: 原文
translation: 译文
Returns:
是否通过验证
"""
# 检查是否是失败标记
if translation.startswith('[翻译失败') or translation.startswith('[解析失败'):
return False
# 检查基本长度
if len(translation) < len(original) * 0.1:
logger.warning("翻译过短")
return False
if len(translation) > len(original) * 8:
logger.warning("翻译过长")
return False
# 检查是否包含中文
if not re.search(r'[\u4e00-\u9fff]', translation):
logger.warning("翻译不包含中文")
return False
return True
def get_statistics(self, paragraphs: List[Dict]) -> Dict:
"""
获取段落统计信息
Args:
paragraphs: 段落列表
Returns:
统计信息字典
"""
if not paragraphs:
return {}
total_chars = sum(p['length'] for p in paragraphs)
avg_length = total_chars / len(paragraphs)
# 按来源文件分组统计
by_source = {}
for p in paragraphs:
source = p['source_file']
if source not in by_source:
by_source[source] = 0
by_source[source] += 1
return {
'total_paragraphs': len(paragraphs),
'total_characters': total_chars,
'average_length': round(avg_length, 1),
'min_length': min(p['length'] for p in paragraphs),
'max_length': max(p['length'] for p in paragraphs),
'by_source_file': by_source
}