""" 测试发送消息和表格到投放审批群聊(简化版) """ 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 config import FEISHU_APP_ID, FEISHU_APP_SECRET, FEISHU_OPERATOR_CHAT_ID # 添加 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 ( _get_tenant_token, _upload_media, _create_import_task, _wait_import_result, _set_permission, _send_link_message, ) 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 ) # 步骤1:发送测试消息 print("🚀 步骤1:发送测试消息...") test_message = """📊 **广告调控测试消息** 这是一条测试消息,验证以下功能: ✅ 消息发送到投放审批群聊 ✅ 在线表格导入和分享 ✅ 外部访问权限设置 接下来会发送在线表格链接... --- 测试时间: 2026-04-17 21:45 """ 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}") sys.exit(1) # 步骤2:查找测试表格 print("🚀 步骤2:准备测试表格...") reports_dir = Path("outputs/reports") # 查找已有的 xlsx 文件 xlsx_files = sorted(list(reports_dir.glob("*.xlsx")), reverse=True) if not xlsx_files: print("⚠️ 未找到现有表格文件,创建测试表格...") # 创建简单的测试 Excel import pandas as pd test_data = pd.DataFrame({ "广告ID": [12345, 67890, 11111], "广告名称": ["测试广告1", "测试广告2", "测试广告3"], "ROI": [1.5, 2.3, 0.8], "日均消耗(元)": [500, 800, 300], "决策": ["保持", "提价5%", "降价10%"], "原因": ["ROI正常", "高ROI低消耗", "ROI过低"], }) reports_dir.mkdir(parents=True, exist_ok=True) test_xlsx = reports_dir / "test_decision_20260417.xlsx" test_data.to_excel(test_xlsx, index=False, engine='openpyxl') print(f"✅ 测试表格已创建: {test_xlsx.name}") xlsx_path = test_xlsx else: xlsx_path = xlsx_files[0] print(f"✅ 使用现有表格: {xlsx_path.name}") print() # 步骤3:导入为飞书在线表格并发送到群聊 print("🚀 步骤3:导入飞书在线表格...") try: # 3.1 获取 token print(" 3.1 获取飞书 token...") token = _get_tenant_token() # 3.2 上传素材 print(" 3.2 上传 Excel 文件...") file_token = _upload_media(token, xlsx_path) # 3.3 创建导入任务 print(" 3.3 创建导入任务...") ticket = _create_import_task(token, file_token, xlsx_path.name) # 3.4 等待导入完成 print(" 3.4 等待导入完成...") result = _wait_import_result(token, ticket) url = result.get("url", "") sheet_token = result.get("token", "") file_type = result.get("type", "sheet") if not url: print("❌ 导入失败:未获取到表格 URL") sys.exit(1) print(f" ✅ 导入成功") # 3.5 设置权限 print(" 3.5 设置访问权限...") _set_permission(token, sheet_token, file_type) # 3.6 发送链接到群聊 print(" 3.6 发送表格链接到群聊...") title = xlsx_path.stem _send_link_message(FEISHU_OPERATOR_CHAT_ID, url, title) print() print("=" * 60) print("✅ 测试完成") print("=" * 60) print() print(f"📊 在线表格 URL:") print(f" {url}") print() print("📱 请在飞书群聊中确认:") print(" 1. ✓ 收到测试消息") print(" 2. ✓ 收到表格链接消息(应该显示为卡片)") print(" 3. ✓ 点击链接能打开表格") print() print("🧪 测试外部访问(重要):") print(" 1. 复制上面的表格 URL") print(" 2. 退出飞书账号(或使用隐私模式浏览器)") print(" 3. 访问 URL") print(" 4. 如果能直接查看 → 外部访问配置成功 ✅") print(" 5. 如果提示需要登录 → 需要配置 drive:permission:manage 权限 ⚠️") print() except Exception as e: print() print(f"❌ 导入失败: {e}") import traceback traceback.print_exc() sys.exit(1)