xinyouhe/yunke
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: anti-risk-control stealth layer (browser headers, jittered throttle, backoff, volume cap)

Browse Source 
Centralize scraping stealth in YunkeClient.request() so automated traffic
resembles the normal CRM web app and is paced conservatively to avoid
triggering 云客 risk control (pulls M6 rate-limit work forward):

- app/config.py: add 6 tunable settings with conservative defaults
  (request_min_interval=1.0, request_jitter=0.8, max_requests_per_run=0,
  risk_backoff_{base=2.0, max=60.0, attempts=4}).
- app/scraper/base.py: RateLimiter gains optional jitter + a new wait()
  method (fixed min+jitter per-call sleep); gap-based acquire() is unchanged
  for backward compat (default jitter=0.0).
- app/session/yunke_client.py: every request now carries browser-identical
  headers (UA/sec-ch-ua/accept/accept-language/referer) merged with the auth
  headers (per-request overrides win); awaits the shared jittered limiter
  before sending; enforces a per-run request cap (RequestCapReached); and
  backs off exponentially on risk signals (HTTP 429 or 200 body carrying
  频繁/风控/限制/稍后), distinct from the existing 401 relogin-once path.
- tests/conftest.py: zero the throttle env in the test suite so the
  conservative prod defaults don't slow unit tests.
- tests/test_yunke_client_stealth.py: 9 new tests (browser headers,
  override, limiter spy, 429 backoff, 200-marker backoff, persistent-429
  raise, volume cap, unlimited, 401-relogin preserved). TDD red->green.

