152 lines
4.7 KiB
Python
152 lines
4.7 KiB
Python
#!/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() |