A Telegram bot for restocks and price drops: JSON-LD first, CSS selectors as a fallback

Written with AI assistance; the template itself was built and tested by the author's AI assistant.

You want a ping when a sold-out item comes back, or when a price falls by a meaningful amount. You don't need a scraping farm for that. Most shops already publish the data you need in a machine-readable form, and a small asyncio process with SQLite can watch a few dozen product pages.

This article walks through the design of a bot built as a reusable template (Python, aiogram 3, httpx, selectolax, SQLite). Every idea below can be reimplemented in an afternoon without buying anything.

Step 1: read structured data before touching CSS

Most shops emit schema.org Product markup in a <script type="application/ld+json"> tag, because search engines want it. That's far more stable than a CSS class that changes with each redesign. So the first adapter just looks for it.

JSON-LD shows up in a few shapes: a single object, a list, or an object with @graph. A small generator flattens them:

import json

def iter_json_ld_blocks(html_parser):
    for node in html_parser.css('script[type="application/ld+json"]'):
        raw = node.text(deep=True)
        if not raw or not raw.strip():
            continue
        try:
            data = json.loads(raw)
        except json.JSONDecodeError:
            continue
        if isinstance(data, list):
            yield from data
        elif isinstance(data, dict):
            if isinstance(data.get("@graph"), list):
                yield from data["@graph"]
            else:
                yield data

Broken JSON in one block is skipped rather than fatal, since real pages often carry several blocks from different plugins.

Offers, availability and variants

Availability is a URL such as https://schema.org/InStock. Take the last path segment, normalise it, and map it to three states: in stock, out of stock, unknown.

AVAILABILITY_MAP = {
    "instock": "in_stock",
    "limitedavailability": "in_stock",
    "preorder": "in_stock",
    "backorder": "out_of_stock",
    "outofstock": "out_of_stock",
    "soldout": "out_of_stock",
    "discontinued": "out_of_stock",
}

def normalize_availability(raw):
    if not raw:
        return "unknown"
    key = raw.rsplit("/", 1)[-1].strip().lower()
    key = key.replace(" ", "").replace("_", "").replace("-", "")
    return AVAILABILITY_MAP.get(key, "unknown")

(The real map has a few more entries such as OnlineOnly, InStoreOnly and PreSale. Whether a pre-order counts as "in stock" is a judgment call, so change it if you disagree.)

The "unknown" state matters. If a page says nothing clear, you don't want to alert on it.

Some ProductGroup pages, commonly Shopify themes, carry no top-level offer. Price and availability sit on each variant under hasVariant. The adapter handles this in the simplest possible way:

offer = first_offer(node.get("offers")) or {}
if not offer and isinstance(node.get("hasVariant"), list) and node["hasVariant"]:
    first_variant = node["hasVariant"][0]
    if isinstance(first_variant, dict):
        offer = first_offer(first_variant.get("offers")) or {}
price = parse_price(offer.get("price") or offer.get("lowPrice"))
currency = offer.get("priceCurrency")

Be aware of the limitation: it reads the first variant only. If you care about one specific size or colour, point the bot at that variant's own URL, or extend this to match on SKU.

If no Product JSON-LD exists, the adapter also tries OpenGraph tags (product:price:amount, product:availability) before giving up with a parse error.

Prices need cleaning too, since "1 234,56" and "1,234.56" both occur. The JSON-LD adapter strips non-numeric characters and treats a lone comma as a decimal separator. It is a heuristic, so test it on your own sites.

Step 2: CSS selectors for sites without markup

Some storefronts, usually hand-built ones, have no structured data. For those, you declare selectors once per site in a YAML file, matched by a glob on the URL:

sites:
  - name: example-generic-shop
    url_pattern: "https://shop.example.com/product/*"
    adapter: generic_css
    interval_seconds: 1800
    jitter_seconds: 300
    selectors:
      name: "h1.product-title"
      price: "span.price"
      availability: "div.stock-status"
    in_stock_text: "in stock"
    out_of_stock_text: "sold out"

