6e84429ad1
Implements a Matrix bot that posts PC gaming deals and free game alerts: - CheapShark API integration (Steam, GOG, Humble, GMG) polled every 2 hours - Epic Games Store free games detection polled daily - IsThereAnyDeal historical low price flagging (optional) - SQLite deduplication with 30-day pruning - First-run population (records existing deals without posting) - Matrix-flavored HTML + plain text fallback messages - Dockerfile for containerized deployment https://claude.ai/code/session_01LPpSZFfyh6vdV5HGFWjoQX
136 lines
4.3 KiB
Python
136 lines
4.3 KiB
Python
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
|