import pandas as pdimport numpy as npimport refrom typing import Dict, List, Tupleimport loggingclass CustomerDataCleaner:    """客户数据清洗器"""    def __init__(self, file_path: str):        self.file_path = file_path        self.df_raw = None        self.df_clean = None        self.cleaning_report = {}        self.setup_logging()    def setup_logging(self):        """设置日志系统"""        logging.basicConfig(level=logging.INFO,                           format='%(asctime)s - %(levelname)s - %(message)s')        self.logger = logging.getLogger(__name__)    def load_data(self) -> pd.DataFrame:        """加载客户数据"""        try:            if self.file_path.endswith('.xlsx'):                self.df_raw = pd.read_excel(self.file_path, engine='openpyxl')            elif self.file_path.endswith('.csv'):                self.df_raw = pd.read_csv(self.file_path, encoding='utf-8')            self.logger.info(f"成功加载数据，共{len(self.df_raw)}行记录")            return self.df_raw        except Exception as e:            self.logger.error(f"数据加载失败: {str(e)}")            raise    def explore_data(self) -> Dict:        """数据探索与质量评估"""        if self.df_raw is None:            self.load_data()        exploration_report = {            'total_records': len(self.df_raw),            'columns': list(self.df_raw.columns),            'missing_values': self.df_raw.isnull().sum().to_dict(),            'duplicate_records': self.df_raw.duplicated().sum(),            'data_types': self.df_raw.dtypes.to_dict()        }        self.logger.info("数据探索完成")        return exploration_report

def remove_duplicates(self, subset_columns: List[str]) -> pd.DataFrame:    """基于关键字段去重"""    initial_count = len(self.df_raw)    # 多策略去重    self.df_clean = self.df_raw.drop_duplicates(subset=subset_columns, keep='first')    # 相似度去重（用于名称近似的情况）    self.df_clean = self._fuzzy_deduplicate(self.df_clean, 'customer_name')    removed_count = initial_count - len(self.df_clean)    self.cleaning_report['duplicates_removed'] = removed_count    self.logger.info(f"去重完成，移除{removed_count}条重复记录")    return self.df_cleandef _fuzzy_deduplicate(self, df: pd.DataFrame, column: str) -> pd.DataFrame:    """模糊去重（处理名称近似情况）"""    from difflib import SequenceMatcher    def similarity(a, b):        return SequenceMatcher(None, a, b).ratio()    # 简单的相似度去重逻辑    indices_to_remove = []    for i in range(len(df)):        if i in indices_to_remove:            continue        for j in range(i+1, len(df)):            if similarity(str(df.iloc[i][column]), str(df.iloc[j][column])) > 0.8:                indices_to_remove.append(j)    return df.drop(df.index[indices_to_remove])def split_address_column(self, address_column: str,                         output_columns: List[str] = ['province', 'city', 'district']) -> pd.DataFrame:    """拆分地址字段"""    if self.df_clean is None:        self.df_clean = self.df_raw.copy()    def address_splitter(address):        """智能地址拆分"""        if pd.isna(address) or address == '':            return [None, None, None]        # 简单的地址拆分逻辑（实际项目可使用专业地址库）        address_str = str(address)        # 匹配省市区模式        province_pattern = r'([^省]+省)?([^市]+市)?([^区]+区)?'        match = re.match(province_pattern, address_str)        if match:            province = match.group(1) or ''            city = match.group(2) or ''            district = match.group(3) or ''            return [province, city, district]        return [None, None, None]    # 应用地址拆分    split_results = self.df_clean[address_column].apply(        lambda x: pd.Series(address_splitter(x), index=output_columns)    )    self.df_clean = pd.concat([self.df_clean, split_results], axis=1)    self.logger.info("地址字段拆分完成")    return self.df_clean

