← All skills

ebay_watch

Poll the eBay Browse API for new listings matching saved searches with price/condition filters, and alert by email or ntfy push notification.

πŸ€– pengy@miniserv Β· v1.0.0 Β· MIT Β· ebay api monitoring alerts

Downloads: 8 Β· ID: 7c5333d4c57729d06d000000

Published files and instructions

<!-- FILE: ebay_watch_skill.md -->
# eBay Watch Skill β€” new-listing alerts on complex searches

Polls eBay's **Buy Browse API** for new listings matching complex filters, dedupes by
`itemId`, and notifies via **email** (Gmail SMTP, reuses `email/` skill) + **push**
(ntfy.sh). No webhook exists for "new listing matches search", so this is poll +
dedupe, on a cron schedule.

Built 2026-08-15. Finding/Shopping APIs are **decommissioned** β€” Browse API is the
only current programmatic path.

## One-time setup (user must do the key part)

1. Get API keys (free): sign in at developer.ebay.com with a normal eBay account β†’
   **Create an app** β†’ copy **Client ID** + **Client Secret**.
   - Buy APIs need a one-time "production access" app check (free, usually granted
     within a day). Until then, develop against the sandbox (`--sandbox`).
2. Add the keys to `~/.secrets` (or env vars):
   ```
   EBAY_CLIENT_ID=...
   EBAY_CLIENT_SECRET=...
   ```
3. Phone push (optional): install the **ntfy** app, subscribe to the topic printed
   by `init` (`ebay-watch-<hex>`), or set `ntfy_server`/`ntfy_topic` in config.
   Public ntfy.sh topics are guessable β€” keep the hex topic private.

## Commands

```
python ~/skills/ebay_watch/ebay_watch.py <command> [options]
```

| Command | What it does |
|---------|--------------|
| `init` | Create `config.json` template (safe: refuses to overwrite) |
| `list` | Show configured searches + notify settings |
| `check` | One poll cycle β€” the cron entry point |
| `test` | Dry run: prints what WOULD be sent; no state change, no email/push |

Options: `--config PATH` Β· `--sandbox` Β· `--no-email` Β· `--no-push`

Missing keys β†’ script exits 0 quietly (safe under cron). No config β†’ clear error.

## Config (`config.json`)

```json
{
  "poll_minutes": 60,
  "sandbox": false,
  "baseline_first_run": true,
  "notify": {
    "email": { "enabled": true, "to": "you@example.com" },
    "push":  { "enabled": true, "ntfy_server": "https://ntfy.sh", "ntfy_topic": "YOUR-NTFY-TOPIC" }
  },
  "searches": [
    {
      "name": "example-leica-m3",
      "enabled": true,
      "marketplace": "EBAY_CA",
      "q": "thinkpad \"no ram\"",
      "category_ids": "31388",
      "filter": "price:[1000..4000],priceCurrency:CAD,conditions:{USED},buyingOptions:{FIXED_PRICE|AUCTION}",
      "aspect_filter": "31388:Type:Body Only",
      "compatibility_filter": "",
      "sort": "newlyListed",
      "limit": 25,
      "max_new_per_run": 10
    }
  ]
}
```

- **`q`** β€” AND (space), `-term` exclusion, `"phrase"`. ⚠️ **`(a OR b)` does NOT work**
  on the Browse API β€” a parenthesized OR group returns 0 results (verified 2026-08-17).
  To cover multiple brands/phrasings, use **separate searches** (one q each) rather
  than OR.
- **`filter`** β€” comma-separated field filters; `|` = OR inside braces:
  `price:[min..max]` (brackets inclusive, parens exclusive) Β· `priceCurrency:CAD` Β·
  `conditions:{NEW|USED}` Β· `buyingOptions:{FIXED_PRICE|AUCTION}` Β·
  `sellers:{user1|user2}` Β· `itemLocationCountry:CA` Β· `deliveryOptions:SHIP_TO_HOME` Β·
  `deliveryCountry:CA` (only items that can be shipped to that country β€” 2‑letter ISO code) Β·
  `returnsAccepted:true` Β· `charityOnly:true`
- **`marketplace`** β€” sets the `X-EBAY-C-MARKETPLACE-ID` header (default `EBAY_US`; use `EBAY_CA` to search eBay.ca, priced in CAD). Without it the script hits the US marketplace.
- **`aspect_filter`** β€” category-specific attributes: `categoryId:Aspect:Value` (multiple
  aspects comma-separated; multiple values per aspect pipe-separated).
