Sub SplitText_Basic()    ' 🚨 问题1：硬编码列号，换个文件就失效    Dim lastRow As Long    lastRow = Cells(Rows.Count, "A").End(xlUp).Row    ' 🚨 问题2：没有进度提示，用户不知道要等多久    Application.ScreenUpdating = False    Dim i As Long, j As Long    Dim sourceText As String    Dim maxLength As Long    maxLength = 32767  ' 限制值    ' 🚨 问题3：简单循环，性能极差    For i = 1 To lastRow        sourceText = Cells(i, 1).Value        ' 🚨 问题4：直接用Len()，不处理空值和异常        If Len(sourceText) > maxLength Then            ' 🚨 问题5：暴力截断，破坏数据完整性            Cells(i, 2).Value = Left(sourceText, maxLength)            Cells(i, 3).Value = Mid(sourceText, maxLength + 1, maxLength)            ' 如果有第三段？第四段？不处理！        Else            Cells(i, 2).Value = sourceText        End If    Next i    Application.ScreenUpdating = True    MsgBox "处理完成！"  ' 🚨 没有告知实际处理了多少条End Sub

Sub SplitText_Intermediate()    ' 🩹 尝试修复一些明显问题    Dim lastRow As Long    lastRow = Cells(Rows.Count, "A").End(xlUp).Row    ' 关闭屏幕刷新    Application.ScreenUpdating = False    Application.Calculation = xlCalculationManual    Application.EnableEvents = False    ' 🩹 添加进度条    Dim progressBar As Object    Set progressBar = CreateProgressBar()    progressBar.Show    Dim i As Long, j As Long    Dim sourceText As String    Dim maxLength As Long    maxLength = 32766  ' 留1字符余地    ' 🩹 使用数组提高性能    Dim dataArray() As Variant    dataArray = Range("A1:A" & lastRow).Value    ' 🩹 预估需要多少列    Dim maxParts As Long    maxParts = 0    For i = LBound(dataArray) To UBound(dataArray)        If Len(dataArray(i, 1)) > maxLength Then            Dim parts As Long            parts = Application.WorksheetFunction.RoundUp(Len(dataArray(i, 1)) / maxLength, 0)            If parts > maxParts Then maxParts = parts        End If    Next i    ' 🩹 重新定义输出数组    Dim outputArray() As Variant    ReDim outputArray(1 To UBound(dataArray), 1 To maxParts)    ' 主处理循环    For i = LBound(dataArray) To UBound(dataArray)        sourceText = CStr(dataArray(i, 1))        If Len(sourceText) > 0 Then            ' 🩹 尝试按句子拆分（但仍有问题）            Dim textParts As Collection            Set textParts = SplitBySentence(sourceText, maxLength)            For j = 1 To textParts.Count                If j <= maxParts Then                    outputArray(i, j) = textParts(j)                End If            Next j        End If        ' 更新进度条        If i Mod 100 = 0 Then            progressBar.Update i / lastRow * 100            DoEvents        End If    Next i    ' 输出到工作表    Range("B1").Resize(UBound(outputArray), maxParts).Value = outputArray    ' 清理    progressBar.Hide    Set progressBar = Nothing    Application.ScreenUpdating = True    Application.Calculation = xlCalculationAutomatic    Application.EnableEvents = True    MsgBox "处理完成！共处理 " & lastRow & " 行数据。"End Sub' 🩹 辅助函数：按句子拆分Function SplitBySentence(text As String, maxLen As Long) As Collection    Dim col As Collection    Set col = New Collection    ' 🚨 复杂逻辑开始    Dim sentences() As String    sentences = Split(text, "。")  ' 中文句号    Dim currentPart As String    currentPart = ""    Dim i As Long    For i = 0 To UBound(sentences)        Dim sentence As String        sentence = sentences(i)        If i < UBound(sentences) Then            sentence = sentence & "。"        End If        ' 🚨 问题：这里没有处理英文句子        ' 🚨 问题：没有处理问号、感叹号        ' 🚨 问题：没有处理换行符        If Len(currentPart) + Len(sentence) <= maxLen Then            currentPart = currentPart & sentence        Else            If Len(currentPart) > 0 Then                col.Add currentPart            End If            currentPart = sentence            ' 🚨 问题：单个句子就可能超过maxLen            If Len(currentPart) > maxLen Then                ' 需要进一步拆分                Dim subParts As Collection                Set subParts = SplitByLength(currentPart, maxLen)                Dim k As Long                For k = 1 To subParts.Count                    col.Add subParts(k)                Next k                currentPart = ""            End If        End If    Next i    If Len(currentPart) > 0 Then        col.Add currentPart    End If    Set SplitBySentence = colEnd Function' 🩹 辅助函数：按长度拆分Function SplitByLength(text As String, maxLen As Long) As Collection    Dim col As Collection    Set col = New Collection    Dim startPos As Long    startPos = 1    Do While startPos <= Len(text)        ' 🚨 问题：中文字符可能被切断        Dim endPos As Long        endPos = startPos + maxLen - 1        If endPos > Len(text) Then            endPos = Len(text)        End If        ' 🚨 问题：尝试不在单词中间切断        If endPos < Len(text) Then            ' 寻找合适的分割点            Dim tempPos As Long            tempPos = endPos            ' 向前找空格或标点            Dim foundBreak As Boolean            foundBreak = False            Dim maxLookback As Long            maxLookback = 50  ' 最多向前找50个字符            Dim lookbackStart As Long            lookbackStart = Application.WorksheetFunction.Max(startPos, endPos - maxLookback)            For tempPos = endPos To lookbackStart Step -1                Dim char As String                char = Mid(text, tempPos, 1)                ' 🚨 问题：这个逻辑对中文不友好                If char = " " Or char = "," Or char = "." Or char = ";" Then                    endPos = tempPos                    foundBreak = True                    Exit For                End If            Next tempPos            ' 如果没找到，就在endPos处切断            If Not foundBreak Then                ' 再尝试找中文字符边界                For tempPos = endPos To lookbackStart Step -1                    char = Mid(text, tempPos, 1)                    ' 🚨 问题：这个中文字符检测不准确                    If AscW(char) < 0 Or AscW(char) > 255 Then                        ' 可能是中文字符                        endPos = tempPos                        Exit For                    End If                Next tempPos            End If        End If        col.Add Mid(text, startPos, endPos - startPos + 1)        startPos = endPos + 1    Loop    Set SplitByLength = colEnd Function' 🩹 创建进度条（又需要50行代码）Function CreateProgressBar() As Object    ' ... 此处省略50行进度条代码End Function

