import requests

r = requests.get("https://httpbin.org/get", timeout=10)
print(r.status_code, r.elapsed.total_seconds())
print(r.json())          # если ответ JSON
print(r.headers['Content-Type'])

import requests

def make_session() -> requests.Session:
    s = requests.Session()
    s.headers.update({
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                      "AppleWebKit/537.36 (KHTML, like Gecko) "
                      "Chrome/120.0 Safari/537.36",
        "Accept-Language": "ru-RU,ru;q=0.9",
    })
    return s

session = make_session()
resp = session.get("https://example.com", timeout=(3.05, 27))  # (connect, read)

from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

retry = Retry(
    total=5,
    backoff_factor=1,                 # 1s, 2s, 4s, 8s...
    status_forcelist=[429, 500, 502, 503, 504],
    allowed_methods=["GET", "POST"],
    respect_retry_after_header=True,
)
session.mount('https://', HTTPAdapter(max_retries=retry))
session.mount('http://', HTTPAdapter(max_retries=retry))

with session.get("https://example.com/big.csv", stream=True) as r:
    r.raise_for_status()
    with open("big.csv", "wb") as f:
        for chunk in r.iter_content(chunk_size=8192):
            f.write(chunk)

# Basic Auth
session.get(url, auth=("user", "password"))

# Bearer-токен
session.headers["Authorization"] = "Bearer <token>"

# Формы логина (CSRF-токен часто прячут в HTML)
login_page = session.get("https://example.com/login")
# ... извлекаем csrf из login_page ...
session.post("https://example.com/login",
             data={"username": "u", "password": "p", "csrf": csrf})

from bs4 import BeautifulSoup

soup = BeautifulSoup(html, "lxml")

# По тегу/классу/id
soup.find("h1")
soup.find_all("div", class_="product", limit=10)
soup.find(id="main")

# По нескольким классам и атрибутам
soup.find_all("a", class_=["btn", "primary"])
soup.find_all(attrs={"data-id": True})            # есть атрибут data-id
soup.find_all("input", attrs={"type": "hidden"})

# CSS-селекторы (мощно)
soup.select("div.card > h2.title")
soup.select("article[data-id] span.price")
soup.select_one("nav a:nth-of-type(2)")

def safe_text(node, selector, default=None):
    """Возвращает текст по CSS-селектору или default, не падая на None."""
    el = node.select_one(selector)
    return el.get_text(strip=True) if el else default

def safe_attr(node, selector, attr, default=None):
    el = node.select_one(selector)
    return el.get(attr, default) if el else default

# Использование
for card in soup.select('article.product'):
    product = {
        "title": safe_text(card, "h2.title"),
        "price": safe_text(card, "span.price", default="0"),
        "url": safe_attr(card, "a", "href"),
    }
    print(product)

price = soup.find("span", class_="price")
price.parent                 # родитель
price.find_parent('article') # ближайший предок-article
price.next_sibling           # следующий узел
price.find_next('a')         # следующая ссылка в документе
[c for c in price.parents]   # все предки до корня

import pandas as pd

# pandas сам распарсит все <table> на странице в список DataFrame
tables = pd.read_html(html)
df = tables[0]

from lxml import html

tree = html.fromstring(page_html)

# Извлекаем сразу списками
titles = tree.xpath('//article[@class="product"]/h2/text()')
prices = tree.xpath('//span[@class="price"]/text()')
links  = tree.xpath('//a[contains(@class,"more")]/@href')

# Относительные пути от найденного узла
for card in tree.xpath('//article[@class="product"]'):
    title = card.xpath('.//h2/text()')          # точка = относительно card
    price = card.xpath('.//span[@class="price"]/text()')
    print(title, price)

# XPath по тексту и осям
tree.xpath('//th[text()="Цена"]/following-sibling::td/text()')

from selectolax.parser import HTMLParser

tree = HTMLParser(page_html)
for node in tree.css('article.product'):
    title = node.css_first('h2.title')
    print(title.text(strip=True) if title else None)

import re

text = "Тел: +7 (999) 123-45-67, email: a.user@mail.ru, цена 59 990 ₽, скидка 15%"

# Именованные группы — читаемее
phone = re.search(r"\+7\s?\(?(?P<code>\d{3})\)?[\s-]?(?P<num>\d{3}[\s-]?\d{2}[\s-]?\d{2})", text)
if phone:
    print(phone.group("code"), phone.group("num"))

# Цена: убираем пробелы-разделители и приводим к int
raw = re.search(r"\d[\d\s]*\d", text).group()
price = int(re.sub(r"\s", "", raw))   # 59990