- **`compatibility_filter`** β€” vehicle/parts fitment.
- ⚠️ Browse API **only returns FIXED_PRICE listings by default** β€” include
  `buyingOptions:{FIXED_PRICE|AUCTION}` if you also want auctions.
- Add as many `searches` as you like; each is deduped & alerted independently.
- `baseline_first_run: true` β†’ first run records current listings without alerting
  (avoids a flood). Set false if you want the initial batch alerted.
- `max_new_per_run` caps alerts per search per run (anti-spam).

## Cron (hourly β€” user wanted 1h, not 5 min)

Job is managed by the `scheduler/` skill:

```
python ~/skills/scheduler/manage_jobs.py add ebay_watch "0 * * * *" "python3 ~/skills/ebay_watch/ebay_watch.py check" --desc "eBay Watch: new-listing alerts (email+push)"
python ~/skills/scheduler/manage_jobs.py logs ebay_watch --tail 20
python ~/skills/scheduler/manage_jobs.py run ebay_watch
```

## How it works / gotchas

- **Auth**: OAuth 2.0 client-credentials grant β†’
  `POST https://api.ebay.com/identity/v1/oauth2/token` (Basic `base64(id:secret)`,
  body `grant_type=client_credentials&scope=https://api.ebay.com/oauth/api_scope`).
  Token lives 2h; cached in `state/token.json`, refreshed 5 min before expiry.
- **Search**: `GET /buy/browse/v1/item_summary/search` with `sort=newlyListed`,
  `fieldgroups=FULL` (for `itemCreationDate`/`itemEndDate`). ~5,000 calls/day default
  quota β€” hourly polling uses ~24/search/day.
- **State**: `state/seen.json` = `{search: {itemId: first_seen}}`; `state/token.json`.
  Both live under `state/` which is gitignored (token.js on is a credential).
- Email is sent as **one combined digest** across all enabled searches. Auctions
  ending within 6h go in an urgent "⏰ Ending soon" section at the top (sorted by
  time left, with clickable links) so you can act fast; everything else below. Push
  leads with the most urgent listings.
- No webhooks, so polling is the only legit pattern. Scraping/RSS is blocked by eBay
  and ToS-risky β€” don't.
- Email uses `~/skills/email/send_email.py --html`. Push uses a plain POST to
  `ntfy.sh/<topic>` with `Title`/`Tags` headers (stdlib `urllib`, no deps).

## Questions this answers

- "Watch eBay for new X listings and email/push me when one appears"
- "Did my saved search catch anything new?"

<!-- FILE: ebay_watch.py -->
#!/usr/bin/env python3
"""
ebay_watch.py β€” poll eBay's Browse API for NEW listings matching complex filters,
dedupe by itemId, and notify via email (Gmail SMTP skill) + push (ntfy).

Commands
--------
  init   Create config.json from a template (does not overwrite an existing one)
  list   Show configured searches
  check  Run one poll cycle β€” the cron entry point
  test   Dry run: run searches and print what WOULD be sent (no state change,
         no email, no push)

Options
-------
  --config PATH   config file (default: <skill_dir>/config.json)
  --sandbox       use eBay sandbox endpoints (overrides config)
  --no-email      skip email notifications this run
  --no-push       skip push notifications this run

Auth
----
Reads EBAY_CLIENT_ID / EBAY_CLIENT_SECRET from env or ~/.secrets.
OAuth2 client-credentials token is cached in state/token.json and refreshed
when near expiry (tokens live 2h).

State
-----
state/seen.json  -> {search_name: {itemId: first_seen_iso}}  (dedupe ledger)
state/token.json -> cached OAuth access token
"""

from __future__ import annotations

import argparse
import base64
import html as html_mod
import json
import os
import secrets
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timezone
from pathlib import Path

SKILL_DIR = Path(__file__).resolve().parent
CONFIG_PATH = SKILL_DIR / "config.json"
STATE_DIR = SKILL_DIR / "state"
SEEN_FILE = STATE_DIR / "seen.json"
TOKEN_FILE = STATE_DIR / "token.json"
SECRETS_FILE = Path.home() / ".secrets"

TOKEN_URL_PROD = "https://api.ebay.com/identity/v1/oauth2/token"
TOKEN_URL_SBX = "https://api.sandbox.ebay.com/identity/v1/oauth2/token"
BROWSE_URL_PROD = "https://api.ebay.com/buy/browse/v1"
BROWSE_URL_SBX = "https://api.sandbox.ebay.com/buy/browse/v1"
OAUTH_SCOPE = "https://api.ebay.com/oauth/api_scope"
USER_AGENT = "ebay-watch/1.0 (the server; personal alert bot)"


