Roman
d5fa546dd7
- Sans Sonarr, les séries/films téléchargés donnent leurs genres via leur fiche French-Stream (page_url des téléchargements), convertis vers les catégories Kitsu (mapping FR→EN ; genres sans équivalent ignorés) - Aucun historique (ni téléchargement, ni favori, ni Sonarr) : réponse cold_start non cachée → carte d'amorçage « téléchargez un premier titre » - Fin de téléchargement → invalidation immédiate des recommandations - genres_of_page ignore les fiches hors French-Stream (URLs d'animés)
430 lines
18 KiB
Python
430 lines
18 KiB
Python
"""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",
|
|
},
|
|
# Parcours par genre (page Explorer) : type → clé → {path, label}.
|
|
# Chemins vérifiés en live : films = /films/<genre>/, séries = /<genre>-series-/.
|
|
"browse": {
|
|
"serie": {
|
|
"aventure": {"path": "/aventure-series-/", "label": "Aventure"},
|
|
"familles": {"path": "/familles-series-/", "label": "Famille"},
|
|
"fantastique": {"path": "/fantastique-series-/", "label": "Fantastique"},
|
|
"judiciaire": {"path": "/judiciare-series-/", "label": "Judiciaire"}, # coquille du site
|
|
"medical": {"path": "/medical-series-/", "label": "Médical"},
|
|
"romance": {"path": "/romance-series-/", "label": "Romance"},
|
|
"science-fiction": {"path": "/science-fiction-series-/", "label": "Science-Fiction"},
|
|
"thriller": {"path": "/thriller-series-/", "label": "Thriller"},
|
|
"western": {"path": "/western-series-/", "label": "Western"},
|
|
},
|
|
"film": {
|
|
"actions": {"path": "/films/actions/", "label": "Action"},
|
|
"animations": {"path": "/films/animations/", "label": "Animation"},
|
|
"aventures": {"path": "/films/aventures/", "label": "Aventure"},
|
|
"biopics": {"path": "/films/biopics/", "label": "Biopic"},
|
|
"comedies": {"path": "/films/comedies/", "label": "Comédie"},
|
|
"cultes": {"path": "/films/cultes/", "label": "Culte"},
|
|
"documentaires": {"path": "/films/documentaires/", "label": "Documentaire"},
|
|
"drames": {"path": "/films/drames/", "label": "Drame"},
|
|
"epouvante-horreurs": {"path": "/films/epouvante-horreurs/", "label": "Épouvante-Horreur"},
|
|
"espionnages": {"path": "/films/espionnages/", "label": "Espionnage"},
|
|
"familles": {"path": "/films/familles/", "label": "Famille"},
|
|
"fantastiques": {"path": "/films/fantastiques/", "label": "Fantastique"},
|
|
"guerres": {"path": "/films/guerres/", "label": "Guerre"},
|
|
"historiques": {"path": "/films/historiques/", "label": "Historique"},
|
|
"policiers": {"path": "/films/policiers/", "label": "Policier"},
|
|
"romances": {"path": "/films/romances/", "label": "Romance"},
|
|
"science-fictions": {"path": "/films/science-fictions/", "label": "Science-Fiction"},
|
|
"thrillers": {"path": "/films/thrillers/", "label": "Thriller"},
|
|
"westerns": {"path": "/films/westerns/", "label": "Western"},
|
|
},
|
|
},
|
|
}
|
|
|
|
_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()
|
|
soup = await fetch_soup(f"{self.base_url}{config['latest']['path']}")
|
|
results = self._short_blocks(soup)
|
|
logger.info("french_stream : %d nouveautés récupérées", len(results))
|
|
return results
|
|
|
|
async def browse(self, media_type: str, category: str) -> list[SearchResult]:
|
|
"""Parcours par genre — pages /films/<genre>/ et /<genre>-series-/.
|
|
|
|
``media_type`` (« serie » | « film ») et ``category`` (clé du catalogue
|
|
YAML, ex. « thriller »). Le type est forcé sur les résultats : une page
|
|
genre séries ne liste que des séries, une page genre films que des films.
|
|
"""
|
|
catalog = _merged_config().get("browse", {}).get(media_type, {})
|
|
entry = catalog.get(category)
|
|
if entry is None:
|
|
raise ScrapeError(f"Catégorie inconnue : {media_type}/{category}")
|
|
soup = await fetch_soup(f"{self.base_url}{entry['path']}")
|
|
results = self._short_blocks(soup, force_type=media_type)
|
|
logger.info("french_stream : %d titres dans %s/%s", len(results), media_type, category)
|
|
return results
|
|
|
|
def _short_blocks(self, soup: BeautifulSoup, force_type: str | None = None) -> list[SearchResult]:
|
|
"""Bloc `div.short` → SearchResult (structure partagée nouveautés/genres)."""
|
|
config = _merged_config()
|
|
latest_cfg = config["latest"]
|
|
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=force_type or _media_type(title, href),
|
|
)
|
|
)
|
|
return results
|
|
|
|
def browse_catalog(self) -> dict[str, dict[str, str]]:
|
|
"""Catalogue des genres parcourables : ``{type: {clé: libellé}}``."""
|
|
catalog = _merged_config().get("browse", {})
|
|
return {
|
|
media_type: {key: entry.get("label", key) for key, entry in genres.items()}
|
|
for media_type, genres in catalog.items()
|
|
}
|
|
|
|
|
|
async def genres_of_page(self, page_url: str) -> list[str]:
|
|
"""Genres (libellés FR) de la fiche pointée par l'URL d'un téléchargement.
|
|
|
|
Utilisé par les recommandations « Pour toi » pour les séries/films,
|
|
absents de Kitsu (et donc sans signal de genre sans Sonarr).
|
|
"""
|
|
if self.base_url.split("//")[-1].split("/")[0] not in page_url:
|
|
return [] # fiche d'une autre source (animé) : rien à lire ici
|
|
source_id = _newsid_from_url(page_url)
|
|
try:
|
|
details = await self.get_details(source_id)
|
|
except ScrapeError:
|
|
return []
|
|
return details.genres or []
|
|
# ------------------------------------------------------------- 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
|