# Все email
emails = re.findall(r"[\w.+-]+@[\w-]+\.[\w.-]+", text)

# Компиляция + флаги для многократного использования
TAG_RE = re.compile(r"<[^>]+>")
clean = TAG_RE.sub("", "<b>Привет</b>")     # удалить теги: Привет

re.findall(r"(?<=id=)\d+", "id=42 id=99")   # lookbehind -> [42, 99]
re.split(r"[,;]\s*", "a, b; c")             # ["a","b","c"]

import requests

def fetch_all(base_url: str, session: requests.Session) -> list[dict]:
    items, page = [], 1
    while True:
        r = session.get(base_url, params={'page': page, 'per_page': 100}, timeout=15)
        r.raise_for_status()
        data = r.json()
        if not data['results']:
            break
        items.extend(data['results'])
        # курсорная пагинация:
        if not data.get('next'):
            break
        page += 1
    return items

def deep_get(d: dict, path: str, default=None):
    """deep_get(obj, "user.address.city")"""
    cur = d
    for key in path.split("."):
        if isinstance(cur, dict) and key in cur:
            cur = cur[key]
        else:
            return default
    return cur

city = deep_get(payload, "user.address.city", default="—")

query = {
    "query": "{ products(first: 50) { edges { node { title price } } } }"
}
r = session.post("https://shop.example/graphql", json=query, timeout=15)
nodes = [e['node'] for e in r.json()['data']['products']['edges']]

from playwright.sync_api import sync_playwright

def scrape_spa(url: str) -> list[str]:
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        ctx = browser.new_context(
            user_agent="Mozilla/5.0 ... Chrome/120.0 Safari/537.36",
            viewport={'width': 1920, 'height': 1080},
            locale='ru-RU',
        )
        page = ctx.new_page()
        page.goto(url, wait_until='networkidle', timeout=30000)

        # Ждём конкретный селектор, а не фиксированный sleep
        page.wait_for_selector('article.product')

        # Бесконечный скролл
        prev = 0
        while True:
            page.mouse.wheel(0, 5000)
            page.wait_for_timeout(1000)
            count = page.locator('article.product').count()
            if count == prev:
                break
            prev = count

        titles = page.locator('h2.title').all_inner_texts()
        browser.close()
        return titles

# Вместо парсинга DOM — ловим JSON, который грузит сама страница
captured = []

def handle_response(response):
    if '/api/products' in response.url:
        captured.append(response.json())

page.on('response', handle_response)
page.goto(url, wait_until='networkidle')
# captured теперь содержит чистый JSON без всякого HTML

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

opts = webdriver.ChromeOptions()
opts.add_argument('--headless=new')
opts.add_argument('--disable-blink-features=AutomationControlled')
driver = webdriver.Chrome(options=opts)
try:
    driver.get(url)
    el = WebDriverWait(driver, 15).until(
        EC.presence_of_element_located((By.CSS_SELECTOR, 'article.product')))
    for p in driver.find_elements(By.CSS_SELECTOR, 'article.product'):
        print(p.find_element(By.CSS_SELECTOR, '.title').text)
finally:
    driver.quit()

import asyncio
import aiohttp
from bs4 import BeautifulSoup

class AsyncScraper:
    def __init__(self, concurrency: int = 10, delay: float = 0.2):
        self.sem = asyncio.Semaphore(concurrency)
        self.delay = delay

    async def fetch(self, session: aiohttp.ClientSession, url: str) -> str | None:
        async with self.sem:                       # ограничение параллелизма
            try:
                async with session.get(url, timeout=aiohttp.ClientTimeout(total=20)) as r:
                    if r.status == 429:
                        await asyncio.sleep(int(r.headers.get('Retry-After', 5)))
                        return await self.fetch(session, url)
                    r.raise_for_status()
                    await asyncio.sleep(self.delay)
                    return await r.text()
            except (aiohttp.ClientError, asyncio.TimeoutError) as e:
                print(f'[!] {url}: {e}')
                return None

    def parse(self, html: str) -> dict:
        soup = BeautifulSoup(html, 'lxml')
        h1 = soup.find('h1')
        return {'title': h1.get_text(strip=True) if h1 else None}

    async def run(self, urls: list[str]) -> list[dict]:
        connector = aiohttp.TCPConnector(limit=20, ttl_dns_cache=300)
        async with aiohttp.ClientSession(connector=connector) as session:
            htmls = await asyncio.gather(*(self.fetch(session, u) for u in urls))
        return [self.parse(h) for h in htmls if h]

urls = [f'https://example.com/page/{i}' for i in range(1, 1001)]
results = asyncio.run(AsyncScraper(concurrency=15).run(urls))
print(f'Собрано: {len(results)}')

