""" Excel文件质量验证器 用于在导入前检查文件数据质量，防止不合格数据进入系统 """ import logging import re from datetime import datetime, timedelta from typing import Dict, List, Any, Optional, Tuple from pathlib import Path import pandas as pd import numpy as np logger = logging.getLogger(__name__) class ValidationResult: """验证结果类""" def __init__(self): self.is_valid = True self.errors: List[str] = [] self.warnings: List[str] = [] self.statistics: Dict[str, Any] = {} def add_error(self, message: str): """添加错误信息""" self.errors.append(message) self.is_valid = False def add_warning(self, message: str): """添加警告信息""" self.warnings.append(message) def to_dict(self) -> Dict[str, Any]: """转换为字典格式""" return { 'is_valid': self.is_valid, 'errors': self.errors, 'warnings': self.warnings, 'statistics': self.statistics } class FileValidator: """Excel文件验证器""" # 必需的列 REQUIRED_COLUMNS = [ '原始订单编号', '线上宝贝名称', '数量', '订单单价' ] # 重要但可选的列 IMPORTANT_COLUMNS = [ '线上销售属性', '线上商家编码', '网店名称', '交易状态', '退款状态', '付款时间' # 移到可选字段，允许空值 ] # 数值型字段 NUMERIC_FIELDS = [ '数量', '订单单价', '订单金额', '实付金额', '退款金额' ] # 日期型字段 DATE_FIELDS = [ '付款时间', '下载时间', '拍下时间', '交易完成时间', '发货时间' ] def __init__(self): """初始化验证器""" self.result = ValidationResult() def validate_excel_file(self, file_path: Path) -> ValidationResult: """ 验证Excel文件的完整流程 Args: file_path: Excel文件路径 Returns: ValidationResult: 验证结果 """ self.result = ValidationResult() try: # 1. 检查文件是否存在和可读 if not file_path.exists(): self.result.add_error(f"文件不存在: {file_path}") return self.result if not file_path.is_file(): self.result.add_error(f"不是有效的文件: {file_path}") return self.result # 2. 读取Excel文件 try: df = self._read_excel_safely(file_path) except Exception as e: self.result.add_error(f"无法读取Excel文件: {str(e)}") return self.result # 3. 检查文件是否为空 if df.empty: self.result.add_error("Excel文件为空，没有数据可导入") return self.result # 记录统计信息 self.result.statistics['total_rows'] = len(df) self.result.statistics['total_columns'] = len(df.columns) # 4. 验证必需列 self._validate_required_columns(df) # 5. 验证数据质量 if self.result.is_valid: # 只有列完整时才验证数据 self._validate_data_quality(df) # 6. 验证业务逻辑 if self.result.is_valid: self._validate_business_logic(df) # 7. 检查潜在的数据问题 self._check_potential_issues(df) except Exception as e: self.result.add_error(f"验证过程中发生错误: {str(e)}") logger.error(f"File validation error: {e}", exc_info=True) return self.result def _read_excel_safely(self, file_path: Path) -> pd.DataFrame: """ 安全读取Excel文件 Args: file_path: 文件路径 Returns: DataFrame: 读取的数据 """ # 尝试不同的读取方式 try: # 首先尝试读取所有sheet excel_file = pd.ExcelFile(file_path) # 如果有多个sheet，使用第一个包含数据的sheet for sheet_name in excel_file.sheet_names: df = pd.read_excel(file_path, sheet_name=sheet_name, header=None) # 查找标题行（包含"原始订单编号"的行） for idx, row in df.iterrows(): if '原始订单编号' in str(row.values): # 找到标题行，重新读取 df = pd.read_excel(file_path, sheet_name=sheet_name, header=idx) return df # 如果没找到标题行，使用第一行作为标题 return pd.read_excel(file_path, header=0) except Exception as e: raise ValueError(f"读取Excel文件失败: {str(e)}") def _validate_required_columns(self, df: pd.DataFrame): """ 验证必需的列是否存在 Args: df: 数据DataFrame """ missing_columns = [] for col in self.REQUIRED_COLUMNS: if col not in df.columns: missing_columns.append(col) if missing_columns: self.result.add_error(f"缺少必需列: {', '.join(missing_columns)}") # 检查重要但可选的列 missing_important = [] for col in self.IMPORTANT_COLUMNS: if col not in df.columns: missing_important.append(col) if missing_important: self.result.add_warning(f"缺少重要列（可选）: {', '.join(missing_important)}") def _validate_data_quality(self, df: pd.DataFrame): """ 验证数据质量 Args: df: 数据DataFrame """ # 1. 检查订单编号 if '原始订单编号' in df.columns: self._validate_order_ids(df['原始订单编号']) # 2. 检查商品名称 if '线上宝贝名称' in df.columns: self._validate_product_names(df['线上宝贝名称']) # 3. 检查数值字段 for field in self.NUMERIC_FIELDS: if field in df.columns: self._validate_numeric_field(df, field) # 4. 检查日期字段 for field in self.DATE_FIELDS: if field in df.columns: self._validate_date_field(df, field) # 5. 检查必需字段的空值 for col in self.REQUIRED_COLUMNS: if col in df.columns: null_count = df[col].isna().sum() if null_count > 0: percentage = (null_count / len(df)) * 100 if percentage > 10: # 超过10%为空 self.result.add_error(f"列 '{col}' 有 {null_count} 个空值 ({percentage:.1f}%)") elif percentage > 0: self.result.add_warning(f"列 '{col}' 有 {null_count} 个空值 ({percentage:.1f}%)") def _validate_order_ids(self, order_ids): """验证订单编号""" # 检查空值 null_count = order_ids.isna().sum() if null_count > 0: self.result.add_error(f"订单编号有 {null_count} 个空值") return # 检查重复 duplicates = order_ids[order_ids.duplicated()] if not duplicates.empty: self.result.add_warning(f"发现 {len(duplicates)} 个重复的订单编号") # 检查格式（应该是数字字符串） invalid_format = [] for idx, oid in order_ids.items(): if pd.notna(oid): oid_str = str(oid).strip() # 订单号通常是纯数字或包含字母数字的组合 if not oid_str or len(oid_str) < 5: invalid_format.append(oid_str) if invalid_format: sample = invalid_format[:3] self.result.add_warning(f"发现 {len(invalid_format)} 个格式异常的订单号，例如: {sample}") def _validate_product_names(self, product_names): """验证商品名称""" # 检查空值 null_count = product_names.isna().sum() if null_count > 0: percentage = (null_count / len(product_names)) * 100 if percentage > 5: self.result.add_error(f"商品名称有 {null_count} 个空值 ({percentage:.1f}%)") # 检查异常字符 invalid_names = [] for idx, name in product_names.items(): if pd.notna(name): name_str = str(name).strip() # 检查是否包含异常字符或太短 if len(name_str) < 3: invalid_names.append((idx, name_str)) # 检查是否包含奇怪的字符 elif re.search(r'[^\w\s\u4e00-\u9fff\-\.\,\(\)\[\]\/&]', name_str): # 允许中文、英文、数字、空格和一些常见符号 pass # 这里可以根据需要添加更严格的检查 if invalid_names: self.result.add_warning(f"发现 {len(invalid_names)} 个可能异常的商品名称") def _validate_numeric_field(self, df: pd.DataFrame, field: str): """验证数值字段""" column = df[field] # 转换为数值类型 numeric_col = pd.to_numeric(column, errors='coerce') # 检查转换失败的值 conversion_errors = column[numeric_col.isna() & column.notna()] if not conversion_errors.empty: self.result.add_warning(f"列 '{field}' 有 {len(conversion_errors)} 个非数值数据") # 检查负数 if field in ['数量', '订单单价', '订单金额']: negative_values = numeric_col[numeric_col < 0] if not negative_values.empty: self.result.add_error(f"列 '{field}' 包含 {len(negative_values)} 个负数") # 检查数量字段是否为整数 if field == '数量': non_integers = numeric_col[ (numeric_col.notna()) & (numeric_col != numeric_col.astype('int64', errors='ignore')) ] if not non_integers.empty: self.result.add_warning(f"数量列包含 {len(non_integers)} 个非整数值") def _validate_date_field(self, df: pd.DataFrame, field: str): """验证日期字段""" column = df[field] # 转换为日期类型 date_col = pd.to_datetime(column, errors='coerce') # 检查转换失败的值 conversion_errors = column[date_col.isna() & column.notna()] if not conversion_errors.empty: self.result.add_warning(f"列 '{field}' 有 {len(conversion_errors)} 个无效的日期格式") # 检查未来日期 future_dates = date_col[date_col > datetime.now() + timedelta(days=1)] if not future_dates.empty: self.result.add_error(f"列 '{field}' 包含 {len(future_dates)} 个未来日期") # 检查过早的日期（比如早于2020年） too_early = date_col[date_col < datetime(2020, 1, 1)] if not too_early.empty: self.result.add_warning(f"列 '{field}' 包含 {len(too_early)} 个早于2020年的日期") def _validate_business_logic(self, df: pd.DataFrame): """ 验证业务逻辑 Args: df: 数据DataFrame """ # 1. 验证金额计算: 订单金额 = 数量 × 单价 if all(col in df.columns for col in ['数量', '订单单价', '订单金额']): df_calc = df[['数量', '订单单价', '订单金额']].copy() df_calc = df_calc.apply(pd.to_numeric, errors='coerce') # 计算期望的订单金额 expected_amount = df_calc['数量'] * df_calc['订单单价'] # 允许一定的误差（比如0.01） diff = abs(df_calc['订单金额'] - expected_amount) errors = diff[diff > 0.01] if not errors.empty: percentage = (len(errors) / len(df)) * 100 if percentage > 10: self.result.add_error( f"有 {len(errors)} 行 ({percentage:.1f}%) 的订单金额计算不正确" ) elif percentage > 0: self.result.add_warning( f"有 {len(errors)} 行 ({percentage:.1f}%) 的订单金额可能有误差" ) # 2. 验证退款金额不超过订单金额 if all(col in df.columns for col in ['订单金额', '退款金额']): df_refund = df[['订单金额', '退款金额']].copy() df_refund = df_refund.apply(pd.to_numeric, errors='coerce') # 检查退款金额是否超过订单金额 over_refund = df_refund[ (df_refund['退款金额'].notna()) & (df_refund['退款金额'] > df_refund['订单金额']) ] if not over_refund.empty: self.result.add_error( f"有 {len(over_refund)} 行的退款金额超过了订单金额" ) # 3. 验证交易状态与退款状态的一致性 if all(col in df.columns for col in ['交易状态', '退款状态']): # 如果有退款状态为"退款成功"，交易状态不应该是"交易成功" inconsistent = df[ (df['退款状态'] == '退款成功') & (df['交易状态'] == '交易成功') ] if not inconsistent.empty: self.result.add_warning( f"有 {len(inconsistent)} 行的交易状态与退款状态不一致" ) def _check_potential_issues(self, df: pd.DataFrame): """ 检查潜在的数据问题 Args: df: 数据DataFrame """ # 1. 检查是否有太多的空列 empty_columns = [] for col in df.columns: if df[col].isna().all(): empty_columns.append(col) if empty_columns: self.result.add_warning(f"发现 {len(empty_columns)} 个完全为空的列") # 2. 检查数据分布 if '网店名称' in df.columns: shop_distribution = df['网店名称'].value_counts() self.result.statistics['shop_count'] = len(shop_distribution) self.result.statistics['top_shop'] = shop_distribution.index[0] if not shop_distribution.empty else None # 3. 检查时间跨度 if '付款时间' in df.columns: payment_times = pd.to_datetime(df['付款时间'], errors='coerce') valid_times = payment_times[payment_times.notna()] if not valid_times.empty: time_span = valid_times.max() - valid_times.min() self.result.statistics['time_span_days'] = time_span.days # 如果时间跨度超过一年，给出警告 if time_span.days > 365: self.result.add_warning( f"数据时间跨度较大: {time_span.days} 天" ) # 4. 统计有效数据行数 key_fields = ['原始订单编号', '线上宝贝名称', '数量', '订单单价'] existing_key_fields = [f for f in key_fields if f in df.columns] if existing_key_fields: # 至少有一个关键字段非空的行 valid_rows = df[existing_key_fields].notna().any(axis=1).sum() self.result.statistics['valid_rows'] = valid_rows if valid_rows < len(df) * 0.5: # 少于50%的有效行 self.result.add_error( f"有效数据行过少: {valid_rows}/{len(df)} ({valid_rows/len(df)*100:.1f}%)" )