weibo_scraper.py
9.47 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
#!/usr/bin/env python3
"""
Weibo Scraper - 微博热点内容爬取工具
"""
import json
import time
import random
import requests
from datetime import datetime
from typing import Optional, List, Dict, Any
import argparse
import os
import sys
# User-Agent 轮换池
USER_AGENTS = [
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit/605.1.15 Mobile/15E148 Safari/604.1",
"Mozilla/5.0 (Linux; Android 13; Pixel 7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/112.0.0.0 Mobile Safari/537.36",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Safari/605.1.15"
]
class WeiboScraper:
"""微博热点内容爬取器"""
def __init__(self, config: Optional[Dict[str, Any]] = None):
self.config = config or {}
self.base_url = "https://m.weibo.cn/api/index/list?type=hot" # 移动端 API 稳定性更高
self.session = requests.Session()
self._update_headers()
self.request_count = 0
self.last_request_time = 0
def _update_headers(self):
"""更新请求头，随机选择 User-Agent"""
self.session.headers.update({
"User-Agent": random.choice(USER_AGENTS),
"Accept": "application/json, text/plain, */*",
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
"Accept-Encoding": "gzip, deflate, br",
"Connection": "keep-alive",
"Referer": "https://weibo.com"
})
def _validate_response(self, response) -> bool:
"""验证返回数据是否有效（检测反爬拦截）"""
# 检查是否返回 HTML（通常是验证码或拦截页面）
if response.headers.get("Content-Type", "").startswith("text/html"):
print("Warning: Received HTML instead of JSON (possible captcha/block)", file=sys.stderr)
return False
# 检查 JSON 解析是否成功
try:
data = response.json()
if data.get("ok") != 1:
print(f"Warning: API returned ok={data.get('ok')}", file=sys.stderr)
return False
return True
except:
return False
def _rate_limit(self):
"""频率控制：请求间隔 3-5 秒随机延迟"""
now = time.time()
elapsed = now - self.last_request_time
if elapsed < 3:
delay = random.uniform(3, 5)
time.sleep(delay)
self.last_request_time = time.time()
self.request_count += 1
# 每分钟请求数 ≤ 20
if self.request_count >= 20:
time.sleep(60)
self.request_count = 0
def get_hot_search(self, limit: int = 50) -> List[Dict[str, Any]]:
"""获取微博热搜榜"""
try:
self._rate_limit()
# 随机更换 UA
self._update_headers()
response = self.session.get(self.base_url, timeout=10)
# 验证返回数据是否有效
if not self._validate_response(response):
print("Response validation failed, trying again with new UA...", file=sys.stderr)
self._update_headers()
time.sleep(5)
response = self.session.get(self.base_url, timeout=10)
if response.status_code == 200:
data = response.json()
if data.get("ok") == 1:
# 移动端 API 数据结构不同
return data.get("data", []).get("realtime", [])[:limit]
elif response.status_code in [403, 429]:
# 反爬处理：等待后重试
print(f"Rate limited, waiting 30s...", file=sys.stderr)
time.sleep(30)
return self.get_hot_search(limit)
return []
except Exception as e:
print(f"Error fetching hot search: {e}", file=sys.stderr)
return []
def format_item(self, item: Dict[str, Any], category: str = "综合") -> Dict[str, Any]:
"""格式化单条热搜数据
注意：热搜 API 返回的字段有限，comments/shares/read_count 无法直接获取
尝试从 item 中提取，如无则标记为 -1 表示"未获取"
"""
# 尝试从 API 返回的数据中提取 engagement 信息
raw_num = item.get("num", 0) # 热度值
raw_hot = item.get("raw_hot", 0) # 原始热度
# 判断是点赞数还是热度值
likes = raw_num if raw_num > 0 else raw_hot
# 热搜 API 不提供 comments/shares，标记为 null 表示未获取
# null 更明确表示数据不可用，而非数据缺失
engagement = {
"likes": likes,
"comments": item.get("comments") if "comments" in item else None, # 热搜 API 不返回此字段
"shares": item.get("shares") if "shares" in item else None,
"read_count": raw_hot if raw_hot > 0 else None
}
# 保留原始数据，便于后续分析和验证
raw_data = {k: v for k, v in item.items() if not k.startswith('_')}
return {
"source": {
"platform": "weibo",
"category": category,
"hot_id": str(item.get("word_scheme", "")),
"url": f"https://s.weibo.com/weibo?q={item.get('word', '')}"
},
"engagement": engagement,
"content": {
"title": item.get("word", ""),
"text": item.get("note", ""),
"media_urls": [],
"category_tags": [category]
},
"author": {
"username": item.get("user_nick") if "user_nick" in item else None, # 热搜 API 不返回作者信息
"followers": item.get("user_followers") if "user_followers" in item else None,
"verified": item.get("verified", False)
},
"timestamp": datetime.now().isoformat(),
"metrics": {
"hot_rank": item.get("rank", 0),
"hot_score": raw_hot,
"trend": "up" if item.get("label_name") else "stable"
},
"metadata": {
"采集时间": datetime.now().isoformat(),
"数据来源": "weibo_hotsearch_api",
"API版本": "v2",
"反爬合规": {
"ua_rotation": True,
"rate_limiting": True,
"response_validation": True
}
},
"raw_data": raw_data # 保留原始数据，便于后续分析
}
def scrape(self, category: str = "综合", limit: int = 50) -> Dict[str, Any]:
"""执行爬取"""
items = self.get_hot_search(limit)
results = []
for item in items:
if item.get("flag", 0) in [1, 2]: # 有效数据
# 根据关键词判断分类
category = self._detect_category(item.get("word", ""))
results.append(self.format_item(item, category))
return {
"success": True,
"timestamp": datetime.now().isoformat(),
"count": len(results),
"data": results
}
def _detect_category(self, keyword: str) -> str:
"""根据关键词检测分类"""
keyword = keyword.lower()
categories = {
"娱乐": ["明星", "演员", "歌手", "电影", "电视剧", "综艺", "演唱会", "恋情", "结婚", "离婚", "分手"],
"社会": ["事件", "新闻", "事故", "疫情", "政策", "法律", "民生"],
"科技": ["手机", "电脑", "AI", "互联网", "数码", "科技", "发布会"],
"搞笑": ["搞笑", "段子", "视频", "搞笑", "沙雕", "幽默"],
"体育": ["足球", "篮球", "NBA", "世界杯", "奥运", "比赛", "冠军"],
"财经": ["股票", "基金", "经济", "财经", "房价", "油价"]
}
for cat, keywords in categories.items():
if any(kw in keyword for kw in keywords):
return cat
return "综合"
def save_results(data: Dict[str, Any], output_path: str):
"""保存结果到文件"""
os.makedirs(os.path.dirname(output_path), exist_ok=True)
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
def main():
parser = argparse.ArgumentParser(description="微博热点内容爬取工具")
parser.add_argument("--category", type=str, default="综合", help="分类筛选")
parser.add_argument("--limit", type=int, default=50, help="爬取数量限制")
parser.add_argument("--output", type=str, default="workspace/weibo_output.json", help="输出文件路径")
parser.add_argument("--cron", action="store_true", help="定时模式(仅输出不打印)")
args = parser.parse_args()
scraper = WeiboScraper()
result = scraper.scrape(category=args.category, limit=args.limit)
if args.cron:
output_path = args.output
else:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
output_path = f"workspace/weibo_{timestamp}.json"
save_results(result, output_path)
if not args.cron:
print(json.dumps(result, ensure_ascii=False, indent=2))
else:
print(f"Saved to {output_path}")
if __name__ == "__main__":
main()