from tqdm.asyncio import tqdm_asyncio

htmls = await tqdm_asyncio.gather(*(self.fetch(s, u) for u in urls))

import scrapy

class BooksSpider(scrapy.Spider):
    name = 'books'
    start_urls = ['https://books.toscrape.com/']
    custom_settings = {
        'DOWNLOAD_DELAY': 0.5,
        'AUTOTHROTTLE_ENABLED': True,
        'CONCURRENT_REQUESTS': 8,
    }

    def parse(self, response):
        for book in response.css('article.product_pod'):
            detail_url = book.css('h3 a::attr(href)').get()
            yield response.follow(detail_url, callback=self.parse_book)
        next_page = response.css('li.next a::attr(href)').get()
        if next_page:
            yield response.follow(next_page, callback=self.parse)

    def parse_book(self, response):
        yield {
            'title': response.css('h1::text').get(),
            'price': response.css('p.price_color::text').get(),
            'stock': response.css('p.availability::text').re_first(r'' + bs+'d+'),
            'desc':  response.css('#product_description + p::text').get(),
        }

# pipelines.py
from itemadapter import ItemAdapter

class PricePipeline:
    def process_item(self, item, spider):
        adapter = ItemAdapter(item)
        price = adapter.get('price', '')
        adapter['price'] = float(price.replace('£', '').strip() or 0)
        if adapter['price'] <= 0:
            from scrapy.exceptions import DropItem
            raise DropItem('Нет цены')
        return item

# settings.py
ITEM_PIPELINES = {'shop.pipelines.PricePipeline': 300}

from fake_useragent import UserAgent
ua = UserAgent()

def realistic_headers() -> dict:
    return {
        'User-Agent': ua.random,
        'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
        'Accept-Language': 'ru-RU,ru;q=0.9,en;q=0.8',
        'Accept-Encoding': 'gzip, deflate, br',
        'Connection': 'keep-alive',
        'Upgrade-Insecure-Requests': '1',
    }

import random, itertools

class ProxyPool:
    def __init__(self, proxies: list[str]):
        self.proxies = proxies
        self.bad = set()

    def get(self) -> str | None:
        alive = [p for p in self.proxies if p not in self.bad]
        return random.choice(alive) if alive else None

    def mark_bad(self, proxy: str):
        self.bad.add(proxy)

pool = ProxyPool(['http://1.2.3.4:8080', 'http://5.6.7.8:3128'])
proxy = pool.get()
try:
    r = requests.get(url, proxies={'http': proxy, 'https': proxy}, timeout=10)
except requests.RequestException:
    pool.mark_bad(proxy)

import time, random

class RateLimiter:
    def __init__(self, rps: float = 1.0):
        self.min_interval = 1.0 / rps
        self.last = 0.0

    def wait(self):
        elapsed = time.monotonic() - self.last
        sleep_for = self.min_interval - elapsed + random.uniform(0, 0.3)
        if sleep_for > 0:
            time.sleep(sleep_for)
        self.last = time.monotonic()

# Playwright: убираем navigator.webdriver
context.add_init_script(
    "Object.defineProperty(navigator,'webdriver',{get:()=>undefined})")
# Либо используйте playwright-stealth / undetected-chromedriver

from pydantic import BaseModel, field_validator, HttpUrl

class Product(BaseModel):
    title: str
    price: float
    url: HttpUrl
    in_stock: bool = True

    @field_validator('price')
    @classmethod
    def price_positive(cls, v):
        if v < 0:
            raise ValueError('Цена не может быть отрицательной')
        return round(v, 2)

# Невалидные записи отсеются автоматически
raw = {'title': 'Книга', 'price': '19.99', 'url': 'https://shop/x'}
product = Product(**raw)        # price приведётся к float
print(product.model_dump())

import csv, json

# JSON Lines — удобно дозаписывать построчно, не держа всё в памяти
with open('data.jsonl', 'a', encoding='utf-8') as f:
    for item in items:
        f.write(json.dumps(item, ensure_ascii=False) + '\n')

# CSV
with open('data.csv', 'w', newline='', encoding='utf-8-sig') as f:
    w = csv.DictWriter(f, fieldnames=['title', 'price'])
    w.writeheader()
    w.writerows(items)

import sqlite3

conn = sqlite3.connect('shop.db')
conn.execute('''
    CREATE TABLE IF NOT EXISTS products (
        url   TEXT PRIMARY KEY,
        title TEXT,
        price REAL,
        updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )''')