"""工单文本智能拆分系统版本: 3.0核心特性: 语义感知、智能分段、格式保持、并行处理架构: 模块化 + 策略模式 + 流水线处理"""import pandas as pdimport numpy as npfrom typing import List, Tuple, Dict, Optional, Union, Callable, Iteratorfrom dataclasses import dataclass, fieldfrom abc import ABC, abstractmethodimport reimport jiebaimport jieba.posseg as psegfrom concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor, as_completedfrom tqdm import tqdmimport loggingfrom datetime import datetimeimport hashlibimport jsonfrom pathlib import Pathimport sysfrom enum import Enumimport warningswarnings.filterwarnings('ignore')# 配置日志logging.basicConfig(    level=logging.INFO,    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',    handlers=[        logging.FileHandler(f'text_splitter_{datetime.now():%Y%m%d_%H%M%S}.log'),        logging.StreamHandler(sys.stdout)    ])logger = logging.getLogger(__name__)class SplitStrategy(Enum):    """拆分策略枚举"""    FIXED_LENGTH = "fixed_length"      # 固定长度    BY_SENTENCE = "by_sentence"        # 按句子    BY_PARAGRAPH = "by_paragraph"      # 按段落    SEMANTIC = "semantic"              # 语义分割    SMART = "smart"                    # 智能混合class TextSegmenter(ABC):    """文本分段器基类"""    @abstractmethod    def segment(self, text: str, max_length: int = 32767) -> List[str]:        """将文本分割为多个段落"""        pass    @abstractmethod    def get_segment_info(self, text: str) -> Dict:        """获取分段信息"""        passclass FixedLengthSegmenter(TextSegmenter):    """固定长度分段器（基础版）"""    def __init__(self, preserve_words: bool = True):        """        初始化        参数:            preserve_words: 是否尽量保持单词完整        """        self.preserve_words = preserve_words    def segment(self, text: str, max_length: int = 32767) -> List[str]:        """按固定长度分段"""        if not text or max_length <= 0:            return []        segments = []        text_length = len(text)        if text_length <= max_length:            return [text]        start = 0        while start < text_length:            end = min(start + max_length, text_length)            if self.preserve_words and end < text_length:                # 尝试在合适的位置截断                end = self._find_break_point(text, start, end)            segment = text[start:end]            if segment.strip():  # 忽略纯空白段落                segments.append(segment)            start = end        return segments    def _find_break_point(self, text: str, start: int, end: int) -> int:        """寻找合适的分割点"""        # 优先在标点符号后分割        for i in range(end - 1, max(start, end - 100), -1):            if i < len(text) and self._is_good_break(text, i):                return i + 1  # 包括标点        # 其次在空格后分割        for i in range(end - 1, max(start, end - 50), -1):            if i < len(text) and text[i].isspace():                return i + 1        # 最后在字符边界分割（避免切分中文字符）        for i in range(end - 1, max(start, end - 10), -1):            if i < len(text) and self._is_character_boundary(text, i):                return i + 1        return end    def _is_good_break(self, text: str, pos: int) -> bool:        """判断是否是好断点"""        if pos >= len(text):            return False        char = text[pos]        # 中英文标点        return char in '。！？；;.?!\n\r\t,，、'    def _is_character_boundary(self, text: str, pos: int) -> bool:        """判断是否是字符边界（避免切分中文字符）"""        if pos <= 0 or pos >= len(text):            return True        # 检查UTF-8多字节字符边界        prev_char = text[pos-1]        curr_char = text[pos]        # 简单的中文字符检查        is_prev_chinese = '\u4e00' <= prev_char <= '\u9fff'        is_curr_chinese = '\u4e00' <= curr_char <= '\u9fff'        # 不在中文字符中间分割        if is_prev_chinese and is_curr_chinese:            return False        return True    def get_segment_info(self, text: str) -> Dict:        """获取分段信息"""        return {            "strategy": SplitStrategy.FIXED_LENGTH.value,            "preserve_words": self.preserve_words,            "total_length": len(text)        }class SentenceSegmenter(TextSegmenter):    """按句子分段器"""    def __init__(self, language: str = 'zh'):        """        初始化        参数:            language: 语言 ('zh'=中文, 'en'=英文)        """        self.language = language        self.sentence_delimiters = self._get_delimiters()    def _get_delimiters(self) -> List[str]:        """获取句子分隔符"""        if self.language == 'zh':            return ['。', '！', '？', '；', '\n', '\r\n', '……', '…']        else:  # 英文            return ['.', '!', '?', ';', '\n', '\r\n', '...']    def segment(self, text: str, max_length: int = 32767) -> List[str]:        """按句子分段"""        segments = []        current_segment = ""        # 使用正则表达式分割句子        pattern = '|'.join(re.escape(d) for d in self.sentence_delimiters)        sentences = re.split(f'({pattern})', text)        for i in range(0, len(sentences), 2):            sentence = sentences[i]            if i + 1 < len(sentences):                sentence += sentences[i + 1]  # 加上分隔符            if not sentence.strip():                continue            # 如果单个句子就超过最大长度            if len(sentence) > max_length:                # 回退到按长度分段                fallback_segmenter = FixedLengthSegmenter(preserve_words=True)                sub_segments = fallback_segmenter.segment(sentence, max_length)                for sub_seg in sub_segments:                    if len(current_segment) + len(sub_seg) <= max_length:                        current_segment += sub_seg                    else:                        if current_segment:                            segments.append(current_segment)                        current_segment = sub_seg            elif len(current_segment) + len(sentence) <= max_length:                current_segment += sentence            else:                if current_segment:                    segments.append(current_segment)                current_segment = sentence        if current_segment:            segments.append(current_segment)        return segments    def get_segment_info(self, text: str) -> Dict:        """获取分段信息"""        return {            "strategy": SplitStrategy.BY_SENTENCE.value,            "language": self.language,            "sentence_count": len(self.segment(text, 1000000))  # 使用大数获取实际句子数        }class SmartSegmenter(TextSegmenter):    """智能分段器（混合策略）"""    def __init__(self,                  use_jieba: bool = True,                 min_segment_length: int = 100):        """        初始化        参数:            use_jieba: 是否使用结巴分词            min_segment_length: 最小分段长度        """        self.use_jieba = use_jieba        self.min_segment_length = min_segment_length        if use_jieba:            # 加载自定义词典            self._load_custom_dict()        # 初始化各种分段器        self.sentence_segmenter = SentenceSegmenter(language='zh')        self.fixed_segmenter = FixedLengthSegmenter(preserve_words=True)    def _load_custom_dict(self):        """加载自定义词典"""        # 工单系统中常见的专业术语        custom_words = [            '客户服务 ns',            '技术支持 ns',            '问题描述 ns',            '解决方案 ns',            '系统错误 ns',            '网络故障 ns',            '硬件故障 ns',            '软件更新 ns',            '紧急程度 ns',            '优先级 ns',        ]        for word in custom_words:            jieba.add_word(word)    def segment(self, text: str, max_length: int = 32767) -> List[str]:        """智能分段"""        if len(text) <= max_length:            return [text]        # 分析文本特性        text_features = self._analyze_text(text)        # 根据特性选择分段策略        if text_features['paragraph_ratio'] > 0.3:            # 段落结构明显，按段落分段            segments = self._segment_by_paragraph(text, max_length)        elif text_features['sentence_ratio'] > 0.7:            # 句子结构清晰，按句子分段            segments = self.sentence_segmenter.segment(text, max_length)        elif text_features['code_ratio'] > 0.2:            # 包含代码，需要特殊处理            segments = self._segment_with_code(text, max_length)        else:            # 混合文本，使用智能分段            segments = self._smart_segment(text, max_length, text_features)        # 后处理：合并过短的段落        segments = self._merge_short_segments(segments, max_length)        return segments    def _analyze_text(self, text: str) -> Dict:        """分析文本特征"""        features = {            'total_length': len(text),            'paragraph_count': text.count('\n\n') + 1,            'sentence_count': len(re.findall(r'[。！？!?]', text)),            'code_count': len(re.findall(r'```[\s\S]*?```|`[^`]*`', text)),            'url_count': len(re.findall(r'https?://\S+', text)),        }        # 计算比例        features['paragraph_ratio'] = features['paragraph_count'] / max(features['total_length'] / 100, 1)        features['sentence_ratio'] = features['sentence_count'] / max(features['total_length'] / 100, 1)        features['code_ratio'] = features['code_count'] / max(features['total_length'] / 1000, 1)        features['url_ratio'] = features['url_count'] / max(features['total_length'] / 1000, 1)        return features    def _segment_by_paragraph(self, text: str, max_length: int) -> List[str]:        """按段落分段"""        paragraphs = text.split('\n\n')        segments = []        current_segment = ""        for para in paragraphs:            para = para.strip()            if not para:                continue            # 如果段落包含换行，先处理内部换行            if '\n' in para:                lines = para.split('\n')                para = ' '.join(line.strip() for line in lines)            if len(para) > max_length:                # 长段落需要进一步分割                sub_segments = self.fixed_segmenter.segment(para, max_length)                for sub_seg in sub_segments:                    if len(current_segment) + len(sub_seg) + 2 <= max_length:                        current_segment += "\n\n" + sub_seg if current_segment else sub_seg                    else:                        if current_segment:                            segments.append(current_segment)                        current_segment = sub_seg            elif len(current_segment) + len(para) + 2 <= max_length:                current_segment += "\n\n" + para if current_segment else para            else:                if current_segment:                    segments.append(current_segment)                current_segment = para        if current_segment:            segments.append(current_segment)        return segments    def _segment_with_code(self, text: str, max_length: int) -> List[str]:        """处理包含代码的文本"""        # 提取代码块        code_blocks = re.findall(r'```[\s\S]*?```|`[^`]*`', text)        text_without_code = re.sub(r'```[\s\S]*?```|`[^`]*`', '█CODE_BLOCK█', text)        # 分段文本        text_segments = self.sentence_segmenter.segment(text_without_code, max_length)        # 恢复代码块        result_segments = []        code_index = 0        for seg in text_segments:            while '█CODE_BLOCK█' in seg and code_index < len(code_blocks):                seg = seg.replace('█CODE_BLOCK█', code_blocks[code_index], 1)                code_index += 1            result_segments.append(seg)        # 处理剩余的代码块        while code_index < len(code_blocks):            if len(code_blocks[code_index]) > max_length:                # 代码块本身太长，需要分割                code_segments = self.fixed_segmenter.segment(code_blocks[code_index], max_length)                result_segments.extend(code_segments)            else:                if result_segments and len(result_segments[-1]) + len(code_blocks[code_index]) + 1 <= max_length:                    result_segments[-1] += "\n" + code_blocks[code_index]                else:                    result_segments.append(code_blocks[code_index])            code_index += 1        return result_segments    def _smart_segment(self, text: str, max_length: int, features: Dict) -> List[str]:        """智能分段（核心算法）"""        if self.use_jieba:            # 使用分词结果            words = list(jieba.cut(text))            segments = []            current_segment = ""            current_length = 0            for word in words:                word_length = len(word)                if current_length + word_length <= max_length:                    current_segment += word                    current_length += word_length                else:                    if current_segment:                        segments.append(current_segment)                    current_segment = word                    current_length = word_length            if current_segment:                segments.append(current_segment)            return segments        else:            # 回退到句子分割            return self.sentence_segmenter.segment(text, max_length)    def _merge_short_segments(self, segments: List[str], max_length: int) -> List[str]:        """合并过短的段落"""        if len(segments) <= 1:            return segments        merged_segments = []        current_segment = segments[0]        for i in range(1, len(segments)):            if (len(current_segment) + len(segments[i]) <= max_length and                len(current_segment) < self.min_segment_length):                current_segment += "\n\n" + segments[i]            else:                merged_segments.append(current_segment)                current_segment = segments[i]        merged_segments.append(current_segment)        return merged_segments    def get_segment_info(self, text: str) -> Dict:        """获取分段信息"""        features = self._analyze_text(text)        segments = self.segment(text, 32767)        return {            "strategy": SplitStrategy.SMART.value,            "total_length": len(text),            "segment_count": len(segments),            "avg_segment_length": sum(len(s) for s in segments) / len(segments) if segments else 0,            "features": features        }class ExcelTextSplitter:    """Excel文本拆分器"""    def __init__(self,                  strategy: SplitStrategy = SplitStrategy.SMART,                 max_length: int = 32767,                 num_workers: int = 4):        """        初始化        参数:            strategy: 拆分策略            max_length: 最大长度限制            num_workers: 并行工作线程数        """        self.max_length = max_length        self.num_workers = num_workers        self.segmenter = self._create_segmenter(strategy)        # 统计信息        self.stats = {            'total_rows': 0,            'processed_rows': 0,            'split_rows': 0,            'total_segments': 0,            'start_time': None,            'end_time': None        }    def _create_segmenter(self, strategy: SplitStrategy) -> TextSegmenter:        """创建分段器"""        if strategy == SplitStrategy.FIXED_LENGTH:            return FixedLengthSegmenter(preserve_words=True)        elif strategy == SplitStrategy.BY_SENTENCE:            return SentenceSegmenter(language='zh')        elif strategy == SplitStrategy.BY_PARAGRAPH:            return SentenceSegmenter(language='zh')  # 暂用句子分割器        elif strategy == SplitStrategy.SEMANTIC:            return SmartSegmenter(use_jieba=True)        elif strategy == SplitStrategy.SMART:            return SmartSegmenter(use_jieba=True)        else:            raise ValueError(f"不支持的策略: {strategy}")    def split_excel_file(self,                        input_path: str,                        output_path: str,                        text_column: str = '备注',                        id_column: str = '工单号',                        sheet_name: Optional[str] = None) -> pd.DataFrame:        """        拆分Excel文件中的文本        参数:            input_path: 输入文件路径            output_path: 输出文件路径            text_column: 文本列名            id_column: ID列名            sheet_name: 工作表名        返回:            拆分后的DataFrame        """        logger.info(f"开始处理文件: {input_path}")        self.stats['start_time'] = datetime.now()        try:            # 读取Excel文件            if input_path.endswith('.xlsx'):                df = pd.read_excel(input_path, sheet_name=sheet_name, dtype=str)            elif input_path.endswith('.csv'):                df = pd.read_csv(input_path, dtype=str, encoding='utf-8-sig')            else:                raise ValueError(f"不支持的文件格式: {input_path}")            self.stats['total_rows'] = len(df)            # 检查列是否存在            if text_column not in df.columns:                text_column = self._detect_text_column(df)            if id_column not in df.columns:                id_column = self._detect_id_column(df)            # 并行处理文本            logger.info(f"开始拆分文本，使用 {self.num_workers} 个线程")            with ThreadPoolExecutor(max_workers=self.num_workers) as executor:                # 提交任务                future_to_idx = {                    executor.submit(self._split_text, row[text_column], row[id_column] if id_column in row else idx): idx                    for idx, row in df.iterrows()                }                # 收集结果                results = []                futures = list(future_to_idx.keys())                for future in tqdm(as_completed(futures), total=len(futures), desc="处理进度"):                    try:                        result = future.result()                        results.append(result)                    except Exception as e:                        logger.error(f"处理失败: {str(e)}")                        results.append({                            'original_id': future_to_idx[future],                            'segments': ['处理失败'],                            'segment_count': 0,                            'needed_split': False                        })                # 按原始顺序排序                results.sort(key=lambda x: x['original_id'])            # 重构DataFrame            split_df = self._reconstruct_dataframe(df, results, id_column, text_column)            # 保存结果            self._save_results(split_df, output_path)            self.stats['end_time'] = datetime.now()            self._generate_report(results)            logger.info(f"处理完成，结果已保存到: {output_path}")            return split_df        except Exception as e:            logger.error(f"处理文件时出错: {str(e)}", exc_info=True)            raise    def _split_text(self, text: str, original_id: Union[str, int]) -> Dict:        """拆分单个文本"""        if pd.isna(text) or not isinstance(text, str) or not text.strip():            return {                'original_id': original_id,                'segments': [''],                'segment_count': 0,                'needed_split': False            }        text = str(text).strip()        # 检查是否需要拆分        if len(text) <= self.max_length:            return {                'original_id': original_id,                'segments': [text],                'segment_count': 1,                'needed_split': False            }        # 需要拆分        segments = self.segmenter.segment(text, self.max_length)        return {            'original_id': original_id,            'segments': segments,            'segment_count': len(segments),            'needed_split': True        }    def _reconstruct_dataframe(self,                              original_df: pd.DataFrame,                             results: List[Dict],                             id_column: str,                             text_column: str) -> pd.DataFrame:        """重构DataFrame"""        rows = []        for idx, result in enumerate(results):            original_row = original_df.iloc[idx].to_dict()            segments = result['segments']            for seg_idx, segment in enumerate(segments):                new_row = original_row.copy()                # 更新文本                new_row[text_column] = segment                # 添加分段信息                new_row['_segment_index'] = seg_idx + 1                new_row['_segment_total'] = len(segments)                new_row['_original_length'] = len(original_row.get(text_column, ''))                new_row['_segment_length'] = len(segment)                new_row['_needed_split'] = result['needed_split']                rows.append(new_row)        result_df = pd.DataFrame(rows)        # 重新排序列        column_order = [id_column, text_column, '_segment_index', '_segment_total',                        '_original_length', '_segment_length', '_needed_split']        other_columns = [col for col in result_df.columns if col not in column_order]        return result_df[column_order + other_columns]    def _save_results(self, df: pd.DataFrame, output_path: str):        """保存结果"""        if output_path.endswith('.xlsx'):            # 保存到Excel，使用openpyxl引擎避免限制            with pd.ExcelWriter(output_path, engine='openpyxl') as writer:                df.to_excel(writer, index=False, sheet_name='拆分结果')                # 添加摘要工作表                summary = self._create_summary(df)                summary.to_excel(writer, index=False, sheet_name='处理摘要')        elif output_path.endswith('.csv'):            df.to_csv(output_path, index=False, encoding='utf-8-sig')        else:            raise ValueError(f"不支持的输出格式: {output_path}")    def _create_summary(self, df: pd.DataFrame) -> pd.DataFrame:        """创建处理摘要"""        total_rows = self.stats['total_rows']        processed_rows = len(df)        split_rows = df['_needed_split'].sum() if '_needed_split' in df.columns else 0        summary_data = {            '指标': ['总行数', '处理行数', '拆分行数', '总段数', '平均每行段数', '最长段长度', '最短段长度'],            '值': [                total_rows,                processed_rows,                split_rows,                df['_segment_total'].sum() if '_segment_total' in df.columns else processed_rows,                round(processed_rows / max(total_rows, 1), 2),                df['_segment_length'].max() if '_segment_length' in df.columns else 0,                df['_segment_length'].min() if '_segment_length' in df.columns else 0            ]        }        if self.stats['start_time'] and self.stats['end_time']:            duration = (self.stats['end_time'] - self.stats['start_time']).total_seconds()            summary_data['指标'].extend(['开始时间', '结束时间', '处理时长(秒)', '处理速度(行/秒)'])            summary_data['值'].extend([                self.stats['start_time'].strftime('%Y-%m-%d %H:%M:%S'),                self.stats['end_time'].strftime('%Y-%m-%d %H:%M:%S'),                round(duration, 2),                round(processed_rows / max(duration, 0.1), 2)            ])        return pd.DataFrame(summary_data)    def _generate_report(self, results: List[Dict]):        """生成处理报告"""        self.stats['processed_rows'] = len(results)        self.stats['split_rows'] = sum(1 for r in results if r['needed_split'])        self.stats['total_segments'] = sum(r['segment_count'] for r in results)    def _detect_text_column(self, df: pd.DataFrame) -> str:        """自动检测文本列"""        # 可能的文本列名        text_columns = ['备注', '内容', '文本', '描述', '说明', '详情',                        'remark', 'content', 'text', 'description', 'details']        for col in text_columns:            if col in df.columns:                return col        # 基于内容检测        for col in df.columns:            if df[col].dtype == 'object':  # 字符串类型                sample = df[col].dropna().sample(min(100, len(df)), random_state=42)                avg_length = sample.astype(str).str.len().mean()                if avg_length > 100:  # 平均长度超过100字符                    return col        raise ValueError("无法自动检测文本列，请指定text_column参数")    def _detect_id_column(self, df: pd.DataFrame) -> str:        """自动检测ID列"""        # 可能的ID列名        id_columns = ['工单号', '订单号', '编号', 'ID', 'id', '序号',                      'ticket_id', 'order_id', 'serial_no']        for col in id_columns:            if col in df.columns:                return col        # 基于内容检测        for col in df.columns:            if df[col].dtype in ['int64', 'float64']:  # 数值类型                return col            elif df[col].dtype == 'object':                # 检查是否像ID                sample = df[col].dropna().sample(min(100, len(df)), random_state=42)                if sample.astype(str).str.match(r'^\d+$').all():  # 全是数字                    return col        # 如果没有找到，使用第一列        return df.columns[0]# 使用示例def main():    """主函数 - 演示如何使用"""    print("工单文本拆分系统演示")    print("="*60)    # 创建示例数据    sample_data = []    # 生成各种长度的文本    test_texts = [        "这是一段很短的文本。",  # 短文本        "这是一段中等长度的文本。" * 100,  # 中等文本        ("这是一段很长的文本，包含多个段落。\n\n"  # 长文本，有段落         "这是第二段，包含详细的描述。" * 500),        ("错误日志开始：\n"  # 包含代码的文本         "```python\n"         "def process_data(data):\n"         "    try:\n"         "        result = complex_operation(data)\n"         "        return result\n"         "    except Exception as e:\n"         "        logger.error(f'处理失败: {e}')\n"         "        raise\n"         "```\n"         "错误信息：连接超时。") * 100,    ]    for i, text in enumerate(test_texts, 1):        sample_data.append({            '工单号': f'TK{1000 + i:04d}',            '客户姓名': f'客户{i}',            '问题类型': ['技术', '咨询', '投诉', '售后'][i % 4],            '备注': text * (i * 10),  # 使文本越来越长            '创建时间': f'2024-01-{i:02d} 10:00:00'        })    df = pd.DataFrame(sample_data)    print("原始数据示例:")    print(df[['工单号', '备注长度']].assign(        备注长度=df['备注'].apply(len)    ).to_string(index=False))    # 保存示例数据    input_file = '示例工单数据.xlsx'    df.to_excel(input_file, index=False)    print(f"\n示例数据已保存到: {input_file}")    # 使用不同策略拆分    strategies = [        (SplitStrategy.FIXED_LENGTH, "固定长度"),        (SplitStrategy.BY_SENTENCE, "按句子"),        (SplitStrategy.SMART, "智能拆分"),    ]    for strategy, name in strategies:        print(f"\n{'='*60}")        print(f"使用策略: {name}")        print("="*60)        splitter = ExcelTextSplitter(strategy=strategy, max_length=1000)  # 使用1000便于演示        output_file = f'拆分结果_{name}.xlsx'        try:            result_df = splitter.split_excel_file(                input_path=input_file,                output_path=output_file,                text_column='备注',                id_column='工单号'            )            print(f"处理完成！结果保存到: {output_file}")            print(f"原始行数: {len(df)}")            print(f"拆分后行数: {len(result_df)}")            print(f"拆分比例: {len(result_df)/len(df):.2f}:1")            # 显示拆分详情            if '_segment_total' in result_df.columns:                split_details = result_df.groupby('工单号').agg({                    '_segment_total': 'first',                    '_segment_length': ['min', 'max', 'mean']                }).round(2)                print(f"\n拆分详情:")                print(split_details.to_string())        except Exception as e:            print(f"处理失败: {str(e)}")    print("\n" + "="*60)    print("提示: 完整结果已保存到Excel文件，包含详细的分段信息")# 性能测试def performance_test():    """性能测试"""    import time    import random    print("\n性能测试")    print("="*60)    # 生成测试数据    print("生成测试数据...")    test_data = []    for i in range(1000):  # 1000行数据        # 生成随机长度的文本        length = random.randint(100, 50000)  # 100到50000字符        text = "这是一段测试文本。" * (length // 10)        test_data.append({            'ID': i + 1,            '文本': text[:length]  # 确保精确长度        })    test_df = pd.DataFrame(test_data)    test_file = '性能测试数据.xlsx'    test_df.to_excel(test_file, index=False)    print(f"测试数据已生成: {len(test_df)} 行，最大长度: {test_df['文本'].str.len().max()} 字符")    # 测试不同策略    strategies = [        (SplitStrategy.FIXED_LENGTH, "固定长度"),        (SplitStrategy.BY_SENTENCE, "按句子"),        (SplitStrategy.SMART, "智能拆分"),    ]    results = []    for strategy, name in strategies:        print(f"\n测试策略: {name}")        splitter = ExcelTextSplitter(strategy=strategy, num_workers=4)        start_time = time.time()        try:            result_df = splitter.split_excel_file(                input_path=test_file,                output_path=f'性能测试结果_{name}.xlsx',                text_column='文本',                id_column='ID'            )            end_time = time.time()            duration = end_time - start_time            results.append({                '策略': name,                '处理时间(秒)': round(duration, 2),                '原始行数': len(test_df),                '结果行数': len(result_df),                '拆分行数': result_df['_needed_split'].sum() if '_needed_split' in result_df.columns else 0,                '速度(行/秒)': round(len(test_df) / duration, 2),                '内存使用(MB)': round(result_df.memory_usage(deep=True).sum() / 1024 / 1024, 2)            })            print(f"  处理时间: {duration:.2f}秒")            print(f"  处理速度: {len(test_df)/duration:.2f} 行/秒")            print(f"  拆分比例: {len(result_df)/len(test_df):.2f}:1")        except Exception as e:            print(f"  测试失败: {str(e)}")            results.append({                '策略': name,                '处理时间(秒)': None,                '原始行数': len(test_df),                '结果行数': None,                '拆分行数': None,                '速度(行/秒)': None,                '内存使用(MB)': None            })    # 显示结果对比    print("\n" + "="*60)    print("性能对比结果:")    print("="*60)    results_df = pd.DataFrame(results)    print(results_df.to_string(index=False))    # 清理测试文件    import os    if os.path.exists(test_file):        os.remove(test_file)    for _, name in strategies:        result_file = f'性能测试结果_{name}.xlsx'        if os.path.exists(result_file):            os.remove(result_file)    print("\n测试完成！")if __name__ == "__main__":    print("工单文本智能拆分系统")    print("="*60)    # 运行演示    main()    # 运行性能测试    performance_test()    print("\n" + "="*60)    print("总结: Python智能拆分方案在准确性、性能和可维护性上全面超越VBA")