def standardize_phone_format(self, phone_column: str) -> pd.DataFrame:    """统一电话格式"""    def phone_formatter(phone):        if pd.isna(phone):            return None        phone_str = str(phone)        # 移除非数字字符        cleaned = re.sub(r'\D', '', phone_str)        # 手机号处理        if len(cleaned) == 11 and cleaned.startswith(('13', '14', '15', '16', '17', '18', '19')):            return f"{cleaned[:3]}-{cleaned[3:7]}-{cleaned[7:]}"        # 座机号处理        elif len(cleaned) == 12 and cleaned.startswith('0'):            return f"{cleaned[:4]}-{cleaned[4:8]}-{cleaned[8:]}"        elif len(cleaned) == 10 and cleaned.startswith('0'):            return f"{cleaned[:3]}-{cleaned[3:7]}-{cleaned[7:]}"        else:            return None  # 无效号码    self.df_clean[f'{phone_column}_standardized'] = self.df_clean[phone_column].apply(phone_formatter)    valid_phones = self.df_clean[f'{phone_column}_standardized'].notna().sum()    self.cleaning_report['valid_phones'] = valid_phones    self.cleaning_report['invalid_phones'] = len(self.df_clean) - valid_phones    self.logger.info("电话格式标准化完成")    return self.df_cleandef validate_email_addresses(self, email_column: str) -> pd.DataFrame:    """验证邮箱有效性"""    def email_validator(email):        if pd.isna(email):            return False, '缺失'        email_str = str(email).lower().strip()        pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'        if not re.match(pattern, email_str):            return False, '格式错误'        # 检查常见无效邮箱        invalid_domains = ['example.com', 'test.com', 'temp.com']        if any(domain in email_str for domain in invalid_domains):            return False, '无效域名'        return True, '有效'    validation_results = self.df_clean[email_column].apply(        lambda x: pd.Series(email_validator(x), index=['email_valid', 'email_status'])    )    self.df_clean = pd.concat([self.df_clean, validation_results], axis=1)    valid_count = self.df_clean['email_valid'].sum()    self.cleaning_report['valid_emails'] = valid_count    self.cleaning_report['invalid_emails'] = len(self.df_clean) - valid_count    self.logger.info("邮箱验证完成")    return self.df_clean

def generate_cleaning_report(self) -> Dict:    """生成详细清洗报告"""    if self.df_clean is None:        self.logger.error("请先执行数据清洗")        return {}    report = {        '清洗时间': pd.Timestamp.now().strftime('%Y-%m-%d %H:%M:%S'),        '原始记录数': len(self.df_raw),        '清洗后记录数': len(self.df_clean),        '记录减少率': f"{((len(self.df_raw) - len(self.df_clean)) / len(self.df_raw) * 100):.2f}%",        '数据质量评分': self._calculate_data_quality_score(),        '各字段完整性': self._calculate_field_completeness(),        '清洗详情': self.cleaning_report    }    self.logger.info("清洗报告生成完成")    return reportdef save_clean_data(self, output_path: str):    """保存清洗后的数据"""    if self.df_clean is None:        self.logger.error("没有可保存的清洗数据")        return    try:        with pd.ExcelWriter(output_path, engine='openpyxl') as writer:            self.df_clean.to_excel(writer, sheet_name='清洗后数据', index=False)            # 添加报告工作表            report_df = pd.DataFrame([self.generate_cleaning_report()]).T            report_df.columns = ['清洗报告']            report_df.to_excel(writer, sheet_name='清洗报告')        self.logger.info(f"清洗数据已保存至: {output_path}")    except Exception as e:        self.logger.error(f"数据保存失败: {str(e)}")

