Learn the techniques

import requests

# The page is JS-rendered, but the data is here:
r = requests.get(
    "https://target.example/api/megacorp/listings",
    params={"region": "apac", "page": 1},
    headers={"Authorization": "Bearer <token>", "Cookie": "og_session=..."},
)
for job in r.json()["items"]:
    print(job["uuid"], job["title"])

from curl_cffi import requests

# Impersonate the latest Chrome (TLS/JA3 + HTTP2 + headers):
r = requests.get("https://target.example/api/specimen/dossier/<uuid>",
                 impersonate="chrome")           # or "chrome124", "safari17_0", ...
print(r.json())

# A Session keeps cookies AND the impersonated fingerprint across requests:
s = requests.Session(impersonate="chrome")
s.get("https://target.example/")                 # collect cookies
data = s.get("https://target.example/api/asic/lots").json()

from playwright.sync_api import sync_playwright

with sync_playwright() as p:
    browser = p.chromium.launch(headless=True)
    page = browser.new_page()
    page.goto("https://target.example/salvage")
    page.wait_for_selector("table tbody tr")
    rows = page.locator("table tbody tr")
    for i in range(rows.count()):
        print(rows.nth(i).inner_text())
    browser.close()

from patchright.sync_api import sync_playwright   # only the import changes

with sync_playwright() as p:
    ctx = p.chromium.launch_persistent_context(
        user_data_dir="/tmp/pw-profile",
        channel="chrome",        # real Chrome, not bundled Chromium
        headless=False,          # headed is far less detectable
        no_viewport=True,
    )
    page = ctx.new_page()
    page.goto("https://target.example")

# Run a headed browser on a headless box:
xvfb-run -a --server-args="-screen 0 1920x1080x24" python scrape.py
# inside scrape.py:  p.chromium.launch(headless=False)

ctx = browser.new_context(
    user_agent=("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
                "(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"),
    locale="en-US",
    timezone_id="America/Los_Angeles",       # match your proxy's country
    viewport={"width": 1920, "height": 1080},
    geolocation={"latitude": 36.17, "longitude": -115.14},  # Las Vegas
    permissions=["geolocation"],
    extra_http_headers={"Accept-Language": "en-US,en;q=0.9"},
)
# Some sites read navigator.webdriver — hide it (Patchright does this for you):
ctx.add_init_script("Object.defineProperty(navigator,'webdriver',{get:()=>undefined})")

import random, time

def human_move(page, x0, y0, x1, y1, steps=40):
    for i in range(steps):
        t = i / steps
        x = x0 + (x1 - x0) * t + random.uniform(-6, 6)   # curve + jitter
        y = y0 + (y1 - y0) * t + random.uniform(-6, 6)
        page.mouse.move(x, y)
        time.sleep(random.uniform(0.012, 0.05))          # variable cadence

for _ in range(6):
    page.mouse.wheel(0, random.randint(200, 500))   # scroll the page
    time.sleep(random.uniform(0.4, 1.2))            # idle-read
time.sleep(4)                                        # dwell before the cage

import itertools, requests

POOL = [
    "http://user:pass@p1.residential.example:8000",
    "http://user:pass@p2.residential.example:8000",
    # free lists (free-proxy-list.net, ...) work too — just flakier
]
rotor = itertools.cycle(POOL)

def fetch(url):
    proxy = next(rotor)
    return requests.get(url, proxies={"http": proxy, "https": proxy}, timeout=20)

ctx = browser.new_context(proxy={
    "server": "http://p1.example:8000", "username": "user", "password": "pass",
})

s = requests.Session()
s.get("https://target.example/megacorp")                 # sets og_session
for hotel in ("mirage-royale", "oasis-grand", "zenith-tower"):
    s.post(f"https://target.example/api/casino/partner/{hotel}")   # comp cookies
# now the ledger trusts the session:
ledger = s.get("https://target.example/api/casino/ledger?page=1").json()

import requests, re
BASE = "https://target.example"

# 1) solve the signup CAPTCHA image (see §05), then register:
requests.post(f"{BASE}/api/megacorp/account/register", json={
    "email": "ghost@temp-mail.org", "password": "p@ssw0rd",
    "captchaId": cid, "captchaText": solved})

# 2) read the code from your burner inbox (here, the site's own inbox API):
msgs = requests.get(f"{BASE}/api/mail/inbox",
                    params={"address": "ghost@temp-mail.org"}).json()["messages"]
code = re.search(r"\b(\d{6})\b", msgs[0]["text"]).group(1)

# 3) verify + sign in:
requests.post(f"{BASE}/api/megacorp/account/verify",
              json={"email": "ghost@temp-mail.org", "code": code})

from ddgs import DDGS

for hit in DDGS().text("deep-sea salvage shipwreck registry", max_results=5):
    print(hit["title"], hit["href"], hit["body"])

search:
  formats:
    - html
    - json   # required for the JSON API below

import requests
r = requests.get("http://localhost:8080/search",
                 params={"q": "RMS Republic wreck coordinates", "format": "json"})
