cryptorugmunch
a7c30b12cd
Mass ruff auto-fix:
- ruff check --fix: 109 issues fixed (F401 unused imports,
I001 unsorted imports, UP037 quoted annotations, SIM105
suppressible exception, RUF100 unused-noqa)
- ruff check --fix --unsafe-fixes: 22 additional issues
- ruff format: 70 files reformatted
- Manual pass: fix 16 misplaced import httpx lines
- Manual pass: fix remaining E402 (import-after-docstring)
Result: 283 errors -> 30 errors.
The remaining 30 are real issues that need manual review:
5 F401 unused-import (likely auto-generated stubs)
5 F821 undefined-name (real bugs in code that references
redis/pydantic/LLMRegistry without imports)
3 BLE001 (the compliance LLM fallback is intentional; the
other two are real)
3 RUF012 mutable-class-default
3 SIM105, 3 SIM117, 2 E722, 2 E741
1 B007, 1 B025, 1 E402, 1 RUF200 (pyproject.toml issue)
Tests: 436/437 pass (1 pre-existing SSE sandbox failure).
format check + import sort: now clean.
make ci: still gated on the 30 remaining real issues.
Follow-up: triage the 30 issues file-by-file.
192 lines
7.3 KiB
Python
192 lines
7.3 KiB
Python
"""Pry — Cookie Warming and Session Aging.
|
|
Pre-age cookies by browsing legitimate pages first. Aged cookies with
|
|
realistic browsing history bypass anti-bot detection that checks for
|
|
'fresh' cookies. This is a technique used by professional scraping services."""
|
|
|
|
# SPDX-License-Identifier: BSL-1.1
|
|
# Copyright (c) 2026 Rug Munch Media LLC
|
|
#
|
|
# Part of Pry — Stealth / Anti-Detection Module
|
|
# Licensed under Business Source License 1.1 — see LICENSE-BSL-STEALTH.
|
|
# Change Date: 2029-01-01 (converts to MIT).
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import random
|
|
import time
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
WARMER_DIR = Path(__file__).parent / "warmed_cookies"
|
|
WARMER_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
# Realistic pages to "warm" cookies against (these are common referrers/browsing paths)
|
|
WARMING_PAGES = {
|
|
"amazon": [
|
|
"https://www.amazon.com/",
|
|
"https://www.amazon.com/gp/help/customer/contact-us",
|
|
"https://www.amazon.com/privacy",
|
|
"https://www.amazon.com/conditions-of-use",
|
|
],
|
|
"ebay": [
|
|
"https://www.ebay.com/",
|
|
"https://www.ebay.com/help/home",
|
|
"https://www.ebay.com/myp/PurchaseHistory",
|
|
],
|
|
"shopify": ["https://www.shopify.com/", "https://www.shopify.com/pricing"],
|
|
"twitter": [
|
|
"https://twitter.com/",
|
|
"https://twitter.com/explore",
|
|
"https://twitter.com/settings/account",
|
|
],
|
|
"linkedin": ["https://www.linkedin.com/", "https://www.linkedin.com/help/intro"],
|
|
"generic": [
|
|
"https://www.google.com/",
|
|
"https://en.wikipedia.org/wiki/Main_Page",
|
|
"https://github.com/",
|
|
],
|
|
}
|
|
|
|
|
|
class CookieWarmer:
|
|
"""Warm cookies by simulating realistic user browsing before scraping."""
|
|
|
|
def __init__(self, browser_pool: Any = None):
|
|
self._browser_pool = browser_pool
|
|
|
|
async def warm_for_site(
|
|
self,
|
|
target_domain: str,
|
|
session_id: str = "",
|
|
pages_to_visit: int = 3,
|
|
min_delay: float = 2.0,
|
|
max_delay: float = 8.0,
|
|
) -> dict[str, Any]:
|
|
"""Warm cookies for a target domain by visiting legitimate pages first.
|
|
|
|
Args:
|
|
target_domain: Domain to warm cookies for (e.g., "amazon.com")
|
|
session_id: Session identifier (for storing warmed cookies)
|
|
pages_to_visit: Number of pages to visit to warm cookies
|
|
min_delay: Minimum delay between page visits (seconds)
|
|
max_delay: Maximum delay between page visits (seconds)
|
|
"""
|
|
from playwright.async_api import async_playwright
|
|
|
|
if not session_id:
|
|
session_id = f"{target_domain}_{int(time.time())}"
|
|
|
|
# Find warming pages for this domain
|
|
domain_key = next((k for k in WARMING_PAGES if k in target_domain.lower()), "generic")
|
|
pages = WARMING_PAGES[domain_key][:pages_to_visit]
|
|
|
|
cookies: list[dict[str, Any]] = []
|
|
user_agent = (
|
|
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
|
|
"(KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36"
|
|
)
|
|
|
|
try:
|
|
async with async_playwright() as pw:
|
|
browser = await pw.chromium.launch(headless=True, args=["--no-sandbox"])
|
|
context = await browser.new_context(
|
|
viewport={"width": 1920, "height": 1080},
|
|
user_agent=user_agent,
|
|
)
|
|
# Add stealth
|
|
await context.add_init_script("""
|
|
Object.defineProperty(navigator, 'webdriver', {get: () => undefined});
|
|
Object.defineProperty(navigator, 'plugins', {get: () => [1,2,3,4,5]});
|
|
""")
|
|
page = await context.new_page()
|
|
|
|
for warming_url in pages:
|
|
try:
|
|
await page.goto(warming_url, wait_until="domcontentloaded", timeout=30000)
|
|
# Human-like behavior: scroll, hover
|
|
await page.evaluate("window.scrollTo(0, document.body.scrollHeight * 0.3)")
|
|
await page.wait_for_timeout(random.randint(500, 2000))
|
|
await page.evaluate("window.scrollTo(0, 0)")
|
|
await page.wait_for_timeout(random.randint(500, 1500))
|
|
delay = random.uniform(min_delay, max_delay)
|
|
await asyncio.sleep(delay)
|
|
except Exception as e: # noqa: BLE001
|
|
logger.warning("warm_page_failed url=%s err=%s", warming_url, str(e)[:50])
|
|
|
|
# Collect cookies
|
|
storage_cookies = await context.cookies()
|
|
cookies = [
|
|
{
|
|
"name": c["name"],
|
|
"value": c["value"],
|
|
"domain": c["domain"],
|
|
"path": c["path"],
|
|
"expires": c.get("expires", -1),
|
|
"httpOnly": c.get("httpOnly", False),
|
|
"secure": c.get("secure", False),
|
|
}
|
|
for c in storage_cookies
|
|
]
|
|
|
|
await browser.close()
|
|
|
|
# Save warmed cookies
|
|
cookie_file = WARMER_DIR / f"{session_id}.json"
|
|
cookie_data = {
|
|
"domain": target_domain,
|
|
"session_id": session_id,
|
|
"warmed_at": datetime.now(UTC).isoformat(),
|
|
"expires_at": (datetime.now(UTC) + timedelta(days=30)).isoformat(),
|
|
"pages_visited": pages,
|
|
"cookies": cookies,
|
|
"user_agent": user_agent,
|
|
}
|
|
cookie_file.write_text(json.dumps(cookie_data, indent=2))
|
|
return {
|
|
"success": True,
|
|
"session_id": session_id,
|
|
"domain": target_domain,
|
|
"cookie_count": len(cookies),
|
|
"pages_visited": pages,
|
|
"expires_at": cookie_data["expires_at"],
|
|
}
|
|
except (json.JSONDecodeError, ValueError) as e:
|
|
return {"success": False, "error": str(e)[:300]}
|
|
|
|
def get_warmed_cookies(self, session_id: str) -> dict[str, Any] | None:
|
|
"""Get warmed cookies for a session."""
|
|
cookie_file = WARMER_DIR / f"{session_id}.json"
|
|
if not cookie_file.exists():
|
|
return None
|
|
try:
|
|
data = json.loads(cookie_file.read_text())
|
|
# Check expiry
|
|
expires = datetime.fromisoformat(data["expires_at"])
|
|
if expires < datetime.now(UTC):
|
|
return None
|
|
return data
|
|
except (json.JSONDecodeError, ValueError):
|
|
return None
|
|
|
|
def list_sessions(self) -> list[dict[str, Any]]:
|
|
sessions: list[dict[str, Any]] = []
|
|
for f in WARMER_DIR.glob("*.json"):
|
|
try:
|
|
data = json.loads(f.read_text())
|
|
sessions.append(
|
|
{
|
|
"session_id": data["session_id"],
|
|
"domain": data["domain"],
|
|
"warmed_at": data["warmed_at"],
|
|
"expires_at": data["expires_at"],
|
|
"cookie_count": len(data.get("cookies", [])),
|
|
}
|
|
)
|
|
except (json.JSONDecodeError, ValueError):
|
|
pass
|
|
return sessions
|