antissoft / lvqianmeiye_ERP

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

import requests
import json
from datetime import datetime, timedelta
from typing import Dict, Any, Optional

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

# 测试结果记录
test_results = []

def print_result(test_name: str, success: bool, message: str = "", data: Any = None):
    """打印测试结果"""
    status = "✅" if success else "❌"
    print(f"{status} {test_name}")
    if message:
        print(f"   {message}")
    if data and success:
        # 打印关键数据
        if isinstance(data, dict):
            if "totalExpansionCount" in data:
                print(f"   总拓客人数: {data.get('totalExpansionCount', 0)}")
                print(f"   总到店人数: {data.get('totalVisitCount', 0)}")
                print(f"   总开单人数: {data.get('totalBillingCount', 0)}")
                print(f"   大单数量: {data.get('bigOrderCount', 0)}")
            elif "summary" in data:
                print(f"   大单汇总 - 数量: {data['summary'].get('bigOrderCount', 0)}, 金额: {data['summary'].get('bigOrderAmount', 0)}")
            elif isinstance(data, list) and len(data) > 0:
                print(f"   返回数据条数: {len(data)}")
                if "employeeName" in data[0]:
                    print(f"   示例员工: {data[0].get('employeeName', '')}")
        print()
    
    test_results.append({
        'name': test_name,
        'success': success,
        'message': message
    })

def get_token():
    """获取登录token"""
    print("=" * 80)
    print("步骤 1: 获取登录Token")
    print("=" * 80)
    
    login_data = {
        "account": "admin",
        "password": "e10adc3949ba59abbe56e057f20f883e"
    }
    
    try:
        response = requests.post(
            f"{BASE_URL}/api/oauth/Login",
            data=login_data,
            headers={"Content-Type": "application/x-www-form-urlencoded"},
            timeout=10
        )
        
        if response.status_code == 200:
            result = response.json()
            if result.get('code') == 200 and result.get('data') and result.get('data').get('token'):
                token = result['data']['token']
                print(f"✓ Token获取成功: {token[:50]}...")
                print()
                return token
            else:
                print(f"✗ Token获取失败: {result}")
                return None
        else:
            print(f"✗ 登录请求失败: HTTP {response.status_code}")
            return None
    except requests.exceptions.ConnectionError:
        print(f"✗ 无法连接到服务器 {BASE_URL}")
        print("   请确保后端服务已启动，并且运行在 http://localhost:2011")
        return None
    except Exception as e:
        print(f"✗ 登录请求异常: {e}")
        return None

def get_event_list(token: str) -> Optional[str]:
    """获取拓客活动列表，返回第一个活动ID"""
    print("=" * 80)
    print("步骤 2: 获取拓客活动列表")
    print("=" * 80)
    
    try:
        # 获取活动列表
        response = requests.get(
            f"{BASE_URL}/api/Extend/LqEvent",
            headers={
                "Authorization": token,
                "Content-Type": "application/json"
            },
            params={
                "page": 1,
                "rows": 10,
                "sidx": "F_CreateTime",
                "sord": "desc"
            },
            timeout=10
        )
        
        if response.status_code == 200:
            result = response.json()
            if result.get('code') == 200 and result.get('data'):
                events = result['data'].get('records', [])
                if events and len(events) > 0:
                    event = events[0]
                    event_id = event.get('id', '')
                    event_name = event.get('eventName', '')
                    event_type = event.get('eventType', 0)
                    start_time = event.get('startTime', '')
                    end_time = event.get('endTime', '')
                    
                    print(f"✓ 获取到活动列表，共 {len(events)} 个活动")
                    print(f"   选择活动: {event_name}")
                    print(f"   活动ID: {event_id}")
                    print(f"   活动类型: {event_type} ({'全员拓客' if event_type == 3 else '日常拓客'})")
                    print(f"   开始时间: {start_time}")
                    print(f"   结束时间: {end_time}")
                    print()
                    
                    return event_id
                else:
                    print("✗ 活动列表为空")
                    return None
            else:
                print(f"✗ 获取活动列表失败: {result}")
                return None
        else:
            print(f"✗ 请求失败: HTTP {response.status_code}")
            print(f"   响应: {response.text[:200]}")
            return None
    except Exception as e:
        print(f"✗ 获取活动列表异常: {e}")
        return None

