Python脚本怎样定时爬取网页数据

本文目录导读：

1. 使用 schedule 库（推荐）
2. 使用 APScheduler 库（功能更强大）
3. 使用多线程 + 定时器
4. 完整的生产环境示例
5. 使用装饰器的优雅方式
6. 安装依赖
7. 注意事项

我来介绍几种Python定时爬取网页数据的方法：

使用 schedule 库（推荐）

import schedule
import time
import requests
from bs4 import BeautifulSoup
def crawl_data():
    try:
        # 爬取数据
        url = "https://example.com/data"
        response = requests.get(url)
        response.encoding = 'utf-8'
        # 解析数据
        soup = BeautifulSoup(response.text, 'html.parser')
        data = soup.find('div', class_='content')
        # 保存数据
        with open('data.txt', 'a', encoding='utf-8') as f:
            f.write(f"{time.strftime('%Y-%m-%d %H:%M:%S')}: {data.text}\n")
        print(f"数据已爬取: {time.strftime('%Y-%m-%d %H:%M:%S')}")
    except Exception as e:
        print(f"爬取失败: {e}")
# 设置定时任务
schedule.every(1).hours.do(crawl_data)  # 每小时执行一次
# schedule.every().day.at("10:30").do(crawl_data)  # 每天10:30执行
# schedule.every(30).minutes.do(crawl_data)  # 每30分钟执行
if __name__ == "__main__":
    # 先执行一次
    crawl_data()
    while True:
        schedule.run_pending()
        time.sleep(60)  # 每60秒检查一次

使用 APScheduler 库（功能更强大）

from apscheduler.schedulers.blocking import BlockingScheduler
import requests
from datetime import datetime
def crawl_job():
    try:
        url = "https://api.example.com/data"
        headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        }
        response = requests.get(url, headers=headers)
        data = response.json()
        # 保存到文件
        with open('crawled_data.json', 'a', encoding='utf-8') as f:
            f.write(json.dumps(data, ensure_ascii=False) + '\n')
        print(f"爬取成功: {datetime.now()}")
    except Exception as e:
        print(f"爬取失败: {e}")
def start_scheduler():
    scheduler = BlockingScheduler()
    # 添加定时任务
    scheduler.add_job(
        crawl_job,
        'interval',
        minutes=30,  # 每30分钟执行
        id='crawl_job'
    )
    # 也可以使用cron表达式
    # scheduler.add_job(
    #     crawl_job,
    #     'cron',
    #     day_of_week='mon-fri',
    #     hour='9-18',
    #     minute='0'
    # )
    print("定时任务已启动...")
    scheduler.start()
if __name__ == "__main__":
    start_scheduler()

使用多线程 + 定时器

import threading
import time
import requests
class CrawlTimer:
    def __init__(self, interval_seconds):
        self.interval = interval_seconds
        self.running = False
    def crawl(self):
        try:
            url = "https://api.example.com/data"
            response = requests.get(url, timeout=10)
            if response.status_code == 200:
                with open('crawl_log.txt', 'a') as f:
                    f.write(f"{time.ctime()}: 状态码 {response.status_code}\n")
                print(f"爬取成功: {time.ctime()}")
        except Exception as e:
            print(f"爬取出错: {e}")
    def run(self):
        self.running = True
        while self.running:
            self.crawl()
            time.sleep(self.interval)
    def stop(self):
        self.running = False
# 使用示例
if __name__ == "__main__":
    timer = CrawlTimer(300)  # 每5分钟执行一次
    # 启动爬虫线程
    crawl_thread = threading.Thread(target=timer.run)
    crawl_thread.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        timer.stop()
        print("爬虫已停止")

完整的生产环境示例

import schedule
import time
import requests
import json
import logging
from datetime import datetime
from typing import Dict, Any
# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('crawler.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)
class WebCrawler:
    def __init__(self, config: Dict[str, Any]):
        self.url = config['url']
        self.interval = config['interval']  # 单位：秒
        self.headers = config.get('headers', {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        })
        self.output_file = config.get('output_file', 'data.json')
    def fetch_data(self) -> Dict:
        """获取网页数据"""
        try:
            response = requests.get(
                self.url, 
                headers=self.headers,
                timeout=10
            )
            response.raise_for_status()
            # 根据实际情况解析数据
            # 这里以JSON为例
            return response.json()
        except requests.exceptions.RequestException as e:
            logger.error(f"请求失败: {e}")
            return None
    def save_data(self, data: Dict):
        """保存数据"""
        if data:
            with open(self.output_file, 'a', encoding='utf-8') as f:
                record = {
                    'timestamp': datetime.now().isoformat(),
                    'data': data
                }
                f.write(json.dumps(record, ensure_ascii=False) + '\n')
            logger.info(f"数据保存成功: {len(data)} 条记录")
    def crawl(self):
        """执行爬取任务"""
        logger.info("开始爬取数据...")
        data = self.fetch_data()
        if data:
            self.save_data(data)
            logger.info("爬取完成")
        else:
            logger.warning("未获取到数据")
def main():
    # 配置
    config = {
        'url': 'https://api.example.com/data',
        'interval': 3600,  # 每小时爬取一次
        'output_file': 'crawled_data.jsonl'
    }
    crawler = WebCrawler(config)
    # 设置定时任务
    schedule.every(config['interval']).seconds.do(crawler.crawl)
    # 启动时先执行一次
    schedule.run_all()
    logger.info(f"定时爬虫已启动，间隔: {config['interval']}秒")
    try:
        while True:
            schedule.run_pending()
            time.sleep(10)  # 每10秒检查一次
    except KeyboardInterrupt:
        logger.info("爬虫已停止")
if __name__ == "__main__":
    main()

使用装饰器的优雅方式

import schedule
import time
from functools import wraps
def scheduled(interval_seconds):
    """定时任务装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 设置定时任务
            schedule.every(interval_seconds).seconds.do(func, *args, **kwargs)
            # 立即执行一次
            func(*args, **kwargs)
            # 启动调度器
            while True:
                schedule.run_pending()
                time.sleep(1)
        return wrapper
    return decorator
@scheduled(3600)  # 每小时执行一次
def crawl_website():
    print(f"正在爬取数据... {time.ctime()}")
    # 实际的爬取逻辑
    return True
if __name__ == "__main__":
    crawl_website()

安装依赖

# 基础库
pip install requests beautifulsoup4
# 定时任务库
pip install schedule apscheduler

注意事项

1. 异常处理：always添加异常处理
2. 请求头：添加适当的User-Agent避免被屏蔽
3. 延迟：添加随机延迟避免被封IP
4. 日志记录：记录爬取日志便于排查问题
5. 数据持久化：定期保存数据，防止丢失

选择哪种方法取决于你的具体需求,schedule简单易用，APScheduler功能更强大。