Method process

internbootcamp/utils/data_postprocess.py:170–228

处理 jsonl 文件 Args: input_path: 输入文件路径 output_path: 输出文件路径（可选）。如果不提供，将自动生成为 input_path_processed.jsonl verbose: 是否打印详细信息 Returns: Dict[str, int]: 处理统计信息

(
        self,
        input_path: Union[str, Path],
        output_path: Optional[Union[str, Path]] = None,
        verbose: bool = True
    )

        return results
    
    def process(
        self,
        input_path: Union[str, Path],
        output_path: Optional[Union[str, Path]] = None,
        verbose: bool = True
    ) -> Dict[str, int]:
        """
        处理 jsonl 文件
        
        Args:
            input_path: 输入文件路径
            output_path: 输出文件路径（可选）。如果不提供，将自动生成为 input_path_processed.jsonl
            verbose: 是否打印详细信息
        
        Returns:
            Dict[str, int]: 处理统计信息
        """
        input_path = Path(input_path)
        
        # 如果未提供输出路径，自动生成
        if output_path is None:
            output_path = input_path.parent / f"{input_path.stem}_processed{input_path.suffix}"
        else:
            output_path = Path(output_path)
        
        # 确保输出目录存在
        output_path.parent.mkdir(parents=True, exist_ok=True)
        
        # 重置统计信息
        self.stats = defaultdict(int)
        
        if verbose:
            print(f"📖 正在读取输入文件: {input_path}")
        
        # 处理数据
        with jsonlines.open(input_path) as reader, \
             jsonlines.open(output_path, mode='w') as writer:
            
            for line in reader:
                self.stats['total_input'] += 1
                
                # 应用过滤器
                if not self._apply_filters(line):
                    self.stats['total_filtered'] += 1
                    continue
                
                # 应用转换器
                transformed_items = self._apply_transformers(line)
                
                # 写入结果
                for item in transformed_items:
                    writer.write(item)
                    self.stats['total_output'] += 1
        
        if verbose:
            self._print_stats()
            print(f"✅ 处理完成，结果已保存到: {output_path}")