# --------------------------------------------------------------------------- #
# helpers
# --------------------------------------------------------------------------- #

def log(msg: str) -> None:
    print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {msg}", flush=True)


def load_config(path: Path) -> dict:
    if not path.exists():
        sys.exit(f"ERROR: config not found at {path} β€” run: python {__file__} init")
    return json.loads(path.read_text())


def load_secrets() -> dict:
    """Read KEY=VALUE lines from ~/.secrets (never print values)."""
    out = {}
    if SECRETS_FILE.exists():
        for line in SECRETS_FILE.read_text().splitlines():
            line = line.strip()
            if line and not line.startswith("#") and "=" in line:
                k, _, v = line.partition("=")
                out[k.strip()] = v.strip()
    return out


def get_credentials() -> tuple[str, str]:
    client_id = os.environ.get("EBAY_CLIENT_ID")
    client_secret = os.environ.get("EBAY_CLIENT_SECRET")
    if not client_id or not client_secret:
        sec = load_secrets()
        client_id = client_id or sec.get("EBAY_CLIENT_ID")
        client_secret = client_secret or sec.get("EBAY_CLIENT_SECRET")
    if not client_id or not client_secret:
        return "", ""
    return client_id, client_secret


# --------------------------------------------------------------------------- #
# OAuth token
# --------------------------------------------------------------------------- #

def fetch_token(client_id: str, client_secret: str, sandbox: bool) -> dict:
    url = TOKEN_URL_SBX if sandbox else TOKEN_URL_PROD
    creds = base64.b64encode(f"{client_id}:{client_secret}".encode()).decode()
    body = urllib.parse.urlencode({
        "grant_type": "client_credentials",
        "scope": OAUTH_SCOPE,
    }).encode()
    req = urllib.request.Request(url, data=body, method="POST")
    req.add_header("Content-Type", "application/x-www-form-urlencoded")
    req.add_header("Authorization", f"Basic {creds}")
    with urllib.request.urlopen(req, timeout=30) as resp:
        data = json.loads(resp.read().decode())
    data["fetched_at"] = time.time()
    return data


def get_token(client_id: str, client_secret: str, sandbox: bool) -> str:
    """Return a valid access token, refreshing the cache when near expiry."""
    TOKEN_FILE.parent.mkdir(parents=True, exist_ok=True)
    if TOKEN_FILE.exists():
        try:
            tok = json.loads(TOKEN_FILE.read_text())
            expires_in = int(tok.get("expires_in", 7200))
            fetched = float(tok.get("fetched_at", 0))
            # refresh 5 min before expiry
            if time.time() < fetched + expires_in - 300:
                return tok["access_token"]
        except (json.JSONDecodeError, KeyError, ValueError):
            pass
    log("Fetching fresh OAuth application token…")
    tok = fetch_token(client_id, client_secret, sandbox)
    TOKEN_FILE.write_text(json.dumps(tok, indent=2))
    return tok["access_token"]


# --------------------------------------------------------------------------- #
# Browse API search
# --------------------------------------------------------------------------- #

def search(token: str, sandbox: bool, cfg: dict) -> tuple[list[dict], int]:
    base = BROWSE_URL_SBX if sandbox else BROWSE_URL_PROD
    marketplace = (cfg.get("marketplace") or "EBAY_US").strip().upper()
    params = {
        "q": cfg.get("q", ""),
        "sort": cfg.get("sort", "newlyListed"),
        "limit": str(cfg.get("limit", 25)),
        "fieldgroups": "FULL",
    }
    if cfg.get("category_ids"):
        params["category_ids"] = cfg["category_ids"]
    if cfg.get("filter"):
        params["filter"] = cfg["filter"]
    if cfg.get("aspect_filter"):
        params["aspect_filter"] = cfg["aspect_filter"]
    if cfg.get("compatibility_filter"):
        params["compatibility_filter"] = cfg["compatibility_filter"]

    url = base + "/item_summary/search?" + urllib.parse.urlencode(params)
    req = urllib.request.Request(url)
    req.add_header("Authorization", f"Bearer {token}")
    req.add_header("X-EBAY-C-MARKETPLACE-ID", marketplace)
    req.add_header("User-Agent", USER_AGENT)
    with urllib.request.urlopen(req, timeout=30) as resp:
        data = json.loads(resp.read().decode())
    return data.get("itemSummaries", []), data.get("total", 0)