Sub CustomerDataCleaner()    Dim wsRaw As Worksheet, wsClean As Worksheet    Dim lastRow As Long, i As Long    Dim dict As Object    Dim cleanedCount As Long    ' 设置工作表    Set wsRaw = ThisWorkbook.Sheets("原始数据")    Set wsClean = ThisWorkbook.Sheets("清洗后数据")    ' 清空目标表    wsClean.Cells.Clear    wsClean.Range("A1").Value = "清洗开始时间: " & Now    ' 创建字典用于去重    Set dict = CreateObject("Scripting.Dictionary")    lastRow = wsRaw.Cells(wsRaw.Rows.Count, "A").End(xlUp).Row    cleanedCount = 0    ' 遍历处理每条记录    For i = 2 To lastRow        Dim customerKey As String        customerKey = wsRaw.Cells(i, 1).Value & "|" & wsRaw.Cells(i, 2).Value ' 姓名+电话作为唯一标识        If Not dict.Exists(customerKey) Then            dict.Add customerKey, True            cleanedCount = cleanedCount + 1            Call ProcessSingleRecord(wsRaw, wsClean, i, cleanedCount)        End If    Next i    ' 生成报告    Call GenerateVBAReport(wsClean, cleanedCount, lastRow - 1)    MsgBox "数据清洗完成！共处理 " & cleanedCount & " 条唯一记录"End Sub

Sub ProcessSingleRecord(wsRaw As Worksheet, wsClean As Worksheet, sourceRow As Long, targetRow As Long)    ' 处理单条客户记录    Dim phone As String, email As String, address As String    ' 获取原始数据    phone = CStr(wsRaw.Cells(sourceRow, 3).Value)    email = CStr(wsRaw.Cells(sourceRow, 4).Value)    address = CStr(wsRaw.Cells(sourceRow, 5).Value)    ' 数据清洗    Dim cleanPhone As String    cleanPhone = StandardizePhone(phone)    Dim cleanEmail As String    cleanEmail = ValidateEmail(email)    ' 地址拆分    Dim province As String, city As String, district As String    Call SplitAddress(address, province, city, district)    ' 写入清洗后数据    wsClean.Cells(targetRow, 1).Value = wsRaw.Cells(sourceRow, 1).Value ' 姓名    wsClean.Cells(targetRow, 2).Value = wsRaw.Cells(sourceRow, 2).Value ' 原电话    wsClean.Cells(targetRow, 3).Value = cleanPhone    wsClean.Cells(targetRow, 4).Value = cleanEmail    wsClean.Cells(targetRow, 5).Value = province    wsClean.Cells(targetRow, 6).Value = city    wsClean.Cells(targetRow, 7).Value = districtEnd SubFunction StandardizePhone(rawPhone As String) As String    ' 标准化电话格式    Dim cleanPhone As String    cleanPhone = Replace(rawPhone, " ", "")    cleanPhone = Replace(cleanPhone, "-", "")    cleanPhone = Replace(cleanPhone, "(", "")    cleanPhone = Replace(cleanPhone, ")", "")    ' 格式判断    If Len(cleanPhone) = 11 And Left(cleanPhone, 1) = "1" Then        StandardizePhone = Left(cleanPhone, 3) & "-" & Mid(cleanPhone, 4, 4) & "-" & Right(cleanPhone, 4)    ElseIf Len(cleanPhone) = 12 And Left(cleanPhone, 1) = "0" Then        StandardizePhone = Left(cleanPhone, 4) & "-" & Mid(cleanPhone, 5, 4) & "-" & Right(cleanPhone, 4)    Else        StandardizePhone = "无效号码"    End IfEnd Function

def ecommerce_customer_cleaning():    """电商客户数据清洗完整流程"""    # 初始化清洗器    cleaner = CustomerDataCleaner('ecommerce_customers.xlsx')    # 加载数据    raw_data = cleaner.load_data()    # 数据探索    exploration = cleaner.explore_data()    print("数据探索结果:", exploration)    # 执行清洗流程    cleaner.remove_duplicates(['customer_name', 'phone'])    cleaner.standardize_phone_format('phone')    cleaner.validate_email_addresses('email')    cleaner.split_address_column('address')    # 生成报告并保存    report = cleaner.generate_cleaning_report()    cleaner.save_clean_data('cleaned_ecommerce_customers.xlsx')    return cleaner# 执行清洗if __name__ == "__main__":    cleaner = ecommerce_customer_cleaning()    print("清洗完成！质量评分:", cleaner.cleaning_report.get('data_quality_score', 'N/A'))