Python开发Telegram机器人如何自动更新网站抓取的JSON文件数据

核心实现逻辑 #

直接在现有机器人代码中嵌入后台定时爬取任务，无需依赖系统层面的定时任务，通过文件锁解决读写冲突问题，保证数据更新和用户查询互不干扰。

具体实现步骤 #

1. 封装独立的汇率爬取更新函数 #

把你之前手动爬取目标网站、生成JSON的逻辑封装成可重复调用的函数，注意和原有JSON结构保持完全一致，避免查询逻辑报错：

import json
import time
import threading
from datetime import datetime
import requests
from bs4 import BeautifulSoup # 替换为你实际使用的爬取解析库

# 全局文件锁，避免读写同时操作导致文件损坏/读空
file_lock = threading.Lock()
# 替换为你本地JSON文件的实际存储路径
JSON_FILE_PATH = "exchange_rate.json"

def update_exchange_data():
    """爬取目标网站最新汇率，覆盖更新本地JSON文件"""
    try:
        # ========== 以下替换为你自己的爬取解析逻辑 ==========
        # 示例爬取框架，直接复用你之前手动抓数据的代码即可
        resp = requests.get("目标汇率网站地址", timeout=15)
        resp.encoding = "utf-8"
        soup = BeautifulSoup(resp.text, "html.parser")
        
        # 按原有JSON结构组装解析到的汇率数据
        latest_data = {
            "Exchange_rates": [
                # 示例条目，替换为实际解析结果
                # {"currency": "USD", "price": "72.35", "currency_full": "美元", "amount": "1"}
            ]
        }
        # ========== 爬取逻辑替换结束 ==========

        # 加锁写入文件，和读操作互斥
        with file_lock:
            with open(JSON_FILE_PATH, "w", encoding="utf-8") as f:
                json.dump(latest_data, f, ensure_ascii=False, indent=2)
        print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 汇率数据更新成功")
    except Exception as e:
        print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 数据更新失败: {str(e)}")

2. 启动后台定时更新线程 #

在机器人主程序启动位置（也就是bot.polling()这类启动代码之前）启动守护线程，按固定间隔在后台跑更新任务，不会阻塞机器人正常响应消息：

def backend_schedule(interval_hour=2):
    """后台定时调度，默认每2小时更新一次，可根据目标网站更新频率调整间隔"""
    # 机器人启动时先跑一次更新，避免第一次查的时候用旧数据
    update_exchange_data()
    while True:
        time.sleep(interval_hour * 3600)
        update_exchange_data()

# 启动后台更新线程，daemon=True表示主程序退出时线程自动跟着退出
update_thread = threading.Thread(target=backend_schedule, kwargs={"interval_hour":2}, daemon=True)
update_thread.start()

# 下方保留你原来的机器人启动代码，比如
# if __name__ == "__main__":
#     bot.polling(none_stop=True)

3. 优化原有查询逻辑 #

给读文件操作加锁，同时优化查询效率，不用每次遍历全部汇率条目：

elif message.text == 'Currency💸':
    markup = types.ReplyKeyboardMarkup(resize_keyboard=True)
    item_1 = types.KeyboardButton('EUR')
    item_2 = types.KeyboardButton('USD')
    item_3 = types.KeyboardButton('Back🔙')
    markup.add(item_1, item_2, item_3)
    bot.send_message(message.chat.id,'Choose', reply_markup=markup)

elif message.text in ['USD', 'EUR']:
    # 加锁读文件，避免和写操作冲突
    with file_lock:
        with open(JSON_FILE_PATH, 'r', encoding='utf8') as f: 
            data = json.load(f)
    # 转成字典直接按key查，比循环遍历效率高
    rate_dict = {item['currency']: item for item in data['Exchange_rates']}
    target_rate = rate_dict.get(message.text)
    if target_rate:
        send_text = f"{target_rate['currency']}\n{target_rate['price']}"
        bot.send_message(message.chat.id, send_text)

注意事项 #

• 更新间隔不要设得太短，汇率类网站一般1~6小时更新一次数据即可，请求太频繁容易被目标网站封IP
• 可以额外加失败重试逻辑：如果爬取失败，等待5~10分钟后重新尝试，不用直接等下一个更新周期
• 为了避免爬取过程出错写坏JSON文件，可以先把新数据写到.tmp临时文件，写入完成后再替换正式文件，保证文件始终是完整可解析的
• 如果你的机器人是多进程部署的，把线程锁换成文件锁即可，否则多进程下锁不生效会出现读写冲突

内容的提问来源于stack exchange，提问作者xogeh