Agent/examples/auto_put_ad_mini/test_feishu_approval.py

#!/usr/bin/env python3
"""
测试飞书审批消息发送（不真正执行API）
"""
import asyncio
import sys
import json
from pathlib import Path
from datetime import datetime

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

from examples.auto_put_ad_mini.tools.im_approval import send_approval_request
from examples.auto_put_ad_mini.config import (
    IM_ENABLED,
    FEISHU_OPERATOR_OPEN_ID,
    FEISHU_OPERATOR_CHAT_ID
)

# 模拟ToolContext
class SimpleContext:
    """简单的上下文模拟"""
    def __init__(self):
        self.config = {}

async def test_feishu_message():
    """测试飞书消息发送"""
    print("=" * 70)
    print("  测试飞书审批消息发送")
    print("=" * 70)

    if not IM_ENABLED:
        print("\n❌ IM_ENABLED=False，飞书功能未启用")
        print("   请在 config.py 中设置 IM_ENABLED=True")
        return False

    print(f"\n✅ IM_ENABLED=True，飞书功能已启用")
    print(f"   接收人 OPEN_ID: {FEISHU_OPERATOR_OPEN_ID}")
    print(f"   接收群 CHAT_ID: {FEISHU_OPERATOR_CHAT_ID}")

    # 创建一个测试决策CSV（包含所有必需字段）
    test_decisions = [
        {
            "ad_id": "99999999999",
            "ad_name": "【测试】R500-回流330+-广告测试",
            "action": "bid_down",
            "final_action": "bid_down",  # IM工具需要这个字段
            "dimension": "测试维度",
            "reason": "这是一条测试消息，验证飞书审批功能是否正常工作。动态ROI为2.5，低于R500组中位数3.74的33%，建议降价5%。",
            "confidence": "high",
            "recommended_change_pct": -0.05,
            "current_bid": 100.0,
            "new_bid": 95.0,
            "cost_7d_avg": 500.0,
            "roi": 2.5,
            "tier": 1,
            "audience_tier": "R500",
            "ad_age_days": 15,
            "roi_valid_days": 7,
        }
    ]

    # 保存测试决策CSV
    outputs_dir = Path(__file__).parent / "outputs" / "reports"
    outputs_dir.mkdir(parents=True, exist_ok=True)

    test_csv = outputs_dir / "test_approval_decisions.csv"
    import pandas as pd
    df = pd.DataFrame(test_decisions)
    df.to_csv(test_csv, index=False, encoding='utf-8-sig')

    print(f"\n📝 创建测试决策文件: {test_csv}")
    print(f"   决策数量: {len(test_decisions)}")
    print(f"   测试广告ID: {test_decisions[0]['ad_id']}")

    # 发送审批请求
    print(f"\n📤 发送飞书审批请求...")
    ctx = SimpleContext()

    try:
        result = await send_approval_request(
            ctx,
            validated_csv=str(test_csv),
            timeout_minutes=30
        )

        print(f"\n✅ 飞书消息发送成功！")
        print(f"   {result.title}")
        print(f"\n📋 输出信息:")
        for line in result.output.split('\n'):
            print(f"   {line}")

        # 检查metadata
        if result.metadata:
            print(f"\n📊 元数据:")
            for key, value in result.metadata.items():
                print(f"   {key}: {value}")

        print(f"\n💡 请检查您的飞书:")
        print(f"   1. 打开飞书App")
        print(f"   2. 查找来自「增长投放」机器人的消息")
        print(f"   3. 消息中应包含测试广告的决策信息")

        return True

    except Exception as e:
        print(f"\n❌ 发送失败: {e}")
        import traceback
        traceback.print_exc()

        print(f"\n🔍 可能的原因:")
        print(f"   1. 飞书应用凭据不正确（APP_ID/APP_SECRET）")
        print(f"   2. 接收人/群聊ID不正确")
        print(f"   3. 网络问题或代理设置")
        print(f"   4. 飞书应用权限不足")

        return False

async def main():
    print("\n" + "🧪" * 35)
    print(" " * 15 + "飞书审批功能测试")
    print("🧪" * 35)

    success = await test_feishu_message()

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

    if success:
        print("✅ 飞书消息发送成功，请检查飞书App")
        print("\n📝 下一步:")
        print("   1. 在飞书中确认收到测试消息")
        print("   2. 如需实际执行决策，设置 EXECUTION_ENABLED=True")
        print("   3. 运行完整流程: python3 execute_once.py")
    else:
        print("❌ 飞书消息发送失败，请检查配置")

    return 0 if success else 1

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