用脚本实现定时抓取并分析

wen 实用脚本 2026-07-25 88

本文目录导读：

用脚本实现定时抓取并分析

基础定时抓取脚本
配置文件版本
增强版分析功能
使用方法

我来帮你创建一个完整的定时抓取和分析脚本方案,这里提供一个使用Python的示例，包含数据抓取、存储和分析功能。

基础定时抓取脚本

import requests
import time
import schedule
import json
from datetime import datetime
import pandas as pd
import sqlite3
from collections import Counter
import re
class DataCrawler:
    def __init__(self, db_name='crawler_data.db'):
        self.db_name = db_name
        self.init_database()
    def init_database(self):
        """初始化数据库"""
        conn = sqlite3.connect(self.db_name)
        cursor = conn.cursor()
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS crawled_data (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                title TEXT,
                content TEXT,
                url TEXT,
                timestamp DATETIME,
                source TEXT
            )
        ''')
        conn.commit()
        conn.close()
    def fetch_data(self, url, headers=None):
        """抓取网页数据"""
        try:
            if headers is None:
                headers = {
                    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
                }
            response = requests.get(url, headers=headers, timeout=10)
            response.raise_for_status()
            return response.text
        except Exception as e:
            print(f"抓取失败: {e}")
            return None
    def save_to_database(self, data_list):
        """保存数据到数据库"""
        conn = sqlite3.connect(self.db_name)
        cursor = conn.cursor()
        for data in data_list:
            cursor.execute('''
                INSERT INTO crawled_data (title, content, url, timestamp, source)
                VALUES (?, ?, ?, ?, ?)
            ''', (
                data.get('title', ''),
                data.get('content', ''),
                data.get('url', ''),
                datetime.now(),
                data.get('source', 'unknown')
            ))
        conn.commit()
        conn.close()
        print(f"已保存 {len(data_list)} 条数据")
class DataAnalyzer:
    def __init__(self, db_name='crawler_data.db'):
        self.db_name = db_name
    def load_data(self):
        """从数据库加载数据"""
        conn = sqlite3.connect(self.db_name)
        query = "SELECT * FROM crawled_data ORDER BY timestamp DESC"
        df = pd.read_sql_query(query, conn)
        conn.close()
        return df
    def analyze_content(self):
        """分析抓取的内容"""
        df = self.load_data()
        if df.empty:
            print("暂无数据可供分析")
            return
        print("\n" + "="*50)
        print("数据分析报告")
        print("="*50)
        # 1. 基本统计
        print(f"\n1. 数据总量: {len(df)} 条")
        print(f"2. 时间范围: {df['timestamp'].min()} 至 {df['timestamp'].max()}")
        # 2. 来源分析
        if 'source' in df.columns:
            source_counts = df['source'].value_counts()
            print(f"\n3. 数据来源分布:")
            for source, count in source_counts.items():
                print(f"   - {source}: {count} 条")
        # 3. 标题长度分析
        if 'title' in df.columns:
            df['title_length'] = df['title'].str.len()
            print(f"\n4. 标题长度分析:")
            print(f"   - 平均长度: {df['title_length'].mean():.2f} 字")
            print(f"   - 最短标题: {df['title'].loc[df['title_length'].idxmin()]}")
            print(f"   - 最长标题: {df['title'].loc[df['title_length'].idxmax()]}")
        # 4. 关键词频率分析
        all_titles = ' '.join(df['title'].dropna())
        words = re.findall(r'\w+', all_titles)
        word_freq = Counter(words).most_common(10)
        print(f"\n5. 高频关键词 TOP 10:")
        for word, count in word_freq:
            print(f"   - {word}: {count} 次")
        # 5. 每小时抓取量统计
        df['hour'] = pd.to_datetime(df['timestamp']).dt.hour
        hourly_counts = df['hour'].value_counts().sort_index()
        print(f"\n6. 每小时抓取量:")
        for hour in range(24):
            count = hourly_counts.get(hour, 0)
            bar = '█' * count
            print(f"   {hour:02d}:00 | {bar} {count}")
def crawl_and_analyze():
    """执行抓取和分析任务"""
    print(f"\n开始执行任务: {datetime.now()}")
    # 示例目标网站（替换为实际目标）
    target_urls = [
        {
            'url': 'https://example.com/news',
            'source': 'example_news'
        }
    ]
    crawler = DataCrawler()
    analyzer = DataAnalyzer()
    for target in target_urls:
        print(f"正在抓取: {target['source']}")
        html_content = crawler.fetch_data(target['url'])
        if html_content:
            # 这里需要根据实际网站结构解析数据
            # 以下为示例数据
            parsed_data = [
                {
                    'title': f"新闻标题 {datetime.now().strftime('%H:%M')}",
                    'content': "新闻内容示例",
                    'url': target['url'],
                    'source': target['source']
                }
            ]
            crawler.save_to_database(parsed_data)
    # 执行分析
    analyzer.analyze_content()
def main():
    """主函数 - 设置定时任务"""
    # 1. 立即执行一次
    crawl_and_analyze()
    # 2. 设置定时任务
    # 每30分钟执行一次
    schedule.every(30).minutes.do(crawl_and_analyze)
    # 或者特定时间执行
    # schedule.every().day.at("08:00").do(crawl_and_analyze)
    # schedule.every().day.at("12:00").do(crawl_and_analyze)
    # schedule.every().day.at("18:00").do(crawl_and_analyze)
    # 或者每周执行
    # schedule.every().monday.at("09:00").do(crawl_and_analyze)
    print("定时任务已启动...")
    print("按 Ctrl+C 停止程序")
    try:
        while True:
            schedule.run_pending()
            time.sleep(1)
    except KeyboardInterrupt:
        print("\n程序已停止")
if __name__ == "__main__":
    main()

配置文件版本

# config.py
import json
class Config:
    def __init__(self, config_file='config.json'):
        self.config_file = config_file
        self.load_config()
    def load_config(self):
        """加载配置文件"""
        try:
            with open(self.config_file, 'r', encoding='utf-8') as f:
                self.config = json.load(f)
        except FileNotFoundError:
            self.config = self.default_config()
            self.save_config()
    def default_config(self):
        """默认配置"""
        return {
            "sites": [
                {
                    "name": "示例新闻",
                    "url": "https://example.com/news",
                    "headers": {
                        "User-Agent": "Mozilla/5.0"
                    },
                    "selector": {
                        "title": "h2.title",
                        "content": "div.content",
                        "link": "a.title-link"
                    }
                }
            ],
            "schedule": {
                "interval_minutes": 30,
                "specific_times": ["08:00", "12:00", "18:00"],
                "day_of_week": ["monday", "wednesday", "friday"]
            },
            "analysis": {
                "keywords_count": 10,
                "export_format": ["csv", "json"],
                "export_path": "./exports"
            },
            "database": {
                "name": "crawler_data.db",
                "max_records": 10000
            }
        }
    def save_config(self):
        """保存配置"""
        with open(self.config_file, 'w', encoding='utf-8') as f:
            json.dump(self.config, f, indent=2, ensure_ascii=False)

增强版分析功能

# advanced_analysis.py
import numpy as np
import matplotlib.pyplot as plt
from wordcloud import WordCloud
import jieba
from datetime import datetime, timedelta
class AdvancedAnalyzer(DataAnalyzer):
    def __init__(self, db_name='crawler_data.db'):
        super().__init__(db_name)
    def generate_wordcloud(self, output_path='wordcloud.png'):
        """生成词云图"""
        df = self.load_data()
        all_content = ' '.join(df['title'].dropna() + ' ' + df['content'].dropna())
        # 中文分词
        words = jieba.cut(all_content)
        text = ' '.join(words)
        # 生成词云
        wordcloud = WordCloud(
            width=800,
            height=400,
            background_color='white',
            font_path='simhei.ttf',  # 如果有中文字体文件
            max_words=100
        ).generate(text)
        # 保存图片
        plt.figure(figsize=(10, 5))
        plt.imshow(wordcloud, interpolation='bilinear')
        plt.axis('off')
        plt.savefig(output_path, dpi=300, bbox_inches='tight')
        plt.close()
        print(f"词云图已保存至: {output_path}")
    def time_series_analysis(self, days=7):
        """时间序列分析"""
        df = self.load_data()
        df['timestamp'] = pd.to_datetime(df['timestamp'])
        # 按天统计
        end_date = datetime.now()
        start_date = end_date - timedelta(days=days)
        mask = (df['timestamp'] >= start_date) & (df['timestamp'] <= end_date)
        filtered_df = df.loc[mask]
        daily_counts = filtered_df.groupby(
            filtered_df['timestamp'].dt.date
        ).size()
        # 绘制趋势图
        plt.figure(figsize=(12, 6))
        daily_counts.plot(kind='line', marker='o')
        plt.title(f'{days} 天数据抓取趋势')
        plt.xlabel('日期')
        plt.ylabel('抓取数量')
        plt.grid(True, alpha=0.3)
        plt.xticks(rotation=45)
        plt.savefig('trend_analysis.png', dpi=300, bbox_inches='tight')
        plt.close()
        print(f"趋势图已保存至: trend_analysis.png")
        # 打印统计信息
        print(f"\n最近 {days} 天统计:")
        print(f"总抓取量: {len(filtered_df)}")
        print(f"日均抓取: {len(filtered_df)/days:.1f}")
        print(f"峰值: {daily_counts.max()} ({daily_counts.idxmax()})")

使用方法

安装依赖

pip install requests schedule pandas numpy matplotlib wordcloud jieba

运行脚本

# 基础版本
python crawler.py
# 带配置文件的版本
python crawler_with_config.py

输出示例

开始执行任务: 2024-01-15 14:30:00
正在抓取: example_news
已保存 1 条数据
==================================================
数据分析报告
==================================================
1. 数据总量: 5 条
2. 时间范围: 2024-01-15 14:00:00 至 2024-01-15 14:30:00
3. 数据来源分布:
   - example_news: 5 条长度分析:
   - 平均长度: 8.50 字
5. 高频关键词 TOP 10:
   - 新闻: 5 次 3 次
6. 每小时抓取量:
   14:00 | █████ 5

定时任务配置示例

# 不同的定时方式
schedule.every(10).minutes.do(job)           # 每10分钟
schedule.every().hour.do(job)                # 每小时
schedule.every().day.at("10:30").do(job)     # 每天10:30
schedule.every().monday.at("09:00").do(job)  # 每周一9点
schedule.every().wednesday.at("13:15").do(job) # 每周三13:15

这个脚本可以根据你的具体需求进行调整,