### 实现定时任务与数据推送的完整方案 要实现每天 11 点从远程数据库提取最新记录并通过飞书以卡片形式推送，可以结合 `APScheduler` 定时调度、`pymysql` 连接数据库以及飞书机器人 Webhook 接口完成整个流程。 #### 使用 Apscheduler 创建定时任务 Python 提供了强大的定时任务库 `APScheduler`，其支持 cron 表达式进行任务调度。通过设置每天 11:00 执行函数，可确保在指定时间触发数据获取与推送操作[^2]。以下是一个示例： ```python from apscheduler.schedulers.blocking import BlockingScheduler import datetime def job_function(): current_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") print(f"任务执行时间：{current_time}") data = fetch_data_from_database() send_card_to_feishu(data) # 创建调度器并添加每日 11:00 的任务 scheduler = BlockingScheduler() scheduler.add_job(job_function, 'cron', hour=11, minute=0) try: scheduler.start() except (KeyboardInterrupt, SystemExit): scheduler.shutdown() ``` #### 从远程数据库提取最新数据 使用 `pymysql` 可以连接远程 MySQL 数据库，并执行查询语句获取当天更新的数据。假设存在一个名为 `table_name` 的表，其中包含字段 `update_time`，可以通过如下方式筛选出最新的那条记录： ```python import pymysql def fetch_data_from_database(): connection = pymysql.connect( host='remote_host', user='username', password='password', database='database_name' ) try: with connection.cursor() as cursor: # 查询当天更新的数据中最新的那一条 sql = """ SELECT * FROM table_name WHERE update_time >= CURDATE() ORDER BY update_time DESC LIMIT 1 """ cursor.execute(sql) result = cursor.fetchone() return result finally: connection.close() ``` #### 飞书卡片消息推送 飞书提供了丰富的 API 支持，包括富文本格式的消息卡片（Message Card）。相比于普通文本消息，卡片形式更易于展示结构化内容。以下是一个使用飞书 Webhook 发送卡片消息的示例： ```python import requests import json def send_card_to_feishu(data): webhook_url = "https://open.feishu.cn/open-apis/bot/v2/hook/your-webhook-token" message = { "msg_type": "interactive", "card": { "config": { "wide_screen_mode": True }, "elements": [ { "tag": "div", "text": { "content": f"**今日最新数据：**\n\n- ID: {data[0]}\n- 内容: {data[1]}\n- 更新时间: {data[2]}", "tag": "lark_md" } } ], "header": { "template": "blue", "title": { "content": "📢 最新数据推送", "tag": "plain_text" } } } } response = requests.post(webhook_url, data=json.dumps(message)) if response.status_code == 200: print("卡片消息已成功发送至飞书") else: print("卡片消息发送失败") ``` #### 完整脚本整合 将上述代码整合为一个完整的 Python 脚本，即可实现从远程数据库获取最新记录，并每天 11:00 自动通过飞书卡片形式推送： ```python from apscheduler.schedulers.blocking import BlockingScheduler import pymysql import requests import json import datetime def fetch_data_from_database(): connection = pymysql.connect( host='remote_host', user='username', password='password', database='database_name' ) try: with connection.cursor() as cursor: sql = """ SELECT * FROM table_name WHERE update_time >= CURDATE() ORDER BY update_time DESC LIMIT 1 """ cursor.execute(sql) result = cursor.fetchone() return result finally: connection.close() def send_card_to_feishu(data): webhook_url = "https://open.feishu.cn/open-apis/bot/v2/hook/your-webhook-token" message = { "msg_type": "interactive", "card": { "config": { "wide_screen_mode": True }, "elements": [ { "tag": "div", "text": { "content": f"**今日最新数据：**\n\n- ID: {data[0]}\n- 内容: {data[1]}\n- 更新时间: {data[2]}", "tag": "lark_md" } } ], "header": { "template": "blue", "title": { "content": "📢 最新数据推送", "tag": "plain_text" } } } } response = requests.post(webhook_url, data=json.dumps(message)) if response.status_code == 200: print("卡片消息已成功发送至飞书") else: print("卡片消息发送失败") def job_function(): current_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") print(f"任务执行时间：{current_time}") data = fetch_data_from_database() if data: send_card_to_feishu(data) else: print("未找到当天更新的数据") scheduler = BlockingScheduler() scheduler.add_job(job_function, 'cron', hour=11, minute=0) try: scheduler.start() except (KeyboardInterrupt, SystemExit): scheduler.shutdown() ```