我是怎么用 Python 爬新浪新闻搜索结果的
前段时间有个需求，要批量采集某个关键词相关的新闻数据做舆情分析。选了一圈数据源，最后落在新浪新闻搜索上。原因有几个：收录量大，搜索结果带时间戳和来源媒体字段，数据结构相对规整，适合做时间序列分析。另外返回的是服务端渲染的 HTML，不需要处理 JavaScript 动态加载，采集成本低。 这篇文章记录整个爬取过程的技术细节，包括请求构造、响应解析、反爬策略和代理池设计。 ## 搜索接口分析 新浪新闻搜索的入口地址是 `https://search.sina.com.cn/`。在浏览器中搜索一个关键词，通过 Chrome DevTools 抓包，可以看到结果是通过一个 GET 请求返回的： ``` GET https://search.sina.com.cn/?q=人工智能&c=news&sort=rel&range=all&num=10&page=1 ``` |参数|含义|可选值| | --- | --- | --- | |q|搜索关键词，需 URL 编码|任意字符串| |c|频道分类|news / video / blog 等| |sort|排序方式|rel（相关性）/ time（时间）| |range|时间范围|all / 1d / 1w / 1m 等| |num|每页条数|默认 10| |page|页码|从 1 开始| 新浪对 `User-Agent` 做了基本校验，缺失或为空会直接返回 403。`Referer` 字段建议携带，模拟从搜索页面发起的正常浏览行为。`Accept-Encoding` 设为 `gzip, deflate` 可以减少传输体积，`requests` 库会自动处理解压。 ## 请求封装 用 `requests.Session` 复用 TCP 连接，减少握手开销。在批量翻页场景下能明显减少请求耗时。配合 `HTTPAdapter` 设置重试策略，对 429 和 5xx 自动重试并指数退避，避免对目标服务器造成瞬时压力： ``` import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def create_session() -> requests.Session: session = requests.Session() retry = Retry(total=3, backoff_factor=1, status_forcelist=[429, 500, 502, 503], allowed_methods=["GET"]) adapter = HTTPAdapter(max_retries=retry, pool_connections=10, pool_maxsize=10) session.mount("http://", adapter) session.mount("https://", adapter) return session def search_sina_news(session, keyword, page=1, sort="time", num=10, timeout=15): url = "https://search.sina.com.cn/" params = {"q": keyword, "c": "news", "sort": sort, "range": "all", "num": num, "page": page} headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/120.0.0.0 Safari/537.36", "Referer": "https://search.sina.com.cn/" } resp = session.get(url, params=params, headers=headers, timeout=timeout) if resp.encoding and resp.encoding.lower() == "iso-8859-1": resp.encoding = resp.apparent_encoding resp.encoding = resp.encoding or "utf-8" return resp.text ``` 编码处理这段值得说明。`requests` 在响应头没有声明 `charset` 时会默认用 ISO-8859-1，导致中文乱码。这里检测到这个默认值就回退到 `apparent_encoding`（基于 `chardet` 的自动检测），再兜底用 UTF-8。新浪搜索结果页实际是 UTF-8，但这个防御性写法可以兼容后续可能遇到的非标准页面。 ## 结果解析 分析返回的 HTML 结构，每条搜索结果包裹在 `` 中，包含标题链接、摘要和来源时间信息。用 CSS 选择器提取字段，同时做基本的数据清洗： ``` from bs4 import BeautifulSoup from dataclasses import dataclass, asdict import re @dataclass class NewsItem: title: str url: str summary: str source: str pub_time: str def parse_results(html): soup = BeautifulSoup(html, "lxml") results = [] for item in soup.select(".box-result"): title_tag = item.select_one("h2 a") if not title_tag: continue meta = item.select_one(".fgray_time") meta_text = meta.get_text(strip=True) if meta else "" m = re.match(r"^(.*?)\s+(\d{4}[-/年]\d{1,2}[-/月]\d{1,2}日?.*)$", meta_text) source, pub_time = (m.group(1).strip(), m.group(2).strip()) if m else (meta_text, "") summary_tag = item.select_one("p.content") results.append(NewsItem( title=title_tag.get_text(strip=True), url=title_tag.get("href", ""), summary=summary_tag.get_text(strip=True) if summary_tag else "", source=source, pub_time=pub_time )) return results ``` 用 `dataclass` 定义数据模型比裸字典安全，字段类型在编码阶段就约束好，后续序列化到 CSV 或数据库不容易出问题。`_parse_meta` 里的正则兼容了两种常见时间格式（`2024-01-15` 和 `2024年01月15日`），比简单字符串拆分健壮。 ## 反爬策略与代理池 新浪的反爬分两个层级。第一层是基于 IP 的请求频率统计，同一 IP 短时间发送过多请求会触发限流，返回 403 或跳转验证码。第二层是行为检测，请求间隔过于均匀也会被标记为爬虫。对应策略是随机延迟加代理 IP 轮换。 纯延迟策略在采集量大时效率太低。100 个关键词每个 30 页，单 IP 延迟 3 秒要跑 2.5 小时。引入代理后多线程并发能压到 20 分钟以内。免费代理可用率通常低于 20%，建议用付费代理，关注代理类型和并发限制即可。 代理池的核心是健康检查和淘汰机制，不能拿到代理就直接用，需要在使用中记录成功率和响应时间： ``` import time import random from typing import Optional @dataclass class ProxyNode: address: str success_count: int = 0 fail_count: int = 0 avg_latency: float = 0.0 @property def score(self): total = self.success_count + self.fail_count if total == 0: return 0.5 return self.success_count / total * 0.7 + max(0, 1 - self.avg_latency / 5.0) * 0.3 class ProxyPool: def __init__(self, fetch_func, min_size=5): self._fetch = fetch_func self._min = min_size self._nodes = [] def refresh(self): for addr in self._fetch(): if not any(n.address == addr for n in self._nodes): self._nodes.append(ProxyNode(address=addr)) def get_proxy(self): available = [n for n in self._nodes if n.fail_count = 5: self._nodes.remove(node) ``` 按健康分加权选择代理，成功率高的优先调度，连续失败 5 次自动淘汰。`avg_latency` 用滑动平均（系数 0.7）平滑波动，避免单次偶发慢响应拉偏评分。代理请求的 timeout 建议设短一点（10 秒左右），质量差的代理经常建立连接后不返回数据，timeout 太长会阻塞整个采集线程。 ## 完整采集流程 把前面的模块组装起来，加上数据存储和去重： ``` import csv import logging import time import random import requests logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") logger = logging.getLogger(__name__) # 亿牛云隧道代理配置 TUNNEL_HOST = "proxy.16yun.cn" TUNNEL_PORT = "3128" TUNNEL_USER = "xxxxx" TUNNEL_PASS = "xxxxx" def get_tunnel_proxies() -> dict: """构建亿牛云隧道代理，每次请求由隧道自动切换出口 IP""" proxy_url = f"http://{TUNNEL_USER}:{TUNNEL_PASS}@{TUNNEL_HOST}:{TUNNEL_PORT}" return {"http": proxy_url, "https": proxy_url} def _is_blocked(html: str) -> bool: return any(k in html for k in ["验证码", "访问受限", "暂时无法访问"]) def search_with_tunnel( session: requests.Session, keyword: str, page: int, proxies: dict, max_retries: int = 3 ) -> str: """通过亿牛云隧道代理请求搜索接口，被拦截时自动重试换 IP""" url = "https://search.sina.com.cn/" params = {"q": keyword, "c": "news", "sort": "time", "range": "all", "num": 10, "page": page} headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/120.0.0.0 Safari/537.36", "Referer": "https://search.sina.com.cn/" } for attempt in range(max_retries): try: resp = session.get(url, params=params, headers=headers, proxies=proxies, timeout=10) if resp.encoding and resp.encoding.lower() == "iso-8859-1": resp.encoding = resp.apparent_encoding resp.encoding = resp.encoding or "utf-8" if resp.status_code == 200 and not _is_blocked(resp.text): return resp.text logger.warning(f"page={page} attempt={attempt+1} 被拦截或状态码异常") except (requests.Timeout, requests.ConnectionError) as e: logger.warning(f"page={page} attempt={attempt+1} 请求失败: {e}") time.sleep(random.uniform(1.0, 2.0)) return "" def crawl_and_save(keyword, max_pages=30, use_proxy=True): session = create_session() proxies = get_tunnel_proxies() if use_proxy else None results, seen = [], set() for page in range(1, max_pages + 1): if use_proxy: html = search_with_tunnel(session, keyword, page, proxies) else: html = search_sina_news(session, keyword, page=page) if not html or _is_blocked(html): logger.warning(f"page={page} 触发反爬，跳过") continue items = parse_results(html) if not items: break new = [r for r in items if r.url not in seen] seen.update(r.url for r in new) results.extend(new) logger.info(f"page={page} new={len(new)} total={len(results)}") time.sleep(random.uniform(1.5, 3.0)) if results: with open(f"sina_news_{keyword}.csv", "w", newline="", encoding="utf-8-sig") as f: writer = csv.DictWriter(f, fieldnames=asdict(results[0]).keys()) writer.writeheader() writer.writerows([asdict(r) for r in results]) return results ``` ``` URL 去重用 `set` 存储已抓链接，新浪搜索结果偶尔跨页重复，尤其是按相关性排序时同一篇新闻可能出现在多个页码。CSV 用 `utf-8-sig` 编码（带 BOM），Windows 下 Excel 直接打开不乱码。 ## 踩过的坑 编码不一致。搜索页是 UTF-8，但跳转到详情页后编码可能是 GB2312 或 GBK，`` 和实际字节流偶尔对不上。不要信任单一来源的编码声明，用 `chardet.detect()` 检测字节流再解码。 搜索结果有页码上限。新浪新闻搜索最多返回约 760 条（76 页），超过后返回空页面。关键词命中量大时需要用 `range` 参数按时间分段查询再合并。 短效代理的生存周期。短效代理有效期通常 1 到 5 分钟，拉取后必须尽快消费。池子不要囤太多，快用完时再拉下一批，配合实际消费速度。 摘要字段是截断的。搜索页返回的 `content` 是正文前 100 到 200 字，不是完整正文。需要全文要额外请求详情页，详情页反爬比搜索页严格。 ## 最后 新浪新闻搜索采集整体难度不高。请求是标准 GET 参数拼接，响应是服务端渲染 HTML，不需要 Selenium 或 Playwright 这类浏览器自动化工具。主要工程挑战在反爬应对：频率控制、代理池管理、健康检查和淘汰策略。这些模块设计好了，换目标站点只需改请求参数和解析逻辑，底层框架可以复用。 单线程顺序采集对多数舆情分析场景够用。需要提速的话用 `concurrent.futures.ThreadPoolExecutor`，线程数控制在 5 到 10 个，配合代理池的并发限制。不建议开太多线程，一方面代理有并发上限，另一方面对目标站点压力过大可能触发更严格的封禁策略。 代码里用 `logging` 替代 `print` 输出日志，带时间戳和级别标签，方便后续接入日志收集系统。如果采集任务需要长期运行，建议加上异常捕获和断点续采逻辑：把已完成的页码持久化到本地文件，程序中断后从断点恢复，避免重复采集。
