Agent/examples/auto_put_ad_mini/test_api_simple.py

"""
腾讯广告平台 API 简化测试脚本

不依赖 agent 框架,直接测试腾讯广告 API

使用方法:
    python3 examples/auto_put_ad_mini/test_api_simple.py
"""

import json
import os
import sys
import time
import uuid
from pathlib import Path
from urllib.parse import urlencode

# 加载 .env 文件
try:
    from dotenv import load_dotenv
    # 加载项目根目录的 .env
    root_dir = Path(__file__).parent.parent.parent
    env_path = root_dir / ".env"
    if env_path.exists():
        load_dotenv(env_path)
        print(f"✅ 已加载配置文件: {env_path}\n")
except ImportError:
    print("⚠️  未安装 python-dotenv, 将尝试从系统环境变量读取\n")

# 禁用代理 (测试时直连腾讯广告 API)
for proxy_var in ['HTTP_PROXY', 'HTTPS_PROXY', 'http_proxy', 'https_proxy']:
    if proxy_var in os.environ:
        del os.environ[proxy_var]
        print(f"⚠️  已禁用代理: {proxy_var}")

try:
    import httpx
except ImportError:
    print("❌ 缺少 httpx 库，请安装: pip3 install httpx")
    exit(1)


# 配置
BASE_URL = os.getenv("TENCENT_AD_BASE_URL", "https://api.e.qq.com/v3.0")
ACCESS_TOKEN = os.getenv("TENCENT_AD_ACCESS_TOKEN", "")
ACCOUNT_ID = os.getenv("TENCENT_AD_ACCOUNT_ID", "")
TIMEOUT = 30


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

    # 检查 ACCESS_TOKEN
    if ACCESS_TOKEN:
        display_token = ACCESS_TOKEN[:10] + "..." if len(ACCESS_TOKEN) > 10 else "***"
        print(f"✅ TENCENT_AD_ACCESS_TOKEN: {display_token}")
    else:
        print(f"❌ TENCENT_AD_ACCESS_TOKEN: 未设置")
        print("\n请设置环境变量:")
        print("  export TENCENT_AD_ACCESS_TOKEN='your_access_token'")
        print("  export TENCENT_AD_ACCOUNT_ID='your_account_id'")
        return False

    # 检查 ACCOUNT_ID
    if ACCOUNT_ID:
        print(f"✅ TENCENT_AD_ACCOUNT_ID: {ACCOUNT_ID}")
    else:
        print(f"❌ TENCENT_AD_ACCOUNT_ID: 未设置")
        return False

    # 检查 BASE_URL
    print(f"✅ TENCENT_AD_BASE_URL: {BASE_URL}")

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


