try:
    df = pd.read_csv(input_file, encoding=encoding)
except FileNotFoundError:
    logger.error(f"輸入文件不存在: {input_file}")
    raise DataProcessingError(f"文件未找到: {input_file}")
except pd.errors.EmptyDataError:
    logger.error(f"輸入文件為空: {input_file}")
    raise DataProcessingError(f"空文件: {input_file}")
except UnicodeDecodeError:
    logger.error(f"文件編碼錯誤: {input_file}")
    raise DataProcessingError(f"編碼錯誤，請嘗試其他編碼格式")

try:
    # 數(shù)據(jù)類型轉(zhuǎn)換
    df[column] = pd.to_numeric(df[column], errors='coerce')
except Exception as e:
    logger.warning(f"列 {column} 數(shù)據(jù)類型轉(zhuǎn)換失敗: {str(e)}")
    # 記錄錯誤但繼續(xù)處理其他列

try:
    df = pd.read_csv(input_file, chunksize=chunk_size)
    # 分塊處理大文件
except MemoryError:
    logger.error("內(nèi)存不足，無法處理大文件")
    raise DataProcessingError("內(nèi)存不足，請減小文件大小或增加系統(tǒng)內(nèi)存")

if not os.path.exists(config_file):
    logger.error(f"配置文件不存在: {config_file}")
    raise DataProcessingError(f"配置文件缺失: {config_file}")

try:
    with open(config_file, 'r', encoding='utf-8') as f:
        config = json.load(f)
