diff --git a/Dockerfile b/Dockerfile index a93adf0..abc4c6b 100644 --- a/Dockerfile +++ b/Dockerfile @@ -10,7 +10,7 @@ RUN pip install -r requirements.txt ENV TZ=Europe/Brussels RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone - +RUN playwright install --with-deps chromium COPY update_and_run.sh /app # Normalize line endings (Windows CRLF -> LF) and ensure readable RUN sed -i 's/\r$//' /app/update_and_run.sh && chmod a+r /app/update_and_run.sh diff --git a/presquegratos.py b/presquegratos.py index 015e76d..6c0f373 100644 --- a/presquegratos.py +++ b/presquegratos.py @@ -24,9 +24,20 @@ import zoneinfo from storage import Storage from keys import xgp_key + +try: + from playwright.async_api import async_playwright +except Exception: + async_playwright = None + LOG = logging.getLogger("bot_weekly") LOG_PATTERN = logging.Formatter("%(asctime)s:%(levelname)s: [%(filename)s] %(message)s") +PRIME_HOME = "https://gaming.amazon.com/home" + +CARD_COLUMNS = 2 +CARD_IMG_H = 180 # même hauteur partout +CARD_TITLE_EM = 3 # ~2 lignes de texte mini def setuplogger(): stream_handler = logging.StreamHandler() @@ -128,6 +139,42 @@ def _fmt_dt(iso_dt: Optional[str]) -> Optional[str]: except Exception: return iso_dt + +def _egs_is_free_now(item: Dict[str, Any], now: Optional[datetime] = None) -> bool: + """Vrai si l'offre est GRATUITE en ce moment (pas juste en promo payante).""" + now = now or datetime.now(timezone.utc) + + # 1) Prix courant = 0 ? + price = (item.get("price") or {}).get("totalPrice") or {} + try: + # certains champs sont des int, d'autres des str + discount_price = int(str(price.get("discountPrice") or 0)) + if discount_price == 0: + # si Epic renvoie 0, c'est bien 'Free Now' + return True + except Exception: + pass + + # 2) Promo 100% active dans la fenêtre de dates ? + promos = (item.get("promotions") or {}) + for group_key in ("promotionalOffers", "upcomingPromotionalOffers"): + for group in promos.get(group_key) or []: + for p in group.get("promotionalOffers") or []: + try: + start = datetime.fromisoformat(p.get("startDate").replace("Z", "+00:00")) + end = datetime.fromisoformat(p.get("endDate").replace("Z", "+00:00")) + except Exception: + start = end = None + discount = ((p.get("discountSetting") or {}).get("discountPercentage")) + # Plusieurs backends EGS renvoient 0 pour 'gratuit', d'autres 100. + # On accepte les deux mais on exige que la fenêtre englobe 'now'. + if discount in (0, 100): + if (start is None or start <= now) and (end is None or now <= end): + return True + return False + + + def _in_last_7_days(start_iso: Optional[str], end_iso: Optional[str]) -> bool: now = datetime.now(timezone.utc) window_start = now - timedelta(days=7) @@ -147,6 +194,231 @@ def _in_last_7_days(start_iso: Optional[str], end_iso: Optional[str]) -> bool: e = e or now return (s <= now) and (e >= window_start) or (s >= window_start and s <= now) + +# Prime gaming + +def _prime_key(it: Dict[str, Any]) -> str: + # priorité à la clé GraphQL si dispo, sinon fallback stable sur (title|url) + k = it.get("key") + if isinstance(k, str) and k.strip(): + return k + t = (it.get("title") or "").strip().lower() + u = (it.get("url") or "").strip().lower() + return f"{t}|{u}" + +def _pick_img(it: Dict[str, Any]) -> Optional[str]: + # quelques champs possibles venant du payload GraphQL + for cand in ("image", "hero", "boxArt", "thumb", "thumbnail", "cover"): + v = it.get(cand) + if isinstance(v, str) and v.startswith(("http://", "https://")): + return _sanitize_url(v) + return None + +def _coerce_items_from_graphql(data: dict) -> list[dict]: + """Extract Prime Gaming Item records from several response shapes.""" + items: List[Dict[str, Any]] = [] + + def push_from_item(it: dict): + if not isinstance(it, dict): + return + iid = it.get("id") or it.get("itemId") + assets = it.get("assets") or {} + title = (assets.get("title") or it.get("title") or "").strip() + url = ( + assets.get("externalClaimLink") + or it.get("claimUrl") + or f"https://gaming.amazon.com/home?itemId={iid}" + ) + media = (assets.get("cardMedia") or {}).get("defaultMedia") or {} + img = media.get("src1x") or media.get("src2x") or None + end = None + offers = it.get("offers") + if isinstance(offers, list) and offers: + end = offers[0].get("endTime") or offers[0].get("endDate") + + items.append( + { + "key": str(iid) if iid else title, # used for de-dupe & DB + "platform": "PRIME", + "title": title + or (it.get("game", {}) or {}).get("assets", {}).get("title") + or "Unknown", + "game": title or "", + "end": end, + "url": _sanitize_url(url), + "image": _sanitize_url(img), + "thumbnail": _sanitize_url(img), + "category": it.get("category"), + "isFGWP": it.get("isFGWP"), + } + ) + + # shape 1: {"data":{"items":{"items":[...]}}} + page_items = (((data or {}).get("data") or {}).get("items") or {}).get("items") + if isinstance(page_items, list): + for it in page_items: + push_from_item(it) + + # shape 2: sections with ItemsPage + sections = [ + "inGameLoot", + "expiring", + "popular", + "games", + "featuredContent", + "eventRow1", + "eventRow2", + ] + root = (data or {}).get("data") or {} + for sec in sections: + arr = (root.get(sec) or {}).get("items") + if isinstance(arr, list): + for it in arr: + if isinstance(it, dict) and it.get("assets"): + push_from_item(it) + + return items + + + +async def _fetch_prime_offers_playwright() -> List[Dict[str, Any]]: + """Open Prime Gaming in a real browser and capture GraphQL responses.""" + if async_playwright is None: + raise RuntimeError( + "Playwright not installed. `pip install playwright` then `playwright install`" + ) + + async with async_playwright() as pw: + browser = await pw.chromium.launch(headless=True) + context = await browser.new_context( + user_agent=UA["User-Agent"], + extra_http_headers={k: v for k, v in UA.items() if k.lower() != "user-agent"}, + ) + + page = await context.new_page() + offers: List[Dict[str, Any]] = [] + + async def handle_response(resp): + try: + if "/graphql" in resp.url and resp.request.method == "POST" and resp.status == 200: + data = await resp.json() + extracted = _coerce_items_from_graphql(data) + offers.extend(extracted) + except Exception: + LOG.exception("PRIME: parse error on %s", resp.url) + + page.on("response", handle_response) + + await page.goto(PRIME_HOME, wait_until="domcontentloaded", timeout=45000) + try: + await page.wait_for_load_state("networkidle", timeout=20000) + except Exception: + pass + + # Nudge lazy sections + if not offers: + try: + await page.evaluate("window.scrollTo(0, document.body.scrollHeight)") + await page.wait_for_timeout(2000) + await page.evaluate("window.scrollTo(0, 0)") + await page.wait_for_timeout(1000) + except Exception: + pass + + await context.close() + await browser.close() + + # de-dupe by key + uniq = {} + for it in offers: + k = it.get("key") + if k and k not in uniq: + uniq[k] = it + + # keep only FULL_GAME (drop in-game loot) and items with a title + url + final = [ + v + for v in uniq.values() + if (v.get("category") == "FULL_GAME" or v.get("isFGWP")) + and v.get("title") + and v.get("url") + ] + return final + + +# Prime Gaming +PRIME_HOME = "https://gaming.amazon.com/" + +def _prime_key(it: Dict[str, Any]) -> str: + # priorité à la clé GraphQL si dispo, sinon fallback stable sur (title|url) + k = it.get("key") + if isinstance(k, str) and k.strip(): + return k + t = (it.get("title") or "").strip().lower() + u = (it.get("url") or "").strip().lower() + return f"{t}|{u}" + +def _pick_img(it: Dict[str, Any]) -> Optional[str]: + # quelques champs possibles venant du payload GraphQL + for cand in ("image", "hero", "boxArt", "thumb", "thumbnail", "cover"): + v = it.get(cand) + if isinstance(v, str) and v.startswith(("http://", "https://")): + return _sanitize_url(v) + return None + +async def fetch_primegaming_week() -> Optional[Dict[str, Any]]: + """ + Ouvre Prime Gaming (Playwright), capte les réponses GraphQL, + garde les FULL_GAME/FGWP et filtre sur 7 jours si possible. + Ne touche pas à la DB ici. + """ + offers = await _fetch_prime_offers_playwright() # fourni par toi + + if not offers: + return None + + # filtre temporel (si dates présentes) + windowed: List[Dict[str, Any]] = [] + for it in offers: + start_iso = it.get("start") or it.get("startDate") or it.get("availabilityStartDate") + end_iso = it.get("end") or it.get("endDate") or it.get("availabilityEndDate") + if (start_iso or end_iso): + if _in_last_7_days(start_iso, end_iso): + windowed.append(it) + else: + windowed.append(it) + + if not windowed: + return None + + # dédup locale + clés calculées pour le filtrage DB ultérieur + uniq: Dict[str, Dict[str, Any]] = {} + for it in windowed: + k = _prime_key(it) + if k not in uniq: + it["_dedup_key"] = k + uniq[k] = it + + items = list(uniq.values()) + if not items: + return None + + # Construire un bloc "brut" (les titres seront re-dérivés après filtrage DB) + hero = _pick_img(items[0]) + + return { + "title": "This Month on Prime Gaming", + "date": datetime.now(timezone.utc).isoformat(), + "url": PRIME_HOME, + "games": [ (it.get("title") or "").strip() for it in items if it.get("title") ][:12], + "image": hero, + "thumbnail": hero, + "_items": items, # pour filtrage DB/remember au call-site + } + + + + # -------------------- EGS freebies (async, via official endpoint) -------------------- async def fetch_egs_week(session: aiohttp.ClientSession, @@ -161,6 +433,9 @@ async def fetch_egs_week(session: aiohttp.ClientSession, out: List[Dict[str, Any]] = [] elements = (((data or {}).get("data") or {}).get("Catalog") or {}).get("searchStore", {}).get("elements", []) or [] for item in elements: + + if not _egs_is_free_now(item): + continue title = item.get("title") slug = item.get("productSlug") if not slug: @@ -181,6 +456,8 @@ async def fetch_egs_week(session: aiohttp.ClientSession, thumb = hero promos = (item.get("promotions") or {}) + + for section in ("promotionalOffers", "upcomingPromotionalOffers"): lst = promos.get(section) or [] for entry in lst: @@ -456,9 +733,57 @@ YOUTUBE_EMBED_TMPL = ( 'allowfullscreen>' ) + +def render_cards_grid(items, columns=2, fixed_h=180, title_em=3, fallback_url="#"): + if not items: + return "

