""" 测试发送消息和表格到投放审批群聊 """ import asyncio import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).parent.parent.parent)) from dotenv import load_dotenv load_dotenv() from agent.tools.builtin.feishu.feishu_client import FeishuClient from agent.tools.models import ToolContext from config import FEISHU_APP_ID, FEISHU_APP_SECRET, FEISHU_OPERATOR_CHAT_ID print("=" * 60) print("测试发送消息和表格到投放审批群聊") print("=" * 60) print() print(f"飞书应用 ID: {FEISHU_APP_ID}") print(f"投放审批群聊 ID: {FEISHU_OPERATOR_CHAT_ID}") print() # 创建飞书客户端 client = FeishuClient( app_id=FEISHU_APP_ID, app_secret=FEISHU_APP_SECRET ) async def main(): # 步骤1:发送测试消息 print("🚀 步骤1:发送测试消息...") test_message = """📊 **广告调控测试消息** 这是一条测试消息,验证以下功能: ✅ 消息发送到投放审批群聊 ✅ 在线表格导入和分享 ✅ 外部访问权限设置 --- 测试时间: 2026-04-17 21:40 """ try: result = client.send_message(to=FEISHU_OPERATOR_CHAT_ID, text=test_message) print(f"✅ 测试消息发送成功") print(f" 消息 ID: {result.message_id}") print() except Exception as e: print(f"❌ 测试消息发送失败: {e}") return # 步骤2:查找或创建测试表格 print("🚀 步骤2:准备测试表格...") reports_dir = Path("outputs/reports") # 查找已有的 xlsx 文件 xlsx_files = list(reports_dir.glob("*.xlsx")) if not xlsx_files: print("⚠️ 未找到现有表格文件,创建测试表格...") # 创建简单的测试 Excel import pandas as pd test_data = pd.DataFrame({ "广告ID": [12345, 67890], "广告名称": ["测试广告1", "测试广告2"], "ROI": [1.5, 2.3], "日均消耗": [500, 800], "决策": ["保持", "提价5%"], }) reports_dir.mkdir(parents=True, exist_ok=True) test_xlsx = reports_dir / "test_decision_20260417.xlsx" test_data.to_excel(test_xlsx, index=False) print(f"✅ 测试表格已创建: {test_xlsx.name}") xlsx_path = test_xlsx else: xlsx_path = xlsx_files[0] print(f"✅ 使用现有表格: {xlsx_path.name}") print() # 步骤3:导入为飞书在线表格并发送到群聊 print("🚀 步骤3:导入飞书在线表格...") # 添加 tools 目录到路径 tools_dir = Path(__file__).parent / "tools" if str(tools_dir) not in sys.path: sys.path.insert(0, str(tools_dir)) from feishu_doc import import_to_feishu ctx = ToolContext(uid="test", step_count=1, trace_id="test-trace") result = await import_to_feishu( ctx=ctx, xlsx_path=str(xlsx_path), send_im=True, chat_id=FEISHU_OPERATOR_CHAT_ID ) print() print("=" * 60) print("📊 测试结果") print("=" * 60) print(f"标题: {result.title}") print(f"输出: {result.output}") print() if result.metadata and result.metadata.get("url"): sheet_url = result.metadata["url"] print(f"✅ 在线表格已生成") print(f" URL: {sheet_url}") print() print("📱 请在飞书群聊中确认:") print(" 1. 收到测试消息") print(" 2. 收到表格链接消息") print(" 3. 点击链接能打开表格") print() print("🧪 测试外部访问:") print(" 1. 复制表格 URL") print(" 2. 退出飞书账号(或使用隐私模式浏览器)") print(" 3. 访问 URL,看是否能直接查看") print() else: print("❌ 在线表格导入失败") print(f" 错误: {result.output}") if __name__ == "__main__": asyncio.run(main())