Agent/examples/auto_put_ad_mini/test_tencent_api.py

"""
腾讯广告平台 API 接口测试脚本

测试内容:
1. 环境变量检查
2. API 连接测试(获取账户信息)
3. 广告列表查询测试
4. 数据报表查询测试

使用方法:
    cd /Users/liulidong/project/agent/Agent
    python3 examples/auto_put_ad_mini/test_tencent_api.py
"""

import asyncio
import os
import sys
from pathlib import Path

# 添加项目根目录到 Python 路径
sys.path.insert(0, str(Path(__file__).parent.parent.parent))

from dotenv import load_dotenv
load_dotenv()

# 导入腾讯广告 API 工具
from examples.auto_put_ad_mini.tools.ad_api import (
    account_get_info,
    ad_get_list,
    ad_get_report,
)


def check_env_vars():
    """检查必需的环境变量"""
    print("=" * 70)
    print("【1/4】环境变量检查")
    print("=" * 70)

    required_vars = {
        "TENCENT_AD_ACCESS_TOKEN": "腾讯广告 OAuth2 访问令牌",
        "TENCENT_AD_ACCOUNT_ID": "腾讯广告账户 ID",
    }

    optional_vars = {
        "TENCENT_AD_BASE_URL": f"API 基础 URL (默认: https://api.e.qq.com/v3.0)",
    }

    missing = []

    # 检查必需变量
    for var, desc in required_vars.items():
        value = os.getenv(var)
        if value:
            # 脱敏显示
            if "TOKEN" in var:
                display_value = value[:10] + "..." if len(value) > 10 else "***"
            else:
                display_value = value
            print(f"✅ {var}: {display_value}")
            print(f"   {desc}")
        else:
            print(f"❌ {var}: 未设置")
            print(f"   {desc}")
            missing.append(var)

    # 检查可选变量
    print("\n可选配置:")
    for var, desc in optional_vars.items():
        value = os.getenv(var)
        if value:
            print(f"✅ {var}: {value}")
        else:
            print(f"⚪ {var}: 未设置 (将使用默认值)")
        print(f"   {desc}")

    if missing:
        print(f"\n❌ 缺少必需环境变量: {', '.join(missing)}")
        print("\n请在 .env 文件中配置:")
        print("-" * 70)
        for var in missing:
            print(f"{var}=your_value_here")
        print("-" * 70)
        return False

    print("\n✅ 环境变量检查通过")
    return True


async def test_account_info():
    """测试获取账户信息"""
    print("\n" + "=" * 70)
    print("【2/4】账户信息查询测试")
    print("=" * 70)

    try:
        result = await account_get_info()

        if "失败" in result.title:
            print(f"❌ 测试失败: {result.title}")
            print(f"   错误信息: {result.output}")
            return False
        else:
            print(f"✅ {result.title}")
            print(result.output)
            return True

    except Exception as e:
        print(f"❌ 测试异常: {e}")
        import traceback
        traceback.print_exc()
        return False


async def test_ad_list():
    """测试查询广告列表"""
    print("\n" + "=" * 70)
    print("【3/4】广告列表查询测试")
    print("=" * 70)

    try:
        # 查询前 5 个广告
        result = await ad_get_list(page=1, page_size=5)

        if "失败" in result.title:
            print(f"❌ 测试失败: {result.title}")
            print(f"   错误信息: {result.output}")
            return False
        else:
            print(f"✅ {result.title}")
            print(result.output)

            # 显示更多统计信息
            if result.metadata and "list" in result.metadata:
                ads = result.metadata["list"]
                if ads:
                    print(f"\n📊 统计信息:")
                    print(f"   - 广告数量: {len(ads)}")

                    statuses = {}
                    for ad in ads:
                        status = ad.get("configured_status", "UNKNOWN")
                        statuses[status] = statuses.get(status, 0) + 1

                    print(f"   - 状态分布: {statuses}")

            return True

    except Exception as e:
        print(f"❌ 测试异常: {e}")
        import traceback
        traceback.print_exc()
        return False


async def test_report():
    """测试查询数据报表"""
    print("\n" + "=" * 70)
    print("【4/4】数据报表查询测试")
    print("=" * 70)

    try:
        from datetime import datetime, timedelta

        # 查询最近 7 天的数据
        end_date = datetime.now()
        start_date = end_date - timedelta(days=6)

        date_range = {
            "start_date": start_date.strftime("%Y-%m-%d"),
            "end_date": end_date.strftime("%Y-%m-%d"),
        }

        print(f"查询日期范围: {date_range['start_date']} ~ {date_range['end_date']}")

        result = await ad_get_report(
            date_range=date_range,
            level="adgroup",
            page=1,
            page_size=5,
        )

        if "失败" in result.title:
            print(f"❌ 测试失败: {result.title}")
            print(f"   错误信息: {result.output}")
            return False
        else:
            print(f"✅ {result.title}")
            print(result.output)
            return True

    except Exception as e:
        print(f"❌ 测试异常: {e}")
        import traceback
        traceback.print_exc()
        return False


async def main():
    """主测试流程"""
    print("\n" + "=" * 70)
    print("腾讯广告平台 API 接口测试")
    print("=" * 70)
    print()

    # 1. 检查环境变量
    if not check_env_vars():
        print("\n❌ 测试中止: 环境变量未配置")
        return

    # 2. 测试账户信息
    test1 = await test_account_info()

    # 3. 测试广告列表
    test2 = await test_ad_list()

    # 4. 测试数据报表
    test3 = await test_report()

    # 总结
    print("\n" + "=" * 70)
    print("测试结果汇总")
    print("=" * 70)

    results = {
        "环境变量检查": "✅ 通过",
        "账户信息查询": "✅ 通过" if test1 else "❌ 失败",
        "广告列表查询": "✅ 通过" if test2 else "❌ 失败",
        "数据报表查询": "✅ 通过" if test3 else "❌ 失败",
    }

    for test_name, status in results.items():
        print(f"{status}  {test_name}")

    all_passed = all([test1, test2, test3])

    if all_passed:
        print("\n🎉 所有测试通过! 腾讯广告平台接口可用")
    else:
        print("\n⚠️  部分测试失败，请检查:")
        print("   1. ACCESS_TOKEN 是否有效")
        print("   2. ACCOUNT_ID 是否正确")
        print("   3. 网络连接是否正常")
        print("   4. 账户权限是否充足")


if __name__ == "__main__":
    asyncio.run(main())