except json.JSONDecodeError:
    logger.error(f"配置文件格式錯誤: {config_file}")
    raise DataProcessingError(f"配置文件格式無效")

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
CSV數(shù)據(jù)清洗和轉(zhuǎn)換工具
功能：清洗、轉(zhuǎn)換和驗證CSV格式數(shù)據(jù)
作者：Cline
版本：1.0
"""

import pandas as pd
import numpy as np
import argparse
import sys
import json
import logging
import os
from datetime import datetime
import chardet
from typing import Dict, List, Any, Optional
import warnings

# 忽略警告信息
warnings.filterwarnings('ignore')

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('csv_cleaner.log'),
        logging.StreamHandler(sys.stdout)
    ]
)
logger = logging.getLogger(__name__)

class DataProcessingError(Exception):
    """數(shù)據(jù)處理異常類"""
    pass

class CSVCleaner:
    def __init__(self, config: Dict[str, Any]):
        self.input_file = config.get('input_file')
        self.output_file = config.get('output_file', 'cleaned_data.csv')
        self.encoding = config.get('encoding', 'auto')
        self.separator = config.get('separator', ',')
        self.rules_file = config.get('rules_file')
        self.output_format = config.get('output_format', 'csv')
        self.chunk_size = config.get('chunk_size', 10000)
        self.backup_original = config.get('backup_original', True)
        
        # 清洗規(guī)則
        self.cleaning_rules = config.get('cleaning_rules', {})
        
        # 處理結(jié)果統(tǒng)計
        self.stats = {
            'total_rows': 0,
            'cleaned_rows': 0,
            'removed_rows': 0,
            'modified_cells': 0,
            'errors': []
        }
        
    def detect_encoding(self, file_path: str) -> str:
        """自動檢測文件編碼"""
        try:
            with open(file_path, 'rb') as f:
                raw_data = f.read(10000)  # 讀取前10KB數(shù)據(jù)
                result = chardet.detect(raw_data)
                encoding = result['encoding']
                confidence = result['confidence']
                
                logger.info(f"檢測到文件編碼: {encoding} (置信度: {confidence:.2f})")
                return encoding if confidence > 0.7 else 'utf-8'
                
        except Exception as e:
            logger.warning(f"編碼檢測失敗，使用默認編碼: {str(e)}")
            return 'utf-8'
            
    def load_data(self) -> pd.DataFrame:
        """加載CSV數(shù)據(jù)"""
        logger.info(f"開始加載數(shù)據(jù)文件: {self.input_file}")
        
        try:
            # 自動檢測編碼
            if self.encoding == 'auto':
                self.encoding = self.detect_encoding(self.input_file)
                
            # 加載數(shù)據(jù)
            df = pd.read_csv(
                self.input_file,
                sep=self.separator,
                encoding=self.encoding,
                low_memory=False
            )
            
            self.stats['total_rows'] = len(df)
            logger.info(f"成功加載 {len(df)} 行數(shù)據(jù)，{len(df.columns)} 列")
            return df
            
        except FileNotFoundError:
            logger.error(f"輸入文件不存在: {self.input_file}")
            raise DataProcessingError(f"文件未找到: {self.input_file}")
        except pd.errors.EmptyDataError:
            logger.error(f"輸入文件為空: {self.input_file}")
            raise DataProcessingError(f"空文件: {self.input_file}")
        except UnicodeDecodeError as e:
            logger.error(f"文件編碼錯誤: {str(e)}")
            raise DataProcessingError(f"編碼錯誤，請嘗試指定其他編碼格式")
        except Exception as e:
            logger.error(f"加載數(shù)據(jù)時發(fā)生錯誤: {str(e)}")
            raise DataProcessingError(f"數(shù)據(jù)加載失敗: {str(e)}")
            
    def backup_original(self):
        """備份原始文件"""
        if not self.backup_original:
            return
            
        try:
            backup_name = f"{self.input_file}.backup_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
            import shutil
            shutil.copy2(self.input_file, backup_name)
            logger.info(f"原始文件已備份到: {backup_name}")
        except Exception as e:
            logger.warning(f"備份原始文件失敗: {str(e)}")
            
    def load_cleaning_rules(self) -> Dict[str, Any]:
        """加載清洗規(guī)則"""
        if not self.rules_file or not os.path.exists(self.rules_file):
            logger.info("未指定清洗規(guī)則文件，使用默認規(guī)則")
            return self.cleaning_rules
            
        try:
            with open(self.rules_file, 'r', encoding='utf-8') as f:
                rules = json.load(f)
            logger.info(f"成功加載清洗規(guī)則文件: {self.rules_file}")
            return rules
        except Exception as e:
            logger.error(f"加載清洗規(guī)則文件失敗: {str(e)}")
            return self.cleaning_rules
            
    def clean_missing_values(self, df: pd.DataFrame, rules: Dict[str, Any]) -> pd.DataFrame:
        """處理缺失值"""
        missing_rules = rules.get('missing_values', {})
        
        for column, strategy in missing_rules.items():
            if column not in df.columns:
                logger.warning(f"列 {column} 不存在于數(shù)據(jù)中")
                continue
                
            logger.info(f"處理列 {column} 的缺失值，策略: {strategy}")
            
            if strategy == 'drop_rows':
                # 刪除包含缺失值的行
                original_count = len(df)
                df = df.dropna(subset=[column])
                removed_count = original_count - len(df)
                self.stats['removed_rows'] += removed_count
                logger.info(f"刪除了 {removed_count} 行包含缺失值的數(shù)據(jù)")
                
            elif strategy == 'fill_zero':
                # 用0填充
                filled_count = df[column].isna().sum()
                df[column] = df[column].fillna(0)
                self.stats['modified_cells'] += filled_count
                logger.info(f"用0填充了 {filled_count} 個缺失值")
                
            elif strategy == 'fill_mean':
                # 用均值填充（僅適用于數(shù)值列）
                if df[column].dtype in ['int64', 'float64']:
                    mean_value = df[column].mean()
                    filled_count = df[column].isna().sum()
                    df[column] = df[column].fillna(mean_value)
                    self.stats['modified_cells'] += filled_count
                    logger.info(f"用均值 {mean_value:.2f} 填充了 {filled_count} 個缺失值")
                else:
                    logger.warning(f"列 {column} 不是數(shù)值類型，無法使用均值填充")
                    
            elif strategy == 'fill_forward':
                # 向前填充
                filled_count = df[column].isna().sum()
                df[column] = df[column].fillna(method='ffill')
                self.stats['modified_cells'] += filled_count
                logger.info(f"向前填充了 {filled_count} 個缺失值")
                
            elif isinstance(strategy, str):
                # 用指定值填充
                filled_count = df[column].isna().sum()
                df[column] = df[column].fillna(strategy)
                self.stats['modified_cells'] += filled_count
                logger.info(f"用 '{strategy}' 填充了 {filled_count} 個缺失值")
                
        return df
        
    def remove_duplicates(self, df: pd.DataFrame, rules: Dict[str, Any]) -> pd.DataFrame:
        """去除重復記錄"""
        duplicate_rules = rules.get('duplicates', {})
        subset = duplicate_rules.get('subset')
        keep = duplicate_rules.get('keep', 'first')
        
        original_count = len(df)
        
        if subset:
            # 基于指定列去重
            if isinstance(subset, str):
                subset = [subset]
            df = df.drop_duplicates(subset=subset, keep=keep)
        else:
            # 基于所有列去重
            df = df.drop_duplicates(keep=keep)
            
        removed_count = original_count - len(df)
        self.stats['removed_rows'] += removed_count
        logger.info(f"去除了 {removed_count} 行重復數(shù)據(jù)")
        return df
        
    def clean_text_data(self, df: pd.DataFrame, rules: Dict[str, Any]) -> pd.DataFrame:
        """清理文本數(shù)據(jù)"""
        text_rules = rules.get('text_cleaning', {})
        
        for column, cleaning_ops in text_rules.items():
            if column not in df.columns:
                logger.warning(f"列 {column} 不存在于數(shù)據(jù)中")
                continue
                
            if df[column].dtype != 'object':
                logger.warning(f"列 {column} 不是文本類型，跳過文本清理")
                continue
                
            logger.info(f"清理列 {column} 的文本數(shù)據(jù)")
            original_modified = self.stats['modified_cells']
            
            # 轉(zhuǎn)換為字符串類型
            df[column] = df[column].astype(str)
            
            for op in cleaning_ops:
                if op == 'strip_whitespace':
                    # 去除首尾空格
                    modified_mask = df[column].str.strip() != df[column]
                    modified_count = modified_mask.sum()
                    df.loc[modified_mask, column] = df.loc[modified_mask, column].str.strip()
                    self.stats['modified_cells'] += modified_count
                    
                elif op == 'remove_extra_spaces':
                    # 去除多余空格
                    modified_mask = df[column].str.replace(r'\s+', ' ', regex=True) != df[column]
                    modified_count = modified_mask.sum()
                    df.loc[modified_mask, column] = df.loc[modified_mask, column].str.replace(r'\s+', ' ', regex=True)
                    self.stats['modified_cells'] += modified_count
                    
                elif op == 'lowercase':
                    # 轉(zhuǎn)換為小寫
                    modified_mask = df[column].str.lower() != df[column]
                    modified_count = modified_mask.sum()
                    df.loc[modified_mask, column] = df.loc[modified_mask, column].str.lower()
                    self.stats['modified_cells'] += modified_count
                    
                elif op == 'uppercase':
                    # 轉(zhuǎn)換為大寫
                    modified_mask = df[column].str.upper() != df[column]
                    modified_count = modified_mask.sum()
                    df.loc[modified_mask, column] = df.loc[modified_mask, column].str.upper()
                    self.stats['modified_cells'] += modified_count
                    
                elif op == 'remove_special_chars':
                    # 去除特殊字符
                    modified_mask = df[column].str.replace(r'[^\w\s]', '', regex=True) != df[column]
                    modified_count = modified_mask.sum()
                    df.loc[modified_mask, column] = df.loc[modified_mask, column].str.replace(r'[^\w\s]', '', regex=True)
                    self.stats['modified_cells'] += modified_count
                    
            logger.info(f"列 {column} 文本清理完成，修改了 {self.stats['modified_cells'] - original_modified} 個單元格")
            
        return df
        
    def standardize_dates(self, df: pd.DataFrame, rules: Dict[str, Any]) -> pd.DataFrame:
        """標準化日期格式"""
        date_rules = rules.get('date_standardization', {})
        
        for column, format_info in date_rules.items():
            if column not in df.columns:
                logger.warning(f"列 {column} 不存在于數(shù)據(jù)中")
                continue
                
            logger.info(f"標準化列 {column} 的日期格式")
            original_modified = self.stats['modified_cells']
            
            try:
                target_format = format_info.get('target_format', '%Y-%m-%d')
                source_formats = format_info.get('source_formats', [])
                
                # 嘗試轉(zhuǎn)換日期格式
                if source_formats:
                    for fmt in source_formats:
                        try:
                            df[column] = pd.to_datetime(df[column], format=fmt, errors='coerce')
                            break
                        except:
                            continue
                else:
                    # 自動推斷日期格式
                    df[column] = pd.to_datetime(df[column], infer_datetime_format=True, errors='coerce')
                    
                # 格式化為目標格式
                df[column] = df[column].dt.strftime(target_format)
                modified_count = df[column].notna().sum()
                self.stats['modified_cells'] += modified_count
                
                logger.info(f"列 {column} 日期標準化完成，處理了 {modified_count} 個日期值")
                
            except Exception as e:
                logger.error(f"標準化列 {column} 日期格式時出錯: {str(e)}")
                
        return df
        
    def validate_data(self, df: pd.DataFrame, rules: Dict[str, Any]) -> pd.DataFrame:
        """數(shù)據(jù)驗證"""
        validation_rules = rules.get('validation', {})
        
        for column, validations in validation_rules.items():
            if column not in df.columns:
                logger.warning(f"列 {column} 不存在于數(shù)據(jù)中")
                continue
                
            logger.info(f"驗證列 {column} 的數(shù)據(jù)")
            
            for validation in validations:
                rule_type = validation.get('type')
                rule_value = validation.get('value')
                action = validation.get('action', 'log')
                
                if rule_type == 'range':
                    # 數(shù)值范圍驗證
                    min_val, max_val = rule_value
                    invalid_mask = (df[column] < min_val) | (df[column] > max_val)
                    
                elif rule_type == 'length':
                    # 字符串長度驗證
                    if isinstance(rule_value, list):
                        min_len, max_len = rule_value
                        invalid_mask = (df[column].str.len() < min_len) | (df[column].str.len() > max_len)
                    else:
                        invalid_mask = df[column].str.len() != rule_value
                        
                elif rule_type == 'pattern':
                    # 正則表達式驗證
                    invalid_mask = ~df[column].str.match(rule_value, na=False)
                    
                elif rule_type == 'not_null':
                    # 非空驗證
                    invalid_mask = df[column].isna()
                    
                else:
                    continue
                    
                invalid_count = invalid_mask.sum()
                if invalid_count > 0:
                    logger.warning(f"列 {column} 發(fā)現(xiàn) {invalid_count} 條不符合 {rule_type} 規(guī)則的數(shù)據(jù)")
                    
                    if action == 'remove':
                        # 刪除不符合規(guī)則的行
                        df = df[~invalid_mask]
                        self.stats['removed_rows'] += invalid_count
                        logger.info(f"刪除了 {invalid_count} 行不符合規(guī)則的數(shù)據(jù)")
                    elif action == 'mark':
                        # 標記不符合規(guī)則的數(shù)據(jù)
                        error_msg = f"列 {column} 不符合 {rule_type} 規(guī)則"
                        self.stats['errors'].append({
                            'column': column,
                            'count': invalid_count,
                            'message': error_msg
                        })
                        
        return df
        
    def process_chunk(self, chunk: pd.DataFrame, rules: Dict[str, Any]) -> pd.DataFrame:
        """處理數(shù)據(jù)塊"""
        try:
            # 1. 處理缺失值
            chunk = self.clean_missing_values(chunk, rules)
            
            # 2. 去除重復記錄
            chunk = self.remove_duplicates(chunk, rules)
            
            # 3. 清理文本數(shù)據(jù)
            chunk = self.clean_text_data(chunk, rules)
            
            # 4. 標準化日期格式
            chunk = self.standardize_dates(chunk, rules)
            
            # 5. 數(shù)據(jù)驗證
            chunk = self.validate_data(chunk, rules)
            
            self.stats['cleaned_rows'] += len(chunk)
            return chunk
            
        except Exception as e:
            logger.error(f"處理數(shù)據(jù)塊時出錯: {str(e)}")
            raise DataProcessingError(f"數(shù)據(jù)塊處理失敗: {str(e)}")
            
    def process_large_file(self, df: pd.DataFrame, rules: Dict[str, Any]) -> pd.DataFrame:
        """處理大文件（分塊處理）"""
        logger.info("開始分塊處理大文件...")
        
        processed_chunks = []
        
        # 分塊處理
        for i, chunk in enumerate(np.array_split(df, max(1, len(df) // self.chunk_size))):
            logger.info(f"處理第 {i+1} 塊數(shù)據(jù) ({len(chunk)} 行)")
            cleaned_chunk = self.process_chunk(chunk, rules)
            processed_chunks.append(cleaned_chunk)
            
        # 合并所有塊
        final_df = pd.concat(processed_chunks, ignore_index=True)
        logger.info(f"大文件處理完成，共處理 {len(final_df)} 行數(shù)據(jù)")
        return final_df
        
    def run_cleaning(self):
        """運行數(shù)據(jù)清洗"""
        logger.info("開始CSV數(shù)據(jù)清洗...")
        
        try:
            # 1. 備份原始文件
            self.backup_original()
            
            # 2. 加載數(shù)據(jù)
            df = self.load_data()
            
            # 3. 加載清洗規(guī)則
            rules = self.load_cleaning_rules()
            
            # 4. 處理數(shù)據(jù)
            if len(df) > self.chunk_size:
                # 大文件分塊處理
                cleaned_df = self.process_large_file(df, rules)
            else:
                # 小文件直接處理
                cleaned_df = self.process_chunk(df, rules)
                
            # 5. 保存結(jié)果
            self.save_results(cleaned_df)
            
            # 6. 生成報告
            self.generate_report()
            
            logger.info("CSV數(shù)據(jù)清洗完成")
            return cleaned_df
            
        except Exception as e:
            logger.error(f"數(shù)據(jù)清洗過程中發(fā)生錯誤: {str(e)}")
            raise DataProcessingError(f"清洗失敗: {str(e)}")
            
    def save_results(self, df: pd.DataFrame):
        """保存處理結(jié)果"""
        try:
            # 確保輸出目錄存在
            output_dir = os.path.dirname(self.output_file) if os.path.dirname(self.output_file) else '.'
            os.makedirs(output_dir, exist_ok=True)
            
            if self.output_format == 'csv':
                df.to_csv(self.output_file, index=False, encoding='utf-8-sig')
            elif self.output_format == 'excel':
                df.to_excel(self.output_file, index=False)
            elif self.output_format == 'json':
                df.to_json(self.output_file, orient='records', force_ascii=False, indent=2)
            else:
                logger.error(f"不支持的輸出格式: {self.output_format}")
                raise DataProcessingError(f"不支持的輸出格式: {self.output_format}")
                
            logger.info(f"處理結(jié)果已保存到 {self.output_file}")
            
        except Exception as e:
            logger.error(f"保存處理結(jié)果時出錯: {str(e)}")
            raise DataProcessingError(f"保存失敗: {str(e)}")
            
    def generate_report(self):
        """生成處理報告"""
        try:
            report = {
                'timestamp': datetime.now().isoformat(),
                'input_file': self.input_file,
                'output_file': self.output_file,
                'processing_stats': self.stats,
                'cleaning_rules': self.cleaning_rules
            }
            
            report_file = f"{self.output_file}.report.json"
            with open(report_file, 'w', encoding='utf-8') as f:
                json.dump(report, f, indent=2, ensure_ascii=False)
                
            logger.info(f"處理報告已保存到 {report_file}")
            
            # 打印簡要報告
            print("\n" + "="*50)
            print("CSV數(shù)據(jù)清洗報告")
            print("="*50)
            print(f"處理時間: {report['timestamp']}")
            print(f"輸入文件: {self.input_file}")
            print(f"輸出文件: {self.output_file}")
            print("-"*50)
            print(f"總行數(shù): {self.stats['total_rows']}")
            print(f"清洗后行數(shù): {self.stats['cleaned_rows']}")
            print(f"刪除行數(shù): {self.stats['removed_rows']}")
            print(f"修改單元格數(shù): {self.stats['modified_cells']}")
            print(f"錯誤記錄數(shù): {len(self.stats['errors'])}")
            print("="*50)
            
        except Exception as e:
            logger.error(f"生成處理報告時出錯: {str(e)}")

def create_sample_rules():
    """創(chuàng)建示例清洗規(guī)則文件"""
    sample_rules = {
        "missing_values": {
            "name": "fill_unknown",
            "age": "fill_mean",
            "email": "drop_rows"
        },
        "duplicates": {
            "subset": ["name", "email"],
            "keep": "first"
        },
        "text_cleaning": {
            "name": ["strip_whitespace", "remove_extra_spaces"],
            "description": ["strip_whitespace", "remove_special_chars"]
        },
        "date_standardization": {
            "created_date": {
                "target_format": "%Y-%m-%d",
                "source_formats": ["%Y/%m/%d", "%d-%m-%Y"]
            }
        },
        "validation": {
            "age": [
                {
                    "type": "range",
                    "value": [0, 120],
                    "action": "remove"
                }
            ],
            "email": [
                {
                    "type": "pattern",
                    "value": r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$",
                    "action": "mark"
                }
            ]
        }
    }
    
    with open('sample_cleaning_rules.json', 'w', encoding='utf-8') as f:
        json.dump(sample_rules, f, indent=2, ensure_ascii=False)
    logger.info("示例清洗規(guī)則文件已創(chuàng)建: sample_cleaning_rules.json")

def main():
    parser = argparse.ArgumentParser(description='CSV數(shù)據(jù)清洗和轉(zhuǎn)換工具')
    parser.add_argument('input_file', help='輸入CSV文件路徑')
    parser.add_argument('-o', '--output', help='輸出文件路徑')
    parser.add_argument('-e', '--encoding', default='auto', help='文件編碼 (auto/utf-8/gbk等)')
    parser.add_argument('-s', '--separator', default=',', help='字段分隔符')
    parser.add_argument('-r', '--rules', help='清洗規(guī)則文件路徑')
    parser.add_argument('-f', '--format', choices=['csv', 'excel', 'json'], default='csv', help='輸出格式')
    parser.add_argument('--chunk-size', type=int, default=10000, help='分塊處理大小')
    parser.add_argument('--no-backup', action='store_true', help='不備份原始文件')
    parser.add_argument('--sample-rules', action='store_true', help='創(chuàng)建示例清洗規(guī)則文件')
    
    args = parser.parse_args()
    
    if args.sample_rules:
        create_sample_rules()
        return
        
    # 配置處理參數(shù)
    config = {
        'input_file': args.input_file,
        'output_file': args.output or f"cleaned_{os.path.basename(args.input_file)}",
        'encoding': args.encoding,
        'separator': args.separator,
        'rules_file': args.rules,
        'output_format': args.format,
        'chunk_size': args.chunk_size,
        'backup_original': not args.no_backup
    }
    
    # 創(chuàng)建清洗器實例
    cleaner = CSVCleaner(config)
    
    try:
        # 執(zhí)行清洗
        cleaner.run_cleaning()
        
    except KeyboardInterrupt:
        logger.info("數(shù)據(jù)清洗被用戶中斷")
        sys.exit(1)
    except DataProcessingError as e:
        logger.error(f"數(shù)據(jù)處理錯誤: {str(e)}")
        sys.exit(1)
    except Exception as e:
        logger.error(f"數(shù)據(jù)清洗過程中發(fā)生未知錯誤: {str(e)}")
        sys.exit(1)

if __name__ == '__main__':
    main()

# 基本數(shù)據(jù)清洗
python csv_cleaner.py data.csv

# 指定輸出文件
python csv_cleaner.py data.csv -o cleaned_data.csv

# 指定文件編碼
python csv_cleaner.py data.csv -e gbk

# 指定分隔符
python csv_cleaner.py data.csv -s ';'

# 使用自定義清洗規(guī)則
python csv_cleaner.py data.csv -r cleaning_rules.json

# 創(chuàng)建示例清洗規(guī)則文件
python csv_cleaner.py --sample-rules

# 輸出為Excel格式
python csv_cleaner.py data.csv -f excel -o result.xlsx

# 輸出為JSON格式
python csv_cleaner.py data.csv -f json -o result.json

# 調(diào)整分塊大小處理大文件
python csv_cleaner.py large_data.csv --chunk-size 50000

# 不備份原始文件
python csv_cleaner.py data.csv --no-backup

{
  "missing_values": {
    "name": "fill_unknown",
    "age": "fill_mean",
    "salary": "fill_zero",
    "email": "drop_rows"
  },
  "duplicates": {
    "subset": ["name", "email"],
    "keep": "first"
  },
  "text_cleaning": {
    "name": ["strip_whitespace", "remove_extra_spaces", "lowercase"],
    "description": ["strip_whitespace", "remove_special_chars"]
  },
  "date_standardization": {
    "hire_date": {
      "target_format": "%Y-%m-%d",
      "source_formats": ["%Y/%m/%d", "%d-%m-%Y", "%m/%d/%Y"]
    }
  },
  "validation": {
    "age": [
      {
        "type": "range",
        "value": [0, 120],
        "action": "remove"
      }
    ],
    "email": [
      {
        "type": "pattern",
        "value": "^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}$",
        "action": "mark"
      }
    ],
    "phone": [
      {
        "type": "pattern",
        "value": "^1[3-9]\\d{9}$",
        "action": "mark"
      }
    ]
  }
}

import glob
import os

# 處理目錄下所有CSV文件
csv_files = glob.glob("data/*.csv")

for csv_file in csv_files:
    config = {
        'input_file': csv_file,
        'output_file': f"cleaned_{os.path.basename(csv_file)}",
        'rules_file': 'cleaning_rules.json'
    }
    
    cleaner = CSVCleaner(config)
    cleaner.run_cleaning()

# 檢查數(shù)據(jù)質(zhì)量指標
def check_data_quality(df):
    quality_metrics = {
        'total_rows': len(df),
        'missing_values': df.isnull().sum().to_dict(),
        'duplicate_rows': df.duplicated().sum(),
        'data_types': df.dtypes.astype(str).to_dict()
    }
    return quality_metrics