使用apscheduler定时爬取需登录站点时如何仅执行一次登录?
优化实现方案
核心逻辑是把单次初始化即可复用的资源和登录操作从定时任务中拆分,仅启动时执行一次登录,全局持有会话保持登录状态,同时兼容会话过期自动重登的异常场景。
核心改动点
- 全局持有
requests.Session实例,复用登录后的Cookie,避免每次请求重复登录 - Telegram Bot实例仅初始化一次,减少重复对象创建开销
- 新增会话有效性校验,请求失败/登录失效时自动触发重新登录,保证可用性
- 简化文件读写逻辑,减少不必要的IO操作
import requests from bs4 import BeautifulSoup import os import telegram from apscheduler.schedulers.blocking import BlockingScheduler BASE_DIR = os.path.dirname(os.path.abspath(__file__)) # 全局通用配置,仅加载一次 HEADERS = {'User-Agent':'Mozilla/5.0'} LOGIN_URL = 'Login page url' LOGIN_DATA = { "user_id":"id", "password":"pw", "keep_signed":"Y" } TARGET_URL = "address" TG_TOKEN = "certain value" TG_CHAT_ID = "id" # 全局复用实例,仅初始化一次 s = requests.Session() bot = telegram.Bot(token=TG_TOKEN) def login(): """单独封装登录逻辑,启动时/会话失效时调用""" login_req = s.post(LOGIN_URL, data=LOGIN_DATA, headers=HEADERS) # 可根据站点返回增加登录成功校验,比如判断状态码、返回内容关键字 login_req.raise_for_status() def scraping(): # 先请求目标页面 req = s.get(TARGET_URL, headers=HEADERS) # 校验会话是否有效,比如返回内容包含登录页特征/状态码403则触发重登 if req.status_code in (401, 403) or "登录" in req.text: login() req = s.get(TARGET_URL, headers=HEADERS) html = req.text soup = BeautifulSoup(html, 'html.parser') title = soup.select('#css values') latest_title = title[0].text # 对比最新标题 latest_file_path = os.path.join(BASE_DIR, 'latest.txt') # 先读历史记录 if os.path.exists(latest_file_path): with open(latest_file_path, 'r', encoding='utf-8') as f: before = f.readline().strip() else: before = "" if before != latest_title: bot.sendMessage(chat_id=TG_CHAT_ID, text=latest_title) # 仅更新时写入文件,减少IO with open(latest_file_path, 'w', encoding='utf-8') as f: f.write(latest_title) if __name__ == "__main__": # 启动时先执行一次登录 login() # 启动定时任务 scheduler = BlockingScheduler() scheduler.add_job(scraping, 'interval', seconds=30) scheduler.start()
补充说明
- 可以根据目标站点的实际登录有效期调整重登校验逻辑,比如固定每24小时主动调用一次
login()刷新会话,进一步降低校验开销 - 代码中已处理会话过期的异常场景,不会因为Cookie失效导致任务中断
- 调整为仅检测到标题更新时才写入文件,相比原逻辑每次读写两次文件,IO开销更低
内容的提问来源于stack exchange,提问作者vantabeam
相关产品推荐
相关产品推荐

