本文目录导读：
我来帮你创建一个完整的定时抓取和分析脚本方案,这里提供一个使用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
这个脚本可以根据你的具体需求进行调整,
- 修改目标URL和解析逻辑
- 调整定时任务的频率和时间
- 添加更多的分析功能
- 配置邮件通知等自动化功能