for hit in r.json()["results"][:5]:
    print(hit["title"], hit["url"])

import subprocess
from faster_whisper import WhisperModel

subprocess.run(["ffmpeg", "-y", "-i", "intercept.wav",
                "-ar", "16000", "-ac", "1", "clean.wav"], check=True)

# device="cuda", compute_type="float16"  on GPU; "cpu"/"int8" works too.
model = WhisperModel("large-v3", device="cpu", compute_type="int8")
segments, info = model.transcribe("clean.wav", beam_size=5)
transcript = " ".join(s.text for s in segments).strip()
print(transcript)        # -> "victor echo zulu kilo"

import pytesseract, re
from PIL import Image

txt = pytesseract.image_to_string(Image.open("challenge.png"))
nums = list(map(int, re.findall(r"\d+", txt)))   # "9 + 9 = ?" -> [9, 9]
answer = nums[0] + nums[1]                          # -> 18

import torch, clip
from PIL import Image

device = "cuda" if torch.cuda.is_available() else "cpu"
model, preprocess = clip.load("ViT-B/32", device=device)

img = preprocess(Image.open("tile.png")).unsqueeze(0).to(device)
labels = ["a bus", "a traffic light", "a crosswalk", "a bicycle"]
text = clip.tokenize([f"a photo of {l}" for l in labels]).to(device)

with torch.no_grad():
    logits_per_image, _ = model(img, text)
    probs = logits_per_image.softmax(dim=-1).cpu().numpy()[0]   # softmax: sums to 1
print(labels[int(probs.argmax())])

import torch
from transformers import AutoProcessor, AutoModel
from PIL import Image

model = AutoModel.from_pretrained("google/siglip2-base-patch16-224")
processor = AutoProcessor.from_pretrained("google/siglip2-base-patch16-224")

image = Image.open("tile.png")
labels = ["a bus", "a traffic light", "a crosswalk", "a bicycle"]
# SigLIP was trained with this template + padding="max_length":
inputs = processor(text=[f"a photo of {l}" for l in labels],
                   images=image, padding="max_length", return_tensors="pt")

with torch.no_grad():
    outputs = model(**inputs)

# sigmoid (NOT softmax) -> independent per-label probabilities:
probs = torch.sigmoid(outputs.logits_per_image)[0]
present = [l for l, p in zip(labels, probs) if p > 0.5]   # select ALL matches
print(present, "| best:", labels[int(probs.argmax())])

import pytesseract
from PIL import Image

text = pytesseract.image_to_string(Image.open("salary.png"))
print(text)   # "$109,000 - $178,000  /  h.recruiter@omnicorp.example"

from rapidocr_onnxruntime import RapidOCR

engine = RapidOCR()
result, elapse = engine("sonar.png")
for box, text, score in (result or []):
    print(text, round(score, 2))   # "LAT -22.706", "LON 73.705", ...

import fitz  # PyMuPDF
import pytesseract
from PIL import Image
import io

doc = fitz.open("scan.pdf")
page = doc[0]
text = page.get_text()
if not text.strip():                       # image-only (scanned) PDF
    pix = page.get_pixmap(dpi=200)
    img = Image.open(io.BytesIO(pix.tobytes("png")))
    text = pytesseract.image_to_string(img)
print(text)

import cairosvg, pytesseract, io, requests
from PIL import Image

svg = requests.get(f"{BASE}/api/syndicate/dossier/{uuid}").content
png = cairosvg.svg2png(bytestring=svg, scale=2)   # rasterize (or use resvg / a headless screenshot)
text = pytesseract.image_to_string(Image.open(io.BytesIO(png)))
print(text)                                        # "REAL NAME ...  STANDING BOUNTY $260,000"

import asyncio, random, httpx

async def fetch(client, url, sem):
    async with sem:
        for attempt in range(5):
            r = await client.get(url)
            if r.status_code in (429, 503):
                await asyncio.sleep(2 ** attempt + random.random())  # backoff + jitter
                continue
            return r.json()

async def main(urls):
    sem = asyncio.Semaphore(8)              # tune below the breaker threshold
    async with httpx.AsyncClient(timeout=30) as client:
        return await asyncio.gather(*(fetch(client, u, sem) for u in urls))

import redis, json
r = redis.Redis(host="coordinator.internal", port=6379)

# producer (once):
for uuid in all_uuids:
    r.lpush("queue:lots", uuid)

# worker (run on each VM, with a DISTINCT proxy/identity):
while (item := r.rpop("queue:lots")):
    data = scrape_lot(item.decode(), proxy=MY_PROXY)   # your scraper + your IP
    r.lpush("results", json.dumps(data))

import asyncio, asyncssh

HOSTS = ["10.0.0.11", "10.0.0.12", "10.0.0.13"]   # worker VMs
PROXIES = {h: f"http://proxy-{i}.pool:8000" for i, h in enumerate(HOSTS)}

