Remove Python codebase and clean up gitignore
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
prosolis
@@ -1,113 +0,0 @@
|
||||
"""Entry point — sets up logging, scheduler, and runs the bot."""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import signal
|
||||
import sys
|
||||
|
||||
from apscheduler.schedulers.asyncio import AsyncIOScheduler
|
||||
from dotenv import load_dotenv
|
||||
|
||||
from .bot import DealsBot
|
||||
from .config import Config
|
||||
from .preflight import run_preflight
|
||||
|
||||
|
||||
def setup_logging(*, debug: bool = False):
|
||||
logging.basicConfig(
|
||||
level=logging.DEBUG if debug else logging.INFO,
|
||||
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
|
||||
stream=sys.stdout,
|
||||
)
|
||||
if not debug:
|
||||
# Keep noisy third-party loggers quiet even in debug mode
|
||||
logging.getLogger("httpx").setLevel(logging.WARNING)
|
||||
logging.getLogger("httpcore").setLevel(logging.WARNING)
|
||||
|
||||
|
||||
async def main():
|
||||
load_dotenv()
|
||||
debug_mode = "--debug" in sys.argv
|
||||
setup_logging(debug=debug_mode)
|
||||
logger = logging.getLogger("gaming_deals_bot")
|
||||
|
||||
check_mode = "--check" in sys.argv
|
||||
|
||||
try:
|
||||
config = Config()
|
||||
except ValueError as exc:
|
||||
logger.error("Configuration error: %s", exc)
|
||||
sys.exit(1)
|
||||
|
||||
if check_mode:
|
||||
ok = await run_preflight(config)
|
||||
sys.exit(0 if ok else 1)
|
||||
|
||||
bot = DealsBot(config)
|
||||
await bot.start()
|
||||
|
||||
scheduler = AsyncIOScheduler()
|
||||
|
||||
# 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(
|
||||
bot.check_epic_free_games,
|
||||
"interval",
|
||||
hours=24,
|
||||
id="epic_free",
|
||||
name="Epic free games check",
|
||||
)
|
||||
|
||||
scheduler.start()
|
||||
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
|
||||
stop_event = asyncio.Event()
|
||||
|
||||
def handle_signal():
|
||||
logger.info("Shutdown signal received")
|
||||
stop_event.set()
|
||||
|
||||
loop = asyncio.get_running_loop()
|
||||
for sig in (signal.SIGINT, signal.SIGTERM):
|
||||
loop.add_signal_handler(sig, handle_signal)
|
||||
|
||||
await stop_event.wait()
|
||||
|
||||
logger.info("Shutting down...")
|
||||
scheduler.shutdown(wait=False)
|
||||
await bot.stop()
|
||||
logger.info("Bot stopped")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -1,276 +0,0 @@
|
||||
"""Main bot orchestration — ties together API clients, database, and Matrix posting."""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
|
||||
import httpx
|
||||
|
||||
from .cheapshark import CheapSharkDeal, fetch_deals as fetch_cheapshark_deals
|
||||
from .config import Config
|
||||
from .currency import configure as configure_currency, refresh_rates
|
||||
from .database import Database
|
||||
from .epic import EpicFreeGame, fetch_free_games
|
||||
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
|
||||
from .threads import ThreadCategory, format_thread_root, itad_type_to_category
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class DealsBot:
|
||||
def __init__(self, config: Config):
|
||||
self.config = config
|
||||
self.db = Database(config.database_path)
|
||||
self.matrix = MatrixDealsClient(
|
||||
homeserver_url=config.matrix_homeserver_url,
|
||||
user_id=config.matrix_bot_user_id,
|
||||
access_token=config.matrix_bot_access_token,
|
||||
room_id=config.matrix_deals_room_id,
|
||||
)
|
||||
self._http = httpx.AsyncClient(timeout=30)
|
||||
self._first_run_done = False
|
||||
|
||||
async def start(self):
|
||||
"""Initialize database, fetch exchange rates, and run first-run population."""
|
||||
await self.db.connect()
|
||||
|
||||
# Configure currency display before fetching rates
|
||||
configure_currency(self.config.default_currency, self.config.extra_currencies)
|
||||
|
||||
# Pre-fetch exchange rates so the first deal post has conversions
|
||||
await refresh_rates(self._http)
|
||||
|
||||
first_run = await self.db.get_config("first_run_done")
|
||||
if first_run != "true":
|
||||
logger.info("First run detected — populating database without posting")
|
||||
await self._populate_initial_state()
|
||||
await self.db.set_config("first_run_done", "true")
|
||||
self._first_run_done = True
|
||||
else:
|
||||
self._first_run_done = True
|
||||
|
||||
# Start presence heartbeat so the bot shows as "online"
|
||||
await self.matrix.start_presence_heartbeat()
|
||||
|
||||
async def stop(self):
|
||||
"""Clean shutdown."""
|
||||
await self._http.aclose()
|
||||
await self.matrix.close()
|
||||
await self.db.close()
|
||||
|
||||
async def _populate_initial_state(self):
|
||||
"""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.
|
||||
"""
|
||||
total = 0
|
||||
|
||||
if "cheapshark" in self.config.deal_sources:
|
||||
deals = await fetch_cheapshark_deals(
|
||||
self._http,
|
||||
max_price=self.config.cheapshark_max_price,
|
||||
min_rating=self.config.cheapshark_min_rating,
|
||||
min_discount=self.config.cheapshark_min_discount,
|
||||
)
|
||||
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,
|
||||
countries=self.config.itad_countries,
|
||||
max_price=self.config.itad_max_price,
|
||||
min_discount=self.config.itad_min_discount,
|
||||
limit=self.config.itad_deals_limit,
|
||||
)
|
||||
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."""
|
||||
if not self.config.send_intro_message:
|
||||
return
|
||||
await self.matrix.send_notice("The deals must flow.")
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Thread helpers
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
async def _get_or_create_thread(self, category: ThreadCategory) -> str | None:
|
||||
"""Return the event ID of the thread root for *category*, creating it if needed."""
|
||||
event_id = await self.db.get_thread_root(category.value)
|
||||
if event_id:
|
||||
return event_id
|
||||
|
||||
plain_text, html = format_thread_root(category)
|
||||
event_id = await self.matrix.create_thread_root(plain_text, html)
|
||||
if event_id:
|
||||
await self.db.set_thread_root(category.value, event_id)
|
||||
logger.info("Created thread root for %s: %s", category.value, event_id)
|
||||
return event_id
|
||||
|
||||
async def _send_to_thread_or_room(
|
||||
self, plain_text: str, html: str, category: ThreadCategory
|
||||
) -> bool:
|
||||
"""Send a message — into a thread if threads are enabled, otherwise to the room."""
|
||||
if not self.config.matrix_use_threads:
|
||||
return await self.matrix.send_deal(plain_text, html)
|
||||
|
||||
thread_root_id = await self._get_or_create_thread(category)
|
||||
if not thread_root_id:
|
||||
logger.warning(
|
||||
"Could not obtain thread root for %s — falling back to room", category.value
|
||||
)
|
||||
return await self.matrix.send_deal(plain_text, html)
|
||||
|
||||
return await self.matrix.send_deal_in_thread(plain_text, html, thread_root_id)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# CheapShark
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
async def check_cheapshark(self):
|
||||
"""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_cheapshark_deals(
|
||||
self._http,
|
||||
max_price=self.config.cheapshark_max_price,
|
||||
min_rating=self.config.cheapshark_min_rating,
|
||||
min_discount=self.config.cheapshark_min_discount,
|
||||
)
|
||||
|
||||
for deal in deals:
|
||||
await self._process_deal(deal)
|
||||
|
||||
# Prune old records
|
||||
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):
|
||||
return
|
||||
|
||||
# Check if this is a historical low via ITAD
|
||||
is_historical_low = False
|
||||
if self.config.itad_api_key and deal.steam_app_id:
|
||||
is_historical_low = await check_single_historical_low(
|
||||
self._http,
|
||||
self.config.itad_api_key,
|
||||
deal.steam_app_id,
|
||||
)
|
||||
|
||||
plain_text, html = format_deal(deal, is_historical_low)
|
||||
success = await self._send_to_thread_or_room(
|
||||
plain_text, html, ThreadCategory.GAME_DEALS
|
||||
)
|
||||
|
||||
if success:
|
||||
await self.db.mark_posted(deal.dedup_id, "cheapshark", deal.title)
|
||||
logger.info("Posted deal: %s", deal.title)
|
||||
else:
|
||||
logger.warning("Failed to post deal: %s — will retry next cycle", deal.title)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# IsThereAnyDeal
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
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,
|
||||
countries=self.config.itad_countries,
|
||||
max_price=self.config.itad_max_price,
|
||||
min_discount=self.config.itad_min_discount,
|
||||
limit=self.config.itad_deals_limit,
|
||||
)
|
||||
|
||||
for deal in deals:
|
||||
await self._process_itad_deal(deal)
|
||||
|
||||
await self.db.prune_old(days=30)
|
||||
|
||||
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
|
||||
|
||||
category = itad_type_to_category(deal.deal_type)
|
||||
|
||||
# When threads are off, preserve original behaviour: skip non-game content
|
||||
if not self.config.matrix_use_threads and category == ThreadCategory.NON_GAME_DEALS:
|
||||
logger.debug("Skipping non-game ITAD deal (threads disabled): %s", deal.title)
|
||||
return
|
||||
|
||||
plain_text, html = format_itad_deal(deal)
|
||||
success = await self._send_to_thread_or_room(plain_text, html, category)
|
||||
|
||||
if success:
|
||||
await self.db.mark_posted(deal.dedup_id, "itad", deal.title)
|
||||
logger.info("Posted ITAD deal: %s [%s]", deal.title, category.value)
|
||||
else:
|
||||
logger.warning("Failed to post ITAD deal: %s — will retry next cycle", deal.title)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Epic Games Store
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
async def check_epic_free_games(self):
|
||||
"""Poll Epic Games Store for free games and post new ones."""
|
||||
if not self._first_run_done:
|
||||
return
|
||||
|
||||
logger.info("Checking Epic Games Store for free games...")
|
||||
current_free, upcoming = await fetch_free_games(self._http)
|
||||
|
||||
for game in current_free:
|
||||
await self._process_epic_game(game, is_upcoming=False)
|
||||
|
||||
for game in upcoming:
|
||||
await self._process_epic_game(game, is_upcoming=True)
|
||||
|
||||
async def _process_epic_game(self, game: EpicFreeGame, *, is_upcoming: bool):
|
||||
"""Check dedup, format, and post a single Epic free game."""
|
||||
if await self.db.has_been_posted(game.dedup_id):
|
||||
return
|
||||
|
||||
if is_upcoming:
|
||||
plain_text, html = format_epic_upcoming(game)
|
||||
else:
|
||||
plain_text, html = format_epic_free(game)
|
||||
|
||||
success = await self._send_to_thread_or_room(
|
||||
plain_text, html, ThreadCategory.EPIC_FREE
|
||||
)
|
||||
|
||||
if success:
|
||||
await self.db.mark_posted(game.dedup_id, "epic", game.title)
|
||||
logger.info("Posted Epic game: %s (upcoming=%s)", game.title, is_upcoming)
|
||||
else:
|
||||
logger.warning(
|
||||
"Failed to post Epic game: %s — will retry next cycle", game.title
|
||||
)
|
||||
@@ -1,108 +0,0 @@
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
|
||||
import httpx
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
BASE_URL = "https://www.cheapshark.com/api/1.0"
|
||||
|
||||
# CheapShark store IDs for PC/digital storefronts
|
||||
STORE_IDS = {
|
||||
"1": "Steam",
|
||||
"7": "GOG",
|
||||
"11": "Humble Store",
|
||||
"23": "GreenManGaming",
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class CheapSharkDeal:
|
||||
deal_id: str
|
||||
game_id: str
|
||||
title: str
|
||||
sale_price: str
|
||||
normal_price: str
|
||||
savings: float # percentage off
|
||||
deal_rating: float
|
||||
store_id: str
|
||||
last_change: int # unix timestamp
|
||||
steam_app_id: str | None
|
||||
|
||||
@property
|
||||
def store_name(self) -> str:
|
||||
return STORE_IDS.get(self.store_id, f"Store {self.store_id}")
|
||||
|
||||
@property
|
||||
def deal_url(self) -> str:
|
||||
return f"https://www.cheapshark.com/redirect?dealID={self.deal_id}"
|
||||
|
||||
@property
|
||||
def dedup_id(self) -> str:
|
||||
return f"cheapshark-{self.game_id}-{self.last_change}"
|
||||
|
||||
|
||||
async def fetch_deals(
|
||||
client: httpx.AsyncClient,
|
||||
*,
|
||||
max_price: float = 20,
|
||||
min_rating: float = 8.0,
|
||||
min_discount: float = 50,
|
||||
page_size: int = 10,
|
||||
) -> list[CheapSharkDeal]:
|
||||
"""Fetch top deals from CheapShark across configured stores."""
|
||||
store_ids = ",".join(STORE_IDS.keys())
|
||||
params = {
|
||||
"storeID": store_ids,
|
||||
"upperPrice": str(int(max_price)),
|
||||
"sortBy": "recent",
|
||||
"desc": "1",
|
||||
"pageSize": str(page_size),
|
||||
}
|
||||
|
||||
try:
|
||||
resp = await client.get(f"{BASE_URL}/deals", params=params)
|
||||
resp.raise_for_status()
|
||||
raw_deals = resp.json()
|
||||
except (httpx.HTTPError, ValueError) as exc:
|
||||
logger.error("CheapShark API error: %s", exc)
|
||||
return []
|
||||
|
||||
logger.info(
|
||||
"CheapShark returned %d raw deals (before filtering)", len(raw_deals),
|
||||
)
|
||||
|
||||
deals = []
|
||||
for d in raw_deals:
|
||||
savings = float(d.get("savings", 0))
|
||||
rating = float(d.get("dealRating", 0))
|
||||
title = d.get("title", "?")
|
||||
|
||||
if savings < min_discount:
|
||||
logger.debug("Filtered out %s: savings %.1f%% < %.1f%%", title, savings, min_discount)
|
||||
continue
|
||||
if rating > 0 and rating < min_rating:
|
||||
logger.debug("Filtered out %s: rating %.1f < %.1f", title, rating, min_rating)
|
||||
continue
|
||||
|
||||
steam_app_id = d.get("steamAppID") or None
|
||||
if steam_app_id == "0":
|
||||
steam_app_id = None
|
||||
|
||||
deals.append(
|
||||
CheapSharkDeal(
|
||||
deal_id=d["dealID"],
|
||||
game_id=d["gameID"],
|
||||
title=d["title"],
|
||||
sale_price=d["salePrice"],
|
||||
normal_price=d["normalPrice"],
|
||||
savings=savings,
|
||||
deal_rating=rating,
|
||||
store_id=d["storeID"],
|
||||
last_change=int(d.get("lastChange", 0)),
|
||||
steam_app_id=steam_app_id,
|
||||
)
|
||||
)
|
||||
|
||||
logger.info("CheapShark returned %d deals after filtering", len(deals))
|
||||
return deals
|
||||
@@ -1,83 +0,0 @@
|
||||
import os
|
||||
|
||||
|
||||
class Config:
|
||||
"""Bot configuration loaded from environment variables."""
|
||||
|
||||
def __init__(self):
|
||||
# Matrix settings
|
||||
self.matrix_homeserver_url = self._require("MATRIX_HOMESERVER_URL")
|
||||
self.matrix_bot_user_id = self._require("MATRIX_BOT_USER_ID")
|
||||
self.matrix_bot_access_token = self._require("MATRIX_BOT_ACCESS_TOKEN")
|
||||
self.matrix_deals_room_id = self._require("MATRIX_DEALS_ROOM_ID")
|
||||
|
||||
# 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()
|
||||
]
|
||||
|
||||
# ITAD countries: comma-separated ISO 3166-1 alpha-2 country codes
|
||||
# Deals are fetched for each country and merged (first country has priority
|
||||
# when the same game appears in multiple regions).
|
||||
raw_countries = os.environ.get("ITAD_COUNTRIES", "US")
|
||||
self.itad_countries: list[str] = [
|
||||
c.strip().upper() for c in raw_countries.split(",") if c.strip()
|
||||
]
|
||||
|
||||
# Currency display
|
||||
self.default_currency = os.environ.get("DEFAULT_CURRENCY", "USD").upper()
|
||||
raw_extra = os.environ.get("EXTRA_CURRENCIES", "CAD,EUR,GBP")
|
||||
self.extra_currencies: list[str] = [
|
||||
c.strip().upper() for c in raw_extra.split(",") if c.strip()
|
||||
]
|
||||
|
||||
# Shared filter fallbacks (support legacy env vars)
|
||||
_max_price = os.environ.get(
|
||||
"MAX_PRICE", os.environ.get("MAX_PRICE_USD", "20")
|
||||
)
|
||||
_min_discount = os.environ.get("MIN_DISCOUNT_PERCENT", "50")
|
||||
_min_rating = os.environ.get("MIN_DEAL_RATING", "8.0")
|
||||
|
||||
# CheapShark-specific filtering (falls back to shared values)
|
||||
self.cheapshark_min_discount = int(
|
||||
os.environ.get("CHEAPSHARK_MIN_DISCOUNT", _min_discount)
|
||||
)
|
||||
self.cheapshark_min_rating = float(
|
||||
os.environ.get("CHEAPSHARK_MIN_RATING", _min_rating)
|
||||
)
|
||||
self.cheapshark_max_price = float(
|
||||
os.environ.get("CHEAPSHARK_MAX_PRICE", _max_price)
|
||||
)
|
||||
|
||||
# ITAD-specific filtering (falls back to shared values)
|
||||
self.itad_min_discount = int(
|
||||
os.environ.get("ITAD_MIN_DISCOUNT", _min_discount)
|
||||
)
|
||||
self.itad_max_price = float(
|
||||
os.environ.get("ITAD_MAX_PRICE", _max_price)
|
||||
)
|
||||
self.itad_deals_limit = int(os.environ.get("ITAD_DEALS_LIMIT", "200"))
|
||||
|
||||
# Matrix threads — post deals into per-category threads
|
||||
self.matrix_use_threads = os.environ.get(
|
||||
"MATRIX_USE_THREADS", "false"
|
||||
).lower() in ("true", "1", "yes")
|
||||
|
||||
# Intro message on startup
|
||||
self.send_intro_message = os.environ.get(
|
||||
"SEND_INTRO_MESSAGE", "false"
|
||||
).lower() in ("true", "1", "yes")
|
||||
|
||||
# Database
|
||||
self.database_path = os.environ.get("DATABASE_PATH", "deals.db")
|
||||
|
||||
@staticmethod
|
||||
def _require(name: str) -> str:
|
||||
value = os.environ.get(name)
|
||||
if not value:
|
||||
raise ValueError(f"Required environment variable {name} is not set")
|
||||
return value
|
||||
@@ -1,189 +0,0 @@
|
||||
"""Currency conversion using the Frankfurter API (ECB rates, no API key required).
|
||||
|
||||
Rates are cached in memory and refreshed at most twice per day.
|
||||
Call ``configure()`` before ``refresh_rates()`` to set the display currencies.
|
||||
"""
|
||||
|
||||
import logging
|
||||
import time
|
||||
|
||||
import httpx
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
FRANKFURTER_URL = "https://api.frankfurter.dev/v1/latest"
|
||||
CACHE_TTL_SECONDS = 43200 # 12 hours — be nice to the free service
|
||||
|
||||
# Module-level state -------------------------------------------------------
|
||||
_rates: dict[str, float] = {}
|
||||
_last_fetched: float = 0.0
|
||||
|
||||
# Display preferences (set via configure())
|
||||
_default_currency: str = "USD"
|
||||
_extra_currencies: list[str] = ["CAD", "EUR", "GBP"]
|
||||
|
||||
# Currency display symbols
|
||||
SYMBOLS: dict[str, str] = {
|
||||
"USD": "$",
|
||||
"CAD": "C$",
|
||||
"EUR": "€",
|
||||
"GBP": "£",
|
||||
"AUD": "A$",
|
||||
"BRL": "R$",
|
||||
"CHF": "CHF ",
|
||||
"CNY": "¥",
|
||||
"CZK": "Kč",
|
||||
"DKK": "kr",
|
||||
"HUF": "Ft",
|
||||
"INR": "₹",
|
||||
"JPY": "¥",
|
||||
"KRW": "₩",
|
||||
"MXN": "MX$",
|
||||
"NOK": "kr",
|
||||
"NZD": "NZ$",
|
||||
"PLN": "zł",
|
||||
"SEK": "kr",
|
||||
"TRY": "₺",
|
||||
"ZAR": "R",
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Configuration
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def configure(default_currency: str = "USD", extra_currencies: list[str] | None = None) -> None:
|
||||
"""Set the primary display currency and additional currencies.
|
||||
|
||||
Must be called **before** ``refresh_rates()`` so the correct symbols are
|
||||
requested from Frankfurter.
|
||||
"""
|
||||
global _default_currency, _extra_currencies
|
||||
_default_currency = default_currency.upper()
|
||||
if extra_currencies is not None:
|
||||
_extra_currencies = [c.upper() for c in extra_currencies]
|
||||
|
||||
|
||||
def _all_target_currencies() -> list[str]:
|
||||
"""Return every non-USD currency we need exchange rates for."""
|
||||
currencies: set[str] = set()
|
||||
if _default_currency != "USD":
|
||||
currencies.add(_default_currency)
|
||||
for c in _extra_currencies:
|
||||
if c != "USD":
|
||||
currencies.add(c)
|
||||
return sorted(currencies)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Rate fetching
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
async def refresh_rates(client: httpx.AsyncClient) -> bool:
|
||||
"""Fetch latest USD-based exchange rates from Frankfurter.
|
||||
|
||||
Returns True if rates were successfully updated.
|
||||
"""
|
||||
global _rates, _last_fetched
|
||||
|
||||
needed = _all_target_currencies()
|
||||
if not needed:
|
||||
# Only USD display — no conversion needed
|
||||
_last_fetched = time.monotonic()
|
||||
return True
|
||||
|
||||
symbols = ",".join(needed)
|
||||
try:
|
||||
resp = await client.get(
|
||||
FRANKFURTER_URL,
|
||||
params={"base": "USD", "symbols": symbols},
|
||||
)
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
except (httpx.HTTPError, ValueError) as exc:
|
||||
logger.warning("Failed to fetch exchange rates: %s", exc)
|
||||
return False
|
||||
|
||||
new_rates = data.get("rates", {})
|
||||
if not new_rates:
|
||||
logger.warning("Frankfurter returned empty rates")
|
||||
return False
|
||||
|
||||
_rates = {k: float(v) for k, v in new_rates.items()}
|
||||
_last_fetched = time.monotonic()
|
||||
logger.info("Exchange rates updated: %s", _rates)
|
||||
return True
|
||||
|
||||
|
||||
async def _ensure_rates(client: httpx.AsyncClient) -> None:
|
||||
"""Refresh rates if the cache is stale or empty."""
|
||||
if _all_target_currencies() and (
|
||||
not _rates or (time.monotonic() - _last_fetched) > CACHE_TTL_SECONDS
|
||||
):
|
||||
await refresh_rates(client)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Conversion helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _convert_from_usd(usd_amount: float, currency: str) -> float | None:
|
||||
"""Convert a USD amount to *currency* using cached rates."""
|
||||
if currency == "USD":
|
||||
return usd_amount
|
||||
rate = _rates.get(currency)
|
||||
if rate is None:
|
||||
return None
|
||||
return round(usd_amount * rate, 2)
|
||||
|
||||
|
||||
def convert_to_usd(amount: float, source_currency: str) -> float:
|
||||
"""Convert *amount* from *source_currency* to USD using cached rates.
|
||||
|
||||
Falls back to a 1:1 conversion with a warning when rates are unavailable.
|
||||
"""
|
||||
if source_currency == "USD":
|
||||
return amount
|
||||
rate = _rates.get(source_currency)
|
||||
if rate is None or rate == 0:
|
||||
logger.warning("No rate for %s → USD, assuming 1:1", source_currency)
|
||||
return amount
|
||||
return round(amount / rate, 2)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Price formatting
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def format_price(usd_amount: float) -> str:
|
||||
"""Return a formatted multi-currency price string.
|
||||
|
||||
The configured *default_currency* is shown first, followed by each
|
||||
*extra_currency*. Example with defaults::
|
||||
|
||||
$14.99 · C$20.54 · €13.78 · £11.98
|
||||
|
||||
Falls back to USD-only if rates aren't available.
|
||||
"""
|
||||
display_order = [_default_currency] + [
|
||||
c for c in _extra_currencies if c != _default_currency
|
||||
]
|
||||
|
||||
parts: list[str] = []
|
||||
for cur in display_order:
|
||||
converted = _convert_from_usd(usd_amount, cur)
|
||||
if converted is not None:
|
||||
symbol = SYMBOLS.get(cur, f"{cur} ")
|
||||
parts.append(f"{symbol}{converted:.2f}")
|
||||
|
||||
if not parts:
|
||||
# Fallback when no rates are loaded yet
|
||||
parts = [f"${usd_amount:.2f}"]
|
||||
|
||||
return " · ".join(parts)
|
||||
|
||||
|
||||
async def format_price_async(client: httpx.AsyncClient, usd_amount: float) -> str:
|
||||
"""Ensure rates are fresh, then format a multi-currency price string."""
|
||||
await _ensure_rates(client)
|
||||
return format_price(usd_amount)
|
||||
@@ -1,115 +0,0 @@
|
||||
import logging
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
import aiosqlite
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
SCHEMA = """
|
||||
CREATE TABLE IF NOT EXISTS posted_deals (
|
||||
id TEXT PRIMARY KEY,
|
||||
source TEXT NOT NULL,
|
||||
title TEXT NOT NULL,
|
||||
posted_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS config (
|
||||
key TEXT PRIMARY KEY,
|
||||
value TEXT
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS thread_roots (
|
||||
category TEXT PRIMARY KEY,
|
||||
event_id TEXT NOT NULL,
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||||
);
|
||||
"""
|
||||
|
||||
|
||||
class Database:
|
||||
def __init__(self, path: str):
|
||||
self.path = path
|
||||
self._db: aiosqlite.Connection | None = None
|
||||
|
||||
async def connect(self):
|
||||
self._db = await aiosqlite.connect(self.path)
|
||||
await self._db.executescript(SCHEMA)
|
||||
await self._db.commit()
|
||||
logger.info("Database initialized at %s", self.path)
|
||||
|
||||
async def close(self):
|
||||
if self._db:
|
||||
await self._db.close()
|
||||
|
||||
async def has_been_posted(self, deal_id: str) -> bool:
|
||||
assert self._db is not None
|
||||
cursor = await self._db.execute(
|
||||
"SELECT 1 FROM posted_deals WHERE id = ?", (deal_id,)
|
||||
)
|
||||
row = await cursor.fetchone()
|
||||
return row is not None
|
||||
|
||||
async def mark_posted(self, deal_id: str, source: str, title: str):
|
||||
assert self._db is not None
|
||||
await self._db.execute(
|
||||
"INSERT OR IGNORE INTO posted_deals (id, source, title) VALUES (?, ?, ?)",
|
||||
(deal_id, source, title),
|
||||
)
|
||||
await self._db.commit()
|
||||
|
||||
async def prune_old(self, days: int = 30):
|
||||
assert self._db is not None
|
||||
cutoff = datetime.now(timezone.utc) - timedelta(days=days)
|
||||
result = await self._db.execute(
|
||||
"DELETE FROM posted_deals WHERE posted_at < ?",
|
||||
(cutoff.isoformat(),),
|
||||
)
|
||||
await self._db.commit()
|
||||
if result.rowcount:
|
||||
logger.info("Pruned %d old deal records", result.rowcount)
|
||||
|
||||
async def get_config(self, key: str) -> str | None:
|
||||
assert self._db is not None
|
||||
cursor = await self._db.execute(
|
||||
"SELECT value FROM config WHERE key = ?", (key,)
|
||||
)
|
||||
row = await cursor.fetchone()
|
||||
return row[0] if row else None
|
||||
|
||||
async def set_config(self, key: str, value: str):
|
||||
assert self._db is not None
|
||||
await self._db.execute(
|
||||
"INSERT OR REPLACE INTO config (key, value) VALUES (?, ?)",
|
||||
(key, value),
|
||||
)
|
||||
await self._db.commit()
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Thread root management
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
async def get_thread_root(self, category: str) -> str | None:
|
||||
"""Return the Matrix event ID for a thread root, or None."""
|
||||
assert self._db is not None
|
||||
cursor = await self._db.execute(
|
||||
"SELECT event_id FROM thread_roots WHERE category = ?", (category,)
|
||||
)
|
||||
row = await cursor.fetchone()
|
||||
return row[0] if row else None
|
||||
|
||||
async def set_thread_root(self, category: str, event_id: str):
|
||||
"""Store the Matrix event ID for a thread root."""
|
||||
assert self._db is not None
|
||||
await self._db.execute(
|
||||
"INSERT OR REPLACE INTO thread_roots (category, event_id) VALUES (?, ?)",
|
||||
(category, event_id),
|
||||
)
|
||||
await self._db.commit()
|
||||
|
||||
async def clear_thread_root(self, category: str):
|
||||
"""Remove a stored thread root (e.g. if the event was redacted)."""
|
||||
assert self._db is not None
|
||||
await self._db.execute(
|
||||
"DELETE FROM thread_roots WHERE category = ?", (category,)
|
||||
)
|
||||
await self._db.commit()
|
||||
@@ -1,135 +0,0 @@
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import httpx
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
FREE_GAMES_URL = (
|
||||
"https://store-site-backend-static.ak.epicgames.com/freeGamesPromotions"
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class EpicFreeGame:
|
||||
title: str
|
||||
game_id: str
|
||||
description: str
|
||||
end_date: str | None # ISO datetime when the offer expires
|
||||
url: str
|
||||
|
||||
@property
|
||||
def dedup_id(self) -> str:
|
||||
return f"epic-{self.game_id}"
|
||||
|
||||
|
||||
def _parse_promotions(
|
||||
elements: list[dict],
|
||||
) -> tuple[list[EpicFreeGame], list[EpicFreeGame]]:
|
||||
"""Parse Epic promotions into current free games and upcoming free games."""
|
||||
current: list[EpicFreeGame] = []
|
||||
upcoming: list[EpicFreeGame] = []
|
||||
now = datetime.now(timezone.utc)
|
||||
|
||||
for elem in elements:
|
||||
title = elem.get("title", "Unknown")
|
||||
game_id = elem.get("id", "")
|
||||
description = elem.get("description", "")
|
||||
|
||||
# Build the store URL from the slug or product slug
|
||||
slug = (
|
||||
elem.get("productSlug")
|
||||
or elem.get("urlSlug")
|
||||
or elem.get("catalogNs", {}).get("mappings", [{}])[0].get("pageSlug", "")
|
||||
)
|
||||
url = f"https://store.epicgames.com/en-US/p/{slug}" if slug else ""
|
||||
|
||||
# Check current promotional offers
|
||||
promotions = elem.get("promotions")
|
||||
if not promotions:
|
||||
continue
|
||||
|
||||
# Current offers
|
||||
for offer_group in promotions.get("promotionalOffers", []):
|
||||
for offer in offer_group.get("promotionalOffers", []):
|
||||
discount = offer.get("discountSetting", {})
|
||||
if discount.get("discountPercentage", 100) == 0:
|
||||
end_date = offer.get("endDate")
|
||||
# Verify the offer is currently active
|
||||
start = offer.get("startDate")
|
||||
if start:
|
||||
start_dt = datetime.fromisoformat(
|
||||
start.replace("Z", "+00:00")
|
||||
)
|
||||
if start_dt > now:
|
||||
continue
|
||||
if end_date:
|
||||
end_dt = datetime.fromisoformat(
|
||||
end_date.replace("Z", "+00:00")
|
||||
)
|
||||
if end_dt < now:
|
||||
continue
|
||||
|
||||
current.append(
|
||||
EpicFreeGame(
|
||||
title=title,
|
||||
game_id=game_id,
|
||||
description=description,
|
||||
end_date=end_date,
|
||||
url=url,
|
||||
)
|
||||
)
|
||||
|
||||
# Upcoming offers
|
||||
for offer_group in promotions.get("upcomingPromotionalOffers", []):
|
||||
for offer in offer_group.get("promotionalOffers", []):
|
||||
discount = offer.get("discountSetting", {})
|
||||
if discount.get("discountPercentage", 100) == 0:
|
||||
end_date = offer.get("endDate")
|
||||
upcoming.append(
|
||||
EpicFreeGame(
|
||||
title=title,
|
||||
game_id=game_id,
|
||||
description=description,
|
||||
end_date=end_date,
|
||||
url=url,
|
||||
)
|
||||
)
|
||||
|
||||
return current, upcoming
|
||||
|
||||
|
||||
async def fetch_free_games(
|
||||
client: httpx.AsyncClient,
|
||||
) -> tuple[list[EpicFreeGame], list[EpicFreeGame]]:
|
||||
"""Fetch current and upcoming free games from Epic Games Store.
|
||||
|
||||
Returns (current_free_games, upcoming_free_games).
|
||||
"""
|
||||
try:
|
||||
resp = await client.get(FREE_GAMES_URL, params={"locale": "en-US"})
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
except (httpx.HTTPError, ValueError) as exc:
|
||||
logger.error("Epic Games Store API error: %s", exc)
|
||||
return [], []
|
||||
|
||||
elements = (
|
||||
data.get("data", {})
|
||||
.get("Catalog", {})
|
||||
.get("searchStore", {})
|
||||
.get("elements", [])
|
||||
)
|
||||
|
||||
if not elements:
|
||||
logger.warning("No elements found in Epic free games response")
|
||||
return [], []
|
||||
|
||||
current, upcoming = _parse_promotions(elements)
|
||||
logger.info(
|
||||
"Epic: %d current free games, %d upcoming",
|
||||
len(current),
|
||||
len(upcoming),
|
||||
)
|
||||
return current, upcoming
|
||||
@@ -1,156 +0,0 @@
|
||||
"""Message formatting for Matrix deal posts.
|
||||
|
||||
Produces both a plain-text fallback and HTML formatted body for Matrix messages.
|
||||
HTML is built directly using the Matrix-supported HTML subset rather than going
|
||||
through a Markdown-to-HTML converter, which collapses line breaks.
|
||||
"""
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from html import escape
|
||||
|
||||
from .cheapshark import CheapSharkDeal
|
||||
from .currency import format_price
|
||||
from .epic import EpicFreeGame
|
||||
from .itad_deals import ITADDeal
|
||||
|
||||
|
||||
def format_deal(deal: CheapSharkDeal, is_historical_low: bool = False) -> tuple[str, str]:
|
||||
"""Format a CheapShark deal into (plain_text, html) for Matrix.
|
||||
|
||||
Returns (body, formatted_body).
|
||||
"""
|
||||
discount = int(deal.savings)
|
||||
sale_price = float(deal.sale_price)
|
||||
normal_price = float(deal.normal_price)
|
||||
|
||||
sale_multi = format_price(sale_price)
|
||||
normal_display = format_price(normal_price)
|
||||
title = escape(deal.title)
|
||||
|
||||
html_lines = [
|
||||
f"<strong>🎮 [DEAL] {title}</strong>",
|
||||
f"{discount}% off on {escape(deal.store_name)} <del>{escape(normal_display)}</del>",
|
||||
f"💰 <strong>{escape(sale_multi)}</strong>",
|
||||
]
|
||||
if is_historical_low:
|
||||
html_lines.append("🏆 <em>All-time low!</em>")
|
||||
html_lines.append(f'🔗 <a href="{escape(deal.deal_url)}">View Deal</a>')
|
||||
html = "<br>\n".join(html_lines)
|
||||
|
||||
plain_lines = [
|
||||
f"🎮 [DEAL] {deal.title}",
|
||||
f" {discount}% off on {deal.store_name} (was {normal_display})",
|
||||
f" 💰 {sale_multi}",
|
||||
]
|
||||
if is_historical_low:
|
||||
plain_lines.append(" 🏆 All-time low!")
|
||||
plain_lines.append(f" 🔗 {deal.deal_url}")
|
||||
plain_text = "\n".join(plain_lines)
|
||||
|
||||
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)
|
||||
title = escape(deal.title)
|
||||
|
||||
html_lines = [
|
||||
f"<strong>🎮 [DEAL] {title}</strong>",
|
||||
f"{deal.discount}% off on {escape(deal.shop_name)} <del>{escape(normal_display)}</del>",
|
||||
f"💰 <strong>{escape(sale_multi)}</strong>",
|
||||
]
|
||||
if deal.is_historical_low:
|
||||
html_lines.append("🏆 <em>All-time low!</em>")
|
||||
html_lines.append(f'🔗 <a href="{escape(deal.url)}">View Deal</a>')
|
||||
html = "<br>\n".join(html_lines)
|
||||
|
||||
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.
|
||||
|
||||
Returns (body, formatted_body).
|
||||
"""
|
||||
title = escape(game.title)
|
||||
|
||||
html_lines = [
|
||||
f"<strong>🆓 [FREE] {title}</strong>",
|
||||
"Free on Epic Games Store",
|
||||
]
|
||||
|
||||
if game.end_date:
|
||||
try:
|
||||
end_dt = datetime.fromisoformat(game.end_date.replace("Z", "+00:00"))
|
||||
date_str = end_dt.strftime("%B %-d")
|
||||
html_lines.append(f"📅 <em>Free until {date_str}</em>")
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
|
||||
if game.url:
|
||||
html_lines.append(f'🔗 <a href="{escape(game.url)}">Claim Now</a>')
|
||||
html = "<br>\n".join(html_lines)
|
||||
|
||||
plain_lines = [
|
||||
f"🆓 [FREE] {game.title}",
|
||||
" Free on Epic Games Store",
|
||||
]
|
||||
if game.end_date:
|
||||
try:
|
||||
end_dt = datetime.fromisoformat(game.end_date.replace("Z", "+00:00"))
|
||||
date_str = end_dt.strftime("%B %-d")
|
||||
plain_lines.append(f" 📅 Free until {date_str}")
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
if game.url:
|
||||
plain_lines.append(f" 🔗 {game.url}")
|
||||
plain_text = "\n".join(plain_lines)
|
||||
|
||||
return plain_text, html
|
||||
|
||||
|
||||
def format_epic_upcoming(game: EpicFreeGame) -> tuple[str, str]:
|
||||
"""Format an upcoming Epic free game into (plain_text, html) for Matrix."""
|
||||
title = escape(game.title)
|
||||
|
||||
html_lines = [
|
||||
f"<strong>📢 [UPCOMING FREE] {title}</strong>",
|
||||
"Coming soon — Free on Epic Games Store",
|
||||
]
|
||||
|
||||
if game.end_date:
|
||||
try:
|
||||
end_dt = datetime.fromisoformat(game.end_date.replace("Z", "+00:00"))
|
||||
date_str = end_dt.strftime("%B %-d")
|
||||
html_lines.append(f"📅 <em>Free until {date_str}</em>")
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
|
||||
if game.url:
|
||||
html_lines.append(f'🔗 <a href="{escape(game.url)}">Store Page</a>')
|
||||
html = "<br>\n".join(html_lines)
|
||||
|
||||
plain_lines = [
|
||||
f"📢 [UPCOMING FREE] {game.title}",
|
||||
" Coming soon — Free on Epic Games Store",
|
||||
]
|
||||
if game.url:
|
||||
plain_lines.append(f" 🔗 {game.url}")
|
||||
plain_text = "\n".join(plain_lines)
|
||||
|
||||
return plain_text, html
|
||||
@@ -1,122 +0,0 @@
|
||||
import logging
|
||||
|
||||
import httpx
|
||||
|
||||
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,
|
||||
steam_app_ids: list[str],
|
||||
) -> dict[str, bool]:
|
||||
"""Check if current prices are historical lows for a batch of Steam app IDs.
|
||||
|
||||
Returns a mapping of steam_app_id -> is_historical_low.
|
||||
"""
|
||||
if not api_key or not steam_app_ids:
|
||||
return {}
|
||||
|
||||
# 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.post(
|
||||
f"{BASE_URL}/games/overview/v2",
|
||||
params={"key": api_key, "shops": [61]},
|
||||
json=list(itad_id_to_steam.keys()),
|
||||
)
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
except (httpx.HTTPError, ValueError) as exc:
|
||||
logger.error("ITAD API error: %s", exc)
|
||||
return {}
|
||||
|
||||
results: dict[str, bool] = {}
|
||||
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",
|
||||
len(steam_app_ids),
|
||||
sum(results.values()),
|
||||
)
|
||||
return results
|
||||
|
||||
|
||||
async def check_single_historical_low(
|
||||
client: httpx.AsyncClient,
|
||||
api_key: str,
|
||||
steam_app_id: str,
|
||||
) -> bool:
|
||||
"""Check if a single game is at its historical low price."""
|
||||
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.post(
|
||||
f"{BASE_URL}/games/overview/v2",
|
||||
params={"key": api_key},
|
||||
json=[itad_id],
|
||||
)
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
except (httpx.HTTPError, ValueError) as exc:
|
||||
logger.error("ITAD API error for app %s: %s", steam_app_id, exc)
|
||||
return False
|
||||
|
||||
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
|
||||
@@ -1,198 +0,0 @@
|
||||
"""IsThereAnyDeal deals list — fetch current deals from the ITAD API."""
|
||||
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
|
||||
import httpx
|
||||
|
||||
from .currency import convert_to_usd
|
||||
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
|
||||
deal_type: str # ITAD type: "game", "dlc", or other
|
||||
sale_price: float # current deal price (normalised to USD)
|
||||
normal_price: float # regular (non-sale) price (normalised to USD)
|
||||
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,
|
||||
*,
|
||||
countries: list[str] | None = None,
|
||||
max_price: float = 20,
|
||||
min_discount: float = 50,
|
||||
limit: int = 200,
|
||||
) -> list[ITADDeal]:
|
||||
"""Fetch current deals from IsThereAnyDeal across one or more countries.
|
||||
|
||||
Defaults to the API maximum of 200 deals per country so that client-side
|
||||
filtering (discount, price, type) has the widest possible pool to work
|
||||
with — this avoids missing deals that wouldn't appear in a smaller window
|
||||
sorted only by discount.
|
||||
|
||||
Deals are fetched per-country, merged (first country in the list wins
|
||||
when the same game+shop appears in multiple regions), and finally sorted
|
||||
by timestamp so that the newest deals appear first.
|
||||
"""
|
||||
if not api_key:
|
||||
return []
|
||||
|
||||
if countries is None:
|
||||
countries = ["US"]
|
||||
|
||||
seen: set[str] = set() # game_id-shop_id keys for cross-country dedup
|
||||
all_deals: list[ITADDeal] = []
|
||||
|
||||
for country in countries:
|
||||
country_deals = await _fetch_country_deals(
|
||||
client,
|
||||
api_key,
|
||||
country=country,
|
||||
max_price=max_price,
|
||||
min_discount=min_discount,
|
||||
limit=limit,
|
||||
)
|
||||
for deal in country_deals:
|
||||
key = f"{deal.game_id}-{deal.shop_id}"
|
||||
if key not in seen:
|
||||
seen.add(key)
|
||||
all_deals.append(deal)
|
||||
else:
|
||||
logger.debug(
|
||||
"Skipping duplicate %s from country %s", deal.title, country
|
||||
)
|
||||
|
||||
# Sort newest first — the API only supports sorting by discount or price,
|
||||
# so we sort client-side by the deal timestamp.
|
||||
all_deals.sort(key=lambda d: d.timestamp, reverse=True)
|
||||
|
||||
logger.info(
|
||||
"ITAD: %d deals across %d country/countries after dedup",
|
||||
len(all_deals),
|
||||
len(countries),
|
||||
)
|
||||
return all_deals
|
||||
|
||||
|
||||
async def _fetch_country_deals(
|
||||
client: httpx.AsyncClient,
|
||||
api_key: str,
|
||||
*,
|
||||
country: str = "US",
|
||||
max_price: float = 20,
|
||||
min_discount: float = 50,
|
||||
limit: int = 200,
|
||||
) -> list[ITADDeal]:
|
||||
"""Fetch deals for a single country from the ITAD ``/deals/v2`` endpoint."""
|
||||
params: dict = {
|
||||
"key": api_key,
|
||||
"country": country,
|
||||
"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 for country %s: %s", country, exc)
|
||||
return []
|
||||
|
||||
raw_list = data.get("list", [])
|
||||
logger.info(
|
||||
"ITAD returned %d raw deals for %s (before filtering)", len(raw_list), country
|
||||
)
|
||||
|
||||
deals: list[ITADDeal] = []
|
||||
for entry in raw_list:
|
||||
deal_data = entry.get("deal", {})
|
||||
if not deal_data:
|
||||
continue
|
||||
|
||||
entry_type = entry.get("type", "game")
|
||||
title = entry.get("title", "?")
|
||||
|
||||
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")
|
||||
|
||||
# Normalise to USD so prices are comparable across regions
|
||||
price_usd = convert_to_usd(price_amount, currency)
|
||||
regular_usd = convert_to_usd(regular_amount, currency)
|
||||
|
||||
# Apply ITAD-specific filters
|
||||
if cut < min_discount:
|
||||
logger.debug(
|
||||
"Filtered out %s: discount %d%% < %d%%",
|
||||
title,
|
||||
cut,
|
||||
int(min_discount),
|
||||
)
|
||||
continue
|
||||
if price_usd > max_price:
|
||||
logger.debug(
|
||||
"Filtered out %s: price $%.2f > $%.2f", title, price_usd, 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,
|
||||
deal_type=entry_type,
|
||||
sale_price=price_usd,
|
||||
normal_price=regular_usd,
|
||||
discount=cut,
|
||||
currency="USD", # normalised
|
||||
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 for %s after filtering", len(deals), country
|
||||
)
|
||||
return deals
|
||||
@@ -1,184 +0,0 @@
|
||||
"""Matrix client wrapper for sending deal messages."""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
|
||||
from nio import AsyncClient, PresenceSetError, RoomSendError, RoomSendResponse
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
PRESENCE_HEARTBEAT_INTERVAL = 60 # seconds
|
||||
|
||||
|
||||
class MatrixDealsClient:
|
||||
def __init__(
|
||||
self,
|
||||
homeserver_url: str,
|
||||
user_id: str,
|
||||
access_token: str,
|
||||
room_id: str,
|
||||
):
|
||||
self.room_id = room_id
|
||||
self._client = AsyncClient(homeserver_url, user_id)
|
||||
self._client.access_token = access_token
|
||||
self._client.user_id = user_id
|
||||
self._heartbeat_task: asyncio.Task | None = None
|
||||
|
||||
async def send_deal(self, plain_text: str, html: str) -> bool:
|
||||
"""Send a deal message to the configured room.
|
||||
|
||||
Returns True on success, False on failure.
|
||||
"""
|
||||
content = {
|
||||
"msgtype": "m.text",
|
||||
"format": "org.matrix.custom.html",
|
||||
"body": plain_text,
|
||||
"formatted_body": html,
|
||||
}
|
||||
|
||||
try:
|
||||
resp = await self._client.room_send(
|
||||
room_id=self.room_id,
|
||||
message_type="m.room.message",
|
||||
content=content,
|
||||
)
|
||||
if isinstance(resp, RoomSendError):
|
||||
logger.error("Failed to send message: %s", resp.message)
|
||||
return False
|
||||
logger.debug("Message sent: %s", resp.event_id)
|
||||
return True
|
||||
except Exception as exc:
|
||||
logger.error("Matrix send error: %s", exc)
|
||||
return False
|
||||
|
||||
async def send_notice(self, text: str) -> bool:
|
||||
"""Send a plain m.notice (non-highlight) message to the configured room."""
|
||||
content = {
|
||||
"msgtype": "m.notice",
|
||||
"body": text,
|
||||
}
|
||||
|
||||
try:
|
||||
resp = await self._client.room_send(
|
||||
room_id=self.room_id,
|
||||
message_type="m.room.message",
|
||||
content=content,
|
||||
)
|
||||
if isinstance(resp, RoomSendError):
|
||||
logger.error("Failed to send notice: %s", resp.message)
|
||||
return False
|
||||
return True
|
||||
except Exception as exc:
|
||||
logger.error("Matrix send error: %s", exc)
|
||||
return False
|
||||
|
||||
async def send_deal_in_thread(
|
||||
self, plain_text: str, html: str, thread_root_id: str
|
||||
) -> bool:
|
||||
"""Send a deal message as a reply inside a Matrix thread.
|
||||
|
||||
``thread_root_id`` is the ``$event_id`` of the thread root message.
|
||||
"""
|
||||
content = {
|
||||
"msgtype": "m.text",
|
||||
"format": "org.matrix.custom.html",
|
||||
"body": plain_text,
|
||||
"formatted_body": html,
|
||||
"m.relates_to": {
|
||||
"rel_type": "m.thread",
|
||||
"event_id": thread_root_id,
|
||||
# Fallback for clients that don't support threads
|
||||
"is_falling_back": True,
|
||||
"m.in_reply_to": {"event_id": thread_root_id},
|
||||
},
|
||||
}
|
||||
|
||||
try:
|
||||
resp = await self._client.room_send(
|
||||
room_id=self.room_id,
|
||||
message_type="m.room.message",
|
||||
content=content,
|
||||
)
|
||||
if isinstance(resp, RoomSendError):
|
||||
logger.error("Failed to send threaded message: %s", resp.message)
|
||||
return False
|
||||
logger.debug("Threaded message sent: %s (thread %s)", resp.event_id, thread_root_id)
|
||||
return True
|
||||
except Exception as exc:
|
||||
logger.error("Matrix threaded send error: %s", exc)
|
||||
return False
|
||||
|
||||
async def create_thread_root(self, plain_text: str, html: str) -> str | None:
|
||||
"""Send a message that will serve as a thread root.
|
||||
|
||||
Returns the ``$event_id`` on success, or ``None`` on failure.
|
||||
"""
|
||||
content = {
|
||||
"msgtype": "m.text",
|
||||
"format": "org.matrix.custom.html",
|
||||
"body": plain_text,
|
||||
"formatted_body": html,
|
||||
}
|
||||
|
||||
try:
|
||||
resp = await self._client.room_send(
|
||||
room_id=self.room_id,
|
||||
message_type="m.room.message",
|
||||
content=content,
|
||||
)
|
||||
if isinstance(resp, RoomSendError):
|
||||
logger.error("Failed to create thread root: %s", resp.message)
|
||||
return None
|
||||
assert isinstance(resp, RoomSendResponse)
|
||||
logger.info("Thread root created: %s", resp.event_id)
|
||||
return resp.event_id
|
||||
except Exception as exc:
|
||||
logger.error("Matrix thread root error: %s", exc)
|
||||
return None
|
||||
|
||||
async def set_presence(self, presence: str) -> bool:
|
||||
"""Set the bot's presence status on the homeserver.
|
||||
|
||||
Returns True on success, False on failure.
|
||||
"""
|
||||
try:
|
||||
resp = await self._client.set_presence(presence)
|
||||
if isinstance(resp, PresenceSetError):
|
||||
logger.error("Failed to set presence to %s: %s", presence, resp.message)
|
||||
return False
|
||||
logger.debug("Presence set to %s", presence)
|
||||
return True
|
||||
except Exception as exc:
|
||||
logger.error("Presence set error: %s", exc)
|
||||
return False
|
||||
|
||||
async def start_presence_heartbeat(self):
|
||||
"""Set presence to online and spawn a background task to keep it alive."""
|
||||
await self.set_presence("online")
|
||||
self._heartbeat_task = asyncio.create_task(self._presence_heartbeat_loop())
|
||||
logger.info("Presence heartbeat started (every %ds)", PRESENCE_HEARTBEAT_INTERVAL)
|
||||
|
||||
async def _presence_heartbeat_loop(self):
|
||||
"""Re-send 'online' presence every PRESENCE_HEARTBEAT_INTERVAL seconds."""
|
||||
try:
|
||||
while True:
|
||||
await asyncio.sleep(PRESENCE_HEARTBEAT_INTERVAL)
|
||||
await self.set_presence("online")
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
|
||||
async def stop_presence_heartbeat(self):
|
||||
"""Cancel the heartbeat task and set presence to offline."""
|
||||
if self._heartbeat_task is not None:
|
||||
self._heartbeat_task.cancel()
|
||||
try:
|
||||
await self._heartbeat_task
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
self._heartbeat_task = None
|
||||
await self.set_presence("offline")
|
||||
logger.info("Presence heartbeat stopped, set to offline")
|
||||
|
||||
async def close(self):
|
||||
await self.stop_presence_heartbeat()
|
||||
await self._client.close()
|
||||
@@ -1,205 +0,0 @@
|
||||
"""Preflight checks — validate configuration and connectivity before running the bot."""
|
||||
|
||||
import logging
|
||||
import sys
|
||||
|
||||
import httpx
|
||||
from nio import AsyncClient, LoginError, RoomResolveAliasError
|
||||
|
||||
from .config import Config
|
||||
from .cheapshark import BASE_URL as CHEAPSHARK_URL
|
||||
from .currency import FRANKFURTER_URL, configure as configure_currency, _all_target_currencies
|
||||
from .epic import FREE_GAMES_URL
|
||||
from .itad import BASE_URL as ITAD_URL
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# ANSI colours for terminal output
|
||||
_GREEN = "\033[32m"
|
||||
_RED = "\033[31m"
|
||||
_YELLOW = "\033[33m"
|
||||
_BOLD = "\033[1m"
|
||||
_RESET = "\033[0m"
|
||||
|
||||
|
||||
def _pass(label: str, detail: str = "") -> bool:
|
||||
suffix = f" — {detail}" if detail else ""
|
||||
print(f" {_GREEN}✓{_RESET} {label}{suffix}")
|
||||
return True
|
||||
|
||||
|
||||
def _fail(label: str, detail: str = "") -> bool:
|
||||
suffix = f" — {detail}" if detail else ""
|
||||
print(f" {_RED}✗{_RESET} {label}{suffix}")
|
||||
return False
|
||||
|
||||
|
||||
def _skip(label: str, detail: str = "") -> bool:
|
||||
suffix = f" — {detail}" if detail else ""
|
||||
print(f" {_YELLOW}–{_RESET} {label}{suffix}")
|
||||
return True # skips don't count as failures
|
||||
|
||||
|
||||
async def run_preflight(config: Config) -> bool:
|
||||
"""Run all preflight checks. Returns True if everything critical passes."""
|
||||
print(f"\n{_BOLD}Pastel — preflight checks{_RESET}\n")
|
||||
all_ok = True
|
||||
|
||||
# Configure currency display so the Frankfurter check fetches the right symbols
|
||||
configure_currency(config.default_currency, config.extra_currencies)
|
||||
|
||||
async with httpx.AsyncClient(timeout=15) as http:
|
||||
# --- Matrix ---
|
||||
print(f"{_BOLD}Matrix{_RESET}")
|
||||
all_ok &= await _check_matrix(config)
|
||||
|
||||
# --- CheapShark ---
|
||||
print(f"\n{_BOLD}CheapShark{_RESET}")
|
||||
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}")
|
||||
all_ok &= await _check_epic(http)
|
||||
|
||||
# --- Frankfurter (exchange rates) ---
|
||||
print(f"\n{_BOLD}Frankfurter (exchange rates){_RESET}")
|
||||
all_ok &= await _check_frankfurter(http, config)
|
||||
|
||||
# --- IsThereAnyDeal ---
|
||||
itad_required = "itad" in config.deal_sources
|
||||
print(f"\n{_BOLD}IsThereAnyDeal{_RESET}")
|
||||
all_ok &= await _check_itad(http, config.itad_api_key, required=itad_required)
|
||||
|
||||
# --- Summary ---
|
||||
print()
|
||||
if all_ok:
|
||||
print(f"{_GREEN}{_BOLD}All checks passed.{_RESET} The bot is ready to run.")
|
||||
else:
|
||||
print(f"{_RED}{_BOLD}Some checks failed.{_RESET} Review the errors above before starting the bot.")
|
||||
print()
|
||||
|
||||
return all_ok
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Individual checks
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def _check_matrix(config: Config) -> bool:
|
||||
"""Verify the access token is valid and the bot can see the target room."""
|
||||
ok = True
|
||||
client = AsyncClient(config.matrix_homeserver_url, config.matrix_bot_user_id)
|
||||
client.access_token = config.matrix_bot_access_token
|
||||
client.user_id = config.matrix_bot_user_id
|
||||
|
||||
try:
|
||||
# whoami — validates the token
|
||||
resp = await client.whoami()
|
||||
if hasattr(resp, "user_id"):
|
||||
ok &= _pass("Authentication", f"logged in as {resp.user_id}")
|
||||
else:
|
||||
ok &= _fail("Authentication", f"token rejected: {resp}")
|
||||
await client.close()
|
||||
return False
|
||||
|
||||
# joined_rooms — check that the bot is in the target room
|
||||
rooms_resp = await client.joined_rooms()
|
||||
if hasattr(rooms_resp, "rooms"):
|
||||
if config.matrix_deals_room_id in rooms_resp.rooms:
|
||||
ok &= _pass("Room access", f"bot is a member of {config.matrix_deals_room_id}")
|
||||
else:
|
||||
ok &= _fail(
|
||||
"Room access",
|
||||
f"bot is NOT in {config.matrix_deals_room_id} — invite the bot first",
|
||||
)
|
||||
else:
|
||||
ok &= _fail("Room access", f"could not list joined rooms: {rooms_resp}")
|
||||
except Exception as exc:
|
||||
ok &= _fail("Homeserver connection", str(exc))
|
||||
finally:
|
||||
await client.close()
|
||||
|
||||
return ok
|
||||
|
||||
|
||||
async def _check_cheapshark(http: httpx.AsyncClient) -> bool:
|
||||
"""Hit the CheapShark deals endpoint to confirm it's reachable."""
|
||||
try:
|
||||
resp = await http.get(f"{CHEAPSHARK_URL}/deals", params={"pageSize": "1"})
|
||||
resp.raise_for_status()
|
||||
deals = resp.json()
|
||||
if isinstance(deals, list):
|
||||
return _pass("API reachable", f"{len(deals)} deal(s) in response")
|
||||
return _fail("API reachable", "unexpected response format")
|
||||
except Exception as exc:
|
||||
return _fail("API reachable", str(exc))
|
||||
|
||||
|
||||
async def _check_epic(http: httpx.AsyncClient) -> bool:
|
||||
"""Hit the Epic free-games endpoint."""
|
||||
try:
|
||||
resp = await http.get(FREE_GAMES_URL, params={"locale": "en-US"})
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
elements = (
|
||||
data.get("data", {})
|
||||
.get("Catalog", {})
|
||||
.get("searchStore", {})
|
||||
.get("elements", [])
|
||||
)
|
||||
return _pass("API reachable", f"{len(elements)} game(s) in catalog")
|
||||
except Exception as exc:
|
||||
return _fail("API reachable", str(exc))
|
||||
|
||||
|
||||
async def _check_frankfurter(http: httpx.AsyncClient, config: Config) -> bool:
|
||||
"""Fetch exchange rates to confirm Frankfurter is reachable."""
|
||||
needed = _all_target_currencies()
|
||||
if not needed:
|
||||
return _pass(
|
||||
"Skipped",
|
||||
f"only {config.default_currency} configured — no conversion needed",
|
||||
)
|
||||
try:
|
||||
symbols = ",".join(needed)
|
||||
resp = await http.get(FRANKFURTER_URL, params={"base": "USD", "symbols": symbols})
|
||||
resp.raise_for_status()
|
||||
rates = resp.json().get("rates", {})
|
||||
if rates:
|
||||
parts = [f"{k}: {v}" for k, v in rates.items()]
|
||||
return _pass("API reachable", ", ".join(parts))
|
||||
return _fail("API reachable", "response contained no rates")
|
||||
except Exception as exc:
|
||||
return _fail("API reachable", str(exc))
|
||||
|
||||
|
||||
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 the lookup endpoint (GET) to validate the key with a known game
|
||||
resp = await http.get(
|
||||
f"{ITAD_URL}/games/lookup/v1",
|
||||
params={"key": api_key, "appid": 220}, # Half-Life 2
|
||||
)
|
||||
if resp.status_code in (401, 403):
|
||||
return _fail("API key", "rejected by ITAD (401/403)")
|
||||
resp.raise_for_status()
|
||||
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))
|
||||
Reference in New Issue
Block a user