Suite: 171 -> 181 passing. ruff clean.

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
浅念
2026-08-01 14:51:46 +08:00
parent 83b1aeb97c
commit 46e3880c0b
6 changed files with 603 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
yunke-server/app/config.py
+21
View File
@@ -28,6 +28,27 @@ class Settings(BaseSettings):
# <token>`` or ``?token=<token>``. See app/api/auth.py.
api_token: str = ""
# --- Anti-risk-control / stealth tuning ---------------------------------
# These are the knobs that pace automated scraping so it resembles the
# normal web app. Defaults are deliberately CONSERVATIVE (slow + jittered):
# roughly one request every ~1.0..1.8s. Loosen these only once we have
# evidence the CRM tolerates a faster pace. See app.session.yunke_client.
# Lower bound on seconds between successive upstream requests.
request_min_interval: float = 1.0
# Random extra 0..jitter seconds added on top of ``request_min_interval``
# per request, so the cadence is not a perfectly periodic signature.
request_jitter: float = 0.8
# Hard cap on the number of upstream requests a single client may send in
# one run. 0 = unlimited; >0 = enforce, then raise RequestCapReached.
max_requests_per_run: int = 0
# Exponential backoff parameters applied when the CRM signals risk control
# (HTTP 429 or a 200 body carrying a 风控/限流 marker). Delay for retry N
# (0-indexed) is ``risk_backoff_base * 2**N`` clamped to ``risk_backoff_max``.
risk_backoff_base: float = 2.0
risk_backoff_max: float = 60.0
# Number of retries attempted after the initial request before giving up.
risk_backoff_attempts: int = 4
@lru_cache
def get_settings() -> Settings:
yunke-server/app/scraper/base.py
+40 -7
View File
@@ -34,11 +34,16 @@ job is just to advance the cursor and stop on ``has_more == False``.
from __future__ import annotations
import asyncio
import random
import time
from collections.abc import AsyncIterator, Awaitable, Callable
from typing import Any
from typing import TYPE_CHECKING, Any
from app.session.yunke_client import YunkeClient
if TYPE_CHECKING:
# ``YunkeClient`` is referenced only in the paginator signatures. Kept
# under TYPE_CHECKING so ``app.session.yunke_client`` may import
# :class:`RateLimiter` at runtime without forming an import cycle.
from app.session.yunke_client import YunkeClient
__all__ = ["RateLimiter", "paginate_by_cursor", "paginate_by_page"]
@@ -175,21 +180,49 @@ async def paginate_by_cursor(
class RateLimiter:
"""Minimal asyncio limiter enforcing a minimum interval between calls.
Call ``await limiter.acquire()`` before each upstream request. The first
call never waits. Successive calls wait just long enough that the gap
between the previous call's dispatch and this one is at least
``min_interval`` seconds.
Two pacing modes are supported:
* :meth:`acquire` — gap-based spacing. Call it before each upstream
request; the first call never waits and successive calls wait just long
enough that the gap between dispatches is at least ``min_interval``
seconds. Used by the list paginators (:func:`paginate_by_page` /
:func:`paginate_by_cursor`).
* :meth:`wait` — fixed per-call throttle. Sleeps ``min_interval`` plus a
uniform random jitter in ``[0, jitter)`` every time it is awaited. Used
by :class:`app.session.yunke_client.YunkeClient` to pace every upstream
request with a conservative, slightly randomized cadence (the jitter
prevents a perfectly periodic signature that risk control could finger).
This is deliberately tiny: no tokens, no concurrency cap, no skew handling
— just a floor on inter-call spacing, which is all the yunke scrapers need
to stay polite to the CRM. If you need a semaphore or token bucket, that
belongs in a future helper, not here.
``jitter`` defaults to ``0.0`` so existing callers that pass only
``min_interval`` keep their exact prior behaviour (gap-based ``acquire``
never consults ``jitter``).
"""
def __init__(self, min_interval: float = 0.0) -> None:
def __init__(self, min_interval: float = 0.0, jitter: float = 0.0) -> None:
self._min_interval = float(min_interval)
self._jitter = float(jitter)
self._last: float = 0.0
async def wait(self) -> None:
"""Sleep ``min_interval`` plus a random ``[0, jitter)`` every call.
Unlike :meth:`acquire` this is not gap-aware — it is a flat, slightly
randomized per-call delay, which is what the yunke client wants ahead
of each upstream request. With ``min_interval == 0`` and
``jitter == 0`` it is a no-op.
"""
delay = self._min_interval
if self._jitter > 0:
delay += random.uniform(0.0, self._jitter)
if delay > 0:
await asyncio.sleep(delay)
async def acquire(self) -> None:
"""Block until at least ``min_interval`` has elapsed since the last call."""
if self._min_interval <= 0:
yunke-server/app/session/yunke_client.py
+192 -9
View File
@@ -7,9 +7,18 @@ CRM (see design doc §6.1 会话层). It wraps an :class:`httpx.AsyncClient` and
session is present, or when constructed with credentials.
* Injects the 5 known auth headers (``userid`` / ``company`` / ``source`` /
``channel`` / ``content-type``) on every request.
* **Anti-risk-control / stealth** (see §6.1 风控规避): every request also
carries browser-identical headers (UA / sec-ch-ua / accept / accept-language
/ referer) so automated traffic looks like the normal web app; a shared
jittered :class:`app.scraper.base.RateLimiter` paces each request; a
per-run volume cap guards against runaway loops; and risk-control signals
(HTTP 429 or a 200 body carrying a 风控/限流 marker) trigger exponential
backoff + retry.
* Detects HTTP 401 (or a body-level session-expired marker), re-logs-in
*once*, and retries. A persistent 401 raises after the single retry — no
infinite loop.
infinite loop. The 401 path is deliberately distinct from the 429/limit
path: a 401 is a session death (-> relogin), a 429 is throttling
(-> backoff); they are never confused.
Phase-0 note: the session is held in-memory only. Future work will persist
the encrypted session blob to ``YkAccount.session_blob_enc`` (see
@@ -19,11 +28,14 @@ session instead of always logging in fresh.
from __future__ import annotations
import asyncio
from collections.abc import Mapping
from typing import Any
import httpx
from app.config import get_settings
from app.scraper.base import RateLimiter
from app.session.login import (
LoginResult,
get_secure_key,
@@ -38,6 +50,46 @@ _STATIC_AUTH_HEADERS: dict[str, str] = {
"content-type": "application/json",
}
# Browser-identical headers captured from the real CRM web app. Sent on EVERY
# request (merged ahead of the auth headers; per-request overrides win) so the
# automated client presents the same fingerprint as a normal Chrome-on-Windows
# user. ``referer`` defaults to the app root; callers can override it per call
# when they know the precise cms path. See §6.1 风控规避.
_BROWSER_HEADERS: dict[str, str] = {
"user-agent": (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/150.0.0.0 Safari/537.36"
),
"accept": "application/json, text/plain, */*",
"accept-language": "zh-CN",
"sec-ch-ua": '"Not;A=Brand";v="8", "Chromium";v="150", "Google Chrome";v="150"',
"sec-ch-ua-mobile": "?0",
"sec-ch-ua-platform": '"Windows"',
"referer": "https://crm2.yunkecn.com/",
}
# Substrings that, when found in a HTTP 200 response body, indicate the CRM is
# throttling / risk-controlling us even though it returned 200 (a common
# Chinese-CRM idiom). Kept deliberately conservative — the explicit markers
# observed in 风控 messages. See §6.1 风控规避.
_RISK_BODY_MARKERS: tuple[str, ...] = ("频繁", "风控", "限制", "稍后")
class RequestCapReached(Exception):
"""Raised when a run exceeds ``max_requests_per_run`` upstream calls.
Carries the cap and the count so callers can surface a useful message.
The exception is a control-flow signal for orchestrators (a run that hits
the cap should stop scheduling more work, not crash the process).
"""
def __init__(self, cap: int, count: int) -> None:
self.cap = cap
self.count = count
super().__init__(
f"request cap reached: {count} requests issued, cap is {cap}"
)
class YunkeClient:
"""Authenticated async HTTP client for the yunke CRM API."""
@@ -52,6 +104,12 @@ class YunkeClient:
company: str | None = None,
cookies: Mapping[str, str] | None = None,
client: httpx.AsyncClient | None = None,
# --- stealth / anti-risk-control knobs (override Settings defaults) ---
rate_limiter: RateLimiter | None = None,
max_requests_per_run: int | None = None,
backoff_base: float | None = None,
backoff_max: float | None = None,
backoff_attempts: int | None = None,
) -> None:
"""Construct a client either from credentials or an existing session.
@@ -68,6 +126,10 @@ class YunkeClient:
:class:`httpx.AsyncClient` (e.g. for custom transports). When omitted
a new client bound to ``base_url`` is created and owned by this
instance (closed by :meth:`aclose`).
Stealth knobs default to the conservative values declared on
:class:`app.config.Settings`; pass any of them explicitly to override
(tests inject a no-op ``rate_limiter`` and small backoff to stay fast).
"""
has_creds = username is not None and password is not None
has_session = userid is not None and company is not None
@@ -93,6 +155,28 @@ class YunkeClient:
for name, value in (cookies or {}).items():
self._client.cookies.set(name, value)
# --- stealth configuration ----------------------------------------
# Defaults come from Settings; explicit args win. Tests pass explicit
# no-op values so the suite doesn't sleep on the conservative prod
# defaults.
s = get_settings()
self._rate_limiter = (
rate_limiter
if rate_limiter is not None
else RateLimiter(
min_interval=s.request_min_interval, jitter=s.request_jitter
)
)
self._max_requests = (
s.max_requests_per_run if max_requests_per_run is None else max_requests_per_run
)
self._backoff_base = s.risk_backoff_base if backoff_base is None else backoff_base
self._backoff_max = s.risk_backoff_max if backoff_max is None else backoff_max
self._backoff_attempts = (
s.risk_backoff_attempts if backoff_attempts is None else backoff_attempts
)
self._request_count = 0
# -- properties ---------------------------------------------------------
@property
@@ -104,6 +188,11 @@ class YunkeClient:
def logged_in(self) -> bool:
return self.login_result is not None
@property
def request_count(self) -> int:
"""Number of upstream requests issued through this client so far."""
return self._request_count
# -- login --------------------------------------------------------------
async def login(self) -> LoginResult:
@@ -133,20 +222,89 @@ class YunkeClient:
path: str,
*,
headers: Mapping[str, str] | None = None,
**kwargs: Any,
) -> httpx.Response:
"""Send an authed request with the full stealth layer applied.
Per request:
1. **Volume cap** — if ``max_requests_per_run > 0`` and the counter
has reached it, raise :class:`RequestCapReached` (fail fast, before
pacing or sending anything).
2. **Throttle** — await the shared jittered :class:`RateLimiter` once,
so successive calls are spaced at the conservative cadence.
3. **Send + 401 handling** — :meth:`_send_with_relogin` performs the
actual request (with browser + auth headers) and re-logs-in once on
a 401/session-expired signal, exactly as before.
4. **Backoff on risk-control signals** — if the response is a 429 or a
200 carrying a 风控/限流 marker, sleep
``backoff_base * 2**attempt`` (capped at ``backoff_max``) and retry
up to ``backoff_attempts`` times; raise on persistent throttling.
This is intentionally separate from the 401 path.
Per-request ``headers`` override both the browser and the auth
defaults for this call only (per-request values win).
"""
# 1. Volume cap.
if self._max_requests > 0 and self._request_count >= self._max_requests:
raise RequestCapReached(cap=self._max_requests, count=self._request_count)
self._request_count += 1
# 2. Throttle ahead of every request.
await self._rate_limiter.wait()
user_headers = dict(headers or {})
# 3 + 4. Send, then back off on risk-control signals. The 401-relogin
# behaviour lives inside ``_send_with_relogin`` and is unaffected by
# this loop — a 401 is never treated as a rate-limit signal.
last_resp: httpx.Response | None = None
for attempt in range(self._backoff_attempts + 1):
resp = await self._send_with_relogin(method, path, user_headers, **kwargs)
if not _is_rate_limited(resp):
return resp
last_resp = resp
if attempt >= self._backoff_attempts:
break # retries exhausted -> raise below
delay = min(
self._backoff_base * (2 ** attempt), self._backoff_max
)
await asyncio.sleep(delay)
# Persistent risk-control signal after all retries — surface it.
assert last_resp is not None # loop ran at least once
raise httpx.HTTPStatusError(
"yunke risk-control signal persisted after"
f" {self._backoff_attempts} backoff retries",
request=last_resp.request,
response=last_resp,
)
async def _send_with_relogin(
self,
method: str,
path: str,
user_headers: Mapping[str, str],
*,
_retried: bool = False,
**kwargs: Any,
) -> httpx.Response:
"""Send an authed request, auto-logging-in and re-logging-in on 401.
"""Perform one actual send with browser + auth headers and 401-relogin.
Per-request ``headers`` override the auth defaults for this call
only. ``_retried`` is an internal guard against infinite re-login
loops and must not be set by callers.
Auto-logs-in on the first call when no session is present. On a
session-expired signal (401 or body code 401) re-logs-in *once* and
retries; a persistent 401 raises so a dead session cannot loop. This
preserves the pre-stealth behaviour verbatim — the only addition is
the browser headers merged into the outbound request.
"""
if self.login_result is None:
await self.login()
user_headers = dict(headers or {})
merged = {**self._auth_headers(), **user_headers}
merged = {
**_BROWSER_HEADERS,
**self._auth_headers(),
**user_headers,
}
resp = await self._client.request(method, path, headers=merged, **kwargs)
@@ -160,8 +318,8 @@ class YunkeClient:
response=resp,
)
await self.login()
return await self.request(
method, path, headers=user_headers, _retried=True, **kwargs
return await self._send_with_relogin(
method, path, user_headers, _retried=True, **kwargs
)
return resp
@@ -198,6 +356,9 @@ def _is_session_expired(resp: httpx.Response) -> bool:
idiom) -> yes. We deliberately stay narrow here — only the explicit
auth-failed code, not generic non-zero codes, to avoid re-login storms
on ordinary business errors.
Note: this is deliberately disjoint from :func:`_is_rate_limited`. A 401
is relogin territory; a 429/limit is backoff territory.
"""
if resp.status_code == 401:
return True
@@ -206,3 +367,25 @@ def _is_session_expired(resp: httpx.Response) -> bool:
except (ValueError, httpx.DecodingError):
return False
return isinstance(body, dict) and body.get("code") in (401, "401")
def _is_rate_limited(resp: httpx.Response) -> bool:
"""Heuristic: does this response indicate risk-control / throttling?
Triggers backoff when:
* HTTP 429 (the canonical "too many requests" status), OR
* HTTP 200 with a body whose text contains any of the conservative
risk-control markers (频繁/风控/限制/稍后). Many Chinese CRMs signal
throttling inline in a 200 envelope rather than via a proper status.
Deliberately conservative: we do NOT flag generic non-zero business codes
(those are ordinary errors, not throttling) and we never flag 401 (that
is session-expiry, handled by :func:`_is_session_expired` + relogin).
"""
if resp.status_code == 429:
return True
if resp.status_code != 200:
return False
text = resp.text
return any(marker in text for marker in _RISK_BODY_MARKERS)