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

Add IsThereAnyDeal as a deal source alongside CheapShark

Browse Source 
Users can now choose between CheapShark, ITAD, or both via the
DEAL_SOURCES env var (comma-separated, default: "cheapshark"). When
"itad" is included, the bot polls GET /deals/v2 for current deals with
discount/price filtering and posts them with the same formatting style.
ITAD deals include built-in historical low detection via the API's
deal flags. Preflight checks enforce ITAD_API_KEY when ITAD is a
configured deal source.

https://claude.ai/code/session_01B7YPGrE3NatkwadCXVjv2j
This commit is contained in:
Claude
2026-02-28 04:59:27 +00:00
parent 72d86fd4e5
commit 34da126507
7 changed files with 281 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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.

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

20
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,9 +168,15 @@ 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: