weibo_scraper.py
10 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
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
#!/usr/bin/env python3
"""
Weibo Scraper - 微博热点内容爬取工具
Compliance notes:
- robots.txt: respect weibo.com Crawl-delay spirit (≥10s intervals)
- Data retention: purge outputs after 90 days
- Privacy: do not collect UID, mobile numbers, or private messages
"""
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
from urllib.parse import quote
# 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:
"""验证返回数据是否有效（检测反爬拦截）"""
try:
# fallback: check if response is HTML instead of JSON
if response.headers.get("Content-Type", "").startswith("text/html"):
print("Warning: Received HTML instead of JSON (possible captcha/block)", file=sys.stderr)
return False
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 Exception:
return False
except Exception:
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]:
# 反爬处理：等待后 retry
print(f"Rate limited, waiting 30s then retry...", 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 无法直接获取
标记为 null 表示"未获取"
"""
raw_num = item.get("num", 0)
raw_hot = item.get("raw_hot", 0)
likes = raw_num if raw_num > 0 else raw_hot
word = item.get("word", "")
word_scheme = item.get("word_scheme", "")
hot_id = word_scheme if word_scheme else word
engagement = {
"likes": likes,
"comments": item.get("comments") if "comments" in item else None,
"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('_')}
now_iso = datetime.now().isoformat()
return {
"source": {
"platform": "weibo",
"category": category,
"hot_id": str(hot_id),
"url": f"https://s.weibo.com/weibo?q={quote(word)}"
},
"engagement": engagement,
"content": {
"title": word,
"text": item.get("note", "") if item.get("note") else word,
"media_urls": [],
"category_tags": [category]
},
"author": {
"username": item.get("user_nick") if "user_nick" in item else None,
"followers": item.get("user_followers") if "user_followers" in item else None,
"verified": item.get("verified", False)
},
"timestamp": now_iso,
"metrics": {
"hot_rank": item.get("rank", 0),
"real_pos": item.get("realpos") if item.get("realpos") is not None else item.get("rank", 0) + 1,
"hot_score": raw_hot,
"trend": "up" if item.get("label_name") else "stable",
"topic_flag": bool(item.get("topic_flag", False))
},
"metadata": {
"source_url": "https://weibo.com/ajax/side/hotSearch",
"crawl_time": now_iso,
"data_source": "weibo_hotsearch_api",
"api_version": "v2",
"flag_desc": item.get("flag_desc") if item.get("flag_desc") else None,
"icon_url": item.get("icon") if item.get("icon") else None,
"compliance": {
"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 = []
seen_words = set()
for item in items:
if item.get("is_ad") is True or item.get("topic_ad") is True:
continue
if item.get("flag", 0) not in [0, 1, 2]:
continue
word = item.get("word", "")
if word in seen_words:
continue
seen_words.add(word)
cat = self._detect_category(word)
results.append(self.format_item(item, cat))
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()