conn.executemany('''
    INSERT INTO products (url, title, price) VALUES (:url, :title, :price)
    ON CONFLICT(url) DO UPDATE SET
        title=excluded.title, price=excluded.price,
        updated_at=CURRENT_TIMESTAMP
''', items)
conn.commit()

import pandas as pd

df = pd.DataFrame(items)
df = df.drop_duplicates(subset='url').dropna(subset=['price'])
df['price'] = pd.to_numeric(df['price'], errors='coerce')
df.to_parquet('data.parquet')      # компактно и быстро для больших данных

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import requests

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=2, max=30),
    retry=retry_if_exception_type(requests.RequestException),
    reraise=True,
)
def fetch(url: str, session: requests.Session) -> str:
    r = session.get(url, timeout=15)
    r.raise_for_status()
    return r.text

from loguru import logger

logger.add('scraper.log', rotation='10 MB', retention='7 days', level='INFO')

logger.info('Старт: {} URL', len(urls))
try:
    html = fetch(url, session)
except Exception as e:
    logger.error('Ошибка на {}: {}', url, e)

import json, os

def load_done(path='done.json') -> set:
    return set(json.load(open(path))) if os.path.exists(path) else set()

def save_done(done: set, path='done.json'):
    json.dump(list(done), open(path, 'w'))

done = load_done()
for url in urls:
    if url in done:
        continue            # пропускаем уже обработанное
    process(url)
    done.add(url)
    if len(done) % 50 == 0:
        save_done(done)     # периодически сохраняем прогресс

# tests/test_parser.py
import pytest
from myparser import parse_product

@pytest.fixture
def sample_html():
    return open('tests/fixtures/product.html', encoding='utf-8').read()

def test_parse_extracts_title(sample_html):
    result = parse_product(sample_html)
    assert result['title'] == 'Ноутбук Pro'
    assert result['price'] == 59990.0

def test_parse_handles_missing_price():
    html = '<article><h2>Без цены</h2></article>'
    result = parse_product(html)
    assert result['price'] is None      # не падает, возвращает None

import responses, requests

@responses.activate
def test_fetch_retries_on_500():
    responses.add(responses.GET, 'https://x.com', status=500)
    responses.add(responses.GET, 'https://x.com', body='OK', status=200)
    # ваш fetch с ретраями должен вернуть 'OK'

import urllib.robotparser as urp

rp = urp.RobotFileParser()
rp.set_url('https://example.com/robots.txt')
rp.read()
if rp.can_fetch('*', 'https://example.com/catalog'):
    ...   # парсим
crawl_delay = rp.crawl_delay('*')   # уважайте указанную задержку

# pipeline.py — оркестратор
from loguru import logger
from .client import HttpClient
from .parsers import parse_product
from .models import Product
from .storage import Storage

class Pipeline:
    def __init__(self, client: HttpClient, storage: Storage):
        self.client = client
        self.storage = storage

    def process(self, url: str) -> None:
        html = self.client.get(url)          # ретраи + лимиты внутри
        if not html:
            return
        raw = parse_product(html)            # чистая функция
        try:
            product = Product(**raw)         # валидация
        except ValueError as e:
            logger.warning('Невалидно {}: {}', url, e)
            return
        self.storage.upsert(product.model_dump())

    def run(self, urls: list[str]) -> None:
        for url in urls:
            try:
                self.process(url)
            except Exception as e:
                logger.exception('Сбой на {}: {}', url, e)

import json, os, requests
from bs4 import BeautifulSoup

STATE = 'prices.json'

def get_price(url: str) -> float:
    html = requests.get(url, timeout=10,
                        headers={'User-Agent': 'Mozilla/5.0'}).text
    soup = BeautifulSoup(html, 'lxml')
    raw = soup.select_one('.price').get_text(strip=True)
    return float(''.join(c for c in raw if c.isdigit() or c == '.'))

def load() -> dict:
    return json.load(open(STATE)) if os.path.exists(STATE) else {}

def check(url: str):
    old = load()
    new_price = get_price(url)
    if url in old and new_price < old[url]:
        print(f'📉 Цена упала: {old[url]} -> {new_price}')
    old[url] = new_price
    json.dump(old, open(STATE, 'w'))

check('https://books.toscrape.com/catalogue/a-light-in-the-attic_1000/index.html')

from curl_cffi import requests

# Имитируем TLS-фингерпринт Chrome
resp = requests.get(
    "https://tls.peet.ws/api/all",
    impersonate="chrome120",
    timeout=20,
)
print(resp.json()["tls"]["ja3"])

import redis, json

r = redis.Redis(host="localhost", port=6379, db=0)

