# ============================================================================== # Author: Simon-Pierre Boucher # File: creaka/connectors/palmares_apple.py # Desc: Connecteur DÉCOUVERTE — balados québécois dans les palmarès canadiens # d'Apple Podcasts : top 100 général (API marketing officielle, # rss.marketingtools.apple.com) + tops 100 PAR GENRE (API RSS iTunes # officielle, itunes.apple.com/ca/rss) + lookup iTunes groupé + tête de # flux RSS pour l'attribution QC. Palier 2 (§9). V4. # ============================================================================== """Palmarès Apple Podcasts Canada → balados québécois qui percent. Évaluation vague 3 des palmarès publics (2026-08-18) : - Spotify : podcastcharts.byspotify.com/api → HTTP 500 sur toutes les variantes (API morte), charts.spotify.com → mur de connexion. Écarté. - ADISQ (palmaresadisq.ca) : SPA JavaScript (pjax renvoie la coquille vide), et un palmarès d'ARTISTES sans liens sociaux → fiches sans lien, rejetées par le schéma. Écarté. - Bible urbaine : articles éditoriaux non structurés (noms en prose, sans handles) → liaison ambiguë interdite (§12). Écarté. - Apple Podcasts : API marketing OFFICIELLE, JSON stable, sans clé. Retenu. Chaîne : top 100 CA (rang = signal de traction) → lookup iTunes GROUPÉ (une requête pour tous les ids : flux RSS, nb d'épisodes, dernier épisode, genres) → attribution QC conservatrice : marqueur québécois dans le nom/artiste, ou dans la TÊTE du flux RSS auto-déclaré (avant le premier : titre, description, auteur — jamais le contenu des épisodes). Le palmarès étant surtout anglo-canadien, le rendement est faible mais de très haute valeur : un balado QC du top 100 canadien est un créateur qui compte. Complémentaire de balados-itunes (recherche thématique) — même clé de compte (podcast:collectionId) → fusion naturelle des fiches. V4 (2026-08-20) : le top 100 général CA est devenu 100 % anglophone (0 balado QC pendant plusieurs passages → alertes). On balaie EN PLUS les tops 100 par genre de l'API RSS iTunes (officielle, sans clé) : le préfiltre nom/artiste/ résumé limite les lectures RSS, et l'attribution finale reste identique (nom/artiste QC, sinon tête de flux fr + marqueur QC — jamais le résumé seul, qui sert uniquement de préfiltre). """ from __future__ import annotations import re import sys import time from ..identity import account from ..schema import Creator from .base import BaseConnector from .balados_itunes import _GENRE_NICHE, _dormant from .youtube_recherche import is_quebec CHARTS_URL = ("https://rss.marketingtools.apple.com/api/v2/ca/podcasts/" "top/100/podcasts.json") GENRE_CHART_URL = ("https://itunes.apple.com/ca/rss/toppodcasts/" "limit=100/genre={gid}/json") LOOKUP_URL = "https://itunes.apple.com/lookup" # genres Apple Podcasts balayés en plus du top général (id officiel → libellé) _GENRE_CHARTS = { 1301: "Arts", 1321: "Affaires", 1303: "Humour", 1304: "Éducation", 1483: "Fiction", 1511: "Gouvernement", 1512: "Santé et mieux-être", 1487: "Histoire", 1305: "Enfants et famille", 1502: "Loisirs", 1310: "Musique", 1489: "Nouvelles", 1314: "Religion et spiritualité", 1533: "Sciences", 1324: "Société et culture", 1545: "Sports", 1318: "Technologies", 1488: "Crimes véritables", 1309: "Télé et cinéma", } _HEAD_BYTES = 40_000 # tête du flux RSS lue pour l'attribution QC def feed_head_is_quebec(head: str) -> bool: """Attribution QC PRUDENTE par l'en-tête du canal RSS (avant ). Exige les DEUX : canal francophone ( fr…) ET marqueur québécois dans les balises descriptives du canal. Un balado anglo qui mentionne « Montreal » (dates de tournée…) ne passe JAMAIS (constaté 2026-08-18 : Giggly Squad/MeidasTouch dans le top 100 CA) ; un balado QC anglophone sans marqueur au nom est manqué — mieux vaut une fiche de moins qu'une attribution erronée (§7/§15). """ channel = (head or "").split("\s*([A-Za-z-]+)", channel, re.I) if not lang or not lang.group(1).lower().startswith("fr"): return False # balises descriptives du canal uniquement fields = re.findall( r"<(?:title|description|itunes:summary|itunes:author|copyright|" r"managingEditor)[^>]*>(.*?) None: super().__init__() self.errors = 0 def _chart(self) -> list[dict]: data = self.get(CHARTS_URL).json() return (data.get("feed") or {}).get("results") or [] def _genre_chart(self, gid: int) -> list[dict]: """Top 100 CA d'un genre (API RSS iTunes) → entrées normalisées.""" data = self.get(GENRE_CHART_URL.format(gid=gid)).json() out: list[dict] = [] for rank, e in enumerate((data.get("feed") or {}).get("entry") or [], start=1): cid = (((e.get("id") or {}).get("attributes") or {}) .get("im:id") or "") out.append({ "id": cid, "name": ((e.get("im:name") or {}).get("label") or "").strip(), "artistName": ((e.get("im:artist") or {}).get("label") or "").strip(), "summary": ((e.get("summary") or {}).get("label") or ""), "url": ((e.get("id") or {}).get("label") or ""), "rank": rank, }) return out def _lookup(self, ids: list[str]) -> dict[str, dict]: """Lookup iTunes GROUPÉ (officiel) : une seule requête pour le top.""" resp = self.get(LOOKUP_URL, params={"id": ",".join(ids), "country": "CA"}) return {str(r.get("collectionId")): r for r in resp.json().get("results") or [] if r.get("collectionId")} def _feed_head(self, url: str) -> str: """Début du flux RSS (40 Ko max) — assez pour l'en-tête du canal.""" self._throttle() resp = self.session.get(url, timeout=15, stream=True) try: resp.raise_for_status() chunk = next(resp.iter_content(_HEAD_BYTES), b"") or b"" finally: resp.close() self._last_request = time.time() return chunk.decode("utf-8", "ignore") def fetch(self) -> list[Creator]: try: chart = self._chart() except Exception as exc: print(f"[crea-ka] ⚠ palmares-apple : palmarès indisponible : {exc}", file=sys.stderr) self.errors += 1 chart = [] candidates: dict[str, dict] = {} for rank, entry in enumerate(chart, start=1): cid = str(entry.get("id") or "") if cid: candidates[cid] = {**entry, "rank": rank, "chart": None} # tops par genre : préfiltre nom/artiste/résumé (limite les lectures # RSS) — l'attribution finale reste nom/artiste ou tête de flux for gid, label in _GENRE_CHARTS.items(): try: genre_entries = self._genre_chart(gid) except Exception: self.errors += 1 continue for entry in genre_entries: cid = str(entry.get("id") or "") if not cid or cid in candidates: continue if not is_quebec(f"{entry.get('name', '')} " f"{entry.get('artistName', '')} " f"{entry.get('summary', '')}"): continue candidates[cid] = {**entry, "chart": label} if not candidates: return [] details: dict[str, dict] = {} try: details = self._lookup(list(candidates)) except Exception: self.errors += 1 # dégradé : on continue avec les palmarès seuls creators: list[Creator] = [] for cid, entry in candidates.items(): rank = entry["rank"] name = (entry.get("name") or "").strip() artist = (entry.get("artistName") or "").strip() if not name: continue info = details.get(cid) or {} feed = info.get("feedUrl") or "" # attribution QC conservatrice : nom/artiste, sinon tête du flux quebec = is_quebec(f"{name} {artist}") if not quebec and feed: try: quebec = feed_head_is_quebec(self._feed_head(feed)) except Exception: pass # flux injoignable → pas d'attribution, on passe if not quebec: continue url = entry.get("url") or info.get("collectionViewUrl") or feed if not url: continue acc = account("podcast", cid, "profil_source", url=url) dormant = _dormant(info.get("releaseDate")) chart = entry.get("chart") # None = top général, sinon genre acc.metrics = {k: v for k, v in { "chart_rank_ca": None if chart else rank, "chart_genre": chart, "chart_rank_genre": rank if chart else None, "episodes": info.get("trackCount"), "genre": info.get("primaryGenreName"), "feed_url": feed, "artist_id": info.get("artistId"), "last_episode": info.get("releaseDate"), "dormant": dormant or None, }.items() if v} genres = [g.get("name") for g in (entry.get("genres") or []) if isinstance(g, dict)] niches = sorted({_GENRE_NICHE[g] for g in [info.get("primaryGenreName"), *genres] if g in _GENRE_NICHE}) or ["actualite-opinion"] creators.append(Creator( source=self.source_id, external_id=cid, display_name=name, bio=f"Balado de {artist}." if artist and artist != name else "", niches=niches, creator_type="podcasteur", avatar_url=info.get("artworkUrl600") or entry.get("artworkUrl100"), status="inactive" if dormant else "active", platforms=[acc], notes=(f"Top 100 balados Apple Podcasts Canada — " f"{chart} (rang {rank})." if chart else f"Top 100 balados Apple Podcasts Canada (rang {rank})."), source_ids=[f"{self.source_id}:{cid}"], )) return creators