### 定时任务的实现 在 Python 中，可以使用 `APScheduler`（Advanced Python Scheduler）来创建定时任务。该库提供了丰富的调度器接口，支持间隔、日期和基于 cron 的任务触发方式。通过设置一个每日 11:00 执行的任务，可以确保每天定时获取远程数据库中的最新记录。 为了实现这一点，需要先安装 `APScheduler` 和 `nonebot-plugin-apscheduler` 插件[^2]： ```bash pip install apscheduler nonebot-plugin-apscheduler ``` 接下来，配置一个定时任务，如下所示： ```python from apscheduler.schedulers.blocking import BlockingScheduler import datetime # 创建调度器实例 scheduler = BlockingScheduler() # 定义定时执行的任务函数 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_to_feishu(data) # 添加每天 11:00 执行的任务 scheduler.add_job(job_function, 'cron', hour=11, minute=0) # 启动调度器 try: scheduler.start() except (KeyboardInterrupt, SystemExit): # 关闭调度器 scheduler.shutdown() ``` ### 数据库连接与查询 为了从远程数据库中获取最新记录，可以使用 Python 的数据库驱动程序，例如 `pymysql` 或 `SQLAlchemy` 来连接 MySQL 数据库。假设目标是获取某张表中最近更新的数据，可以通过 SQL 查询语句筛选出当天新增或更新的记录。 以下是使用 `pymysql` 连接数据库并查询数据的示例代码： ```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() """ cursor.execute(sql) result = cursor.fetchall() return result finally: connection.close() ``` ### 飞书 API 集成 飞书提供了一个 Webhook 接口，允许开发者通过 HTTP 请求向群组发送消息。首先需要在飞书中创建一个自定义机器人，并获取 Webhook URL。然后，可以使用 `requests` 库向该 URL 发送 POST 请求以推送消息。 以下是一个简单的飞书消息发送函数示例： ```python import requests import json def send_to_feishu(data): webhook_url = "https://open.feishu.cn/open-apis/bot/v2/hook/your-webhook-token" message = { "msg_type": "text", "content": { "text": "以下是今日更新的数据：\n\n" } } for row in data: message["content"]["text"] += f"- {row}\n" response = requests.post(webhook_url, data=json.dumps(message)) if response.status_code == 200: print("消息已成功发送至飞书") else: print("消息发送失败") ``` ### 完整脚本整合 将上述代码片段整合为一个完整的 Python 脚本，可以实现定时任务、数据库连接以及飞书消息推送功能。确保所有依赖库已正确安装，并根据实际需求修改数据库连接参数和飞书 Webhook 地址。 ```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() """ cursor.execute(sql) result = cursor.fetchall() return result finally: connection.close() def send_to_feishu(data): webhook_url = "https://open.feishu.cn/open-apis/bot/v2/hook/your-webhook-token" message = { "msg_type": "text", "content": { "text": "以下是今日更新的数据：\n\n" } } for row in data: message["content"]["text"] += f"- {row}\n" 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() send_to_feishu(data) scheduler = BlockingScheduler() scheduler.add_job(job_function, 'cron', hour=11, minute=0) try: scheduler.start() except (KeyboardInterrupt, SystemExit): scheduler.shutdown() ```