import asyncio import argparse import datetime as dt import html import json import logging import os import re import time from logging.handlers import RotatingFileHandler from typing import Optional, List, Dict import feedparser import requests import jwt import zoneinfo # Python 3.9+ from urllib.parse import urlparse, parse_qs, urljoin # ------------- Web Crawler for Images ------------- def extract_image_from_url(url: str, timeout: int = 10) -> Optional[str]: """ Fetch a webpage and extract the best image (og:image, twitter:image, or first large image). Returns the image URL or None. """ try: resp = requests.get( url, timeout=timeout, headers={ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Accept": "text/html,application/xhtml+xml", }, allow_redirects=True, ) resp.raise_for_status() html_content = resp.text # Try OpenGraph image first (most reliable) og_match = re.search(r']+property=["\']og:image["\'][^>]+content=["\']([^"\']+)["\']', html_content, re.IGNORECASE) if not og_match: og_match = re.search(r']+content=["\']([^"\']+)["\'][^>]+property=["\']og:image["\']', html_content, re.IGNORECASE) if og_match: img_url = og_match.group(1) return urljoin(url, img_url) # Try Twitter card image tw_match = re.search(r']+name=["\']twitter:image["\'][^>]+content=["\']([^"\']+)["\']', html_content, re.IGNORECASE) if not tw_match: tw_match = re.search(r']+content=["\']([^"\']+)["\'][^>]+name=["\']twitter:image["\']', html_content, re.IGNORECASE) if tw_match: img_url = tw_match.group(1) return urljoin(url, img_url) # Fallback: look for article/main image article_img = re.search(r']*>.*?]+src=["\']([^"\']+)["\']', html_content, re.IGNORECASE | re.DOTALL) if article_img: img_url = article_img.group(1) # Skip tiny images, icons, avatars if not any(skip in img_url.lower() for skip in ['avatar', 'icon', 'logo', 'emoji', '1x1', 'pixel']): return urljoin(url, img_url) return None except Exception as e: LOG.debug("Failed to extract image from %s: %s", url, e) return None # ------------- YouTube helpers ------------- def fetch_youtube_oembed_html(youtube_url: str, timeout: int = 10) -> Optional[str]: """ Get YouTube oEmbed HTML exactly as provided and wrap it as a Ghost embed card. """ try: resp = requests.get( "https://www.youtube.com/oembed", params={"url": youtube_url, "format": "json"}, headers={"User-Agent": "ghost-bot/1.0"}, timeout=timeout, ) resp.raise_for_status() data = resp.json() html_content = data.get("html") if not html_content: return None # Wrap in Ghost embed card container; do NOT alter the iframe attributes. return f'
{html_content}
' except Exception: return None def youtube_thumbnail_url(video_id: str) -> str: return f"https://i.ytimg.com/vi/{video_id}/hqdefault.jpg" def extract_youtube_id(url: str) -> Optional[str]: try: u = urlparse(url) host = u.netloc.lower() if host.endswith("youtube.com"): if u.path == "/watch": return parse_qs(u.query).get("v", [None])[0] m = re.match(r"^/(shorts/|live/)?([A-Za-z0-9_-]{6,})", u.path) if m: return m.group(2) if host == "youtu.be": slug = u.path.strip("/").split("/")[0] return slug or None except Exception: return None return None # ------------- Logging ------------- LOG = logging.getLogger("bot") LOG_PATTERN = logging.Formatter("%(asctime)s:%(levelname)s: [%(filename)s] %(message)s") def setuplogger(): stream_handler = logging.StreamHandler() stream_handler.setFormatter(LOG_PATTERN) stream_handler.setLevel(logging.DEBUG) file_handler = RotatingFileHandler("bot.log", "a", 1_000_000, 1) file_handler.setFormatter(LOG_PATTERN) LOG.setLevel(logging.DEBUG) LOG.addHandler(stream_handler) LOG.addHandler(file_handler) # ------------- Model ------------- class RSSfeed: def __init__(self, url: str, yt: bool = False): self.url = url self.youtube = yt # ------------- Mistral AI Client ------------- class MistralClient: """Client for Mistral AI API to filter and group news items.""" def __init__(self, api_key: str, model: str = "mistral-small-latest"): self.api_key = api_key self.model = model self.base_url = "https://api.mistral.ai/v1/chat/completions" def _call_api(self, messages: List[Dict], temperature: float = 0.3, json_mode: bool = True) -> Optional[str]: """Make a call to the Mistral API.""" headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json", } payload = { "model": self.model, "messages": messages, "temperature": temperature, } if json_mode: payload["response_format"] = {"type": "json_object"} try: resp = requests.post(self.base_url, headers=headers, json=payload, timeout=120) resp.raise_for_status() return resp.json()["choices"][0]["message"]["content"] except Exception as e: LOG.error("Mistral API error: %s", e) return None def filter_news_items(self, items: List[dict], dry_run: bool = False) -> List[dict]: """ Filter out non-news items (tips, walkthroughs, guides, tutorials). Returns only actual game news items. """ if not items: return [] # Prepare items for analysis items_for_analysis = [] for i, item in enumerate(items): items_for_analysis.append({ "id": i, "title": item.get("title", ""), "link": item.get("link", ""), "summary": (item.get("summary", "") or "")[:300], # Truncate for API }) # Split into batches to avoid token limits batch_size = 50 filtered_ids = set() for batch_start in range(0, len(items_for_analysis), batch_size): batch = items_for_analysis[batch_start:batch_start + batch_size] prompt = f"""Tu analyses des articles de sites de jeux vidéo. Tu dois identifier UNIQUEMENT les articles à EXCLURE. EXCLURE UNIQUEMENT si le titre contient EXPLICITEMENT UN de ces mots-clés: - "guide" (le mot exact) - "soluce" (le mot exact) - "astuce" (le mot exact) - "solution" (le mot exact, pas "résolution") - "code promo" - "bon plan" - "-20%" ou "-30%" etc (réductions) - "tuto" ou "tutoriel" - "comment faire" - "how to" NE JAMAIS EXCLURE: - "Early Access" = news de sortie anticipée, À GARDER - "Test" ou "Review" = critique, À GARDER - "Partie Rapide" = émission/podcast, À GARDER - Tout article de news, annonce, sortie, preview - Tout article d'opinion, éditorial, récap - Tout le reste qui ne contient pas les mots-clés d'exclusion ci-dessus Articles à analyser: {json.dumps(batch, ensure_ascii=False, indent=2)} Retourne un JSON avec "exclude_ids" contenant UNIQUEMENT les IDs des articles guides/soluces/promos. Si aucun article ne correspond aux critères d'exclusion, retourne {{"exclude_ids": []}} Sois TRÈS conservateur - en cas de doute, NE PAS exclure.""" messages = [{"role": "user", "content": prompt}] response = self._call_api(messages) if response: try: result = json.loads(response) excluded_ids = set(result.get("exclude_ids", [])) # Keep all items NOT in excluded_ids batch_ids = {item["id"] for item in batch} kept_ids = batch_ids - excluded_ids filtered_ids.update(kept_ids) except json.JSONDecodeError: LOG.warning("Failed to parse Mistral response for filtering") # Fallback: include all items from this batch filtered_ids.update(item["id"] for item in batch) else: # Fallback: include all items from this batch filtered_ids.update(item["id"] for item in batch) # Log filtered out items in dry-run mode if dry_run: excluded_ids = set(range(len(items))) - filtered_ids if excluded_ids: LOG.debug("=== FILTERED OUT (non-news) ===") for i in sorted(excluded_ids): LOG.debug(" [EXCLUDED] %s", items[i].get("title", "No title")) LOG.debug("=== KEPT (news) ===") for i in sorted(filtered_ids): if i < len(items): LOG.debug(" [KEPT] %s", items[i].get("title", "No title")) return [items[i] for i in sorted(filtered_ids) if i < len(items)] def group_similar_items(self, items: List[dict]) -> List[Dict]: """ Group news items by category (News, Tests/Reviews, Previews, etc.) with sub-groups by game/topic within each category. Returns a list of categories, each with sub-groups containing items. """ if not items: return [] # Prepare items for analysis items_for_analysis = [] for i, item in enumerate(items): items_for_analysis.append({ "id": i, "title": item.get("title", ""), "link": item.get("link", ""), }) prompt = f"""Organise ces articles de jeux vidéo en CATÉGORIES et SOUS-GROUPES. Articles à organiser: {json.dumps(items_for_analysis, ensure_ascii=False, indent=2)} CATÉGORIES (utilise ces noms exacts): 1. "Actualités" - News, annonces, sorties, mises à jour, industrie 2. "Tests & Critiques" - Reviews, tests, avis, notes 3. "Aperçus & Previews" - Previews, impressions, démos, hands-on 4. "Vidéos" - Trailers, gameplay vidéos, podcasts 5. "Autres" - Le reste RÈGLES DE GROUPEMENT (TRÈS IMPORTANT): - Groupe par FRANCHISE ou SÉRIE (ex: tous les "Final Fantasy" ensemble, même FF7, FF16, FF XIV) - Groupe par UNIVERS (ex: "Warhammer 40K" = Space Marine + Dawn of War + Darktide) - Groupe par ÉVÉNEMENT (ex: "Nintendo Direct", "State of Play", "Game Awards") - N'utilise JAMAIS de noms de sites web comme groupes (pas "NoFrag", "JeuxOnline", etc.) EXEMPLES DE GROUPEMENTS CORRECTS: - "Final Fantasy VII Rebirth sur Switch 2" + "Final Fantasy VII Remake Intergrade en tête" → groupe "Final Fantasy" - "Techmarine dans Space Marine 2" + "Dawn of War 4 gameplay Ork" → groupe "Warhammer 40K" - "GTA 6 trailer" + "GTA 6 date de sortie" → groupe "GTA 6" - "Nintendo Direct annoncé" + "Zelda dans le Nintendo Direct" → groupe "Nintendo Direct" Retourne ce JSON: {{ "categories": [ {{ "name": "Actualités", "subgroups": [ {{"title": "Final Fantasy", "item_ids": [0, 3, 7]}}, {{"title": "Warhammer 40K", "item_ids": [1, 2]}}, {{"title": "Steam", "item_ids": [5]}} ] }} ] }} IMPORTANT: Chaque article dans UN SEUL sous-groupe. Titre = nom de franchise/série/univers, PAS nom de site.""" messages = [{"role": "user", "content": prompt}] response = self._call_api(messages, temperature=0.2) if response: try: result = json.loads(response) categories = [] used_ids = set() for cat_data in result.get("categories", []): cat_name = cat_data.get("name", "Autres") subgroups = [] for sg_data in cat_data.get("subgroups", []): sg_title = sg_data.get("title", "Divers") item_ids = sg_data.get("item_ids", []) # Filter to valid, unused IDs valid_ids = [i for i in item_ids if i < len(items) and i not in used_ids] if valid_ids: used_ids.update(valid_ids) subgroups.append({ "title": sg_title, "items": [items[i] for i in valid_ids] }) if subgroups: categories.append({ "name": cat_name, "subgroups": subgroups }) # Add any ungrouped items ungrouped = [items[i] for i in range(len(items)) if i not in used_ids] if ungrouped: # Find or create "Autres" category autres_cat = next((c for c in categories if c["name"] == "Autres"), None) if autres_cat: autres_cat["subgroups"].append({"title": "Divers", "items": ungrouped}) else: categories.append({ "name": "Autres", "subgroups": [{"title": "Divers", "items": ungrouped}] }) return categories except json.JSONDecodeError: LOG.warning("Failed to parse Mistral response for grouping") # Fallback: return all items in a single category/subgroup return [{ "name": "Actualités de la semaine", "subgroups": [{"title": "Toutes les news", "items": items}] }] def select_top_story_ids(self, items: List[dict], count: int = 5) -> List[int]: """ Sélectionne UNIQUEMENT les ids des actualités les plus importantes de la semaine (pas de réécriture : le texte affiché reste celui de l'article original). Retourne une liste d'ids, du plus important au moins important. """ if not items: return [] items_for_analysis = [] for i, item in enumerate(items): raw_summary = html.unescape(item.get("summary", "") or "") clean_summary = re.sub("<[^<]+?>", "", raw_summary).strip() items_for_analysis.append({ "id": i, "title": item.get("title", ""), "summary": clean_summary[:400], }) prompt = f"""Tu es rédacteur pour une newsletter jeux vidéo en français. Parmi ces articles de la semaine, sélectionne les {count} actualités les PLUS IMPORTANTES (grosses annonces, sorties majeures, gros trailers, actu qui va marquer la semaine). Ignore les tests/critiques individuels sauf s'ils concernent un jeu très attendu. Articles à analyser: {json.dumps(items_for_analysis, ensure_ascii=False, indent=2)} Retourne ce JSON, du plus important au moins important, avec au maximum {count} ids: {{"top_ids": [3, 0, 7]}}""" response = self._call_api([{"role": "user", "content": prompt}], temperature=0.2) if response: try: result = json.loads(response) ids: List[int] = [] for idx in (result.get("top_ids") or [])[:count]: if isinstance(idx, int) and 0 <= idx < len(items) and idx not in ids: ids.append(idx) if ids: return ids except json.JSONDecodeError: LOG.warning("Failed to parse Mistral response for top stories") return [] def write_intro(self, top_titles: List[str]) -> Optional[str]: """ Rédige une courte introduction (façon "Cette semaine, ...") qui cite naturellement les titres les plus importants de la semaine. """ if not top_titles: return None prompt = f"""Tu rédiges l'introduction d'une newsletter jeux vidéo hebdomadaire en français. Voici les titres des actualités les plus importantes de la semaine, du plus au moins important: {json.dumps(top_titles, ensure_ascii=False)} Rédige UNE introduction FACTUELLE (2 à 3 phrases maximum, un seul paragraphe) qui énumère ces actus. Reformule légèrement les titres pour qu'ils s'enchaînent naturellement, avec des verbes simples et neutres (dévoile, présente, revient avec, sort, annonce, c'est aussi/également...). Reste fidèle au contenu, n'invente aucune information. INTERDIT: adjectifs et formules d'accroche marketing ("particulièrement riche", "frappe fort", "en force", "sans oublier"), questions rhétoriques ou phrases d'accroche finales ("prêts à...", "de quoi...", "à ne pas manquer"). Pas de ton commercial, pas de superlatifs. Un simple constat factuel de ce qui s'est passé, comme un résumé de dépêches. Exemple de ton attendu: "Cette semaine, Marvel’s Wolverine nous dévoile ses personnages clés, Microsoft présente This is an Xbox, Battlefield 6 revient avec une saison 4 centrée sur la mer. C'est également la sortie de la version remastérisée d’Arma: Cold War Assault et le lancement mondial de DragonSword : Awakening, après le rejet de l’injonction de Webzen." Commence par "Cette semaine, ...". Retourne UNIQUEMENT le texte brut de l'introduction, sans guillemets, sans markdown, sans JSON.""" response = self._call_api([{"role": "user", "content": prompt}], temperature=0.5, json_mode=False) if response: text = response.strip().strip('"').strip() text = re.sub(r'[*_`]+', '', text) # strip stray markdown emphasis (raw HTML post, not markdown) if text: return text return None def write_excerpt(self, top_titles: List[str], max_len: int = 300) -> Optional[str]: """ Version courte de l'intro (une seule phrase) qui tient dans la limite de caractères de Ghost, plutôt que de tronquer l'intro complète du corps du post. """ if not top_titles: return None prompt = f"""Tu rédiges le résumé (excerpt) d'une newsletter jeux vidéo hebdomadaire en français. Voici les titres des actualités les plus importantes de la semaine, du plus au moins important: {json.dumps(top_titles, ensure_ascii=False)} Rédige UNE SEULE phrase factuelle qui tient en moins de {max_len} caractères, résumant les 2 ou 3 actus les plus importantes de cette liste. Verbes simples et neutres (dévoile, présente, sort, annonce...). Reste fidèle au contenu, n'invente rien. INTERDIT: adjectifs et formules d'accroche marketing, questions rhétoriques, phrases d'accroche finales. Pas de ton commercial. Commence par "Cette semaine, ...". Retourne UNIQUEMENT le texte brut, sans guillemets, sans markdown, sans JSON. La phrase DOIT tenir en moins de {max_len} caractères.""" response = self._call_api([{"role": "user", "content": prompt}], temperature=0.5, json_mode=False) if response: text = response.strip().strip('"').strip() text = re.sub(r'[*_`]+', '', text) if text: return text return None # ------------- Ghost Admin API 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 latest_published_date(self, tz_name: str = "Europe/Brussels"): """ Date (aware) de la dernière publication (status=published), ou None. """ url = self.base + "posts/?limit=1&order=published_at%20desc&fields=published_at" resp = requests.get(url, headers=self._headers(), timeout=20) resp.raise_for_status() posts = resp.json().get("posts", []) if not posts or not posts[0].get("published_at"): return None # ISO 8601 → aware UTC → converti tz locale dtu = dt.datetime.fromisoformat(posts[0]["published_at"].replace("Z", "+00:00")) return dtu.astimezone(zoneinfo.ZoneInfo(tz_name)) def get_newsletters(self): url = self.base + "newsletters/" resp = requests.get(url, headers=self._headers(), timeout=20) if resp.status_code >= 400: raise RuntimeError(f"Ghost newsletters error {resp.status_code}: {resp.text}") return resp.json().get("newsletters", []) def pick_newsletter_slug(self, preferred_slug: Optional[str]) -> str: if preferred_slug: return preferred_slug nls = self.get_newsletters() if not nls: raise RuntimeError("No newsletters configured in Ghost (Settings → Newsletters).") actives = [n for n in nls if n.get("status") == "active"] for n in actives: if n.get("is_default"): return n.get("slug") return (actives or nls)[0].get("slug") def create_post_html(self, title: str, html_content: str, status: str = "draft", feature_image: Optional[str] = None, custom_excerpt: Optional[str] = None): url = self.base + "posts/?source=html" post = {"title": title, "html": html_content, "status": status} if feature_image: post["feature_image"] = feature_image if custom_excerpt: post["custom_excerpt"] = custom_excerpt[:300] # Ghost limit resp = requests.post(url, headers=self._headers(), json={"posts": [post]}, 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]): 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] # ------------- Task orchestration ------------- class GhostTask: def __init__(self, feeds: List[RSSfeed], admin_url: str, admin_key: str, mistral_api_key: Optional[str] = None, newsletter_slug: Optional[str] = None, email_segment: Optional[str] = None, dry_run: bool = False): self.ghost = GhostAdmin(admin_url, admin_key) self.feeds = feeds self.newsletter_slug = newsletter_slug self.email_segment = email_segment self.mistral = MistralClient(mistral_api_key) if mistral_api_key else None self.dry_run = dry_run for feed in self.feeds: LOG.info("Adding feed %s", feed.url) # --- startup immediate run if not yet published this week def _published_this_week(self) -> bool: """Check if we already published this week (since last Saturday 12:00).""" tz = zoneinfo.ZoneInfo("Europe/Brussels") last = self.ghost.latest_published_date("Europe/Brussels") if not last: return False now = dt.datetime.now(tz) # Find last Saturday at 12:00 days_since_saturday = (now.weekday() - 5) % 7 # Saturday = 5 last_saturday = (now - dt.timedelta(days=days_since_saturday)).replace( hour=12, minute=0, second=0, microsecond=0 ) return last >= last_saturday async def maybe_run_this_week(self): if not self._published_this_week(): LOG.info("Aucune newsletter publiée cette semaine -> génération immédiate.") await self.weekly_task() else: LOG.info("Déjà publié cette semaine, on attend la prochaine fenêtre.") # --- utils @staticmethod def _fr_week_range() -> str: """Returns a French formatted date range for the past week.""" months = { 'January': 'Janvier', 'February': 'Février', 'March': 'Mars', 'April': 'Avril', 'May': 'Mai', 'June': 'Juin', 'July': 'Juillet', 'August': 'Août', 'September': 'Septembre', 'October': 'Octobre', 'November': 'Novembre', 'December': 'Décembre' } today = dt.datetime.now() week_ago = today - dt.timedelta(days=7) # Format: "24 - 31 Janvier 2025" or "28 Janvier - 4 Février 2025" if week_ago.month == today.month: formatted = f"{week_ago.day} - {today.strftime('%d %B %Y')}" else: formatted = f"{week_ago.strftime('%d %B')} - {today.strftime('%d %B %Y')}" for en, fr in months.items(): formatted = formatted.replace(en, fr) return formatted @staticmethod def _safe_get(url: str, timeout: int = 20) -> Optional[bytes]: try: r = requests.get(url, timeout=timeout, headers={"User-Agent": "ghost-bot/1.0"}) r.raise_for_status() return r.content except Exception as e: LOG.warning("Flux indisponible: %s (%s)", url, e) return None @staticmethod def _entry_datetime(entry) -> Optional[dt.datetime]: """ Tente de récupérer une datetime aware (UTC) pour un item feedparser. """ # Try common fields first if getattr(entry, "published", None): try: # YouTube (ISO) e.g. 2025-09-05T10:20:33+00:00 return dt.datetime.fromisoformat(entry.published.replace("Z", "+00:00")).astimezone(dt.timezone.utc) except Exception: pass try: # RFC822 e.g. Fri, 05 Sep 2025 10:20:33 +0000 return dt.datetime.strptime(entry.published.replace('GMT', '+0000'), '%a, %d %b %Y %H:%M:%S %z').astimezone(dt.timezone.utc) except Exception: pass if getattr(entry, "updated_parsed", None): try: return dt.datetime.fromtimestamp(time.mktime(entry.updated_parsed), tz=dt.timezone.utc) except Exception: pass return None def _select_top_stories(self, items: List[dict], count: int = 5) -> List[Dict]: """ Choisit les news les plus importantes (via Mistral si dispo, sinon les premières de la liste). Sert de base à l'introduction rédigée. """ if not items: return [] ids: List[int] = self.mistral.select_top_story_ids(items, count=count) if self.mistral else [] if not ids: ids = list(range(min(count, len(items)))) return [items[idx] for idx in ids] @staticmethod def _deterministic_summary(titles: List[str], max_len: Optional[int] = None) -> str: """No-AI fallback sentence, optionally shortened (by dropping items, not mid-word) to fit max_len.""" def render(ts: List[str]) -> str: if len(ts) == 1: return f"Cette semaine : {ts[0]}." return "Cette semaine : " + ", ".join(ts[:-1]) + " et " + ts[-1] + "." text = render(titles) if not max_len or len(text) <= max_len: return text for n in range(len(titles) - 1, 0, -1): candidate = render(titles[:n]) if len(candidate) <= max_len: return candidate truncated = text[:max_len - 1] last_space = truncated.rfind(" ") if last_space > 20: truncated = truncated[:last_space] return truncated.rstrip(",.;: ") + "…" def _build_intro(self, top_stories: List[dict]) -> Optional[str]: """Rédige le paragraphe d'intro ('Cette semaine, ...') citant les news les plus importantes.""" titles = [(t.get("title") or "").strip() for t in top_stories] titles = [t for t in titles if t] if not titles: return None if self.mistral: intro = self.mistral.write_intro(titles) if intro: return intro return self._deterministic_summary(titles) def _build_excerpt_text(self, top_stories: List[dict], max_len: int = 300) -> Optional[str]: """ Résumé COURT dédié au champ excerpt de Ghost : rédigé pour tenir dans max_len, plutôt que de tronquer l'intro complète du corps du post. """ titles = [(t.get("title") or "").strip() for t in top_stories] titles = [t for t in titles if t] if not titles: return None if self.mistral: excerpt = self.mistral.write_excerpt(titles, max_len=max_len) if excerpt: if len(excerpt) <= max_len: return excerpt truncated = excerpt[:max_len - 1] last_space = truncated.rfind(" ") if last_space > 20: truncated = truncated[:last_space] return truncated.rstrip(",.;: ") + "…" return self._deterministic_summary(titles, max_len=max_len) # --- HTML builder for grouped content def _build_html_roundup_grouped(self, categories: List[Dict], feeds: List[RSSfeed], intro: Optional[str] = None): """ Construit le HTML avec un paragraphe d'introduction, puis des catégories et sous-groupes thématiques. Retourne (html, feature_image_url_ou_None). """ parts: List[str] = [] first_image: Optional[str] = None # --- Intro paragraph --- if intro: parts.append(f'

{html.escape(intro)}

') parts.append('
') # --- Build Table of Contents --- # parts.append('

📋 Sommaire

') # parts.append('
    ') # for cat in categories: # cat_name = cat.get("name", "Actualités") # cat_anchor = self._make_anchor(cat_name) # subgroups = cat.get("subgroups", []) # total_items = sum(len(sg.get("items", [])) for sg in subgroups) # parts.append(f'
  • {html.escape(cat_name)} ({total_items} articles)') # if len(subgroups) > 1 or (len(subgroups) == 1 and len(subgroups[0].get("items", [])) > 1): # parts.append('
      ') # for sg in subgroups: # sg_title = sg.get("title", "Divers") # sg_anchor = self._make_anchor(f"{cat_name}-{sg_title}") # item_count = len(sg.get("items", [])) # parts.append(f'
    • {html.escape(sg_title)} ({item_count})
      • ') # parts.append('
    ') # parts.append('
    • ') # parts.append('
') # parts.append('
') # --- Build Content by Category --- for cat in categories: cat_name = cat.get("name", "Actualités") cat_anchor = self._make_anchor(cat_name) subgroups = cat.get("subgroups", []) if not subgroups: continue # Category header with emoji cat_emoji = { "Actualités": "📰", "Tests & Critiques": "⭐", "Aperçus & Previews": "👁️", "Vidéos": "🎬", "Autres": "📁" }.get(cat_name, "📌") parts.append(f'

{cat_emoji} {html.escape(cat_name)}

') for sg in subgroups: sg_title = sg.get("title", "Divers") sg_anchor = self._make_anchor(f"{cat_name}-{sg_title}") items = sg.get("items", []) if not items: continue # Sub-group header (only if more than 1 item in subgroup) if len(items) > 1: parts.append(f'

{html.escape(sg_title)}

') for post in items: title = post.get("title", "") or "" linkURL = post.get("link", "") or "" parts.append(f'

{html.escape(title)}

') # --- YouTube embed / fallback vid = post.get("yt_videoid") or extract_youtube_id(linkURL) if vid: watch_url = f"https://www.youtube.com/watch?v={vid}" # Try provider HTML via oEmbed (as Ghost does) embed_html = fetch_youtube_oembed_html(watch_url, timeout=10) if embed_html: parts.append(embed_html) else: # Fallback: leave the plain URL on its own line so Ghost may still auto-embed parts.append(f'\n

{watch_url}

\n') # Minimal fallback link (non-intrusive for email/web) parts.append(f'

Voir sur YouTube

') else: # --- Texte + lien ftext = "" if "summary" in post and post["summary"]: ftext = html.unescape(post["summary"]) ftext = re.sub("<[^<]+?>", "", ftext) ftext = re.sub(r"L'article .* est apparu en premier sur .*", "", ftext) if ftext: parts.append(f"

{html.escape(ftext)}

") if linkURL: esc = html.escape(linkURL) parts.append(f'

{esc}

') # --- Images: first try RSS metadata, then crawl the page found_image = False for link in post.get("links", []) or []: if link.get("type") in ("image/jpg", "image/jpeg", "image/png", "image/webp"): imgUrl = link.get("href") if imgUrl: imgUrl = imgUrl.replace("/250x250/", "/990x320/") if not first_image: first_image = imgUrl parts.append(f'
') found_image = True # If no image from RSS, try to extract from the article page if not found_image and linkURL: crawled_img = extract_image_from_url(linkURL, timeout=8) if crawled_img: if not first_image: first_image = crawled_img parts.append(f'
') parts.append('
') # --- Sources parts.append("

📚 Sources

") for feed in feeds: esc = html.escape(feed.url) parts.append(f'

{esc}

') parts.append('

Abonnez-vous pour recevoir chaque semaine les news et soutenir mon travail.

') return "\n".join(parts), first_image @staticmethod def _make_anchor(text: str) -> str: """Convert text to a valid HTML anchor ID.""" # Remove accents and special chars, lowercase, replace spaces with dashes anchor = text.lower() anchor = re.sub(r'[àáâãäå]', 'a', anchor) anchor = re.sub(r'[èéêë]', 'e', anchor) anchor = re.sub(r'[ìíîï]', 'i', anchor) anchor = re.sub(r'[òóôõö]', 'o', anchor) anchor = re.sub(r'[ùúûü]', 'u', anchor) anchor = re.sub(r'[ýÿ]', 'y', anchor) anchor = re.sub(r'[ç]', 'c', anchor) anchor = re.sub(r'[^a-z0-9\s-]', '', anchor) anchor = re.sub(r'\s+', '-', anchor.strip()) return anchor or "section" @staticmethod def _format_duration(seconds: float) -> str: seconds = int(seconds) days, seconds = divmod(seconds, 86400) hours, seconds = divmod(seconds, 3600) minutes, seconds = divmod(seconds, 60) parts = [] if days: parts.append(f"{days} days") if hours: parts.append(f"{hours} hours") if minutes: parts.append(f"{minutes} minutes") if seconds: parts.append(f"{seconds} seconds") return ", ".join(parts) if parts else "0 seconds" async def run_weekly_on_saturday(self): """Run every Saturday at 12:00 (noon).""" while True: now = dt.datetime.now() # Calculate next Saturday at 12:00 days_until_saturday = (5 - now.weekday()) % 7 # Saturday = 5 if days_until_saturday == 0 and now.hour >= 12: days_until_saturday = 7 # Already past Saturday 12:00, wait for next week next_run = (now + dt.timedelta(days=days_until_saturday)).replace( hour=12, minute=0, second=0, microsecond=0 ) sleep_seconds = (next_run - now).total_seconds() while sleep_seconds > 0: LOG.info("Waiting for %s for next scan (Saturday noon)", self._format_duration(sleep_seconds)) await asyncio.sleep(min(sleep_seconds, 5 * 60)) now = dt.datetime.now() sleep_seconds = (next_run - now).total_seconds() LOG.info("Going to run the weekly task") await self.weekly_task() async def weekly_task(self): """Main weekly task: collect, filter, group, and publish.""" # Log newsletters (debug) try: nls = self.ghost.get_newsletters() LOG.info("Newsletters: %s", ", ".join(f"{n.get('name')}[{n.get('slug')}]" for n in nls)) except Exception as e: LOG.warning("Unable to list newsletters: %s", e) title_post = "Les news de la semaine du " + self._fr_week_range() LOG.info("Running weekly task : %s", title_post) # (Re)charge les feeds feeds_file = os.environ.get("FEEDS_FILE", "/data/feeds.txt") if not os.path.isfile(feeds_file): feeds_file = os.environ.get("FEEDS_FILE_FALLBACK", r"f:\workspace\Substack_JV\feeds.txt") feeds: List[RSSfeed] = [] with open(feeds_file, encoding="utf-8") as f: lines = [line.strip() for line in f if line.strip()] for line in lines: feeds.append(RSSfeed(line, "youtube" in line.lower())) self.feeds = feeds # Fenêtre: depuis 7 jours à 06:00 UTC week_ago_6am_utc = dt.datetime.now(dt.timezone.utc).replace( hour=6, minute=0, second=0, microsecond=0 ) - dt.timedelta(days=7) all_news_posts: List[dict] = [] for feed in self.feeds: LOG.info("Scanning feed %s", feed.url) content = self._safe_get(feed.url, timeout=30) if not content: continue fp = feedparser.parse(content) # Sélection des items de la semaine new_entries = [] for e in fp.entries: dte = self._entry_datetime(e) if dte and dte > week_ago_6am_utc: new_entries.append(e) # Basic URL-based filtering (keep existing logic) filtered = [] for e in new_entries: linkURL = e.get("link", "") or "" if "actugaming" in linkURL and ("puzzle-" in linkURL or "guide-" in linkURL): continue # enrich YouTube id if applicable if feed.youtube and linkURL: vid = extract_youtube_id(linkURL) if vid: e["yt_videoid"] = vid filtered.append(e) all_news_posts.extend(filtered) if not all_news_posts: LOG.warning("Aucun item récupéré (flux down ?). On n'envoie pas cette semaine.") return LOG.info("Collected %d items from feeds", len(all_news_posts)) # Use Mistral AI for filtering and grouping if available if self.mistral: LOG.info("Using Mistral AI to filter non-news content...") filtered_posts = self.mistral.filter_news_items(all_news_posts, dry_run=self.dry_run) LOG.info("After filtering: %d items (removed %d)", len(filtered_posts), len(all_news_posts) - len(filtered_posts)) if filtered_posts: LOG.info("Using Mistral AI to group items by category...") categories = self.mistral.group_similar_items(filtered_posts) total_cats = len(categories) total_subgroups = sum(len(cat.get("subgroups", [])) for cat in categories) LOG.info("Created %d categories with %d sub-groups", total_cats, total_subgroups) LOG.info("Picking the top stories of the week...") top_stories = self._select_top_stories(filtered_posts, count=5) LOG.info("Selected %d top stories", len(top_stories)) else: categories = [] top_stories = [] else: LOG.warning("No Mistral API key configured, skipping AI filtering/grouping") # Fallback: single category with all items categories = [{ "name": "Actualités de la semaine", "subgroups": [{"title": "Toutes les news", "items": all_news_posts}] }] top_stories = self._select_top_stories(all_news_posts, count=5) if not categories or all( len(sg.get("items", [])) == 0 for cat in categories for sg in cat.get("subgroups", []) ): LOG.warning("No news items after filtering. Skipping this week.") return intro = self._build_intro(top_stories) excerpt = self._build_excerpt_text(top_stories, max_len=300) roundup_html, feature_image = self._build_html_roundup_grouped(categories, self.feeds, intro) # 1) Create draft (with feature image if any) created = self.ghost.create_post_html(title_post, roundup_html, status="draft", feature_image=feature_image, custom_excerpt=excerpt) LOG.info("Created draft post: %s (id: %s)", created.get("title"), created.get("id")) # 2) Publish + send email (unless dry-run mode) if self.dry_run: LOG.info("DRY-RUN MODE: Post created as draft but NOT published. URL: %s", created.get("url", "N/A")) LOG.info("DRY-RUN MODE: Review the draft in Ghost admin, then publish manually if satisfied.") return published = self.ghost.publish_post( post_id=created["id"], updated_at=created["updated_at"], newsletter_slug=os.environ.get("GHOST_NEWSLETTER_SLUG"), email_segment=os.environ.get("GHOST_EMAIL_SEGMENT"), ) LOG.info("Published post: %s (emailed via newsletter)", published.get("url")) # ------------- main ------------- async def main(): setuplogger() parser = argparse.ArgumentParser() parser.add_argument("--runonce", action="store_true", help="Run now and exit (no scheduler)") parser.add_argument("--dry-run", action="store_true", dest="dry_run", help="Run immediately, create draft but do NOT publish (for testing)") args = parser.parse_args() # Feeds init (list may be reloaded inside task) feeds: List[RSSfeed] = [] feeds_file = os.environ.get("FEEDS_FILE", "/data/feeds.txt") if not os.path.isfile(feeds_file): feeds_file = os.environ.get("FEEDS_FILE_FALLBACK", r"f:\workspace\Substack_JV\feeds.txt") with open(feeds_file, encoding="utf-8") as f: for line in f: line = line.strip() if not line: continue feeds.append(RSSfeed(line, "youtube" in line.lower())) admin_url = os.environ["GHOST_ADMIN_URL"] # e.g. https://ghostadmin.zep.best/ghost/api/admin/ admin_key = os.environ["GHOST_ADMIN_KEY"] # integration_id:secret_hex mistral_api_key = os.environ.get("MISTRAL_API_KEY") # Optional: for AI filtering/grouping if not mistral_api_key: LOG.warning("MISTRAL_API_KEY not set. AI filtering and grouping will be disabled.") task = GhostTask( feeds=feeds, admin_url=admin_url, admin_key=admin_key, mistral_api_key=mistral_api_key, newsletter_slug=os.environ.get("GHOST_NEWSLETTER_SLUG"), email_segment=os.environ.get("GHOST_EMAIL_SEGMENT"), dry_run=args.dry_run, ) LOG.info("Starting bot (weekly mode%s)", " - DRY RUN" if args.dry_run else "") if args.runonce: await task.weekly_task() return if args.dry_run: LOG.info("DRY-RUN: Running weekly task immediately (will create draft only)") await task.weekly_task() return # Démarrage: publier l'édition de la semaine si elle n'existe pas encore await task.maybe_run_this_week() # Planification hebdomadaire le samedi à 12:00 Europe/Brussels await task.run_weekly_on_saturday() if __name__ == "__main__": asyncio.run(main())