The adapter extracts the text of each selector. Availability is a case-insensitive substring check, with your configured phrases tested first and a few common English and Russian phrases as defaults:

def detect_availability(text, in_stock_text, out_of_stock_text):
    if text is None:
        return "unknown"
    lowered = text.lower()
    if out_of_stock_text and out_of_stock_text in lowered:
        return "out_of_stock"
    if in_stock_text and in_stock_text in lowered:
        return "in_stock"
    if any(p in lowered for p in ("in stock", "available")):
        return "in_stock"
    if any(p in lowered for p in ("out of stock", "sold out", "unavailable")):
        return "out_of_stock"
    return "unknown"

Note the order: "out of stock" is checked before the generic "in stock" patterns would match, which avoids the classic bug where "not available" reads as available. Keep your own out_of_stock_text specific for the same reason. If none of your selectors match, the adapter raises a parse error that tells you to check the YAML, so a redesigned page fails loudly instead of silently reporting nothing.

Step 3: change detection that alerts once

The trick for "alert once" is boring: persist the last known state per product, compare the new reading to it, then overwrite it. The comparison itself is a pure function with no I/O, which makes it easy to unit test.

def detect_changes(old_price, old_available, new_price, new_availability, threshold_percent):
    events = []
    new_available = new_availability == "in_stock"
    if old_available is not None and new_availability != "unknown":
        if not old_available and new_available:
            events.append("back_in_stock")
        elif old_available and not new_available:
            events.append("out_of_stock")
    if old_price and new_price is not None and new_price != old_price:
        change = abs(new_price - old_price) / old_price * 100
        if change >= threshold_percent:
            events.append("price_change")
    return events

(The real function returns ChangeEvent dataclasses carrying the URL, name and old/new price, but the logic is the same.) Details worth copying:

• old_available is None means this is the first check. There is no baseline, so no event. You won't get a flood when you add a product.
• An unknown reading never triggers anything.
• The price threshold is a percentage (default 5.0, set with PRICE_CHANGE_THRESHOLD_PERCENT), and it fires on rises as well as drops. The message text says which direction.

After each check, the new state is written back in SQLite, plus a row in a price_history table:

await conn.execute(
    "UPDATE products SET name = COALESCE(?, name), last_price = ?, "
    "last_currency = ?, last_available = ?, last_checked_at = ? WHERE id = ?",
    (name, price, currency, None if available is None else int(available), now, product_id),
)
await conn.execute(
    "INSERT INTO price_history (product_id, price, available, checked_at) VALUES (?, ?, ?, ?)",
    (product_id, price, None if available is None else int(available), now),
)
await conn.commit()

Because the baseline moves forward on every check, a back-in-stock alert fires once per transition. A caveat I'd rather you hear from me: a slow drift, say 2% per day, never crosses a 5% threshold between two consecutive checks, so it never alerts. If that matters, compare against the price from your last alert, not the last check.

Another quirk: if one check returns unknown, the stored availability becomes null, so the next definite reading is treated as a new baseline and does not alert.

Step 4: Telegram alerts

Subscriptions live in SQLite (users, products, subscriptions), so one product can have several subscribers and gets fetched once. Sending is a loop around aiogram's send_message, and one blocked user must not stop the rest:

async def send_events(bot, events, subscriber_tg_ids):
    for event in events:
        text = event.format_message()
        for tg_id in subscriber_tg_ids:
            try:
                await bot.send_message(tg_id, text)
            except TelegramAPIError as exc:
                logger.warning("failed to notify %s: %s", tg_id, exc)

Users add products with /add <url>, which also runs a first check. /list, /remove <id> and /check cover the rest. Bot replies and alerts come from a small strings table with English and Russian versions, picked by a LANGUAGE=en|ru setting.

Step 5: scheduling and polite fetching

