我是怎么用 Python 爬新浪新闻搜索结果的

原创

小白学大数据

发布于 2026-07-28 16:51:26

1250

前段时间有个需求，要批量采集某个关键词相关的新闻数据做舆情分析。选了一圈数据源，最后落在新浪新闻搜索上。原因有几个：收录量大，搜索结果带时间戳和来源媒体字段，数据结构相对规整，适合做时间序列分析。另外返回的是服务端渲染的 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 结构，每条搜索结果包裹在 <div class="box-result"> 中，包含标题链接、摘要和来源时间信息。用 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 < 3]
        if len(available) < self._min:
            self.refresh()
            available = [n for n in self._nodes if n.fail_count < 3]
        if not available:
            return None
        weights = [max(n.score, 0.01) for n in available]
        return random.choices(available, weights=weights, k=1)[0]

    def report(self, node, success, latency=0):
        if success:
            node.success_count += 1
            node.avg_latency = node.avg_latency * 0.7 + latency * 0.3
        else:
            node.fail_count += 1
            if node.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，<meta charset> 和实际字节流偶尔对不上。不要信任单一来源的编码声明，用 chardet.detect() 检测字节流再解码。

搜索结果有页码上限。新浪新闻搜索最多返回约 760 条（76 页），超过后返回空页面。关键词命中量大时需要用 range 参数按时间分段查询再合并。

短效代理的生存周期。短效代理有效期通常 1 到 5 分钟，拉取后必须尽快消费。池子不要囤太多，快用完时再拉下一批，配合实际消费速度。

摘要字段是截断的。搜索页返回的 content 是正文前 100 到 200 字，不是完整正文。需要全文要额外请求详情页，详情页反爬比搜索页严格。

最后

新浪新闻搜索采集整体难度不高。请求是标准 GET 参数拼接，响应是服务端渲染 HTML，不需要 Selenium 或 Playwright 这类浏览器自动化工具。主要工程挑战在反爬应对：频率控制、代理池管理、健康检查和淘汰策略。这些模块设计好了，换目标站点只需改请求参数和解析逻辑，底层框架可以复用。

单线程顺序采集对多数舆情分析场景够用。需要提速的话用 concurrent.futures.ThreadPoolExecutor，线程数控制在 5 到 10 个，配合代理池的并发限制。不建议开太多线程，一方面代理有并发上限，另一方面对目标站点压力过大可能触发更严格的封禁策略。

代码里用 logging 替代 print 输出日志，带时间戳和级别标签，方便后续接入日志收集系统。如果采集任务需要长期运行，建议加上异常捕获和断点续采逻辑：把已完成的页码持久化到本地文件，程序中断后从断点恢复，避免重复采集。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

python

爬虫