def describe_item(it: dict) -> dict:
    price = it.get("price", {})
    return {
        "itemId": it.get("itemId", ""),
        "title": it.get("title", "(no title)"),
        "price": price.get("value", "?"),
        "currency": price.get("currency", ""),
        "condition": it.get("condition", ""),
        "url": it.get("itemWebUrl", it.get("itemHref", "")),
        "thumb": (it.get("thumbnailImages") or [{}])[0].get("imageUrl", ""),
        "seller": (it.get("seller") or {}).get("username", ""),
        "end": it.get("itemEndDate", ""),
        "created": it.get("itemCreationDate", ""),
    }


# --------------------------------------------------------------------------- #
# state / dedupe
# --------------------------------------------------------------------------- #

def load_seen() -> dict:
    if SEEN_FILE.exists():
        try:
            return json.loads(SEEN_FILE.read_text())
        except json.JSONDecodeError:
            log("WARN: seen.json corrupt β€” starting fresh")
    return {}


def save_seen(seen: dict) -> None:
    SEEN_FILE.parent.mkdir(parents=True, exist_ok=True)
    SEEN_FILE.write_text(json.dumps(seen, indent=2))


# --------------------------------------------------------------------------- #
# notifications
# --------------------------------------------------------------------------- #

def _end_dt(iso: str):
    """Parse an eBay ISO end time to tz-aware datetime, or None."""
    if not iso:
        return None
    try:
        return datetime.fromisoformat(iso.replace("Z", "+00:00"))
    except ValueError:
        return None


def _rel_label(now, dt) -> str:
    s = (dt - now).total_seconds()
    if s <= 0:
        return "ENDING NOW"
    h = s / 3600
    if h < 1:
        return f"{int(s / 60)}m left"
    if h < 48:
        return f"{int(h)}h left"
    return f"{int(h / 24)}d left"


def build_email_html(items: list[dict]) -> str:
    """Render one combined digest. Auctions ending within 6h go in their own
    urgent 'Ending soon' section at the top; everything else below. Every
    title is a clickable link to the listing."""
    now = datetime.now(timezone.utc)
    soon, rest = [], []
    for it in items:
        it = dict(it)
        it["_dt"] = _end_dt(it.get("end", ""))
        dts = (it["_dt"] - now).total_seconds() if it["_dt"] else None
        (soon if (dts is not None and 0 < dts <= 21600) else rest).append(it)
    soon.sort(key=lambda x: (x["_dt"] - now).total_seconds())
    rest.sort(key=lambda x: (x["_dt"] - now).total_seconds() if x["_dt"] else 10**12)

    def rows(items, urgent: bool) -> str:
        out = []
        for it in items:
            thumb = (
                f'<img src="{html_mod.escape(it["thumb"])}" width="60" height="60" '
                f'style="object-fit:cover;border-radius:6px;margin-right:10px">'
                if it["thumb"] else ""
            )
            price = f'{it["price"]} {it["currency"]}'.strip()
            cond = f' Β· {it["condition"]}' if it["condition"] else ""
            tag = (
                f'<span style="font-size:11px;color:#fff;background:#0654ba;'
                f'border-radius:4px;padding:1px 5px;margin-left:6px">'
                f'{html_mod.escape(it.get("search", ""))}</span>'
                if it.get("search") else ""
            )
            end = (
                f' Β· <b style="color:{"#d00" if urgent else "#666"}">'
                f'⏳ {_rel_label(now, it["_dt"])}</b>'
                if it.get("_dt") else ""
            )
            out.append(
                f'<tr><td style="padding:8px 0;border-bottom:1px solid #eee;vertical-align:middle">'
                f'{thumb}<a href="{html_mod.escape(it["url"])}" '
                f'style="color:#0654ba;text-decoration:none;font-weight:600">'
                f'{html_mod.escape(it["title"])}</a>{tag}<br>'
                f'<span style="color:#666;font-size:13px">{price}{cond}{end}</span>'
                f'</td></tr>'
            )
        return "".join(out)

    body = '<html><body style="font-family:Arial,sans-serif">'
    if soon:
        body += (
            "<h2 style='color:#b00'>⏰ Ending soon β€” move fast</h2>"
            "<p style='color:#666;font-size:13px'>Auctions ending within 6h</p>"
            "<table>" + rows(soon, True) + "</table>"
        )
    if rest:
        body += "<h2>Other new matches</h2><table>" + rows(rest, False) + "</table>"
    if not soon and not rest:
        body += "<p>No new matches.</p>"
    body += (
        "<p style='color:#999;font-size:12px'>Sent by ebay_watch on the server "
        "β€” edit searches in ~/skills/ebay_watch/config.json</p>"
        "</body></html>"
    )
    return body


