Умный парсинг: LLM извлекает данные из любого сайта

import httpx
from bs4 import BeautifulSoup

r = httpx.get("https://example.com/catalog",
              headers={"User-Agent": "Mozilla/5.0"})
soup = BeautifulSoup(r.text, "lxml")

products = []
for card in soup.select(".product-card"):
    products.append({
        "name": card.select_one(".product-title").text.strip(),
        "price": card.select_one(".price").text.strip(),
        "sku": card.get("data-sku"),
    })

import httpx
from bs4 import BeautifulSoup
import anthropic
import json

client = anthropic.Anthropic()

def parse_with_llm(html: str, schema: dict, task_description: str) -> dict:
    # Очищаем HTML от мусора
    soup = BeautifulSoup(html, "lxml")
    for tag in soup(["script", "style", "svg", "img"]):
        tag.decompose()
    clean_text = soup.get_text(separator="\n", strip=True)

    # Ограничиваем размер (токены стоят денег)
    clean_text = clean_text[:8000]

    response = client.messages.create(
        model="claude-opus-4-5",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": f"""{task_description}

Извлеки данные из текста и верни JSON строго по схеме:
{json.dumps(schema, ensure_ascii=False, indent=2)}

Если поле не найдено - верни null.
Верни только JSON без пояснений.

Текст:
{clean_text}"""
        }]
    )

    return json.loads(response.content[0].text)

schema = {
    "name": "название товара",
    "price": "цена числом без валюты",
    "currency": "валюта (RUB/USD/EUR)",
    "availability": "в наличии/под заказ/нет в наличии",
    "sku": "артикул если есть",
    "characteristics": {"ключ": "значение"}
}

r = httpx.get("https://shop.example.com/product/123",
              headers={"User-Agent": "Mozilla/5.0"}, timeout=15)

product = parse_with_llm(
    r.text,
    schema,
    "Это страница товара интернет-магазина."
)
print(product)
# {"name": "Дрель Bosch GSB 13 RE", "price": 4590, "currency": "RUB", ...}

schema = {
    "company_name": "название компании",
    "phones": ["список телефонов"],
    "emails": ["список email"],
    "address": "адрес",
    "working_hours": "режим работы",
    "website": "сайт если есть"
}

# Работает даже если контакты написаны в произвольном тексте
contacts = parse_with_llm(html, schema,
    "Это страница компании в бизнес-каталоге.")

import time
import httpx

def monitor_price(url: str) -> float | None:
    try:
        r = httpx.get(url, headers={"User-Agent": "Mozilla/5.0"}, timeout=15)
        result = parse_with_llm(
            r.text,
            {"price": "текущая цена числом", "in_stock": True},
            "Страница товара. Найди актуальную цену."
        )
        return result.get("price")
    except Exception as e:
        print(f"Ошибка для {url}: {e}")
        return None

urls = ["https://...", "https://...", "https://..."]
for url in urls:
    price = monitor_price(url)
    print(f"{url}: {price}")
    time.sleep(2)  # Не перегружаем сервер

def parse_product(html: str) -> dict:
    soup = BeautifulSoup(html, "lxml")

    # Сначала пробуем классический парсинг
    price_el = soup.select_one(".price, [itemprop='price'], .product-price")

    if price_el:
        # Быстро и дёшево
        return {"price": price_el.text.strip(), "source": "classic"}

    # Если не нашли - отдаём в LLM
    return {**parse_with_llm(html, {"price": "цена"}, "Товарная страница"),
            "source": "llm"}

import asyncio
import anthropic

sem = asyncio.Semaphore(10)  # Не более 10 параллельных запросов

async def parse_async(client, html: str) -> dict:
    async with sem:
        # async вызов к API
        ...