#!/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()