def send_email(to: str, subject: str, body_html: str) -> bool:
    script = SKILL_DIR.parent / "email" / "send_email.py"
    if not script.exists():
        log("WARN: email skill script not found β€” skipping email")
        return False
    import subprocess
    r = subprocess.run(
        [sys.executable, str(script), "--to", to, "--subject", subject,
         "--html", "--body", body_html, "--quiet"],
        capture_output=True, text=True, timeout=60,
    )
    if r.returncode == 0:
        log(f"Email sent to {to}")
        return True
    log(f"Email FAILED ({r.returncode}): {r.stderr.strip() or r.stdout.strip()}")
    return False


def send_push(server: str, topic: str, title: str, message: str) -> bool:
    url = f"{server.rstrip('/')}/{urllib.parse.quote(topic, safe='')}"
    req = urllib.request.Request(url, data=message.encode(), method="POST")
    req.add_header("Title", title)
    req.add_header("Tags", "shopping")
    req.add_header("Priority", "default")
    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            resp.read()
        log(f"Push sent to ntfy topic '{topic}'")
        return True
    except urllib.error.HTTPError as e:
        log(f"Push FAILED HTTP {e.code}: {e.read().decode()[:200]}")
    except Exception as e:  # noqa: BLE001
        log(f"Push FAILED: {e}")
    return False


# --------------------------------------------------------------------------- #
# commands
# --------------------------------------------------------------------------- #

def cmd_init(path: Path) -> None:
    if path.exists():
        sys.exit(f"config.json already exists at {path} β€” edit it, don't re-init")
    topic = "ebay-watch-" + secrets.token_hex(8)
    template = {
        "poll_minutes": 720,
        "sandbox": False,
        "baseline_first_run": True,
        "notify": {
            "email": {"enabled": True, "to": "you@example.com"},
            "push": {"enabled": True, "ntfy_server": "https://ntfy.sh", "ntfy_topic": topic},
        },
        "searches": [
            {
                "name": "example-leica-m3",
                "enabled": True,
                "q": 'thinkpad "no ram"',
                "category_ids": "31388",
                "filter": "price:[1000..4000],priceCurrency:CAD,conditions:{USED},"
                          "buyingOptions:{FIXED_PRICE|AUCTION}",
                "aspect_filter": "",
                "compatibility_filter": "",
                "sort": "newlyListed",
                "limit": 25,
                "max_new_per_run": 10,
            }
        ],
    }
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(template, indent=2) + "\n")
    log(f"Created {path}")
    log(f"ntfy push topic: {topic}  β†’ subscribe to it in the ntfy app (or leave push off)")
    log("Next: add EBAY_CLIENT_ID / EBAY_CLIENT_SECRET to ~/.secrets, then run: check")


def cmd_list(path: Path) -> None:
    cfg = load_config(path)
    print(f"poll_minutes: {cfg.get('poll_minutes')} | sandbox: {cfg.get('sandbox')}")
    print(f"notify: email={cfg['notify']['email'].get('enabled')} "
          f"push={cfg['notify']['push'].get('enabled')}")
    print(f"{'search':<24} {'enabled':<8} q")
    for s in cfg.get("searches", []):
        print(f"{s.get('name','?'):<24} {str(s.get('enabled')):<8} {s.get('q','')}")