# 紧急解决方案 - 基础拆分def quick_split_excel(input_file, output_file, column_name):    """快速拆分Excel中的长文本"""    import pandas as pd    # 读取数据    df = pd.read_excel(input_file)    # 简单拆分函数    def split_text(text, max_len=32767):        if not isinstance(text, str):            return [str(text)]        return [text[i:i+max_len] for i in range(0, len(text), max_len)]    # 应用拆分    result_rows = []    for idx, row in df.iterrows():        texts = split_text(row[column_name])        for i, txt in enumerate(texts):            new_row = row.copy()            new_row[column_name] = txt            new_row['_split_index'] = i + 1            new_row['_split_total'] = len(texts)            result_rows.append(new_row)    # 保存结果    result_df = pd.DataFrame(result_rows)    result_df.to_excel(output_file, index=False)    return result_df

def detect_paragraphs(text):    """智能检测段落"""    # 多级段落分隔符    delimiters = [        '\n\n',      # 空行        '\r\n\r\n',  # Windows空行        '\n-\n',     # 分隔线        '\n•\n',     # 项目符号        '\n*\n',     # 星号分隔    ]    for delim in delimiters:        if delim in text:            return text.split(delim)    # 回退到句子分割    return re.split(r'[。！？!?]\s+', text)

def split_preserve_format(text, max_len):    """保留Markdown/HTML格式的拆分"""    # 检测代码块    code_blocks = re.findall(r'```.*?```', text, re.DOTALL)    # 临时替换    for i, block in enumerate(code_blocks):        text = text.replace(block, f'__CODE_BLOCK_{i}__')    # 拆分文本    segments = smart_split(text, max_len)    # 恢复代码块    for i, block in enumerate(code_blocks):        for j, seg in enumerate(segments):            segments[j] = seg.replace(f'__CODE_BLOCK_{i}__', block)    return segments

