如何结合现有代码与Telegram API实现新电影上映自动推送提醒

前置依赖安装 #

执行以下命令安装需要的第三方库：

pip install beautifulsoup4 python-telegram-bot==13.7 schedule

你需要提前准备2个Telegram相关参数：

• Telegram Bot Token：通过Telegram官方@BotFather创建机器人即可获取
• 接收通知的Chat ID：通过Telegram官方@userinfobot发送任意消息即可获取自己的ID

核心逻辑说明 #

1. 新片检测逻辑 #

我们会将每次爬取到的影片名称存在本地文本文件中，下次爬取时和存储的历史列表做对比，存在差异的内容即为新上线的影片。

2. Telegram推送逻辑 #

调用机器人官方API，将新增的影片名称批量发送到指定的Chat ID。

3. 定时任务逻辑 #

使用schedule库配置定时规则，每2小时执行一次爬取+检测+推送的全流程，跨平台通用无需配置系统定时任务。

完整可运行代码 #

import os
import schedule
import time
from bs4 import BeautifulSoup
from urllib.request import Request, urlopen
from telegram import Bot

# -------------------------- 配置项 请自行修改 --------------------------
TARGET_URL = "https://www.thenetnaija.com/videos/movies"
BOT_TOKEN = "替换为你的Telegram Bot Token"
CHAT_ID = "替换为你的接收消息Chat ID"
HISTORY_FILE = "movie_history.txt"
# ----------------------------------------------------------------------

# 初始化Telegram Bot
bot = Bot(token=BOT_TOKEN)

def get_latest_movies():
    """爬取最新电影列表"""
    try:
        req = Request(TARGET_URL, headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36'})
        webpage = urlopen(req, timeout=15)
        b4 = BeautifulSoup(webpage, "html.parser")
        movie_list = b4.find_all("div", {"class" : "video-files"})
        latest_films = []
        for container in movie_list:
            film_name = container.find('img').get('alt')
            if film_name:
                latest_films.append(film_name.strip())
        return latest_films
    except Exception as e:
        print(f"爬取失败：{str(e)}")
        return []

def load_history_movies():
    """加载历史电影列表"""
    if not os.path.exists(HISTORY_FILE):
        return []
    with open(HISTORY_FILE, 'r', encoding='utf-8') as f:
        return [line.strip() for line in f.readlines() if line.strip()]

def save_history_movies(movie_list):
    """保存最新电影列表到历史文件"""
    with open(HISTORY_FILE, 'w', encoding='utf-8') as f:
        for movie in movie_list:
            f.write(f"{movie}\n")

def send_telegram_notification(new_movies):
    """发送新片提醒到Telegram"""
    if not new_movies:
        return
    message = "🎬 检测到新上线电影：\n" + "\n".join([f"- {movie}" for movie in new_movies])
    try:
        bot.send_message(chat_id=CHAT_ID, text=message)
        print("推送通知成功")
    except Exception as e:
        print(f"推送失败：{str(e)}")

def check_new_movies_job():
    """定时任务执行的完整流程"""
    print("开始执行新片检测任务...")
    latest_movies = get_latest_movies()
    if not latest_movies:
        return
    history_movies = load_history_movies()
    # 首次运行直接保存历史不推送
    if not history_movies:
        save_history_movies(latest_movies)
        print("首次运行，已初始化电影历史列表")
        return
    # 对比找新增电影
    new_movies = [movie for movie in latest_movies if movie not in history_movies]
    if new_movies:
        print(f"发现{len(new_movies)}部新电影：{new_movies}")
        send_telegram_notification(new_movies)
        save_history_movies(latest_movies)
    else:
        print("未发现新上线电影")

if __name__ == "__main__":
    # 首次运行立即执行一次检测
    check_new_movies_job()
    # 配置每2小时执行一次任务
    schedule.every(2).hours.do(check_new_movies_job)
    # 保持脚本运行
    while True:
        schedule.run_pending()
        time.sleep(60)

运行说明 #

• 替换代码开头配置项里的BOT_TOKEN和CHAT_ID为你自己的参数
• 直接运行脚本即可，首次运行会初始化本地电影历史列表，不会推送消息，之后每2小时自动检测一次，有新片就会推送通知
• 如果需要在服务器后台长期运行，Linux系统可以使用nohup python 你的脚本文件名.py &命令挂后台，Windows系统可以配置为系统服务或者开机启动

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