Aucun élément disponible.

" + + items = items[:12] + col_pct = 100 // max(1, columns) + # explicit col widths => no leftover gap + colgroup = "".join([f'' for _ in range(columns)]) + + # build cells + cells = [] + for it in items: + title = (it.get("title") or it.get("name") or "Voir l’offre").strip() + url = (it.get("url") or fallback_url).strip() + img = (it.get("image") or it.get("hero") or it.get("thumb") or "") + img = _sanitize_url(img) if img else "" + + cells.append(f""" + + + {(''+html.escape(title)+'') if img else ''} +
{html.escape(title)}
+ +""") + + # pad last row + rows = [] + for i in range(0, len(cells), columns): + row = cells[i:i+columns] + if len(row) < columns: + missing = columns - len(row) + row += [f''] * missing + rows.append("" + "".join(row) + "") + + # IMPORTANT: force fixed layout + zero spacing + return f""" + + {colgroup} + {''.join(rows)} +
""" + def build_html(egs: List[Dict[str, Any]], psplus: Optional[Dict[str, Any]], - xgp: List[Dict[str, Any]]) -> Tuple[str, Optional[str]]: + xgp: List[Dict[str, Any]], + prime: Optional[Dict[str, Any]] = None) -> Tuple[str, Optional[str]]: + parts: List[str] = [] now_local = datetime.now(TZ) title_h2 = f"Presque Gratuit — semaine du {(now_local - timedelta(days=6)).strftime('%d/%m')} au {now_local.strftime('%d/%m')}" @@ -466,30 +791,70 @@ def build_html(egs: List[Dict[str, Any]], feature: Optional[str] = None # --- Epic Games Store - parts.append("

🎁 Epic Games Store — Jeux gratuits (7 jours)

") + parts.append("

🎁 Epic Games Store — Jeux gratuits cette semaine

") if not egs: parts.append("

Aucun jeu gratuit relevé sur la période.

") else: + parts.append(render_cards_grid(egs, columns=CARD_COLUMNS, fixed_h=CARD_IMG_H, fallback_url="https://store.epicgames.com/")) for it in egs: - title = html.escape(it.get("title") or "") - url = it.get("url") or "" - start = _fmt_dt(it.get("start")) or "?" - end = _fmt_dt(it.get("end")) or "?" + # title = html.escape(it.get("title") or "") + # url = it.get("url") or "" + # start = _fmt_dt(it.get("start")) or "?" + # end = _fmt_dt(it.get("end")) or "?" img = it.get("thumbnail") if img and not feature: feature = img - parts.append("
") - if img: - parts.append(f'

') - parts.append(f"

{title}
Période : {start} → {end}

") - if url: - parts.append(f'

{html.escape(url)}

') - parts.append("
") + # parts.append("
") + # if img: + # parts.append(f'

') + # parts.append(f"

{title}
Période : {start} → {end}

") + # if url: + # parts.append(f'

{html.escape(url)}

') + # parts.append("
") + + + # --- Prime Gaming + parts.append("

🟣 Prime Gaming — Jeux offerts

") + if not prime: + parts.append("

Aucun nouveau billet Prime Gaming cette semaine.

") + else: + # utilise l’image de section comme “feature” si rien d’autre n’a été choisi + # if prime.get("image") and not feature: + # feature = prime["image"] + + # titre cliquable vers la page "This Month on Prime Gaming" + title = html.escape(prime.get("title") or "This Month on Prime Gaming") + url = prime.get("url") or "https://gaming.amazon.com/" + #parts.append(f"

{title}

") + + # grille de cartes (image au-dessus, nom en dessous, lien vers l’offre) + parts.append(render_cards_grid(prime.get("_items", []), columns=CARD_COLUMNS, fixed_h=CARD_IMG_H, fallback_url=url)) + + # --- Xbox Game Pass + + parts.append("

🟩 Xbox Game Pass — Récemment ajoutés

") + if not xgp: + parts.append("

Pas d'entrées détectées.

") + else: + parts.append(render_cards_grid(xgp, columns=CARD_COLUMNS, fixed_h=CARD_IMG_H, fallback_url="https://www.xbox.com/xbox-game-pass")) + for it in xgp: + # title = html.escape(it.get("title") or it.get("productId") or "") + # url = it.get("url") or "" + # img = it.get("thumbnail") + if img and not feature: + feature = img + # parts.append("
") + # if img: + # parts.append(f'

') + # parts.append(f"

{title}

") + # if url: + # parts.append(f'

{html.escape(url)}

') + # parts.append("
") # --- PlayStation Plus - parts.append("

🎮 PlayStation Plus — Jeux du mois (si annoncé cette semaine)

") + parts.append("

🎮 PlayStation Plus — Jeux du mois

") if not psplus: - parts.append("

Aucun nouveau billet PS+ dans la fenêtre des 7 jours.

") + parts.append("

Aucune nouvelle annonce PS+.

") else: if psplus.get("image") and not feature: feature = psplus["image"] @@ -507,26 +872,6 @@ def build_html(egs: List[Dict[str, Any]], parts.append("") parts.append("") - # --- Xbox Game Pass - - parts.append("

🟩 Xbox Game Pass — Récemment ajoutés

") - if not xgp: - parts.append("

Pas d'entrées détectées.

") - else: - for it in xgp: - title = html.escape(it.get("title") or it.get("productId") or "") - url = it.get("url") or "" - img = it.get("thumbnail") - if img and not feature: - feature = img - parts.append("
") - if img: - parts.append(f'

') - parts.append(f"

{title}

") - if url: - parts.append(f'

{html.escape(url)}

') - parts.append("
") - parts.append("

Newsletter hebdomadaire — envoyée automatiquement chaque dimanche à midi.

") return "\n".join(parts), feature @@ -543,6 +888,9 @@ async def run_weekly(): egs_items: List[Dict[str, Any]] = [] psplus_data: Optional[Dict[str, Any]] = None xgp_items: List[Dict[str, Any]] = [] + prime_data: Optional[Dict[str, Any]] = None + + store = Storage() try: async with aiohttp.ClientSession(headers=UA) as s: @@ -563,6 +911,32 @@ async def run_weekly(): except Exception as e: LOG.warning("HTTP session failed: %s", e) + #prime + try: + prime_data = await fetch_primegaming_week() + except Exception as e: + LOG.warning("prime fetch failed: %s", e) + + + if prime_data: + # 2.a Filtre DB (ne garder que ce qui n’a jamais été publié) + fresh = [] + for it in prime_data["_items"]: + k = it.get("_dedup_key") or _prime_key(it) + if not store.seen(k): # ← usage DB (filtrage) + it["_dedup_key"] = k + fresh.append(it) + + if fresh: + # régénère la liste "games" et l'image en fonction du filtrage + prime_data["_items"] = fresh + prime_data["games"] = [ (it.get("title") or "").strip() for it in fresh if it.get("title") ][:12] + prime_data["image"] = prime_data.get("image") or _pick_img(fresh[0]) + prime_data["thumbnail"] = prime_data["image"] + else: + prime_data = None # rien de neuf → pas de section + + # XGP (requests, sync) try: market = os.environ.get("XGP_MARKET", "US") @@ -572,15 +946,11 @@ async def run_weekly(): LOG.warning("XGP fetch failed: %s", e) # Build HTML - store = Storage() - def keep_new(items, key_fn): fresh = [] for it in items: k = key_fn(it) - if not store.seen(k): - it["_dedup_key"] = k fresh.append(it) return fresh @@ -589,20 +959,25 @@ async def run_weekly(): xgp_items = keep_new(xgp_items, xgp_key) - html_body, feature = build_html(egs_items, psplus_data, xgp_items) + html_body, feature = build_html(egs_items, psplus_data, xgp_items, prime_data) # Title (FR) start = (datetime.now(TZ) - timedelta(days=6)).strftime("%d/%m/%Y") end = datetime.now(TZ).strftime("%d/%m/%Y") - title = f"Récap hebdo — EGS, PS Plus, Game Pass ({start} → {end})" + title = f"Récap hebdo — EGS, Prime Gaming, PS Plus, Game Pass ({start} → {end})" # Create + publish + email created = ghost.create_post_html(title, html_body, status="draft", feature_image=feature) #ghost.publish_post(created["id"], created["updated_at"], newsletter_slug=newsletter_slug) post_id = created["id"] - for it in xgp_items: - store.remember("xgp", it["_dedup_key"], post_id) + if xgp_items and post_id: + for it in xgp_items: + store.remember("xgp", it["_dedup_key"], post_id) + # 2.b Marquer comme publié (clé + post_id) + if prime_data and post_id: + store.bulk_remember("prime", [(it["_dedup_key"], post_id) for it in prime_data["_items"]]) + LOG.info("Published weekly newsletter: %s", created.get("url")) # -------------------- Scheduler -------------------- diff --git a/requirements.txt b/requirements.txt index dd5746a..201ba88 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,4 +3,5 @@ PyJWT>=2.7,<3 requests>=2.31 feedparser>=6.0 aiohttp -bs4 \ No newline at end of file +bs4 +playwright \ No newline at end of file