class IncrementalSplitter:    """增量式拆分器，处理超大文件"""    def __init__(self, chunk_size=10000):        self.chunk_size = chunk_size    def process_large_file(self, input_file, output_file):        """分批处理大文件"""        # 使用迭代器逐块读取        reader = pd.read_excel(input_file, chunksize=self.chunk_size)        with pd.ExcelWriter(output_file, engine='openpyxl') as writer:            for i, chunk in enumerate(reader):                processed = self.process_chunk(chunk)                processed.to_excel(writer, sheet_name=f'chunk_{i}', index=False)                # 释放内存                del processed        return self.merge_results(output_file)

from multiprocessing import cpu_count, Pooldef parallel_split(texts, max_len, n_jobs=None):    """并行拆分文本列表"""    if n_jobs is None:        n_jobs = cpu_count() - 1    with Pool(n_jobs) as pool:        # 分块处理避免大内存        chunk_size = len(texts) // (n_jobs * 4) or 1        results = []        for i in range(0, len(texts), chunk_size):            chunk = texts[i:i+chunk_size]            result = pool.starmap(                smart_split_text,                [(text, max_len) for text in chunk]            )            results.extend(result)        return results

def split_with_memory_optimize(text, max_len):    """内存优化的拆分"""    # 使用生成器避免一次性加载    def text_generator(text, max_len):        start = 0        while start < len(text):            # 智能寻找断点            end = find_optimal_break(text, start, start + max_len)            yield text[start:end]            start = end    return list(text_generator(text, max_len))def find_optimal_break(text, start, proposed_end):    """寻找最优断点"""    # 1. 首先尝试在句子边界    for i in range(proposed_end, start, -1):        if i < len(text) and text[i] in '。！？!?':            return i + 1    # 2. 其次在标点后    for i in range(proposed_end, start, -1):        if i < len(text) and text[i] in '，,;；':            return i + 1    # 3. 然后在空格后    for i in range(proposed_end, start, -1):        if i < len(text) and text[i].isspace():            return i + 1    # 4. 最后在字符边界    for i in range(proposed_end, start, -1):        if is_character_boundary(text, i):            return i    return proposed_end

