update docker

post_rss_to_ghost.py

@@ -414,6 +414,10 @@ async def main():
 
     LOG.info("Starting bot")
 
+    if args.run_once:
+        await task.daily_task()
+        return
+
     # Démarrage: publier l'édition du jour si elle n'existe pas encore
     await task.maybe_run_today()
 

presquegratos.py

@@ -0,0 +1,597 @@
+# weekly_games_roundup.py
+# -*- coding: utf-8 -*-
+import asyncio
+import dataclasses
+import html
+import json
+import logging
+import os
+import random
+import re
+import time
+from dataclasses import dataclass
+from typing import Any, Dict, List, Optional, Tuple
+from urllib.parse import urlparse, parse_qs
+
+import aiohttp
+import jwt  # PyJWT
+import requests
+from bs4 import BeautifulSoup
+from datetime import datetime, timedelta, timezone
+import zoneinfo
+
+LOG = logging.getLogger("weekly")
+TZ = zoneinfo.ZoneInfo("Europe/Brussels")
+UA = {"User-Agent": "Mozilla/5.0 (compatible; weekly-games-roundup/1.0)"}
+
+# -------------------- Ghost Admin client --------------------
+
+class GhostAdmin:
+    def __init__(self, admin_url: str, admin_key: str, accept_version: str = "v6.0"):
+        self.base = admin_url.rstrip("/") + "/"
+        self.key_id, self.key_secret_hex = admin_key.split(":")
+        self.accept_version = accept_version
+
+    def _jwt(self) -> str:
+        iat = int(time.time())
+        payload = {"iat": iat, "exp": iat + 5 * 60, "aud": "/admin/"}
+        headers = {"alg": "HS256", "typ": "JWT", "kid": self.key_id}
+        token = jwt.encode(payload, bytes.fromhex(self.key_secret_hex), algorithm="HS256", headers=headers)
+        return token if isinstance(token, str) else token.decode("utf-8")
+
+    def _headers(self):
+        return {"Authorization": f"Ghost {self._jwt()}",
+                "Accept-Version": self.accept_version,
+                "Content-Type": "application/json"}
+
+    def pick_newsletter_slug(self, preferred_slug: Optional[str]) -> str:
+        if preferred_slug:
+            return preferred_slug
+        url = self.base + "newsletters/"
+        resp = requests.get(url, headers=self._headers(), timeout=20)
+        resp.raise_for_status()
+        newsletters = resp.json().get("newsletters", [])
+        if not newsletters:
+            raise RuntimeError("No newsletters configured in Ghost.")
+        actives = [n for n in newsletters if n.get("status") == "active"]
+        for n in actives:
+            if n.get("is_default"):
+                return n["slug"]
+        return (actives or newsletters)[0]["slug"]
+
+    def create_post_html(self, title: str, html_content: str,
+                         status: str = "draft", feature_image: Optional[str] = None) -> Dict[str, Any]:
+        url = self.base + "posts/?source=html"
+        body = {"posts": [{"title": title, "html": html_content, "status": status,
+                           **({"feature_image": feature_image} if feature_image else {})}]}
+        resp = requests.post(url, headers=self._headers(), json=body, timeout=30)
+        if resp.status_code >= 400:
+            raise RuntimeError(f"Ghost create error {resp.status_code}: {resp.text}")
+        return resp.json()["posts"][0]
+
+    def publish_post(self, post_id: str, updated_at: str,
+                     newsletter_slug: Optional[str], email_segment: Optional[str] = None) -> Dict[str, Any]:
+        slug = self.pick_newsletter_slug(newsletter_slug)
+        params = [f"newsletter={requests.utils.quote(slug)}"]
+        if email_segment:
+            params.append(f"email_segment={requests.utils.quote(email_segment)}")
+        url = self.base + f"posts/{post_id}/?{'&'.join(params)}"
+        body = {"posts": [{"updated_at": updated_at, "status": "published"}]}
+        resp = requests.put(url, headers=self._headers(), json=body, timeout=30)
+        if resp.status_code >= 400:
+            raise RuntimeError(f"Ghost publish error {resp.status_code}: {resp.text}")
+        return resp.json()["posts"][0]
+
+# -------------------- Helpers (URLs, images, dates) --------------------
+
+def _sanitize_url(u: Optional[str]) -> Optional[str]:
+    if not u or not isinstance(u, str):
+        return None
+    u = u.strip().replace("\\", "/")
+    if not u:
+        return None
+    if u.startswith("//"):
+        u = "https:" + u
+    if " " in u:
+        u = u.replace(" ", "%20")
+    p = urlparse(u)
+    if p.scheme not in ("http", "https") or not p.netloc:
+        return None
+    return u
+
+def _fmt_dt(iso_dt: Optional[str]) -> Optional[str]:
+    if not iso_dt:
+        return None
+    try:
+        dt = datetime.fromisoformat(iso_dt.replace("Z", "+00:00")).astimezone(timezone.utc)
+        return dt.strftime("%Y-%m-%d %H:%M UTC")
+    except Exception:
+        return iso_dt
+
+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)
+    try:
+        if start_iso:
+            s = datetime.fromisoformat(start_iso.replace("Z", "+00:00"))
+        else:
+            s = None
+        if end_iso:
+            e = datetime.fromisoformat(end_iso.replace("Z", "+00:00"))
+        else:
+            e = None
+    except Exception:
+        return False
+    # Overlap test: [s,e] intersects [now-7d, now]
+    s = s or now
+    e = e or now
+    return (s <= now) and (e >= window_start) or (s >= window_start and s <= now)
+
+# -------------------- EGS freebies (async, via official endpoint) --------------------
+
+async def fetch_egs_week(session: aiohttp.ClientSession,
+                         locale="en-US", country="US", allow_countries=None) -> List[Dict[str, Any]]:
+    allow = allow_countries or country
+    url = ("https://store-site-backend-static.ak.epicgames.com/freeGamesPromotions"
+           f"?locale={locale}&country={country}&allowCountries={allow}")
+    async with session.get(url, headers=UA, timeout=aiohttp.ClientTimeout(total=25)) as r:
+        r.raise_for_status()
+        data = await r.json()
+
+    out: List[Dict[str, Any]] = []
+    elements = (((data or {}).get("data") or {}).get("Catalog") or {}).get("searchStore", {}).get("elements", []) or []
+    for item in elements:
+        title = item.get("title")
+        slug = item.get("productSlug")
+        if not slug:
+            mappings = (item.get("catalogNs", {}) or {}).get("mappings") or []
+            if mappings:
+                slug = mappings[0].get("pageSlug")
+        url_item = _sanitize_url(f"https://store.epicgames.com/p/{slug}" if slug else None)
+        key_images = item.get("keyImages") or []
+        pref_order = ("OfferImageWide", "DieselStoreFrontWide", "DieselStoreFront", "OfferImageTall", "Thumbnail")
+        chosen = None
+        for t in pref_order:
+            chosen = next((k for k in key_images if (k.get("type") or "").lower() == t.lower()), None)
+            if chosen:
+                break
+        if not chosen and key_images:
+            chosen = key_images[0]
+        hero = _sanitize_url(chosen.get("url")) if chosen else None
+        thumb = hero
+
+        promos = (item.get("promotions") or {})
+        for section in ("promotionalOffers", "upcomingPromotionalOffers"):
+            lst = promos.get(section) or []
+            for entry in lst:
+                for p in entry.get("promotionalOffers", []) or []:
+                    start = p.get("startDate")
+                    end = p.get("endDate")
+                    if _in_last_7_days(start, end):
+                        out.append({
+                            "platform": "EGS",
+                            "title": title,
+                            "start": start,
+                            "end": end,
+                            "url": url_item,
+                            "image": hero,
+                            "thumbnail": thumb,
+                            "is_current": None,
+                        })
+    # keep unique by (title,start)
+    seen = set()
+    uniq = []
+    for it in out:
+        k = (it.get("title"), it.get("start"))
+        if k not in seen:
+            seen.add(k)
+            uniq.append(it)
+    # sort by start desc
+    uniq.sort(key=lambda e: _fmt_dt(e.get("start")) or "", reverse=True)
+    return uniq
+
+# -------------------- PS Plus (async, blog scraping FR) --------------------
+
+PSPLUS_CATEGORY_URL_FR = "https://blog.fr.playstation.com/category/ps-plus/"
+BASE_BLOG = "https://blog.fr.playstation.com/"
+
+def _soup(text: str) -> BeautifulSoup:
+    return BeautifulSoup(text, "html.parser")
+
+def _clean_titles(candidates: List[str], limit: int = 7) -> List[str]:
+    cleaned, seen = [], set()
+    for g in candidates:
+        g = re.sub(r"\s*\|\s*PS[45].*$", "", g).strip()
+        g = re.sub(r"\s*\(PS\+\)$", "", g).strip()
+        if not g or len(g) < 2:
+            continue
+        k = g.lower()
+        if k not in seen:
+            seen.add(k)
+            cleaned.append(g)
+        if len(cleaned) >= limit:
+            break
+    return cleaned
+
+def _abs_url(u: Optional[str], base: str) -> Optional[str]:
+    if not u:
+        return None
+    if u.startswith("http://") or u.startswith("https://"):
+        return u
+    if u.startswith("/"):
+        return BASE_BLOG.rstrip("/") + u
+    return base.rstrip("/") + "/" + u
+
+def _extract_best_image(art: BeautifulSoup, article_url: str) -> Tuple[Optional[str], Optional[str]]:
+    for sel in [
+        'meta[property="og:image"]',
+        'meta[name="og:image"]',
+        'meta[name="twitter:image"]',
+        'meta[property="twitter:image"]',
+    ]:
+        tag = art.select_one(sel)
+        if tag and tag.get("content"):
+            img = _abs_url(tag["content"].strip(), article_url)
+            if img:
+                return (img, img)
+    tag = art.select_one('link[rel="image_src"]')
+    if tag and tag.get("href"):
+        img = _abs_url(tag["href"].strip(), article_url)
+        if img:
+            return (img, img)
+    first_img = art.select_one("article img[src]")
+    if first_img:
+        img = _abs_url(first_img.get("src"), article_url)
+        return (img, img)
+    return (None, None)
+
+async def fetch_psplus_week(session: aiohttp.ClientSession) -> Optional[Dict[str, Any]]:
+    """Dernier billet FR “Les jeux du mois PlayStation Plus …”. On le retient s’il tombe dans T-7..T."""
+    async with session.get(PSPLUS_CATEGORY_URL_FR, headers=UA, timeout=aiohttp.ClientTimeout(total=25)) as r:
+        r.raise_for_status()
+        idx_html = await r.text()
+    idx = _soup(idx_html)
+    candidates: List[Tuple[str, str]] = []
+    for a in idx.select("a[href]"):
+        href = a.get("href") or ""
+        title = (a.get_text() or "").strip()
+        if not href.startswith(BASE_BLOG):
+            continue
+        if title.lower().startswith("les jeux du mois playstation plus"):
+            candidates.append((title, href))
+    if not candidates:
+        return None
+
+    latest: Optional[Tuple[datetime, str, str, BeautifulSoup]] = None
+    for title, href in candidates:
+        try:
+            async with session.get(href, headers=UA, timeout=aiohttp.ClientTimeout(total=25)) as r:
+                r.raise_for_status()
+                art_html = await r.text()
+        except Exception:
+            continue
+        art = _soup(art_html)
+        time_el = art.find("time")
+        date_iso = time_el.get("datetime") if time_el else None
+        if not date_iso:
+            continue
+        try:
+            dt_iso = datetime.fromisoformat(date_iso.replace("Z", "+00:00"))
+        except Exception:
+            continue
+        if latest is None or dt_iso > latest[0]:
+            latest = (dt_iso, title, href, art)
+
+    if latest is None:
+        return None
+
+    dt_iso, title, url, art = latest
+    now = datetime.now(timezone.utc)
+    if (now - dt_iso) > timedelta(days=7):
+        # Hors fenêtre -> pas de section PS+ cette semaine
+        return None
+
+    # extraire quelques titres listés dans l’article
+    games: List[str] = []
+    for node in art.find_all(string=True):
+        t = " ".join((node or "").strip().split())
+        if " | PS" in t and 0 < len(t) <= 120:
+            name = t.split(" | ")[0].strip("—:-– ")
+            if 2 <= len(name.split()) <= 8:
+                games.append(name)
+    if not games:
+        for tag in art.select("strong, em"):
+            t = " ".join(tag.get_text(" ", strip=True).split())
+            if 2 <= len(t.split()) <= 8:
+                games.append(t)
+
+    image_url, thumb_url = _extract_best_image(art, url)
+    return {
+        "title": title,
+        "date": dt_iso.replace(tzinfo=timezone.utc).isoformat(),
+        "url": url,
+        "games": _clean_titles(games),
+        "image": _sanitize_url(image_url),
+        "thumbnail": _sanitize_url(thumb_url),
+    }
+
+# -------------------- Xbox Game Pass (sync, SIGLS + DisplayCatalog) --------------------
+
+def _xbox_locale_from_languages(languages: str) -> str:
+    if not languages or "-" not in languages:
+        return "en-US"
+    lang, region = languages.split("-", 1)
+    return f"{lang.lower()}-{region.upper()}"
+
+def _slugify(s: str) -> str:
+    s = re.sub(r"[^a-zA-Z0-9]+", "-", s).strip("-").lower()
+    return s or "game"
+
+def _pick_images_displaycatalog(prod: Dict[str, Any]) -> Tuple[Optional[str], Optional[str]]:
+    def candidates_of(prod):
+        cands: List[Dict[str, Any]] = []
+        lp = prod.get("LocalizedProperties") or []
+        if isinstance(lp, list) and lp:
+            images = lp[0].get("Images") or []
+            if isinstance(images, list):
+                cands.extend(images)
+        top = prod.get("Images") or []
+        if isinstance(top, list):
+            cands.extend(top)
+        return cands
+
+    def is_pref(img: Dict[str, Any]) -> bool:
+        purpose = (img.get("ImagePurpose") or img.get("Purpose") or "").lower()
+        return any(k in purpose for k in ("poster", "brandedkeyart", "superherowide", "superheroart"))
+
+    cands = candidates_of(prod)
+    if not cands:
+        return None, None
+    pref = [i for i in cands if is_pref(i)] or cands
+    try:
+        hero_raw = max(pref, key=lambda i: i.get("Width", 0)).get("Uri")
+    except Exception:
+        hero_raw = pref[0].get("Uri")
+    try:
+        thumb_raw = min(pref, key=lambda i: i.get("Width", 10**9)).get("Uri")
+    except Exception:
+        thumb_raw = hero_raw
+    hero = _sanitize_url(hero_raw)
+    thumb = _sanitize_url(thumb_raw)
+    return hero or thumb, thumb or hero
+
+def fetch_xgp_recent(limit=20, market="US", languages="en-us") -> List[Dict[str, Any]]:
+    """Best-effort: ‘Recently added’ via SIGLS; fallback All games; enrich via DisplayCatalog."""
+    recently_ids = [
+        "3fdd7f57-7092-4b65-bd40-5a9dac1b2b84",
+        "61d6e1a1-735c-4b97-9d15-22ce8dfb0c03",
+        "7d2d3d36-1c52-4a63-8b3e-3b7ee4d0f62a",
+    ]
+    all_games_id = "29a81209-df6f-41fd-a528-2ae6b91f719c"
+
+    def sigls(collection_id: str) -> List[str]:
+        try:
+            r = requests.get("https://catalog.gamepass.com/sigls/v2",
+                             params={"language": languages, "market": market, "id": collection_id},
+                             headers=UA, timeout=25)
+            r.raise_for_status()
+            payload = r.json()
+        except Exception:
+            return []
+        ids: List[str] = []
+        if isinstance(payload, list):
+            for obj in payload:
+                if isinstance(obj, dict) and "id" in obj and isinstance(obj["id"], str):
+                    ids.append(obj["id"])
+                elif isinstance(obj, str):
+                    ids.append(obj)
+        return ids
+
+    big_ids: List[str] = []
+    for cid in recently_ids:
+        ids = sigls(cid)
+        if ids:
+            big_ids = ids
+            break
+    if not big_ids:
+        big_ids = sigls(all_games_id)
+        if not big_ids:
+            return []
+
+    out: List[Dict[str, Any]] = []
+    seen = set()
+    for i in range(0, len(big_ids), 20):
+        chunk = big_ids[i:i+20]
+        try:
+            dc = requests.get("https://displaycatalog.mp.microsoft.com/v7.0/products",
+                              params={"bigIds": ",".join(chunk), "market": market, "languages": languages},
+                              headers=UA, timeout=25)
+            dc.raise_for_status()
+            data = dc.json()
+        except Exception:
+            continue
+        for p in (data.get("Products") or []):
+            pid = p.get("ProductId")
+            if not pid or pid in seen:
+                continue
+            seen.add(pid)
+            title = pid
+            lp = p.get("LocalizedProperties") or []
+            if lp:
+                title = lp[0].get("ProductTitle") or pid
+            locale_path = _xbox_locale_from_languages("fr-fr")
+            url = f"https://www.xbox.com/{locale_path}/games/store/{_slugify(title)}/{pid}"
+            hero, thumb = _pick_images_displaycatalog(p)
+            out.append({"platform": "XGP", "title": title, "productId": pid, "url": _sanitize_url(url),
+                        "image": hero, "thumbnail": thumb})
+    return out[:limit]
+
+# -------------------- HTML builder --------------------
+
+YOUTUBE_EMBED_TMPL = (
+    '<div class="yt-container" style="position:relative;aspect-ratio:16/9;max-width:800px;margin:1rem 0">'
+    '<iframe src="https://www.youtube.com/embed/{vid}" title="YouTube video" loading="lazy" '
+    'style="position:absolute;inset:0;width:100%;height:100%;border:0" '
+    'allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture; web-share" '
+    'allowfullscreen></iframe></div>'
+)
+
+def build_html(egs: List[Dict[str, Any]],
+               psplus: Optional[Dict[str, Any]],
+               xgp: List[Dict[str, Any]]) -> 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')}"
+    parts.append(f"<h2>{html.escape(title_h2)}</h2>")
+    feature: Optional[str] = None
+
+    # --- Epic Games Store
+    parts.append("<h3>🎁 Epic Games Store — Jeux gratuits (7 jours)</h3>")
+    if not egs:
+        parts.append("<p>Aucun jeu gratuit relevé sur la période.</p>")
+    else:
+        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 "?"
+            img = it.get("thumbnail")
+            if img and not feature:
+                feature = img
+            parts.append("<div style='margin:12px 0'>")
+            if img:
+                parts.append(f'<p><img src="{html.escape(img)}" style="max-width:100%;height:auto;border:0"/></p>')
+            parts.append(f"<p><strong>{title}</strong><br/>Période : {start} → {end}</p>")
+            if url:
+                parts.append(f'<p><a href="{html.escape(url)}">{html.escape(url)}</a></p>')
+            parts.append("</div>")
+
+    # --- PlayStation Plus
+    parts.append("<h3>🎮 PlayStation Plus — Jeux du mois (si annoncé cette semaine)</h3>")
+    if not psplus:
+        parts.append("<p>Aucun nouveau billet PS+ dans la fenêtre des 7 jours.</p>")
+    else:
+        if psplus.get("image") and not feature:
+            feature = psplus["image"]
+        parts.append("<div style='margin:12px 0'>")
+        if psplus.get("image"):
+            parts.append(f'<p><img src="{html.escape(psplus["image"])}" style="max-width:100%;height:auto;border:0"/></p>')
+        title = html.escape(psplus.get("title") or "PlayStation Plus — Jeux du mois")
+        url = psplus.get("url")
+        if url:
+            parts.append(f"<p><strong><a href='{html.escape(url)}'>{title}</a></strong></p>")
+        else:
+            parts.append(f"<p><strong>{title}</strong></p>")
+        games = psplus.get("games") or []
+        if games:
+            parts.append("<ul>" + "".join(f"<li>{html.escape(g)}</li>" for g in games) + "</ul>")
+        parts.append("</div>")
+
+    # --- Xbox Game Pass
+    parts.append("<h3>🟩 Xbox Game Pass — Récemment ajoutés</h3>")
+    if not xgp:
+        parts.append("<p>Pas d'entrées détectées.</p>")
+    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("<div style='margin:12px 0'>")
+            if img:
+                parts.append(f'<p><img src="{html.escape(img)}" style="max-width:100%;height:auto;border:0"/></p>')
+            parts.append(f"<p><strong>{title}</strong></p>")
+            if url:
+                parts.append(f'<p><a href="{html.escape(url)}">{html.escape(url)}</a></p>')
+            parts.append("</div>")
+
+    parts.append("<hr><p><em>Newsletter hebdomadaire — envoyée automatiquement chaque dimanche à midi.</em></p>")
+    return "\n".join(parts), feature
+
+# -------------------- Orchestrator --------------------
+
+async def run_weekly():
+    logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
+    admin_url = os.environ["GHOST_ADMIN_URL"]
+    admin_key = os.environ["GHOST_ADMIN_KEY"]
+    newsletter_slug = os.environ.get("GHOST_NEWSLETTER_SLUG")
+    ghost = GhostAdmin(admin_url, admin_key)
+
+    # Fetch all sources (with timeouts and isolation)
+    egs_items: List[Dict[str, Any]] = []
+    psplus_data: Optional[Dict[str, Any]] = None
+    xgp_items: List[Dict[str, Any]] = []
+
+    try:
+        async with aiohttp.ClientSession(headers=UA) as s:
+            # EGS
+            try:
+                locale = os.environ.get("EGS_LOCALE", "en-US")
+                country = os.environ.get("EGS_COUNTRY", "US")
+                allow = os.environ.get("EGS_ALLOW_COUNTRIES", country)
+                egs_items = await fetch_egs_week(s, locale=locale, country=country, allow_countries=allow)
+            except Exception as e:
+                LOG.warning("EGS fetch failed: %s", e)
+
+            # PS Plus
+            try:
+                psplus_data = await fetch_psplus_week(s)
+            except Exception as e:
+                LOG.warning("PS+ fetch failed: %s", e)
+    except Exception as e:
+        LOG.warning("HTTP session failed: %s", e)
+
+    # XGP (requests, sync)
+    try:
+        market = os.environ.get("XGP_MARKET", "US")
+        languages = os.environ.get("XGP_LANGUAGES", "en-us")
+        xgp_items = fetch_xgp_recent(limit=20, market=market, languages=languages)
+    except Exception as e:
+        LOG.warning("XGP fetch failed: %s", e)
+
+    # Build HTML
+    html_body, feature = build_html(egs_items, psplus_data, xgp_items)
+    # 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})"
+
+    # 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)
+
+    LOG.info("Published weekly newsletter: %s", created.get("url"))
+
+# -------------------- Scheduler --------------------
+
+async def run_forever_sunday_noon():
+    """Run at next Sunday 12:00 Europe/Brussels, then every 7 days."""
+    while True:
+        now = datetime.now(TZ)
+        # days until Sunday (weekday(): Monday=0..Sunday=6)
+        days = (6 - now.weekday()) % 7
+        target = (now + timedelta(days=days)).replace(hour=12, minute=0, second=0, microsecond=0)
+        if target <= now:
+            target = target + timedelta(days=7)
+        wait = (target - now).total_seconds()
+        LOG.info("Next run at %s (in %.0f min)", target.isoformat(), wait/60)
+        await asyncio.sleep(wait)
+        try:
+            await run_weekly()
+        except Exception as e:
+            LOG.exception("Weekly run failed: %s", e)
+        # then sleep 7 days
+        await asyncio.sleep(7 * 24 * 3600)
+
+# -------------------- Entrypoint --------------------
+
+if __name__ == "__main__":
+    import argparse
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--run-once", action="store_true", help="Run now and exit (no scheduler)")
+    args = parser.parse_args()
+    if args.run_once:
+        asyncio.run(run_weekly())
+    else:
+        asyncio.run(run_forever_sunday_noon())

update_and_run.sh

@@ -1,10 +1,14 @@
 #!/bin/bash
+set -euo pipefail
 
+trap 'echo "[stop] shutting down..."; kill 0' SIGINT SIGTERM
 
+git fetch --all || true
+git reset --hard origin/main || true
 
-# Pull the latest changes
+# Lancer les 2 bots en parallèle
-git fetch --all
+python -u post_rss_to_ghost.py & PID1=$!
-git reset --hard origin/main
+python -u presquegratos.py    & PID2=$!
 
-# Run your Python script
+# Attendre que l'un des deux meure (et laisser l'autre se faire tuer par le trap)
-python post_rss_to_ghost.py
+wait -n "$PID1" "$PID2" || true