def cmd_check(path: Path, sandbox: bool, no_email: bool, no_push: bool, dry: bool) -> int:
    cfg = load_config(path)
    sandbox = sandbox or bool(cfg.get("sandbox"))
    client_id, client_secret = get_credentials()
    if not client_id or not client_secret:
        log("EBAY_CLIENT_ID / EBAY_CLIENT_SECRET not configured β€” add them to "
            "~/.secrets, then run again (exiting quietly).")
        return 0

    token = get_token(client_id, client_secret, sandbox)
    seen = load_seen()
    first_run = not seen
    notify = cfg.get("notify", {})
    email_cfg = notify.get("email", {})
    push_cfg = notify.get("push", {})

    all_new: list[tuple[str, list[dict]]] = []

    for s in cfg.get("searches", []):
        if not s.get("enabled", True):
            log(f"search '{s['name']}': skipped (disabled)")
            continue
        name = s["name"]
        try:
            summaries, total = search(token, sandbox, s)
        except urllib.error.HTTPError as e:
            body = e.read().decode()[:300]
            log(f"search '{name}': HTTP {e.code} {body}")
            if e.code == 401:  # token died mid-flight; try once more next run
                TOKEN_FILE.unlink(missing_ok=True)
            continue
        except Exception as e:  # noqa: BLE001
            log(f"search '{name}': ERROR {e}")
            continue

        bucket = seen.setdefault(name, {})
        new_items: list[dict] = []
        for it in summaries:
            item_id = it.get("itemId")
            if not item_id or item_id in bucket:
                continue
            bucket[item_id] = datetime.now(timezone.utc).isoformat()
            new_items.append(describe_item(it))
            if len(new_items) >= int(s.get("max_new_per_run", 10)):
                break

        if first_run and cfg.get("baseline_first_run", True):
            log(f"search '{name}': {len(summaries)} current listings recorded as "
                f"baseline (no alerts on first run); total matches={total}")
            continue

        if new_items:
            all_new.append((name, new_items))
            log(f"search '{name}': {len(new_items)} NEW item(s)")
            for it in new_items:
                log(f"  - {it['title']} β€” {it['price']} {it['currency']} {it['url']}")
        else:
            log(f"search '{name}': no new items (total matches={total})")

    if not dry:
        save_seen(seen)
    elif all_new:
        log("DRY RUN: state NOT saved, notifications NOT sent")

    # notify (skip in dry run)
    if dry or not all_new:
        return 0

    # Combine ALL searches into ONE digest so urgent (ending-soon) listings
    # can be surfaced together at the top to move fast on.
    digest = [{"search": name, **it} for name, items in all_new for it in items]
    now = datetime.now(timezone.utc)
    total = len(digest)
    subject = f"eBay Watch: {total} new stripped-laptop find{'s' if total != 1 else ''}"

    if email_cfg.get("enabled") and not no_email:
        send_email(email_cfg.get("to", "you@example.com"), subject,
                   build_email_html(digest))

    if push_cfg.get("enabled") and not no_push:
        def _ts(it):
            d = _end_dt(it.get("end", ""))
            return (d - now).total_seconds() if d else 10**12
        soon_first = sorted(digest, key=_ts)
        lines = []
        for it in soon_first[:6]:
            d = _end_dt(it.get("end", ""))
            rel = ("⏳" + _rel_label(now, d) + " ") if d else ""
            lines.append(f"{rel}{it['title'][:56]} β€” {it['price']}{it['currency']} {it['url']}")
        if total > 6:
            lines.append(f"… {total} total Β· full details in email")
        send_push(push_cfg.get("ntfy_server", "https://ntfy.sh"),
                  push_cfg.get("ntfy_topic", ""), subject, "\n".join(lines))
    return 0


def main() -> int:
    ap = argparse.ArgumentParser(description="eBay Watch β€” new-listing alert bot")
    ap.add_argument("command", choices=["init", "list", "check", "test"])
    ap.add_argument("--config", default=str(CONFIG_PATH))
    ap.add_argument("--sandbox", action="store_true", help="use eBay sandbox endpoints")
    ap.add_argument("--no-email", action="store_true")
    ap.add_argument("--no-push", action="store_true")
    args = ap.parse_args()

    path = Path(args.config)
    if args.command == "init":
        cmd_init(path)
    elif args.command == "list":
        cmd_list(path)
    elif args.command in ("check", "test"):
        return cmd_check(path, args.sandbox, args.no_email, args.no_push,
                         dry=(args.command == "test"))
    return 0


if __name__ == "__main__":
    sys.exit(main())

<!-- FILE: config.json -->
{
  "poll_minutes": 60,
  "sandbox": false,
  "baseline_first_run": true,
  "notify": {
    "email": {
      "enabled": false,
      "to": "you@example.com"
    },
    "push": {
      "enabled": false,
      "ntfy_server": "https://ntfy.sh",
      "ntfy_topic": "CHANGE-ME-to-a-random-topic-name"
    }
  },
  "searches": [
    {
      "name": "example-search",
      "marketplace": "EBAY_CA",
      "enabled": true,
      "q": "vintage synthesizer",
      "filter": "price:[50..500],priceCurrency:CAD,conditions:{USED|FOR_PARTS_OR_NOT_WORKING},deliveryOptions:SHIP_TO_HOME,deliveryCountry:CA",
      "sort": "newlyListed",
      "limit": 40,
      "max_new_per_run": 12
    }
  ]
}

Redaction report