# 错误：len()计算的是字符数，不是存储长度text = "你好，世界！"  # 5个字符print(len(text))  # 输出5，但UTF-8编码可能是15字节# 正确：考虑编码def get_byte_length(text, encoding='utf-8'):    return len(text.encode(encoding))

# 错误：只处理一种换行符parts = text.split('\n')# 正确：处理所有换行符import reparts = re.split(r'\r\n|\n|\r', text)

# 错误：一次性读取df = pd.read_excel('huge_file.xlsx')  # 内存爆炸！# 正确：分块读取chunk_size = 10000for chunk in pd.read_excel('huge_file.xlsx', chunksize=chunk_size):    process_chunk(chunk)

# 错误：假设所有文件都是utf-8with open('file.txt', 'r') as f:    text = f.read()# 正确：检测编码import chardetwith open('file.txt', 'rb') as f:    raw_data = f.read()    encoding = chardet.detect(raw_data)['encoding']    text = raw_data.decode(encoding)

# 错误：不验证拆分结果segments = split_text(text, max_len)# 正确：严格验证def validate_split(original, segments, max_len):    # 1. 长度一致    assert len(''.join(segments)) == len(original)    # 2. 每段不超过限制    for seg in segments:        assert len(seg) <= max_len    # 3. 内容一致    assert ''.join(segments) == original    # 4. 语义检查    for i in range(len(segments) - 1):        if not is_valid_break(segments[i][-5:], segments[i+1][:5]):            logger.warning(f"可能的语义断点问题: ...{segments[i][-5:]} || {segments[i+1][:5]}...")    return True

