Ajout de la source French-Stream (french-stream.lat) : séries et films

- Scraper DataLife Engine : recherche AJAX, fiches par saison, épisodes via ep-data.php (VF/VOSTFR), films via film_api.php, nouveautés /series/ - Config YAML externalisée (sélecteurs réparables sans coder) - Domaines étendus : uqload.vc, vidzy.cc/.live - 11 tests (126 au total)
2026-09-25 14:08:10 +00:00
parent 89d9a98f80
commit 79a44297fd
7 changed files with 629 additions and 9 deletions
@@ -0,0 +1,348 @@
+"""Source French-Stream (french-stream.lat) — films & séries VF/VOSTFR, moteur DataLife Engine.
+
+Faits structurels (vérifiés en live) :
+- Recherche : POST /engine/ajax/search.php (query, page) → blocs `div.search-item`
+  (lien dans onclick="location.href='...'", poster `.search-poster img`).
+- Fiche : `/index.php?newsid=<id>` (ou la jolie URL `/<id>-<slug>.html`).
+  `source_id` = newsid numérique.
+- Séries : une fiche par saison (« Titre - Saison N »). Épisodes via
+  GET /ep-data.php?id=<newsid>&format=js → JSON
+  `{"vf": {"1": {"vidzy": url, "uqload": url, ...}}, "vostfr": {...}, "info": {...}}`.
+- Films : lecteurs via GET /engine/ajax/film_api.php?id=<newsid> → JSON
+  `{"players": {"vidzy": {"default": url, "vff": url, "vostfr": url, ...}}}`.
+- URL d'épisode interne : `<fiche>#vf-3` (série) ou `<fiche>#film` (film).
+- Nouveautés : /series/ → blocs `div.short` (mix films/séries, « Saison » dans le
+  titre = série).
+"""
+
+import json
+import logging
+import re
+from copy import deepcopy
+
+from bs4 import BeautifulSoup, Tag
+
+from app.scrapers.base import (
+    Episode,
+    ScrapeError,
+    SearchResult,
+    SourceScraper,
+    TitleDetails,
+    register_source,
+)
+from app.scrapers.config_loader import load_scraper_config
+from app.scrapers.http import fetch, fetch_soup
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_CONFIG: dict = {
+    "endpoints": {
+        "search": "/engine/ajax/search.php",
+        "episodes": "/ep-data.php?id={newsid}&format=js",
+        "film": "/engine/ajax/film_api.php?id={newsid}",
+    },
+    "search": {
+        "item": "div.search-item",
+        "title": ".search-title",
+        "poster": ".search-poster img",
+    },
+    "details": {
+        "title": "h1#s-title",
+        "synopsis": "div.fdesc",
+        "synopsis_boilerplate": "p.desc-text",
+        "genres": ".facts .genres",
+        "year": ".facts .release",
+        "poster_serie": ".fposter img",
+        "poster_film": "#film-data",
+        "serie_marker": "#serie-config",
+    },
+    "latest": {
+        "path": "/series/",
+        "item": "div.short",
+        "link": "a.short-poster",
+        "image": "img",
+    },
+}
+
+_NEWSID_RE = re.compile(r"(?:newsid=|/)(\d+)(?:-[^/]*)?\.?html?$|(?:newsid=)(\d+)")
+_ONCLICK_URL_RE = re.compile(r"location\.href='([^']+)'")
+_SEASON_RE = re.compile(r"saison\s*:?\s*(\d+)", re.IGNORECASE)
+_YEAR_RE = re.compile(r"(\d{4})")
+_FRAGMENT_RE = re.compile(r"^(vf|vostfr|vo)-(\d+(?:\.\d+)?)$")
+_VERSION_ORDER = {"vf": 0, "vostfr": 1, "vo": 2}
+
+
+def _merged_config() -> dict:
+    merged = deepcopy(DEFAULT_CONFIG)
+    for key, values in load_scraper_config("french_stream").items():
+        if isinstance(values, dict) and isinstance(merged.get(key), dict):
+            merged[key].update(values)
+        else:
+            merged[key] = values
+    return merged
+
+
+def _newsid_from_url(url: str) -> str | None:
+    match = _NEWSID_RE.search(url)
+    if not match:
+        return None
+    return match.group(1) or match.group(2)
+
+
+def _media_type(title: str, url: str) -> str:
+    if _SEASON_RE.search(title) or "-saison-" in url:
+        return "serie"
+    return "film"
+
+
+@register_source
+class FrenchStreamScraper(SourceScraper):
+    name = "french_stream"
+    label = "French-Stream"
+    base_url = "https://french-stream.lat"
+    media_types = ("serie", "film")
+
+    # ------------------------------------------------------------- helpers
+
+    def _title_url(self, source_id: str) -> str:
+        return f"{self.base_url}/index.php?newsid={source_id}"
+
+    @staticmethod
+    def _text(element: Tag | None) -> str:
+        return element.get_text(" ", strip=True) if element else ""
+
+    def _is_serie(self, soup: BeautifulSoup, title: str, url: str) -> bool:
+        config = _merged_config()
+        if soup.select_one(config["details"]["serie_marker"]):
+            return True
+        return _media_type(title, url) == "serie"
+
+    # ------------------------------------------------------------- search
+
+    async def search(self, query: str) -> list[SearchResult]:
+        config = _merged_config()
+        url = f"{self.base_url}{config['endpoints']['search']}"
+        html = await fetch(
+            url, referer=f"{self.base_url}/", data={"query": query, "page": "1"}
+        )
+        soup = BeautifulSoup(html, "lxml")
+        results: list[SearchResult] = []
+        for block in soup.select(config["search"]["item"]):
+            onclick = block.get("onclick", "")
+            url_match = _ONCLICK_URL_RE.search(onclick)
+            if not url_match:
+                logger.warning("french_stream : résultat sans lien, ignoré")
+                continue
+            href = url_match.group(1)
+            source_id = _newsid_from_url(href)
+            if not source_id:
+                logger.warning("french_stream : newsid introuvable dans %s", href)
+                continue
+            title = self._text(block.select_one(config["search"]["title"]))
+            image = block.select_one(config["search"]["poster"])
+            results.append(
+                SearchResult(
+                    source=self.name,
+                    source_id=source_id,
+                    title=title,
+                    url=f"{self.base_url}{href}" if href.startswith("/") else href,
+                    image_url=image.get("src") if image else None,
+                    media_type=_media_type(title, href),
+                )
+            )
+        logger.info("french_stream : %d résultats pour %r", len(results), query)
+        return results
+
+    # ------------------------------------------------------------- latest
+
+    async def latest(self) -> list[SearchResult]:
+        """Ajouts récents — page /series/ (mix films & séries)."""
+        config = _merged_config()
+        latest_cfg = config["latest"]
+        soup = await fetch_soup(f"{self.base_url}{latest_cfg['path']}")
+        results: list[SearchResult] = []
+        for block in soup.select(latest_cfg["item"]):
+            link = block.select_one(latest_cfg["link"])
+            href = link.get("href") if link else None
+            if not href:
+                continue
+            source_id = _newsid_from_url(href)
+            if not source_id:
+                logger.warning("french_stream : newsid introuvable dans %s", href)
+                continue
+            image = link.select_one(latest_cfg["image"])
+            title = link.get("alt") or self._text(link)
+            results.append(
+                SearchResult(
+                    source=self.name,
+                    source_id=source_id,
+                    title=title,
+                    url=f"{self.base_url}{href}" if href.startswith("/") else href,
+                    image_url=image.get("src") if image else None,
+                    media_type=_media_type(title, href),
+                )
+            )
+        logger.info("french_stream : %d nouveautés récupérées", len(results))
+        return results
+
+    # ------------------------------------------------------------- details
+
+    async def get_details(self, source_id: str) -> TitleDetails:
+        config = _merged_config()
+        details_cfg = config["details"]
+        url = self._title_url(source_id)
+        soup = await fetch_soup(url)
+
+        title = " ".join(self._text(soup.select_one(details_cfg["title"])).split())
+        if not title:
+            raise ScrapeError(f"french_stream : fiche introuvable pour {source_id} ({url})")
+
+        synopsis_el = soup.select_one(details_cfg["synopsis"])
+        if synopsis_el is not None:
+            for boilerplate in synopsis_el.select(details_cfg["synopsis_boilerplate"]):
+                boilerplate.decompose()
+        synopsis = self._text(synopsis_el) or None
+
+        genres_el = soup.select_one(details_cfg["genres"])
+        if genres_el and genres_el.select("a"):
+            genres = [a.get_text(strip=True) for a in genres_el.select("a") if a.get_text(strip=True)]
+        else:
+            genres = [g.strip() for g in self._text(genres_el).split(",") if g.strip()]
+
+        year_match = _YEAR_RE.search(self._text(soup.select_one(details_cfg["year"])))
+        if year_match is None:
+            year_match = _YEAR_RE.search(title)
+
+        poster_el = soup.select_one(details_cfg["poster_serie"])
+        poster = poster_el.get("src") if poster_el else None
+        if not poster:
+            film_data = soup.select_one(details_cfg["poster_film"])
+            poster = film_data.get("data-affiche") if film_data else None
+
+        is_serie = self._is_serie(soup, title, url)
+        episodes = await self._fetch_episodes(source_id, url, title) if is_serie else []
+
+        return TitleDetails(
+            source=self.name,
+            source_id=source_id,
+            title=title,
+            url=url,
+            synopsis=synopsis,
+            image_url=poster,
+            genres=genres,
+            year=int(year_match.group(1)) if year_match else None,
+            episode_count=len(episodes) if is_serie else 1,
+            episodes=episodes if is_serie else [self._film_episode(url)],
+            media_type="serie" if is_serie else "film",
+        )
+
+    # ------------------------------------------------------------ episodes
+
+    async def list_episodes(self, source_id: str) -> list[Episode]:
+        url = self._title_url(source_id)
+        soup = await fetch_soup(url)
+        title = " ".join(
+            self._text(soup.select_one(_merged_config()["details"]["title"])).split()
+        )
+        if not self._is_serie(soup, title, url):
+            return [self._film_episode(url)]
+        episodes = await self._fetch_episodes(source_id, url, title)
+        if not episodes:
+            raise ScrapeError(f"french_stream : aucun épisode trouvé pour {source_id} ({url})")
+        return episodes
+
+    @staticmethod
+    def _film_episode(page_url: str) -> Episode:
+        return Episode(number=1, title="Film", url=f"{page_url}#film", season=1)
+
+    async def _fetch_episodes(
+        self, source_id: str, page_url: str, title: str
+    ) -> list[Episode]:
+        """Épisodes d'une saison via l'API JSON du site (versions vf/vostfr/vo)."""
+        config = _merged_config()
+        endpoint = config["endpoints"]["episodes"].format(newsid=source_id)
+        raw = await fetch(f"{self.base_url}{endpoint}", referer=page_url)
+        try:
+            data = json.loads(raw)
+        except json.JSONDecodeError as exc:
+            raise ScrapeError(
+                f"french_stream : JSON d'épisodes invalide pour {source_id}"
+            ) from exc
+        if not isinstance(data, dict):
+            return []
+
+        season_match = _SEASON_RE.search(title)
+        season = int(season_match.group(1)) if season_match else 1
+        info = data.get("info") if isinstance(data.get("info"), dict) else {}
+
+        episodes: list[Episode] = []
+        for version, eps in data.items():
+            if version == "info" or not isinstance(eps, dict):
+                continue
+            for number_key in eps:
+                number_match = re.match(r"^(\d+(?:\.\d+)?)$", number_key)
+                if not number_match:
+                    continue
+                number = float(number_key)
+                ep_info = info.get(number_key) or info.get(str(int(number))) or {}
+                episodes.append(
+                    Episode(
+                        number=number,
+                        title=ep_info.get("title") if isinstance(ep_info, dict) else None,
+                        url=f"{page_url}#{version}-{number_key}",
+                        season=season,
+                        version=version,
+                    )
+                )
+        episodes.sort(
+            key=lambda e: (e.number, _VERSION_ORDER.get(e.version or "", 99))
+        )
+        return episodes
+
+    # -------------------------------------------------------------- embeds
+
+    async def extract_embed_links(self, episode_url: str) -> list[str]:
+        config = _merged_config()
+        page_url, _, fragment = episode_url.partition("#")
+        source_id = _newsid_from_url(page_url)
+        if not source_id:
+            raise ScrapeError(f"french_stream : newsid introuvable dans {episode_url}")
+
+        if fragment == "film":
+            endpoint = config["endpoints"]["film"].format(newsid=source_id)
+            raw = await fetch(f"{self.base_url}{endpoint}", referer=page_url)
+            try:
+                data = json.loads(raw)
+            except json.JSONDecodeError as exc:
+                raise ScrapeError(
+                    f"french_stream : JSON film invalide pour {source_id}"
+                ) from exc
+            links: list[str] = []
+            for hoster_urls in (data.get("players") or {}).values():
+                if not isinstance(hoster_urls, dict):
+                    continue
+                for embed in hoster_urls.values():
+                    if isinstance(embed, str) and embed.startswith("http") and embed not in links:
+                        links.append(embed)
+        else:
+            fragment_match = _FRAGMENT_RE.match(fragment)
+            if not fragment_match:
+                raise ScrapeError(
+                    f"french_stream : fragment d'épisode invalide dans {episode_url}"
+                )
+            version, number = fragment_match.group(1), fragment_match.group(2)
+            endpoint = config["endpoints"]["episodes"].format(newsid=source_id)
+            raw = await fetch(f"{self.base_url}{endpoint}", referer=page_url)
+            try:
+                data = json.loads(raw)
+            except json.JSONDecodeError as exc:
+                raise ScrapeError(
+                    f"french_stream : JSON d'épisodes invalide pour {source_id}"
+                ) from exc
+            hosters = (data.get(version) or {}).get(number) or {}
+            links = [u for u in hosters.values() if isinstance(u, str) and u.startswith("http")]
+
+        if not links:
+            raise ScrapeError(f"french_stream : aucun lien embed extrait de {episode_url}")
+        logger.info("french_stream : %d liens embed pour %s", len(links), episode_url)
+        return links