def _common_params():
    """公共查询参数"""
    return {
        "access_token": ACCESS_TOKEN,
        "timestamp": str(int(time.time())),
        "nonce": uuid.uuid4().hex,
    }


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

    try:
        # 构建请求
        params = _common_params()
        params["account_id"] = ACCOUNT_ID
        params["fields"] = json.dumps(["balance", "daily_budget", "configured_status"])

        url = f"{BASE_URL}/accounts/get?{urlencode(params)}"

        print(f"请求 URL: {BASE_URL}/accounts/get")
        print(f"请求参数: account_id={ACCOUNT_ID}")

        # 发送请求
        resp = httpx.get(url, timeout=TIMEOUT)

        print(f"响应状态码: {resp.status_code}")

        if resp.status_code != 200:
            print(f"❌ HTTP 错误: {resp.status_code}")
            print(f"响应内容: {resp.text[:500]}")
            return False

        # 解析响应
        data = resp.json()
        print(f"响应内容: {json.dumps(data, ensure_ascii=False, indent=2)}")

        # 检查 API 错误码
        code = data.get("code", -1)
        if code != 0:
            msg = data.get("message_cn") or data.get("message", "未知错误")
            print(f"❌ API 错误 (code={code}): {msg}")
            return False

        # 成功
        account_data = data.get("data", {})
        if isinstance(account_data, dict) and "list" in account_data:
            account_data = account_data["list"][0] if account_data["list"] else {}

        balance = account_data.get("balance", 0)
        daily_budget = account_data.get("daily_budget", 0)
        status = account_data.get("configured_status", "未知")

        print(f"\n✅ 测试通过")
        print(f"账户信息:")
        print(f"  - 账户 ID: {ACCOUNT_ID}")
        print(f"  - 余额: {balance/100:.2f} 元")
        print(f"  - 日限额: {daily_budget/100:.0f} 元")
        print(f"  - 状态: {status}")

        return True

    except httpx.RequestError as e:
        print(f"❌ 网络请求错误: {e}")
        return False
    except Exception as e:
        print(f"❌ 测试异常: {e}")
        import traceback
        traceback.print_exc()
        return False


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

    try:
        # 构建请求
        params = _common_params()
        params["account_id"] = ACCOUNT_ID
        params["page"] = "1"
        params["page_size"] = "5"

        url = f"{BASE_URL}/adgroups/get?{urlencode(params)}"

        print(f"请求 URL: {BASE_URL}/adgroups/get")
        print(f"请求参数: account_id={ACCOUNT_ID}, page=1, page_size=5")

        # 发送请求
        resp = httpx.get(url, timeout=TIMEOUT)

        print(f"响应状态码: {resp.status_code}")

        if resp.status_code != 200:
            print(f"❌ HTTP 错误: {resp.status_code}")
            print(f"响应内容: {resp.text[:500]}")
            return False

        # 解析响应
        data = resp.json()

        # 检查 API 错误码
        code = data.get("code", -1)
        if code != 0:
            msg = data.get("message_cn") or data.get("message", "未知错误")
            print(f"❌ API 错误 (code={code}): {msg}")
            return False

        # 成功
        result_data = data.get("data", {})
        ads = result_data.get("list", [])
        page_info = result_data.get("page_info", {})

        print(f"\n✅ 测试通过")
        print(f"广告列表:")
        print(f"  - 总数量: {page_info.get('total_number', len(ads))}")
        print(f"  - 当前页: {len(ads)} 个广告")

        # 显示前几个广告
        for i, ad in enumerate(ads[:5], 1):
            print(f"\n  [{i}] 广告 ID: {ad.get('adgroup_id')}")
            print(f"      名称: {ad.get('adgroup_name')}")
            print(f"      状态: {ad.get('configured_status')}")
            print(f"      出价: {ad.get('bid_amount', 0)/100:.2f} 元")
            print(f"      日预算: {ad.get('daily_budget', 0)/100:.0f} 元")

        # 统计状态分布
        if ads:
            statuses = {}
            for ad in ads:
                status = ad.get("configured_status", "UNKNOWN")
                statuses[status] = statuses.get(status, 0) + 1
            print(f"\n  状态分布: {statuses}")

        return True

    except httpx.RequestError as e:
        print(f"❌ 网络请求错误: {e}")
        return False
    except Exception as e:
        print(f"❌ 测试异常: {e}")
        import traceback
        traceback.print_exc()
        return False


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"),
        }

        # 构建请求
        params = _common_params()
        params["account_id"] = ACCOUNT_ID
        params["level"] = "ADGROUP"
        params["date_range"] = json.dumps(date_range)
        params["fields"] = json.dumps(["cost", "impression", "click", "ctr", "conversion"])
        params["page"] = "1"
        params["page_size"] = "5"

        url = f"{BASE_URL}/daily_reports/adgroups/get?{urlencode(params)}"

        print(f"请求 URL: {BASE_URL}/daily_reports/adgroups/get")
        print(f"日期范围: {date_range['start_date']} ~ {date_range['end_date']}")

        # 发送请求
        resp = httpx.get(url, timeout=TIMEOUT)

        print(f"响应状态码: {resp.status_code}")

        if resp.status_code != 200:
            print(f"❌ HTTP 错误: {resp.status_code}")
            print(f"响应内容: {resp.text[:500]}")
            return False

        # 解析响应
        data = resp.json()

        # 检查 API 错误码
        code = data.get("code", -1)
        if code != 0:
            msg = data.get("message_cn") or data.get("message", "未知错误")
            print(f"❌ API 错误 (code={code}): {msg}")
            return False

        # 成功
        result_data = data.get("data", {})
        reports = result_data.get("list", [])

        if not reports:
            print(f"\n⚠️  该时间段内无数据")
            return True

        print(f"\n✅ 测试通过")
        print(f"数据报表:")
        print(f"  - 记录数: {len(reports)}")

        # 显示前几条
        for i, report in enumerate(reports[:5], 1):
            cost = report.get("cost", 0)
            impression = report.get("impression", 0)
            click = report.get("click", 0)
            ctr = report.get("ctr", 0)
            conversion = report.get("conversion", 0)

            print(f"\n  [{i}] 广告 ID: {report.get('adgroup_id', '-')}")
            if "date" in report:
                print(f"      日期: {report['date']}")
            print(f"      消耗: {cost/100:.2f} 元")
            print(f"      展示: {impression:,}")
            print(f"      点击: {click:,}")
            print(f"      CTR: {ctr:.2%}")
            print(f"      转化: {conversion}")

        return True

    except httpx.RequestError as e:
        print(f"❌ 网络请求错误: {e}")
        return False
    except Exception as e:
        print(f"❌ 测试异常: {e}")
        import traceback
        traceback.print_exc()
        return False


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

    # 1. 检查环境变量
    if not check_env_vars():
        print("\n❌ 测试中止: 环境变量未配置")
        print("\n配置方法:")
        print("1. 在终端中设置环境变量:")
        print("   export TENCENT_AD_ACCESS_TOKEN='your_access_token'")
        print("   export TENCENT_AD_ACCOUNT_ID='your_account_id'")
        print("\n2. 或者创建 .env 文件:")
        print("   TENCENT_AD_ACCESS_TOKEN=your_access_token")
        print("   TENCENT_AD_ACCOUNT_ID=your_account_id")
        return

    # 2-4. 依次测试
    test1 = test_account_info()
    test2 = test_ad_list()
    test3 = test_report()

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

    results = [
        ("环境变量检查", True),
        ("账户信息查询", test1),
        ("广告列表查询", test2),
        ("数据报表查询", test3),
    ]

    for test_name, passed in results:
        status = "✅ 通过" if passed else "❌ 失败"
        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. 账户权限是否充足")
        print("   5. API 版本是否为 v3.0")


if __name__ == "__main__":
    main()