Files
robotex/src/antibot/humanize.py
T
protokeyandClaude Opus 5.5 437ea27be7 fix: слоты, безопасность, единый формат ответа и валидация
- слоты браузеров: asyncio.Semaphore, размер из BROWSER_SLOTS, при нехватке — 503 + Retry-After;
  устранена утечка счётчика при отказе (#12, #49)
- session_id принимается только как UUID, путь профиля проверяется (#5)
- секреты клиента в логах маскируются, полный дамп — только на DEBUG (#6)
- ответ: status всегда "ok"/"err", поле code (no_slots / request_failed / challenge_failed /
  action_failed), response_model (#7)
- непройденный антибот → challenge_failed, действия не выполняются (#8)
- таймауты провайдеров в мс вынесены в константы; detect переживает навигацию посреди
  проверки; пауза между неудачными solve вместо холостого цикла (#9)
- обязательные поля действий и формат прокси проверяются схемой → 422 до запуска браузера (#13)
- прокси необязателен (#15)
- user_agent сохраняется в профиле сессии и реально передаётся в браузер (#16)
- duration_ms у действий и у запроса (#18), опция screenshot (#19)
- общий конфиг логирования (LOG_LEVEL), print → логгер, CloudflareProvider,
  удалён cached_persistent, GET /health (#22)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-13 14:52:12 +04:00

152 lines
5.7 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
человекоподобные движения мыши (безье+ускорение), тайминги кликов
Раньше это частично закрывал Camoufox (форк Firefox с движением курсора по кривой при humanize=True) —
после перехода на голый Playwright + Chromium это ушло: обычный page.click()/locator.click() телепортирует
курсор мгновенно. Функции ниже — единственный источник человекоподобности сейчас.
"""
import math
import random
import string
from typing import Optional
from patchright.async_api import TimeoutError
from patchright.async_api import Page
# последняя известная позиция курсора на страницу — Playwright не даёт её прочитать из браузера,
# поэтому считаем сами, чтобы следующее движение начиналось не из (0, 0), а оттуда, где реально
# остановились в прошлый раз
_last_mouse_pos: dict[int, tuple[float, float]] = {}
def _ease_in_out(t: float) -> float:
"""smoothstep: медленный старт, разгон к середине, торможение к цели — как у реальной руки."""
return t * t * (3 - 2 * t)
def _cubic_bezier(p0, c1, c2, p1, t: float) -> tuple[float, float]:
mt = 1 - t
x = mt ** 3 * p0[0] + 3 * mt ** 2 * t * c1[0] + 3 * mt * t ** 2 * c2[0] + t ** 3 * p1[0]
y = mt ** 3 * p0[1] + 3 * mt ** 2 * t * c1[1] + 3 * mt * t ** 2 * c2[1] + t ** 3 * p1[1]
return x, y
def _control_points(x0, y0, x1, y1):
"""Два контрольных узла, смещённых перпендикулярно линии движения — даёт дугу вместо прямой,
как у настоящего движения руки/трекпада, а не у синтетического teleport'а."""
dx, dy = x1 - x0, y1 - y0
dist = math.hypot(dx, dy) or 1.0
nx, ny = -dy / dist, dx / dist
curve = random.uniform(0.08, 0.28) * dist * random.choice((-1, 1))
t1 = random.uniform(0.2, 0.4)
t2 = random.uniform(0.6, 0.8)
c1 = (x0 + dx * t1 + nx * curve, y0 + dy * t1 + ny * curve)
c2 = (x0 + dx * t2 + nx * curve * random.uniform(0.5, 1.0), y0 + dy * t2 + ny * curve * random.uniform(0.5, 1.0))
return c1, c2
def _random_start_position(page: Page) -> tuple[float, float]:
vp = page.viewport_size or {"width": 1280, "height": 720}
return (
random.uniform(vp["width"] * 0.3, vp["width"] * 0.7),
random.uniform(vp["height"] * 0.3, vp["height"] * 0.7),
)
async def move_mouse_human(
page: Page,
x: float,
y: float,
*,
steps: Optional[int] = None,
min_duration: float = 0.25,
max_duration: float = 0.9,
) -> None:
"""Двигает курсор к (x, y) по кривой Безье с ease-in-out таймингом и лёгким дрожанием на
промежуточных точках, вместо page.mouse.move(x, y), которое телепортирует курсор мгновенно.
Начальная точка — последняя известная позиция курсора на этой странице (или случайная точка
у центра вьюпорта, если это первое движение в сессии)."""
start = _last_mouse_pos.get(id(page)) or _random_start_position(page)
c1, c2 = _control_points(start[0], start[1], x, y)
dist = math.hypot(x - start[0], y - start[1])
if steps is None:
steps = max(8, min(60, int(dist / 8)))
duration = random.uniform(min_duration, max_duration)
step_delay_ms = (duration / steps) * 1000
for i in range(1, steps + 1):
t = _ease_in_out(i / steps)
px, py = _cubic_bezier(start, c1, c2, (x, y), t)
if i < steps:
px += random.uniform(-1.5, 1.5)
py += random.uniform(-1.5, 1.5)
await page.mouse.move(px, py)
await page.wait_for_timeout(step_delay_ms * random.uniform(0.7, 1.3))
_last_mouse_pos[id(page)] = (x, y)
async def click_human(
page: Page,
x: float,
y: float,
*,
button: str = "left",
**move_kwargs,
) -> None:
"""move_mouse_human + клик с человеческой задержкой между наведением и down/up, вместо
мгновенного page.mouse.click(x, y)."""
await move_mouse_human(page, x, y, **move_kwargs)
await page.wait_for_timeout(random.uniform(50, 180))
await page.mouse.down(button=button)
await page.wait_for_timeout(random.uniform(40, 99))
await page.mouse.up(button=button)
async def wait_for_networkidle(page: Page, timeout: Optional[int] = None):
"""Wait for the page to become idle (no network activity) even if there are never-ending requests."""
try:
await page.wait_for_load_state("networkidle", timeout=timeout)
except TimeoutError:
pass
async def wait_for_page_stability(page: Page, load_dom: bool, network_idle: bool):
await page.wait_for_load_state(state="load")
if load_dom:
await page.wait_for_load_state(state="domcontentloaded")
if network_idle:
await wait_for_networkidle(page)
async def typing(page: Page, locator: str, text: str):
element = page.locator(locator)
await element.scroll_into_view_if_needed()
assert await element.is_visible()
assert await element.is_enabled()
assert await element.is_editable()
await element.focus()
await element.click()
for char in text:
await page.keyboard.press(char)
if char in string.ascii_lowercase:
delays = (50, 100)
elif char in string.ascii_uppercase:
delays = (100, 150)
elif char in string.digits:
delays = (100, 150)
elif char in string.punctuation:
delays = (100, 200)
else:
raise SystemError("character not supported: {}".format(char))
await page.wait_for_timeout(random.uniform(*delays))