def test_get_overview(token: str, event_id: str, start_time: str = None, end_time: str = None):
    """测试1: 获取驾驶舱概览数据"""
    print("=" * 80)
    print("测试 1: GetOverview - 获取驾驶舱概览数据")
    print("=" * 80)
    
    data = {
        "eventId": event_id
    }
    
    if start_time:
        data["startTime"] = start_time
    if end_time:
        data["endTime"] = end_time
    
    try:
        response = requests.post(
            f"{BASE_URL}/api/Extend/LqTkDashboard/GetOverview",
            json=data,
            headers={
                "Authorization": token,
                "Content-Type": "application/json"
            },
            timeout=30
        )
        
        if response.status_code == 200:
            result = response.json()
            if isinstance(result, dict) and result.get('code') == 200:
                data_result = result.get('data', {})
                print_result(
                    "GetOverview",
                    True,
                    "接口调用成功",
                    data_result
                )
                return True, data_result
            else:
                print_result(
                    "GetOverview",
                    False,
                    f"接口返回错误: {result.get('msg', result)}"
                )
                return False, None
        else:
            print_result(
                "GetOverview",
                False,
                f"HTTP状态码错误: {response.status_code}, 响应: {response.text[:200]}"
            )
            return False, None
    except Exception as e:
        print_result(
            "GetOverview",
            False,
            f"接口调用异常: {str(e)}"
        )
        return False, None

def test_get_big_order_statistics(token: str, event_id: str, start_time: str = None, end_time: str = None):
    """测试2: 获取大单统计"""
    print("=" * 80)
    print("测试 2: GetBigOrderStatistics - 获取大单统计")
    print("=" * 80)
    
    data = {
        "eventId": event_id
    }
    
    if start_time:
        data["startTime"] = start_time
    if end_time:
        data["endTime"] = end_time
    
    try:
        response = requests.post(
            f"{BASE_URL}/api/Extend/LqTkDashboard/GetBigOrderStatistics",
            json=data,
            headers={
                "Authorization": token,
                "Content-Type": "application/json"
            },
            timeout=30
        )
        
        if response.status_code == 200:
            result = response.json()
            if isinstance(result, dict) and result.get('code') == 200:
                data_result = result.get('data', {})
                summary = data_result.get('summary', {})
                by_store = data_result.get('byStore', [])
                by_employee = data_result.get('byEmployee', [])
                details = data_result.get('details', [])
                
                print_result(
                    "GetBigOrderStatistics",
                    True,
                    f"接口调用成功 - 汇总数据: 大单数量={summary.get('bigOrderCount', 0)}, "
                    f"按门店统计={len(by_store)}条, 按员工统计={len(by_employee)}条, 明细={len(details)}条",
                    data_result
                )
                return True, data_result
            else:
                print_result(
                    "GetBigOrderStatistics",
                    False,
                    f"接口返回错误: {result.get('msg', result)}"
                )
                return False, None
        else:
            print_result(
                "GetBigOrderStatistics",
                False,
                f"HTTP状态码错误: {response.status_code}, 响应: {response.text[:200]}"
            )
            return False, None
    except Exception as e:
        print_result(
            "GetBigOrderStatistics",
            False,
            f"接口调用异常: {str(e)}"
        )
        return False, None

def test_get_employee_participation_statistics(token: str, event_id: str, start_time: str = None, end_time: str = None):
    """测试3: 获取拓客人员参与统计"""
    print("=" * 80)
    print("测试 3: GetEmployeeParticipationStatistics - 获取拓客人员参与统计")
    print("=" * 80)
    
    data = {
        "eventId": event_id
    }
    
    if start_time:
        data["startTime"] = start_time
    if end_time:
        data["endTime"] = end_time
    
    try:
        response = requests.post(
            f"{BASE_URL}/api/Extend/LqTkDashboard/GetEmployeeParticipationStatistics",
            json=data,
            headers={
                "Authorization": token,
                "Content-Type": "application/json"
            },
            timeout=30
        )
        
        if response.status_code == 200:
            result = response.json()
            if isinstance(result, list):
                # 直接返回列表
                print_result(
                    "GetEmployeeParticipationStatistics",
                    True,
                    f"接口调用成功 - 返回 {len(result)} 条人员统计数据",
                    result
                )
                return True, result
            elif isinstance(result, dict):
                if result.get('code') == 200:
                    data_result = result.get('data', [])
                    print_result(
                        "GetEmployeeParticipationStatistics",
                        True,
                        f"接口调用成功 - 返回 {len(data_result)} 条人员统计数据",
                        data_result
                    )
                    return True, data_result
                else:
                    print_result(
                        "GetEmployeeParticipationStatistics",
                        False,
                        f"接口返回错误: {result.get('msg', result)}"
                    )
                    return False, None
            else:
                print_result(
                    "GetEmployeeParticipationStatistics",
                    False,
                    f"接口返回格式异常: {type(result)}"
                )
                return False, None
        else:
            print_result(
                "GetEmployeeParticipationStatistics",
                False,
                f"HTTP状态码错误: {response.status_code}, 响应: {response.text[:200]}"
            )
            return False, None
    except Exception as e:
        print_result(
            "GetEmployeeParticipationStatistics",
            False,
            f"接口调用异常: {str(e)}"
        )
        return False, None