A single asyncio loop wakes every 30 seconds and asks the database which products are due: never checked, or older than their interval plus a random jitter. The defaults are 1800 s interval and up to 300 s jitter. Jitter keeps products from becoming due in lock-step.

threshold = interval + (random.uniform(0, jitter_seconds) if jitter_seconds else 0)
if (now - last).total_seconds() >= threshold:
    due.append(product)

The fetcher adds the following, all configurable through environment variables:

• a robots.txt check per host, cached, enabled by default (RESPECT_ROBOTS_TXT); a disallowed URL raises an error instead of being fetched. If robots.txt is missing or can't be fetched, the fetcher assumes allowed;
• a per-host limit, default 6 requests per minute;
• up to 3 attempts with linear backoff (2 s, then 4 s) on HTTP errors;
• a 15 s timeout.

A failed fetch or parse is logged and skipped; the loop carries on.

Step 6: JS-rendered prices (optional)

If the price is injected by JavaScript, the HTML contains nothing to parse. The template includes a small Playwright fetcher that loads the page in Chromium and returns the rendered HTML:

async with async_playwright() as p:
    browser = await p.chromium.launch()
    try:
        page = await browser.new_page(user_agent=user_agent)
        await page.goto(url, timeout=20_000, wait_until="networkidle")
        return await page.content()
    finally:
        await browser.close()

Playwright is imported lazily, so it isn't required unless you use it. It is switched on with USE_PLAYWRIGHT=true or per site with playwright: true in sites.yaml. Browser fetches wait on the same per-host rate limiter as plain HTTP requests, and if Playwright is enabled but not installed the bot logs one error and falls back to httpx. Retries and the robots.txt check apply only to the plain HTTP path. Also check whether the site offers a JSON endpoint behind that JavaScript first; it's lighter and kinder to the server.

Step 7: run it

With Docker Compose, the SQLite file lives in a named volume so subscriptions survive restarts:

services:
  stockbot:
    build: .
    image: stockbot:latest
    restart: unless-stopped
    env_file:
      - .env
    volumes:
      - stockbot-data:/app/data

volumes:
  stockbot-data:

Without Docker, a systemd unit does the job:

[Service]
Type=simple
User=stockbot
WorkingDirectory=/opt/stockbot
EnvironmentFile=/opt/stockbot/.env
ExecStart=/opt/stockbot/.venv/bin/python -m stockbot
Restart=on-failure
RestartSec=5
NoNewPrivileges=true
PrivateTmp=true

The bot needs a token from BotFather in BOT_TOKEN and refuses to start without it.

Being a good citizen

Monitoring a page is still automated traffic to someone else's server.

• Read each site's terms of service and robots.txt before adding it. The template honours robots.txt by default, but a permissive robots file is not the same as permission in the terms.
• Keep intervals modest. Half an hour is the default here; for most restocks, hourly or slower is plenty. Don't lower it just because you can.
• Watch a handful of products, not a whole catalogue.
• If a store has an official API, a product feed or an email "notify me" button, use that instead of scraping.
• If a site asks you to stop or starts returning errors, back off.

What this is not

• Not a scalping or checkout bot. It reads public product pages and sends you a message. It doesn't log in, fill carts or buy anything.
• Not an anti-bot toolkit. It does not solve CAPTCHAs, spoof fingerprints or defeat bot protection. For transparency about what it does contain: it rotates the User-Agent from a configurable list, and the fetcher accepts an optional proxy list, which is empty by default. If a site blocks you, treat that as a "no" and don't work around it.
• One process, one SQLite file. There is no queue, no worker pool and no horizontal scaling. That's fine for personal or small-shop use, and not designed for thousands of URLs.
• Heuristic parsing. Price formats, variant handling and availability wording vary by site, so test each one before trusting it.

Get the template

The ideas above are enough to build your own. If you'd rather start from working code, the full template (source, 56 tests, Dockerfile and Compose file, systemd unit, README in English and Russian) is sold for 9 USDT through the Telegram shop. There is also a free monitoring kit for a single VPS.