#!/usr/bin/env python """ 测试数据标准化功能 """ import sys import os sys.path.append('server') def test_load_multi_store_data(): """测试多店铺数据加载和标准化""" print("=== 测试多店铺数据加载和标准化 ===") try: from utils.multi_store_data_utils import load_multi_store_data # 加载数据 df = load_multi_store_data('pharmacy_sales_multi_store.csv', product_id='P001') print(f"数据形状: {df.shape}") print(f"数据列名: {list(df.columns)}") # 检查标准化的列是否存在 required_columns = ['sales', 'price', 'weekday', 'month', 'is_holiday', 'is_weekend', 'is_promotion', 'temperature'] missing_columns = [col for col in required_columns if col not in df.columns] if missing_columns: print(f"FAIL: 缺少标准化列: {missing_columns}") return False else: print("OK: 所有必需的标准化列都存在") # 检查数据类型 print("\n数据类型检查:") for col in required_columns: if col in df.columns: print(f" {col}: {df[col].dtype}, 样本值: {df[col].iloc[0] if len(df) > 0 else 'N/A'}") return True except Exception as e: print(f"FAIL: 数据加载失败: {e}") return False def test_store_product_data(): """测试店铺产品数据获取""" print("\n=== 测试店铺产品数据获取 ===") try: from utils.multi_store_data_utils import get_store_product_sales_data # 获取特定店铺产品数据 df = get_store_product_sales_data('S001', 'P001', 'pharmacy_sales_multi_store.csv') print(f"店铺产品数据形状: {df.shape}") # 检查必需列 required_columns = ['sales', 'price', 'weekday', 'month', 'is_holiday', 'is_weekend', 'is_promotion', 'temperature'] missing_columns = [col for col in required_columns if col not in df.columns] if missing_columns: print(f"FAIL: 缺少列: {missing_columns}") return False else: print("OK: 店铺产品数据标准化成功") return True except Exception as e: print(f"FAIL: 店铺产品数据获取失败: {e}") return False def test_aggregate_data(): """测试聚合数据""" print("\n=== 测试聚合数据 ===") try: from utils.multi_store_data_utils import aggregate_multi_store_data # 聚合数据 df = aggregate_multi_store_data('P001', aggregation_method='sum', file_path='pharmacy_sales_multi_store.csv') print(f"聚合数据形状: {df.shape}") # 检查必需列 required_columns = ['sales', 'price', 'weekday', 'month', 'is_holiday', 'is_weekend', 'is_promotion', 'temperature'] missing_columns = [col for col in required_columns if col not in df.columns] if missing_columns: print(f"FAIL: 缺少列: {missing_columns}") return False else: print("OK: 聚合数据标准化成功") return True except Exception as e: print(f"FAIL: 聚合数据失败: {e}") return False def test_csv_structure(): """检查CSV文件结构""" print("\n=== 检查CSV文件结构 ===") try: import pandas as pd # 直接读取CSV查看原始结构 df = pd.read_csv('pharmacy_sales_multi_store.csv') print(f"原始CSV列名: {list(df.columns)}") print(f"原始数据形状: {df.shape}") print(f"P001产品数据量: {len(df[df['product_id'] == 'P001'])}") return True except Exception as e: print(f"FAIL: CSV读取失败: {e}") return False def main(): """主测试函数""" print("开始数据标准化测试") tests_passed = 0 total_tests = 4 # 检查CSV文件结构 if test_csv_structure(): tests_passed += 1 # 测试多店铺数据加载 if test_load_multi_store_data(): tests_passed += 1 # 测试店铺产品数据 if test_store_product_data(): tests_passed += 1 # 测试聚合数据 if test_aggregate_data(): tests_passed += 1 print(f"\n测试结果: {tests_passed}/{total_tests} 项测试通过") if tests_passed == total_tests: print("SUCCESS: 数据标准化功能正常工作") else: print("WARNING: 部分测试失败,需要进一步检查") if __name__ == "__main__": main()