drwily/Pastel
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #11 from prosolis/claude/fix-isthereanydeal-api-W5umt

Browse Source 
CheapShark kinda sucks.. let's move to ITAD instead
This commit is contained in:
prosolis
2026-02-27 21:06:12 -08:00
committed by GitHub
parent a83f230291 d1a33e97c5
commit 1dcb76a394
9 changed files with 356 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
.env.example
View File

@@ -4,9 +4,13 @@ MATRIX_BOT_USER_ID=@dealsbot:example.com
MATRIX_BOT_ACCESS_TOKEN=syt_...
MATRIX_DEALS_ROOM_ID=!roomid:example.com
# IsThereAnyDeal API key (optional — enables historical low detection)
# IsThereAnyDeal API key (optional — enables historical low detection and ITAD deals)
ITAD_API_KEY=
# Deal sources: comma-separated list (cheapshark, itad, or both)
# Requires ITAD_API_KEY when "itad" is included
DEAL_SOURCES=cheapshark
# Deal filtering
MIN_DEAL_RATING=8.0
MIN_DISCOUNT_PERCENT=50

11
README.md
View File

@@ -7,8 +7,10 @@ Deals are sourced from PC/digital storefronts only (Steam, GOG, Humble Store, Gr
## Data Sources
- **CheapShark** — polled every 2 hours for top deals across Steam, GOG, Humble Store, and GreenManGaming
- **IsThereAnyDeal** — polled every 2 hours for deals across all tracked stores, with built-in historical low detection (requires API key)
- **Epic Games Store** — polled daily for free game promotions
- **IsThereAnyDeal** (optional) — flags deals that are at an all-time historical low price
CheapShark and IsThereAnyDeal can be used individually or together — configure via the `DEAL_SOURCES` variable. When used as a deal source, IsThereAnyDeal provides historical low flags directly. When only CheapShark is active, IsThereAnyDeal can still optionally enrich deals with historical low info via the `ITAD_API_KEY`.
## Quick Start
@@ -59,7 +61,8 @@ All configuration is via environment variables (see `.env.example`):
| `MATRIX_BOT_USER_ID` | Yes | — | Bot's Matrix user ID |
| `MATRIX_BOT_ACCESS_TOKEN` | Yes | — | Bot's access token |
| `MATRIX_DEALS_ROOM_ID` | Yes | — | Room ID to post deals in |
| `ITAD_API_KEY` | No | — | IsThereAnyDeal API key for historical low detection |
| `ITAD_API_KEY` | No | — | IsThereAnyDeal API key (required when `itad` is in `DEAL_SOURCES`, optional otherwise for historical low detection) |
| `DEAL_SOURCES` | No | cheapshark | Comma-separated deal sources: `cheapshark`, `itad`, or `cheapshark,itad` |
| `MIN_DEAL_RATING` | No | 8.0 | Minimum CheapShark deal rating (0-10) |
| `MIN_DISCOUNT_PERCENT` | No | 50 | Minimum discount percentage |
| `MAX_PRICE_USD` | No | 20 | Maximum sale price in USD |
@@ -77,10 +80,10 @@ python -m gaming_deals_bot --check
This verifies:
- **Matrix** — authentication token is valid and bot has joined the target room
- **CheapShark** — API is reachable
- **CheapShark** — API is reachable (skipped if not in `DEAL_SOURCES`)
- **Epic Games Store** — API is reachable
- **Frankfurter** — exchange rate API is reachable
- **IsThereAnyDeal** — API key is valid (skipped if not configured)
- **IsThereAnyDeal** — API key is valid (required when `itad` is in `DEAL_SOURCES`, skipped otherwise if not configured)
The command exits with code 0 on success and 1 on failure, so it works in CI and Docker health-checks.

33
gaming_deals_bot/__main__.py
View File

@@ -48,14 +48,25 @@ async def main():
scheduler = AsyncIOScheduler()
# CheapShark: every 2 hours
scheduler.add_job(
bot.check_cheapshark,
"interval",
hours=2,
id="cheapshark",
name="CheapShark deals check",
)
# CheapShark: every 2 hours (if enabled)
if "cheapshark" in config.deal_sources:
scheduler.add_job(
bot.check_cheapshark,
"interval",
hours=2,
id="cheapshark",
name="CheapShark deals check",
)
# ITAD deals: every 2 hours (if enabled)
if "itad" in config.deal_sources:
scheduler.add_job(
bot.check_itad_deals,
"interval",
hours=2,
id="itad_deals",
name="ITAD deals check",
)
# Epic free games: once daily
scheduler.add_job(
@@ -67,12 +78,16 @@ async def main():
)
scheduler.start()
logger.info("Bot started — scheduler running")
logger.info(
"Bot started — scheduler running (deal sources: %s)",
", ".join(config.deal_sources),
)
await bot.send_intro()
# Run initial checks immediately (after first-run population is done)
await bot.check_cheapshark()
await bot.check_itad_deals()
await bot.check_epic_free_games()
# Keep running until signaled to stop

81
gaming_deals_bot/bot.py
View File

@@ -5,13 +5,14 @@ import logging
import httpx
from .cheapshark import CheapSharkDeal, fetch_deals
from .cheapshark import CheapSharkDeal, fetch_deals as fetch_cheapshark_deals
from .config import Config
from .currency import refresh_rates
from .database import Database
from .epic import EpicFreeGame, fetch_free_games
from .formatter import format_deal, format_epic_free, format_epic_upcoming
from .formatter import format_deal, format_epic_free, format_epic_upcoming, format_itad_deal
from .itad import check_single_historical_low
from .itad_deals import ITADDeal, fetch_deals as fetch_itad_deals
from .matrix_client import MatrixDealsClient
logger = logging.getLogger(__name__)
@@ -53,23 +54,38 @@ class DealsBot:
await self.db.close()
async def _populate_initial_state(self):
"""Fetch current CheapShark deals and record them without posting (avoids spam on first run).
"""Fetch current deals and record them without posting (avoids spam on first run).
Epic free games are intentionally *not* recorded here so that the
subsequent ``check_epic_free_games`` call will post them. There are
only a handful at any time and they are time-limited, so users should
see them immediately rather than waiting for the next cycle.
"""
deals = await fetch_deals(
self._http,
max_price=self.config.max_price_usd,
min_rating=self.config.min_deal_rating,
min_discount=self.config.min_discount_percent,
)
for deal in deals:
await self.db.mark_posted(deal.dedup_id, "cheapshark", deal.title)
total = 0
logger.info("First run: recorded %d existing CheapShark deals", len(deals))
if "cheapshark" in self.config.deal_sources:
deals = await fetch_cheapshark_deals(
self._http,
max_price=self.config.max_price_usd,
min_rating=self.config.min_deal_rating,
min_discount=self.config.min_discount_percent,
)
for deal in deals:
await self.db.mark_posted(deal.dedup_id, "cheapshark", deal.title)
total += len(deals)
if "itad" in self.config.deal_sources and self.config.itad_api_key:
itad_deals = await fetch_itad_deals(
self._http,
self.config.itad_api_key,
max_price=self.config.max_price_usd,
min_discount=self.config.min_discount_percent,
)
for deal in itad_deals:
await self.db.mark_posted(deal.dedup_id, "itad", deal.title)
total += len(itad_deals)
logger.info("First run: recorded %d existing deals", total)
async def send_intro(self):
"""Send an intro message to the Matrix room if configured."""
@@ -81,9 +97,11 @@ class DealsBot:
"""Poll CheapShark for deals and post new ones."""
if not self._first_run_done:
return
if "cheapshark" not in self.config.deal_sources:
return
logger.info("Checking CheapShark for deals...")
deals = await fetch_deals(
deals = await fetch_cheapshark_deals(
self._http,
max_price=self.config.max_price_usd,
min_rating=self.config.min_deal_rating,
@@ -96,6 +114,29 @@ class DealsBot:
# Prune old records
await self.db.prune_old(days=30)
async def check_itad_deals(self):
"""Poll IsThereAnyDeal for deals and post new ones."""
if not self._first_run_done:
return
if "itad" not in self.config.deal_sources:
return
if not self.config.itad_api_key:
logger.warning("ITAD deal source enabled but ITAD_API_KEY is not set")
return
logger.info("Checking IsThereAnyDeal for deals...")
deals = await fetch_itad_deals(
self._http,
self.config.itad_api_key,
max_price=self.config.max_price_usd,
min_discount=self.config.min_discount_percent,
)
for deal in deals:
await self._process_itad_deal(deal)
await self.db.prune_old(days=30)
async def _process_deal(self, deal: CheapSharkDeal):
"""Check dedup, check historical low, format, and post a single deal."""
if await self.db.has_been_posted(deal.dedup_id):
@@ -119,6 +160,20 @@ class DealsBot:
else:
logger.warning("Failed to post deal: %s — will retry next cycle", deal.title)
async def _process_itad_deal(self, deal: ITADDeal):
"""Check dedup, format, and post a single ITAD deal."""
if await self.db.has_been_posted(deal.dedup_id):
return
plain_text, html = format_itad_deal(deal)
success = await self.matrix.send_deal(plain_text, html)
if success:
await self.db.mark_posted(deal.dedup_id, "itad", deal.title)
logger.info("Posted ITAD deal: %s", deal.title)
else:
logger.warning("Failed to post ITAD deal: %s — will retry next cycle", deal.title)
async def check_epic_free_games(self):
"""Poll Epic Games Store for free games and post new ones."""
if not self._first_run_done:

6
gaming_deals_bot/config.py
View File

@@ -14,6 +14,12 @@ class Config:
# ITAD API key (optional — historical low checks disabled without it)
self.itad_api_key = os.environ.get("ITAD_API_KEY", "")
# Deal sources: comma-separated list of "cheapshark", "itad" (default: cheapshark)
raw_sources = os.environ.get("DEAL_SOURCES", "cheapshark")
self.deal_sources: list[str] = [
s.strip().lower() for s in raw_sources.split(",") if s.strip()
]
# Deal filtering
self.min_deal_rating = float(os.environ.get("MIN_DEAL_RATING", "8.0"))
self.min_discount_percent = int(os.environ.get("MIN_DISCOUNT_PERCENT", "50"))

37
gaming_deals_bot/formatter.py
View File

@@ -10,6 +10,7 @@ import markdown
from .cheapshark import CheapSharkDeal
from .currency import format_price
from .epic import EpicFreeGame
from .itad_deals import ITADDeal
_md = markdown.Markdown()
@@ -61,6 +62,42 @@ def format_deal(deal: CheapSharkDeal, is_historical_low: bool = False) -> tuple[
return plain_text, html
def format_itad_deal(deal: ITADDeal) -> tuple[str, str]:
"""Format an ITAD deal into (plain_text, html) for Matrix.
Returns (body, formatted_body).
"""
sale_multi = format_price(deal.sale_price)
normal_display = format_price(deal.normal_price)
lines = [
f"**🎮 [DEAL] {deal.title}**",
f"> {deal.discount}% off on {deal.shop_name} ~~{normal_display}~~",
f"> 💰 **{sale_multi}**",
]
if deal.is_historical_low:
lines.append("> 🏆 _All-time low!_")
lines.append(f"> 🔗 [View Deal]({deal.url})")
md_text = "\n".join(lines)
html = _render_html(md_text)
plain_lines = [
f"🎮 [DEAL] {deal.title}",
f" {deal.discount}% off on {deal.shop_name} (was {normal_display})",
f" 💰 {sale_multi}",
]
if deal.is_historical_low:
plain_lines.append(" 🏆 All-time low!")
plain_lines.append(f" 🔗 {deal.url}")
plain_text = "\n".join(plain_lines)
return plain_text, html
def format_epic_free(game: EpicFreeGame) -> tuple[str, str]:
"""Format an Epic free game into (plain_text, html) for Matrix.

90
gaming_deals_bot/itad.py
View File

@@ -7,6 +7,29 @@ logger = logging.getLogger(__name__)
BASE_URL = "https://api.isthereanydeal.com"
async def _lookup_game_id(
client: httpx.AsyncClient,
api_key: str,
steam_app_id: str,
) -> str | None:
"""Look up the ITAD game UUID for a Steam app ID.
Returns the ITAD game UUID string, or None if not found.
"""
try:
resp = await client.get(
f"{BASE_URL}/games/lookup/v1",
params={"key": api_key, "appid": steam_app_id},
)
resp.raise_for_status()
data = resp.json()
if data.get("found") and data.get("game"):
return data["game"]["id"]
except (httpx.HTTPError, ValueError, KeyError) as exc:
logger.error("ITAD lookup error for app %s: %s", steam_app_id, exc)
return None
async def check_historical_lows(
client: httpx.AsyncClient,
api_key: str,
@@ -19,18 +42,22 @@ async def check_historical_lows(
if not api_key or not steam_app_ids:
return {}
# Build the list of ITAD game IDs from Steam app IDs
# ITAD uses the format "app/{steam_app_id}" for Steam games
game_ids = [f"app/{sid}" for sid in steam_app_ids]
# Look up ITAD game UUIDs for each Steam app ID
itad_id_to_steam: dict[str, str] = {}
for sid in steam_app_ids:
itad_id = await _lookup_game_id(client, api_key, sid)
if itad_id:
itad_id_to_steam[itad_id] = sid
if not itad_id_to_steam:
return {}
try:
resp = await client.get(
resp = await client.post(
f"{BASE_URL}/games/overview/v2",
params={"key": api_key, "shops[]": "steam"},
headers={"Content-Type": "application/json"},
params={"key": api_key, "shops": [61]},
json=list(itad_id_to_steam.keys()),
)
# The v2 endpoint may use POST with body — fall back to per-game lookup
# if batch doesn't work. For now, use the games/info endpoint pattern.
resp.raise_for_status()
data = resp.json()
except (httpx.HTTPError, ValueError) as exc:
@@ -38,19 +65,17 @@ async def check_historical_lows(
return {}
results: dict[str, bool] = {}
# Parse the response — structure depends on the ITAD API version
# The overview endpoint returns price overview with historical low data
if isinstance(data, dict):
for steam_id in steam_app_ids:
game_key = f"app/{steam_id}"
game_data = data.get(game_key) or data.get(steam_id, {})
if isinstance(game_data, dict):
lowest = game_data.get("lowest", {})
current = game_data.get("current", {})
if lowest and current:
lowest_price = lowest.get("price", float("inf"))
current_price = current.get("price", float("inf"))
results[steam_id] = current_price <= lowest_price
for price_entry in data.get("prices", []):
itad_id = price_entry.get("id")
steam_id = itad_id_to_steam.get(itad_id)
if not steam_id:
continue
lowest = price_entry.get("lowest")
current = price_entry.get("current")
if lowest and current:
lowest_price = lowest.get("price", {}).get("amount", float("inf"))
current_price = current.get("price", {}).get("amount", float("inf"))
results[steam_id] = current_price <= lowest_price
logger.info(
"ITAD: checked %d games, %d are at historical low",
@@ -69,13 +94,15 @@ async def check_single_historical_low(
if not api_key or not steam_app_id:
return False
itad_id = await _lookup_game_id(client, api_key, steam_app_id)
if not itad_id:
return False
try:
resp = await client.get(
resp = await client.post(
f"{BASE_URL}/games/overview/v2",
params={
"key": api_key,
"apps[]": f"app/{steam_app_id}",
},
params={"key": api_key},
json=[itad_id],
)
resp.raise_for_status()
data = resp.json()
@@ -83,10 +110,13 @@ async def check_single_historical_low(
logger.error("ITAD API error for app %s: %s", steam_app_id, exc)
return False
# Try to extract historical low info
if isinstance(data, dict) and "prices" in data:
for price_info in data["prices"]:
if price_info.get("isLowest"):
for price_entry in data.get("prices", []):
lowest = price_entry.get("lowest")
current = price_entry.get("current")
if lowest and current:
lowest_price = lowest.get("price", {}).get("amount", float("inf"))
current_price = current.get("price", {}).get("amount", float("inf"))
if current_price <= lowest_price:
return True
return False

126
gaming_deals_bot/itad_deals.py Normal file
View File

@@ -0,0 +1,126 @@
"""IsThereAnyDeal deals list — fetch current deals from the ITAD API."""
import logging
from dataclasses import dataclass
import httpx
from .itad import BASE_URL
logger = logging.getLogger(__name__)
# ITAD shop ID for Steam
STEAM_SHOP_ID = 61
@dataclass
class ITADDeal:
game_id: str # ITAD UUID
slug: str
title: str
sale_price: float # current deal price
normal_price: float # regular (non-sale) price
discount: int # percentage off (0-100)
currency: str
shop_name: str
shop_id: int
url: str # purchase redirect URL
is_historical_low: bool
timestamp: str # ISO datetime of deal
expiry: str | None # ISO datetime when deal expires
@property
def dedup_id(self) -> str:
return f"itad-{self.game_id}-{self.shop_id}-{self.discount}"
@property
def sale_price_usd(self) -> str:
return f"{self.sale_price:.2f}"
@property
def normal_price_usd(self) -> str:
return f"{self.normal_price:.2f}"
async def fetch_deals(
client: httpx.AsyncClient,
api_key: str,
*,
max_price: float = 20,
min_discount: float = 50,
limit: int = 20,
) -> list[ITADDeal]:
"""Fetch current deals from IsThereAnyDeal.
Uses GET /deals/v2 with sorting by highest discount.
"""
if not api_key:
return []
params: dict = {
"key": api_key,
"sort": "-cut",
"limit": min(limit, 200),
"nondeals": "false",
}
try:
resp = await client.get(f"{BASE_URL}/deals/v2", params=params)
resp.raise_for_status()
data = resp.json()
except (httpx.HTTPError, ValueError) as exc:
logger.error("ITAD deals API error: %s", exc)
return []
raw_list = data.get("list", [])
logger.info("ITAD returned %d raw deals (before filtering)", len(raw_list))
deals: list[ITADDeal] = []
for entry in raw_list:
deal_data = entry.get("deal", {})
if not deal_data:
continue
cut = deal_data.get("cut", 0)
price_amount = deal_data.get("price", {}).get("amount", 0)
regular_amount = deal_data.get("regular", {}).get("amount", 0)
currency = deal_data.get("price", {}).get("currency", "USD")
title = entry.get("title", "?")
# Apply filters
if cut < min_discount:
logger.debug(
"Filtered out %s: discount %d%% < %d%%", title, cut, int(min_discount)
)
continue
if price_amount > max_price:
logger.debug(
"Filtered out %s: price %.2f > %.2f", title, price_amount, max_price
)
continue
flag = deal_data.get("flag")
is_historical_low = flag in ("H", "N")
shop = deal_data.get("shop", {})
deals.append(
ITADDeal(
game_id=entry.get("id", ""),
slug=entry.get("slug", ""),
title=title,
sale_price=price_amount,
normal_price=regular_amount,
discount=cut,
currency=currency,
shop_name=shop.get("name", "Unknown"),
shop_id=shop.get("id", 0),
url=deal_data.get("url", ""),
is_historical_low=is_historical_low,
timestamp=deal_data.get("timestamp", ""),
expiry=deal_data.get("expiry"),
)
)
logger.info("ITAD returned %d deals after filtering", len(deals))
return deals

33
gaming_deals_bot/preflight.py
View File

@@ -52,7 +52,10 @@ async def run_preflight(config: Config) -> bool:
# --- CheapShark ---
print(f"\n{_BOLD}CheapShark{_RESET}")
all_ok &= await _check_cheapshark(http)
if "cheapshark" in config.deal_sources:
all_ok &= await _check_cheapshark(http)
else:
_skip("Skipped", "not in DEAL_SOURCES")
# --- Epic Games Store ---
print(f"\n{_BOLD}Epic Games Store{_RESET}")
@@ -62,9 +65,10 @@ async def run_preflight(config: Config) -> bool:
print(f"\n{_BOLD}Frankfurter (exchange rates){_RESET}")
all_ok &= await _check_frankfurter(http)
# --- IsThereAnyDeal (optional) ---
# --- IsThereAnyDeal ---
itad_required = "itad" in config.deal_sources
print(f"\n{_BOLD}IsThereAnyDeal{_RESET}")
all_ok &= await _check_itad(http, config.itad_api_key)
all_ok &= await _check_itad(http, config.itad_api_key, required=itad_required)
# --- Summary ---
print()
@@ -164,20 +168,29 @@ async def _check_frankfurter(http: httpx.AsyncClient) -> bool:
return _fail("API reachable", str(exc))
async def _check_itad(http: httpx.AsyncClient, api_key: str) -> bool:
"""Verify the ITAD API key works (optional — skipped if no key is set)."""
async def _check_itad(http: httpx.AsyncClient, api_key: str, *, required: bool = False) -> bool:
"""Verify the ITAD API key works.
When *required* is True (ITAD is a deal source), a missing key is a failure.
Otherwise it's an optional skip.
"""
if not api_key:
if required:
return _fail("API key", "ITAD_API_KEY is required when 'itad' is in DEAL_SOURCES")
return _skip("Skipped", "no ITAD_API_KEY configured (optional)")
try:
# Use a lightweight endpoint to validate the key
# Use the lookup endpoint (GET) to validate the key with a known game
resp = await http.get(
f"{ITAD_URL}/games/overview/v2",
params={"key": api_key, "apps[]": "app/220"}, # Half-Life 2
f"{ITAD_URL}/games/lookup/v1",
params={"key": api_key, "appid": 220}, # Half-Life 2
)
if resp.status_code == 401 or resp.status_code == 403:
if resp.status_code in (401, 403):
return _fail("API key", "rejected by ITAD (401/403)")
resp.raise_for_status()
return _pass("API key valid", "ITAD responded successfully")
data = resp.json()
if data.get("found"):
return _pass("API reachable", "ITAD responded successfully")
return _pass("API reachable", "ITAD key valid (game not found)")
except Exception as exc:
return _fail("API reachable", str(exc))