antissoft / lvqianmeiye_ERP

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
事业部驾驶舱接口测试脚本
"""

import requests
import json
from datetime import datetime

# API基础URL
BASE_URL = "http://localhost:2011"

# 登录接口获取Token
def login():
    """登录获取Token"""
    url = f"{BASE_URL}/api/oauth/Login"
    data = {
        "account": "admin",
        "password": "e10adc3949ba59abbe56e057f20f883e"
    }
    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }
    
    try:
        response = requests.post(url, data=data, headers=headers)
        response.raise_for_status()
        result = response.json()
        if result.get("code") == 200:
            token = result.get("data", {}).get("token")
            print(f"✅ 登录成功，Token: {token[:50]}...")
            return token
        else:
            print(f"❌ 登录失败: {result.get('msg')}")
            return None
    except Exception as e:
        print(f"❌ 登录异常: {str(e)}")
        return None

# 测试GetStatistics接口
def test_get_statistics(token, business_unit_id=None, store_ids=None, statistics_month="202512"):
    """测试获取统计数据"""
    url = f"{BASE_URL}/api/Extend/LqBusinessUnitDashboard/GetStatistics"
    headers = {
        "Content-Type": "application/json",
        "Authorization": token
    }
    
    data = {
        "statisticsMonth": statistics_month
    }
    if business_unit_id:
        data["businessUnitId"] = business_unit_id
    elif store_ids:
        data["storeIds"] = store_ids
    
    try:
        response = requests.post(url, json=data, headers=headers)
        response.raise_for_status()
        result = response.json()
        
        if result.get("code") == 200 or "billingPerformance" in str(result):
            print("\n✅ GetStatistics 测试成功")
            print(f"   开单业绩: {result.get('billingPerformance', 0):,.2f}")
            print(f"   消耗业绩: {result.get('consumePerformance', 0):,.2f}")
            print(f"   净业绩: {result.get('netPerformance', 0):,.2f}")
            print(f"   目标业绩: {result.get('targetPerformance', 0):,.2f}")
            print(f"   完成率: {result.get('completionRate', 0):.2f}%")
            print(f"   管理的门店数: {result.get('managedStoreCount', 0)}")
            print(f"   活跃门店数: {result.get('activeStoreCount', 0)}")
            return True
        else:
            print(f"\n❌ GetStatistics 测试失败: {result}")
            return False
    except Exception as e:
        print(f"\n❌ GetStatistics 测试异常: {str(e)}")
        return False

# 测试GetPerformanceTrend接口
def test_get_performance_trend(token, business_unit_id=None, statistics_month="202512", month_count=12):
    """测试获取业绩趋势"""
    url = f"{BASE_URL}/api/Extend/LqBusinessUnitDashboard/GetPerformanceTrend"
    headers = {
        "Content-Type": "application/json",
        "Authorization": token
    }
    
    data = {
        "statisticsMonth": statistics_month,
        "monthCount": month_count
    }
    if business_unit_id:
        data["businessUnitId"] = business_unit_id
    
    try:
        response = requests.post(url, json=data, headers=headers)
        response.raise_for_status()
        result = response.json()
        
        if result.get("code") == 200 or "trendData" in str(result):
            trend_data = result.get("trendData", [])
            print(f"\n✅ GetPerformanceTrend 测试成功")
            print(f"   趋势数据点数量: {len(trend_data)}")
            if trend_data:
                latest = trend_data[-1]
                print(f"   最新月份: {latest.get('month')}")
                print(f"   最新开单业绩: {latest.get('billingPerformance', 0):,.2f}")
            return True
        else:
            print(f"\n❌ GetPerformanceTrend 测试失败: {result}")
            return False
    except Exception as e:
        print(f"\n❌ GetPerformanceTrend 测试异常: {str(e)}")
        return False

# 测试GetStoreRanking接口
def test_get_store_ranking(token, business_unit_id=None, statistics_month="202512", ranking_type="Billing", top_count=10):
    """测试获取门店排行"""
    url = f"{BASE_URL}/api/Extend/LqBusinessUnitDashboard/GetStoreRanking"
    headers = {
        "Content-Type": "application/json",
        "Authorization": token
    }
    
    data = {
        "statisticsMonth": statistics_month,
        "rankingType": ranking_type,
        "topCount": top_count
    }
    if business_unit_id:
        data["businessUnitId"] = business_unit_id
    
    try:
        response = requests.post(url, json=data, headers=headers)
        response.raise_for_status()
        result = response.json()
        
        if result.get("code") == 200 or "rankingData" in str(result):
            ranking_data = result.get("rankingData", [])
            print(f"\n✅ GetStoreRanking 测试成功")
            print(f"   排行数据数量: {len(ranking_data)}")
            if ranking_data:
                top1 = ranking_data[0]
                print(f"   第1名: {top1.get('storeName')} - 开单业绩: {top1.get('billingPerformance', 0):,.2f}")
            return True
        else:
            print(f"\n❌ GetStoreRanking 测试失败: {result}")
            return False
    except Exception as e:
        print(f"\n❌ GetStoreRanking 测试异常: {str(e)}")
        return False

# 测试GetOperationStatistics接口
def test_get_operation_statistics(token, business_unit_id=None, statistics_month="202512"):
    """测试获取运营统计"""
    url = f"{BASE_URL}/api/Extend/LqBusinessUnitDashboard/GetOperationStatistics"
    headers = {
        "Content-Type": "application/json",
        "Authorization": token
    }
    
    data = {
        "statisticsMonth": statistics_month
    }
    if business_unit_id:
        data["businessUnitId"] = business_unit_id
    
    try:
        response = requests.post(url, json=data, headers=headers)
        response.raise_for_status()
        result = response.json()
        
        if result.get("code") == 200 or "billingAnalysis" in str(result):
            billing = result.get("billingAnalysis", {})
            consume = result.get("consumeAnalysis", {})
            refund = result.get("refundAnalysis", {})
            print(f"\n✅ GetOperationStatistics 测试成功")
            print(f"   开单次数: {billing.get('billingCount', 0)}")
            print(f"   平均开单金额: {billing.get('averageBillingAmount', 0):,.2f}")
            print(f"   消耗次数: {consume.get('consumeCount', 0)}")
            print(f"   消耗率: {consume.get('consumeRate', 0):.2f}%")
            print(f"   退卡次数: {refund.get('refundCount', 0)}")
            return True
        else:
            print(f"\n❌ GetOperationStatistics 测试失败: {result}")
            return False
    except Exception as e:
        print(f"\n❌ GetOperationStatistics 测试异常: {str(e)}")
        return False

# 测试GetStoreDetailList接口
def test_get_store_detail_list(token, business_unit_id=None, statistics_month="202512", current_page=1, page_size=10):
    """测试获取门店明细列表"""
    url = f"{BASE_URL}/api/Extend/LqBusinessUnitDashboard/GetStoreDetailList"
    headers = {
        "Content-Type": "application/json",
        "Authorization": token
    }
    
    data = {
        "statisticsMonth": statistics_month,
        "currentPage": current_page,
        "pageSize": page_size
    }
    if business_unit_id:
        data["businessUnitId"] = business_unit_id
    
    try:
        response = requests.post(url, json=data, headers=headers)
        response.raise_for_status()
        result = response.json()
        
        if result.get("code") == 200 or "list" in str(result):
            total = result.get("total", 0)
            list_data = result.get("list", [])
            print(f"\n✅ GetStoreDetailList 测试成功")
            print(f"   总记录数: {total}")
            print(f"   当前页数据: {len(list_data)}")
            if list_data:
                first = list_data[0]
                print(f"   第1条: {first.get('storeName')} - 开单业绩: {first.get('billingPerformance', 0):,.2f}")
            return True
        else:
            print(f"\n❌ GetStoreDetailList 测试失败: {result}")
            return False
    except Exception as e:
        print(f"\n❌ GetStoreDetailList 测试异常: {str(e)}")
        return False

# 主测试函数
def main():
    print("=" * 60)
    print("事业部驾驶舱接口测试")
    print("=" * 60)
    
    # 1. 登录获取Token
    token = login()
    if not token:
        print("❌ 无法获取Token，测试终止")
        return
    
    # 测试用的事业部ID（需要根据实际数据调整）
    business_unit_id = "734725299018663173"
    statistics_month = "202512"
    
    # 2. 测试所有接口
    test_results = []
    
    print("\n" + "=" * 60)
    print("开始测试接口...")
    print("=" * 60)
    
    # 测试GetStatistics
    test_results.append(("GetStatistics", test_get_statistics(token, business_unit_id=business_unit_id, statistics_month=statistics_month)))
    
    # 测试GetPerformanceTrend
    test_results.append(("GetPerformanceTrend", test_get_performance_trend(token, business_unit_id=business_unit_id, statistics_month=statistics_month, month_count=6)))
    
    # 测试GetStoreRanking
    test_results.append(("GetStoreRanking", test_get_store_ranking(token, business_unit_id=business_unit_id, statistics_month=statistics_month, ranking_type="Billing", top_count=5)))
    
    # 测试GetOperationStatistics
    test_results.append(("GetOperationStatistics", test_get_operation_statistics(token, business_unit_id=business_unit_id, statistics_month=statistics_month)))
    
    # 测试GetStoreDetailList
    test_results.append(("GetStoreDetailList", test_get_store_detail_list(token, business_unit_id=business_unit_id, statistics_month=statistics_month, current_page=1, page_size=5)))
    
    # 3. 输出测试结果汇总
    print("\n" + "=" * 60)
    print("测试结果汇总")
    print("=" * 60)
    success_count = sum(1 for _, result in test_results if result)
    total_count = len(test_results)
    
    for name, result in test_results:
        status = "✅ 通过" if result else "❌ 失败"
        print(f"{status} - {name}")
    
    print(f"\n总计: {success_count}/{total_count} 通过")
    print("=" * 60)

if __name__ == "__main__":
    main()