# Продюсер: кладём URL-ы в очередь
def enqueue(urls):
    for url in urls:
        r.lpush("scrape:queue", json.dumps({"url": url}))

# Воркер: забираем задачи и обрабатываем
def worker():
    while True:
        _, raw = r.brpop("scrape:queue")
        task = json.loads(raw)
        process(task["url"])  # ваша логика парсинга

import requests_cache

session = requests_cache.CachedSession(
    "http_cache",
    expire_after=3600,        # кэш живёт 1 час
    allowable_methods=["GET"],
)
resp = session.get("https://example.com")
print(resp.from_cache)  # False при первом запросе, True при повторном

from dataclasses import dataclass, field
import time

@dataclass
class Metrics:
    requests: int = 0
    errors: int = 0
    started: float = field(default_factory=time.time)

    def hit(self): self.requests += 1
    def fail(self): self.errors += 1

    def report(self):
        elapsed = time.time() - self.started
        rps = self.requests / elapsed if elapsed else 0
        return f"{self.requests} запросов, {self.errors} ошибок, {rps:.1f} rps"

m = Metrics()

import hashlib

def content_hash(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()

seen = {}  # url -> hash

def changed(url: str, html: str) -> bool:
    h = content_hash(html)
    if seen.get(url) == h:
        return False
    seen[url] = h
    return True

import requests
from lxml import etree

def urls_from_sitemap(sitemap_url: str) -> list[str]:
    xml = requests.get(sitemap_url, timeout=15).content
    root = etree.fromstring(xml)
    ns = {"s": "http://www.sitemaps.org/schemas/sitemap/0.9"}
    # sitemap может ссылаться на вложенные карты
    nested = root.xpath("//s:sitemap/s:loc/text()", namespaces=ns)
    if nested:
        out = []
        for sm in nested:
            out += urls_from_sitemap(sm)
        return out
    return root.xpath("//s:url/s:loc/text()", namespaces=ns)

import json
from bs4 import BeautifulSoup

def extract_jsonld(html: str) -> list[dict]:
    soup = BeautifulSoup(html, "lxml")
    blocks = []
    for tag in soup.find_all("script", type="application/ld+json"):
        try:
            data = json.loads(tag.string or "{}")
            blocks.extend(data if isinstance(data, list) else [data])
        except json.JSONDecodeError:
            continue
    return blocks

# Часто здесь лежит готовый Product с ценой, рейтингом и наличием
products = [b for b in extract_jsonld(html) if b.get("@type") == "Product"]

# ❌ Хрупко — классы меняются при пересборке фронтенда
soup.select_one("div.css-1a2b3c > span.sc-9f8e7d")

# ✅ Надёжно — атрибуты данных и семантика живут дольше
soup.select_one("[data-testid='product-price']")
soup.select_one("[itemprop='price']")
soup.find("meta", attrs={"property": "og:title"})["content"]

from playwright.sync_api import sync_playwright

# Первый запуск: логинимся вручную/программно и сохраняем состояние
with sync_playwright() as p:
    browser = p.chromium.launch(headless=False)
    ctx = browser.new_context()
    page = ctx.new_page()
    page.goto("https://example.com/login")
    # ... выполняем вход ...
    ctx.storage_state(path="state.json")   # сохранили cookies + localStorage
    browser.close()

# Последующие запуски: стартуем уже залогиненными
with sync_playwright() as p:
    browser = p.chromium.launch(headless=True)
    ctx = browser.new_context(storage_state="state.json")
    # сразу открываем закрытые страницы

import requests

# cookies получены из ctx.cookies() в Playwright
browser_cookies = ctx.cookies()

session = requests.Session()
for c in browser_cookies:
    session.cookies.set(c["name"], c["value"], domain=c["domain"])

# Теперь тяжёлые страницы тянем через requests — в десятки раз быстрее
resp = session.get("https://example.com/api/items", timeout=15)

def is_honeypot(node) -> bool:
    style = (node.get("style") or "").replace(" ", "").lower()
    cls = " ".join(node.get("class", [])).lower()
    return (
        "display:none" in style
        or "visibility:hidden" in style
        or node.get("hidden") is not None
        or "hidden" in cls or "trap" in cls
    )

# Берём только видимые ссылки
links = [a for a in soup.select("a[href]") if not is_honeypot(a)]

from concurrent.futures import ProcessPoolExecutor
from functools import partial

def parse_html(html: str) -> dict:
    ...  # тяжёлый разбор: lxml/regex/нормализация

def parse_many(htmls: list[str]) -> list[dict]:
    with ProcessPoolExecutor(max_workers=4) as ex:
        return list(ex.map(parse_html, htmls, chunksize=20))