def test_get_visit_conversion_analysis(token: str, event_id: str, start_time: str = None, end_time: str = None):
    """测试4: 获取到店转化分析"""
    print("=" * 80)
    print("测试 4: GetVisitConversionAnalysis - 获取到店转化分析")
    print("=" * 80)
    
    data = {
        "eventId": event_id
    }
    
    if start_time:
        data["startTime"] = start_time
    if end_time:
        data["endTime"] = end_time
    
    try:
        response = requests.post(
            f"{BASE_URL}/api/Extend/LqTkDashboard/GetVisitConversionAnalysis",
            json=data,
            headers={
                "Authorization": token,
                "Content-Type": "application/json"
            },
            timeout=30
        )
        
        if response.status_code == 200:
            result = response.json()
            if isinstance(result, dict) and result.get('code') == 200:
                data_result = result.get('data', {})
                by_store = data_result.get('byStore', [])
                by_employee = data_result.get('byEmployee', [])
                distribution = data_result.get('visitIntervalDistribution', {})
                
                print_result(
                    "GetVisitConversionAnalysis",
                    True,
                    f"接口调用成功 - 整体到店率={data_result.get('overallVisitRate', 0)}%, "
                    f"平均间隔={data_result.get('averageVisitInterval', 0)}天, "
                    f"按门店统计={len(by_store)}条, 按员工统计={len(by_employee)}条",
                    data_result
                )
                return True, data_result
            else:
                print_result(
                    "GetVisitConversionAnalysis",
                    False,
                    f"接口返回错误: {result.get('msg', result)}"
                )
                return False, None
        else:
            print_result(
                "GetVisitConversionAnalysis",
                False,
                f"HTTP状态码错误: {response.status_code}, 响应: {response.text[:200]}"
            )
            return False, None
    except Exception as e:
        print_result(
            "GetVisitConversionAnalysis",
            False,
            f"接口调用异常: {str(e)}"
        )
        return False, None

def print_summary():
    """打印测试总结"""
    print("=" * 80)
    print("测试总结")
    print("=" * 80)
    
    total = len(test_results)
    passed = sum(1 for r in test_results if r['success'])
    failed = total - passed
    
    print(f"总测试数: {total}")
    print(f"通过: {passed} ✅")
    print(f"失败: {failed} ❌")
    print()
    
    if failed > 0:
        print("失败的测试:")
        for r in test_results:
            if not r['success']:
                print(f"  ❌ {r['name']}: {r['message']}")
        print()
    
    return passed == total

def main():
    """主函数"""
    print("=" * 80)
    print("拓客驾驶舱接口测试")
    print("=" * 80)
    print()
    
    # 1. 获取Token
    token = get_token()
    if not token:
        print("❌ 无法获取Token，测试终止")
        return False
    
    # 2. 获取活动列表
    event_id = get_event_list(token)
    if not event_id:
        print("❌ 无法获取活动ID，测试终止")
        print("   提示: 请确保数据库中存在拓客活动数据")
        return False
    
    # 3. 测试所有接口
    test_get_overview(token, event_id)
    test_get_big_order_statistics(token, event_id)
    test_get_employee_participation_statistics(token, event_id)
    test_get_visit_conversion_analysis(token, event_id)
    
    # 4. 打印总结
    all_passed = print_summary()
    
    return all_passed

if __name__ == "__main__":
    try:
        success = main()
        exit(0 if success else 1)
    except KeyboardInterrupt:
        print("\n\n测试被用户中断")
        exit(1)
    except Exception as e:
        print(f"\n\n测试过程发生异常: {e}")
        import traceback
        traceback.print_exc()
        exit(1)