ylt_diy/test_diagnosis.py

"""
测试脚本：检查数据库连接和历史记录
"""
import sys
from lib.db import execute_query

def test_connection():
    """测试数据库连接"""
    print("=" * 60)
    print("测试1: 数据库连接")
    print("=" * 60)
    try:
        result = execute_query("SELECT 1 as test")
        print(f"✓ 数据库连接成功: {result}")
        return True
    except Exception as e:
        print(f"✗ 数据库连接失败: {e}")
        return False


def test_tables():
    """测试表是否存在"""
    print("\n" + "=" * 60)
    print("测试2: 检查表是否存在")
    print("=" * 60)
    
    tables = ['t_daily_report_config', 't_daily_report_history']
    
    for table in tables:
        try:
            result = execute_query(f"SELECT COUNT(*) as count FROM {table}")
            count = result[0]['count']
            print(f"✓ 表 {table} 存在，当前记录数: {count}")
        except Exception as e:
            print(f"✗ 表 {table} 不存在或查询失败: {e}")


def test_history():
    """检查历史记录"""
    print("\n" + "=" * 60)
    print("测试3: 查看历史记录")
    print("=" * 60)
    
    try:
        result = execute_query("""
            SELECT id, report_date, split_type, split_value, 
                   total_orders, total_amount, status, file_path, create_time
            FROM t_daily_report_history 
            ORDER BY create_time DESC 
            LIMIT 10
        """)
        
        if not result:
            print("⚠ 历史记录表为空，还没有生成过日报")
            print("\n建议操作：")
            print("1. 在浏览器中访问 http://localhost:5000")
            print("2. 创建一个配置")
            print("3. 生成一个日报")
            print("4. 再次运行此测试脚本")
        else:
            print(f"✓ 找到 {len(result)} 条历史记录：\n")
            for item in result:
                status_text = "✓ 成功" if item['status'] == 1 else "✗ 失败"
                print(f"  ID: {item['id']}")
                print(f"  日期: {item['report_date']}")
                print(f"  拆分: {item['split_type']} = {item['split_value']}")
                print(f"  订单数: {item['total_orders']}")
                print(f"  金额: {item['total_amount']}")
                print(f"  状态: {status_text}")
                print(f"  文件: {item['file_path'] or '无'}")
                print(f"  创建时间: {item['create_time']}")
                print("-" * 60)
    except Exception as e:
        print(f"✗ 查询历史记录失败: {e}")


def test_configs():
    """检查配置"""
    print("\n" + "=" * 60)
    print("测试4: 查看配置")
    print("=" * 60)
    
    try:
        result = execute_query("""
            SELECT id, config_name, split_type, split_value, 
                   is_active, selected_fields
            FROM t_daily_report_config 
            ORDER BY sort_order ASC
        """)
        
        if not result:
            print("⚠ 配置表为空，还没有创建过配置")
            print("\n建议操作：")
            print("1. 在浏览器中访问 http://localhost:5000")
            print("2. 点击'配置管理'标签")
            print("3. 点击'创建配置'按钮")
            print("4. 填写配置信息并保存")
        else:
            print(f"✓ 找到 {len(result)} 个配置：\n")
            for item in result:
                status_text = "✓ 启用" if item['is_active'] == 1 else "✗ 禁用"
                print(f"  ID: {item['id']}")
                print(f"  名称: {item['config_name']}")
                print(f"  拆分: {item['split_type']} = {item['split_value']}")
                print(f"  状态: {status_text}")
                print(f"  字段数: {len(item['selected_fields']) if item['selected_fields'] else 0}")
                print("-" * 60)
    except Exception as e:
        print(f"✗ 查询配置失败: {e}")


def test_order_data():
    """测试订单数据"""
    print("\n" + "=" * 60)
    print("测试5: 检查订单数据")
    print("=" * 60)
    
    try:
        # 查询最近的订单
        result = execute_query("""
            SELECT COUNT(*) as total,
                   MIN(report_time) as earliest,
                   MAX(report_time) as latest
            FROM t_equipment_charge_order
            WHERE state = 3
        """)
        
        if result and result[0]['total'] > 0:
            print(f"✓ 找到 {result[0]['total']} 条已完成的订单")
            print(f"  最早订单: {result[0]['earliest']}")
            print(f"  最新订单: {result[0]['latest']}")
            
            # 查询企业列表
            companies = execute_query("""
                SELECT DISTINCT company_id, company_name 
                FROM t_company 
                LIMIT 5
            """)
            
            if companies:
                print(f"\n✓ 找到 {len(companies)} 个企业（前5个）：")
                for c in companies:
                    print(f"  - {c.get('company_name', 'N/A')} (ID: {c.get('company_id', 'N/A')})")
        else:
            print("⚠ 没有找到已完成的订单（state=3）")
            print("\n可能原因：")
            print("1. 订单表中没有数据")
            print("2. 订单状态不是3（已完成）")
            print("3. report_time字段为空")
    except Exception as e:
        print(f"✗ 查询订单数据失败: {e}")


def main():
    print("\n" + "=" * 60)
    print("订单日报系统 - 诊断测试")
    print("=" * 60)
    
    # 运行所有测试
    if not test_connection():
        print("\n✗ 数据库连接失败，无法继续测试")
        print("\n请检查：")
        print("1. 数据库地址和端口是否正确")
        print("2. 用户名和密码是否正确")
        print("3. 防火墙是否放行")
        print("4. 数据库是否正在运行")
        return
    
    test_tables()
    test_configs()
    test_history()
    test_order_data()
    
    print("\n" + "=" * 60)
    print("测试完成")
    print("=" * 60)
    print("\n如果所有测试都通过，但历史记录仍然为空：")
    print("1. 请查看服务器控制台输出的日志")
    print("2. 检查生成日报时是否有错误信息")
    print("3. 确认时间范围内是否有订单数据")


if __name__ == '__main__':
    main()