搜狗微信（weixin.sogou.com）是公众号文章监测的常用入口，信任状态挂在 Cookie 上。过了点选验证码之后，SNUID、SUV 这类凭证有时效，过期就重新弹验证。核心做法是 requests.Session 养长寿命会话、按实效提前续期、尽量别掉回验证那步。下面以代码为主，叙述从简。
- 会话与续期核心
```from future import annotations
import logging
import time
from dataclasses import dataclass, field
from typing import Optional
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
logger = logging.getLogger("sogou_weixin")
@dataclass
class CrawlerConfig:
base_url: str = "https://weixin.sogou.com"
search_path: str = "/weixin"
proxy: Optional[str] = None # "http://user:pass@proxy.yiniu.com:port"
max_retries: int = 3
timeout: float = 10.0
cookie_ttl: int = 600
ua_pool: list[str] = field(default_factory=list)
class SogouSession:
def init(self, config: CrawlerConfig) -> None:
self.config = config
self.session = self._build_session()
self.cookie_issued_at: float = 0.0
def _build_session(self) -> requests.Session:
session = requests.Session()
retry = Retry(total=self.config.max_retries, backoff_factor=0.5,
status_forcelist=[429, 500, 502, 503, 504])
session.mount("https://", HTTPAdapter(max_retries=retry))
if self.config.proxy:
session.proxies.update({"https": self.config.proxy})
ua = self.config.ua_pool[0] if self.config.ua_pool else (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/124.0 Safari/537.36")
session.headers.update({"User-Agent": ua, "Accept-Language": "zh-CN,zh;q=0.9"})
return session
def ensure_session(self) -> None:
if not self.session.cookies or \
time.time() - self.cookie_issued_at > self.config.cookie_ttl:
self._refresh()
def _refresh(self) -> None:
try:
resp = self.session.get(self.config.base_url + self.config.search_path,
params={"type": 2, "query": "测试"},
timeout=self.config.timeout)
resp.raise_for_status()
if self._is_captcha_page(resp.text):
logger.warning("续期被弹验证码，需重新过验证")
self._solve_captcha()
self.cookie_issued_at = time.time()
logger.info("续期成功，Cookie 数量=%d", len(self.session.cookies))
except requests.RequestException as exc:
logger.warning("续期失败：%s", exc)
raise
def _is_captcha_page(self, html: str) -> bool:
return "请输入验证码" in html or "antispider" in html
def _solve_captcha(self) -> None:
raise NotImplementedError("接入点选验证码识别服务")
def fetch(self, query: str, page: int = 1) -> str:
self.ensure_session()
resp = self.session.get(self.config.base_url + self.config.search_path,
params={"type": 2, "query": query, "page": page},
timeout=self.config.timeout)
resp.raise_for_status()
return resp.text
续期阈值别拍脑袋设。先空跑看第几分钟被弹，那个点减一两分钟就是安全线。实测 SNUID 二十分钟出头松动，所以 cookie_ttl 设 600 秒留缓冲。
2. 返回内容校验，防静默失效
Cookie 还在但信任没了很常见，返回页看着正常实则是验证页。fetch 之后必须校验。
```import re
RESULT_ITEM_RE = re.compile(r'class="txt-box"') # 正常结果条目特征
CAPTCHA_HINT_RE = re.compile(r"请输入验证码|antispider")
def validate_result(html: str) -> bool:
"""True 表示页面可信：含正常结果结构且无非验证提示。"""
if CAPTCHA_HINT_RE.search(html):
return False
return bool(RESULT_ITEM_RE.search(html))
命中异常就 session.cookies.clear()，下一轮 ensure_session 强制重握手，代价只是多一次首页请求。
- UA 池与请求节奏
别用默认 UA，准备几个轮换，翻页之间加随机休眠。
```import random
UA_POOL = [
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/124.0 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",
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/123.0 Safari/537.36",
]
def pick_ua() -> str:
return random.choice(UA_POOL)
def human_sleep() -> None:
time.sleep(random.uniform(1.5, 4.0))
CrawlerConfig.ua_pool 填上 UA_POOL，_build_session 会随机取一个；每次 human_sleep() 后再翻页。
4. 多会话 Cookie 池
大批量关键词用单会话扛不住，起一组预热会话并行。
```from threading import Lock
class SogouSessionPool:
def __init__(self, config: CrawlerConfig, size: int = 4) -> None:
self.config = config
self._lock = Lock()
self._pool: list[SogouSession] = []
for _ in range(size):
s = SogouSession(config)
s._refresh() # 启动即握手预热
self._pool.append(s)
def acquire(self) -> SogouSession:
with self._lock:
s = self._pool.pop()
s.ensure_session()
if not validate_result(s.fetch("测试")):
s.session.cookies.clear()
s._refresh()
return s
def release(self, s: SogouSession) -> None:
with self._lock:
self._pool.append(s)
被弹验证码的会话移出池子走 _solve_captcha，解完放回，解不开丢弃换出口重建。
代理与出口切换
搜狗看出口 IP 的稳定和干净。亿牛云的企业级代理支持独享 IP 与会话保持，同一任务绑定固定出口，续期时 IP 不变，验证码触发率更低。配置就是 CrawlerConfig 里那行 proxy，账号域名端口填进去即可，按请求计费对平时静默、偶尔爆发的采集也合适。
出口断了换 IP 再握手：
```def rotate_proxy(config: CrawlerConfig, new_proxy: str) -> None:
config.proxy = new_proxy # 新出口，如亿牛云切换接口返回之后用新 config 重建 SogouSession 即生效
```
验证码钩子示例
_solve_captcha 真实环境接打码平台或自研模型，这里给个签名示例。
```
from dataclasses import dataclass
@dataclass
class ClickPoint:
x: int
y: int
def _solve_captcha(self) -> None:
# 1. 拉取验证图片与提示文字
# 2. 识别点击序列（模型 / 打码平台）
points: list[ClickPoint] = recognize(self.session)
# 3. 按坐标回提交接口
submit(self.session, points)
self.cookie_issued_at = time.time()
7. 完整调用
```def parse(html: str) -> None:
"""从结果页抽取文章条目，按业务入库。"""
# ...
def main() -> None:
config = CrawlerConfig(proxy="http://user:pass@proxy.yiniu.com:port",
ua_pool=UA_POOL)
pool = SogouSessionPool(config, size=4)
keywords = ["双减政策", "研学旅行", "课后服务"]
for kw in keywords:
s = pool.acquire()
try:
for page in range(1, 11):
html = s.fetch(kw, page)
if not validate_result(html):
s.session.cookies.clear()
s._refresh()
continue
parse(html)
human_sleep()
finally:
pool.release(s)
if __name__ == "__main__":
main()
骨架就三件事：Session 养长寿命会话，按实效提前续期，稳定干净的高匿代理守住出口。续期和返回校验做扎实，验证码求解当兜底，采集就能稳稳跑完一轮。