async def deploy(host):
    async with asyncssh.connect(host, username="ops") as c:
        await asyncssh.scp("worker.py", (c, "/opt/worker.py"))     # ship code
        # each VM gets a DISTINCT egress identity -> beats the rotation gate
        await c.run(f"COORD=coordinator.internal PROXY={PROXIES[host]} "
                    f"nohup python /opt/worker.py >/var/log/w.log 2>&1 &", check=True)
        return f"{host}: launched"

print(await asyncio.gather(*(deploy(h) for h in HOSTS)))

import asyncssh, asyncio

async def health(host):
    async with asyncssh.connect(host, username="ops") as c:
        r = await c.run("tail -n1 /var/log/w.log", check=False)
        return host, r.stdout.strip()

async def watch():
    while True:
        rows = await asyncio.gather(*(health(h) for h in HOSTS))
        for host, line in rows:
            print(f"[{host}] {line}")          # rate / errors / last-ok age
        await asyncio.sleep(10)                # 10s poll, jittered in practice

import base64, json

# Specimen: { "encoding": "base64", "data": "<blob>" }
body = session.get(f"/api/specimen/dossier/{uuid}", headers=fp).json()
dossier = json.loads(base64.b64decode(body["data"]))

# ASIC: { "alg": "xor-b64", "enc": "<blob>" }  — key is the lot uuid
def xor_b64(enc: str, key: str) -> dict:
    raw = base64.b64decode(enc)
    out = bytes(b ^ ord(key[i % len(key)]) for i, b in enumerate(raw))
    return json.loads(out)

detail = xor_b64(resp["enc"], lot_uuid)   # buyer id, clearing price, documents

import requests, json

records = []
with requests.get(url, headers=hdrs, stream=True, timeout=120) as r:
    for line in r.iter_lines():        # arrives one wreck at a time
        if not line:
            continue
        obj = json.loads(line)
        if "stream" in obj:            # first line is meta {page,totalPages,total}
            continue
        records.append(obj)            # drip — keep reading to the end

import requests, struct
from argon2.low_level import hash_secret_raw, Type   # pip install argon2-cffi

s = requests.Session()
ch = s.get("/api/casino/pow").json()              # {parameters, signature}
p = ch["parameters"]
salt, nonce, prefix = bytes.fromhex(p["salt"]), bytes.fromhex(p["nonce"]), p["keyPrefix"]

def derive(counter):                               # mirror the server's KDF exactly
    return hash_secret_raw(
        secret=nonce + struct.pack(">I", counter), # message = nonce ‖ uint32be(counter)
        salt=salt,
        time_cost=p["cost"], memory_cost=p["memoryCost"],
        parallelism=p["parallelism"], hash_len=p["keyLength"], type=Type.ID,
    )

def solve():
    for counter in range(0, 5_000):                # ~tens of attempts; each is memory-hard
        key = derive(counter)
        if key.hex().startswith(prefix):
            return counter, key.hex()

counter, derived = solve()
s.post("/api/casino/pow",
       json={"challenge": ch, "solution": {"counter": counter, "derivedKey": derived}})
# pow_ok cookie is now set on the session -> the ledger opens
bets = s.get("/api/casino/ledger?page=1").json()

from bs4 import BeautifulSoup
import re

soup = BeautifulSoup(html, "html.parser")
# 1. skip decoy rows (hidden via opaque utility classes)
for junk in soup.select(".idx-decoy, .tok-x"):
    junk.decompose()
rows = []
for tr in soup.select("tbody tr"):
    cells = [td.get_text(strip=True) for td in tr.select("td")]
    if not cells:
        continue
    rows.append(cells)
# real designations now read clean: SP-3001, not SP-30<junk>01

# inner_text() excludes display:none honeypots; visible rows only
for tr in page.locator("tbody tr:visible").all():
    print(tr.inner_text())     # decoys + junk fragments never appear

import requests, re
from bs4 import BeautifulSoup

s = requests.Session()           # (clear the search CAPTCHA first; route via a US proxy)
html = s.post(URL, data={"txtBusinessName": "", "SearchType": "StartsWith",
                         "ddlNameType": "Business Name"}).text

def hidden(html, name):
    return re.search(rf'name="{name}"\s+value="([^"]*)"', html).group(1)

rows = []
while True:
    soup = BeautifulSoup(html, "html.parser")
    rows += [[td.get_text(strip=True) for td in tr.select("td")] for tr in soup.select("tbody tr")]
    m = re.search(r"page <b>(\d+)</b> of <b>(\d+)</b>", html)
    cur, last = (int(m[1]), int(m[2])) if m else (1, 1)
    if cur >= last:
        break
    html = s.post(URL, data={                       # re-POST the scraped state
        "__VIEWSTATE": hidden(html, "__VIEWSTATE"),
        "__EVENTVALIDATION": hidden(html, "__EVENTVALIDATION"),
        "__EVENTARGUMENT": "Page$Next",
    }).text

def parse_coord(ocr_text: str):
    m = re.search(r"-?\d+\.\d+", ocr_text)
    return float(m.group()) if m else ""   # "———" -> "" (matches empty truth)

wreck["latitude"]  = parse_coord(lat_text)
wreck["longitude"] = parse_coord(lon_text)
# A confident wrong number scores 0; an honest blank scores 1.