weibo_scraper.py
5.32 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
#!/usr/bin/env python3
"""
Weibo Scraper - 微博热点内容爬取工具
"""
import json
import time
import requests
from datetime import datetime
from typing import Optional, List, Dict, Any
import argparse
import os
import sys
class WeiboScraper:
"""微博热点内容爬取器"""
def __init__(self, config: Optional[Dict[str, Any]] = None):
self.config = config or {}
self.base_url = "https://weibo.com/ajax/side/hotSearch"
self.headers = {
"User-Agent": "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",
"Accept": "application/json, text/plain, */*",
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
"Referer": "https://weibo.com"
}
self.session = requests.Session()
self.session.headers.update(self.headers)
def get_hot_search(self, limit: int = 50) -> List[Dict[str, Any]]:
"""获取微博热搜榜"""
try:
response = self.session.get(self.base_url, timeout=10)
if response.status_code == 200:
data = response.json()
if data.get("ok") == 1:
return data.get("data", {}).get("realtime", [])[: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]:
"""格式化单条热搜数据"""
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": {
"likes": item.get("num", 0),
"comments": 0,
"shares": 0,
"read_count": item.get("raw_hot", 0)
},
"content": {
"title": item.get("word", ""),
"text": item.get("note", ""),
"media_urls": [],
"category_tags": [category]
},
"author": {
"username": "",
"followers": 0,
"verified": False
},
"timestamp": datetime.now().isoformat(),
"metrics": {
"hot_rank": item.get("rank", 0),
"hot_score": item.get("raw_hot", 0),
"trend": "up" if item.get("label_name") else "stable"
}
}
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()