基于Python实现埃隆·马斯克新推特自动推送至Telegram的方案求助

核心逻辑规则 #

• 固定两个本地文件路径做数据流转：
  • 爬取脚本输出的最新推文文件：latest_tweets.txt，存储格式为每条推文占单独一行，按最新推文在前的顺序排列
  • 已推送内容存档文件：pushed_tweets_archive.txt，按推送顺序逐行追加存储所有已发送到Telegram的推文
• 对比规则：提取最新推文文件中、未出现在存档文件里的全部内容即为需要推送的新推文，不依赖固定行号对比，能避免爬取脚本偶尔漏行导致的逻辑错乱

分步代码实现 #

1. 通用文本文件读取函数 #

兼容文件不存在的场景，首次运行时若存档文件未生成不会触发报错

def read_txt_lines(file_path):
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            # 过滤空行，去掉每行首尾多余的空格/换行
            return [line.strip() for line in f.readlines() if line.strip()]
    except FileNotFoundError:
        return []

2. 新内容对比逻辑 #

def get_pending_tweets(latest_file, archive_file):
    latest_lines = read_txt_lines(latest_file)
    archive_lines = read_txt_lines(archive_file)
    # 最新文件为空直接返回空列表
    if not latest_lines:
        return []
    # 筛选出最新文件里没被存档记录过的内容，即为待推送的新推文
    return [tweet for tweet in latest_lines if tweet not in archive_lines]

3. 优化Telegram发送接口 #

替换原本URL拼参的写法，用POST方式传参，自动兼容推文里的特殊字符（&、引号、换行等），避免接口调用报错

import requests

def send_telegram_message(message, bot_token='你的机器人Token', bot_chat_id='你的聊天ID'):
    api_url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
    request_params = {
        "chat_id": bot_chat_id,
        "parse_mode": "Markdown",
        "text": message
    }
    response = requests.post(api_url, data=request_params)
    return response.json()

4. 整合全流程推送逻辑 #

保证推送成功的内容才会写入存档，避免发送失败导致的漏推

def run_push_flow():
    LATEST_TWEET_PATH = "latest_tweets.txt"
    ARCHIVE_PATH = "pushed_tweets_archive.txt"
    pending_tweets = get_pending_tweets(LATEST_TWEET_PATH, ARCHIVE_PATH)
    if not pending_tweets:
        print("无新推文待推送")
        return
    # 倒序遍历待推送列表：最新文件里第一条是最新的推文，倒序后从较早的新推文开始发送，Telegram收到的消息顺序符合时间先后
    for tweet in reversed(pending_tweets):
        send_result = send_telegram_message(tweet)
        if send_result.get("ok"):
            print(f"推送成功：{tweet[:30]}...")
            # 发送成功后追加写入存档
            with open(ARCHIVE_PATH, "a", encoding="utf-8") as f:
                f.write(tweet + "\n")
        else:
            print(f"推送失败，接口返回：{send_result}")

if __name__ == "__main__":
    run_push_flow()

实操注意事项 #

• 两个文本文件需和脚本放在同一工作目录，否则需填写文件的绝对路径
• 爬取脚本写入latest_tweets.txt时，需保证单条推文占一行，不要把推文内部的换行写入文件，避免读取时拆分成多行识别错误
• 可配合系统定时任务（Windows任务计划、Linux crontab）按固定频率运行脚本，自动完成爬取、对比、推送全流程
• 不要将机器人Token直接硬编码后上传到公开代码仓库，避免被滥用

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