class EcommerceTicketSplitter(ExcelTextSplitter):    """电商工单专用拆分器"""    def __init__(self):        super().__init__(strategy=SplitStrategy.SMART)        # 电商特定配置        self.max_conv_turns = 20  # 最多20轮对话        self.preserve_emoji = True  # 保留表情符号        self.separate_qa = True  # 问答分开    def split_conversation(self, conversation):        """拆分对话历史"""        # 1. 按说话人分割        parts = re.split(r'【(客服|用户)】', conversation)        # 2. 智能合并短消息        merged = self.merge_short_messages(parts)        # 3. 按语义分段        segments = self.semantic_segment(merged)        return segments

class TechnicalTicketSplitter(ExcelTextSplitter):    """技术工单专用拆分器"""    def __init__(self):        super().__init__(strategy=SplitStrategy.SMART)        # 技术特定配置        self.preserve_code_blocks = True        self.handle_stack_traces = True        self.normalize_paths = True    def split_technical_text(self, text):        """拆分技术文本"""        # 1. 提取并保护代码块        protected = self.protect_code_blocks(text)        # 2. 处理错误堆栈        protected = self.protect_stack_traces(protected)        # 3. 标准化文件路径        protected = self.normalize_file_paths(protected)        # 4. 智能拆分        segments = self.segmenter.segment(protected, self.max_length)        # 5. 恢复保护的内容        segments = self.restore_protected_content(segments)        return segments