搜狗微信点选验证的会话保持与 Cookie 续期实战

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

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))

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))

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)

def rotate_proxy(config: CrawlerConfig, new_proxy: str) -> None:
    config.proxy = new_proxy   # 新出口，如亿牛云切换接口返回
    # 之后用新 config 重建 SogouSession 即生效

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()

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()