Roman/ohm_streaming
1
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

Ajout de la source VoirAnime (voiranime.xyz)

Browse Source 
- Scraper catalogue : recherche /catalogue?q=, fiches + épisodes
  (cartes S·E, multi-saisons), chaîne de lecteurs prepare/fallback
  avec dédoublonnage par id de source, latest via carrousels accueil.
- Extracteur hoster : endpoint prepare (AJAX) → MP4 direct décodé du
  payload (Referer d'origine) ou HLS relayé par le proxy du site
  (CDN signés pour leur backend → accès direct en 403).
- fetch() accepte désormais des en-têtes additionnels (X-Requested-With).
- 9 tests sur fixtures HTML réelles ; README mis à jour.
This commit is contained in:
Roman
2026-09-23 10:51:29 +00:00
parent 5fedc4f9d7
commit 7399f4b781
5 changed files with 616 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+4 -3
View File
@@ -111,11 +111,12 @@ Variables d'environnement (préfixe `OHM_`, voir `.env.example`) :
## Fonctionnalités
- **Recherche unifiée** sur plusieurs sources (Vostfree, French-Manga) — chaque source
- **Recherche unifiée** sur plusieurs sources (Vostfree, French-Manga, VoirAnime) — chaque source
est un module interchangeable activable/désactivable à chaud (page Admin), dont l'URL
est modifiable à la volée (utile si un site change de domaine).
- **Extraction en 2 niveaux** : page d'épisode → lecteurs embarqués → URL directe
(Sibnet, SendVid, VidMoly, Uqload, Vidzy, Luluvdo).
(Sibnet, SendVid, VidMoly, Uqload, Vidzy, Luluvdo ; VoirAnime résout via son
endpoint « prepare » : MP4 direct ou HLS relayé par le proxy du site).
- **Proxy vidéo intégré** (`/api/proxy`) : contourne les protections (tokens liés à l'IP, Referer/UA obligatoires), réécrit les playlists HLS.
- **Streaming HLS** via hls.js ; **téléchargement HLS** via ffmpeg (remux mp4, progression temps réel).
- **Métadonnées enrichies** via Kitsu (synopsis, genres, note, images) avec cache 72 h.
@@ -198,7 +199,7 @@ app/
├── scrapers/
│ ├── base.py # contrats SourceScraper/HosterExtractor + registres
│ ├── configs/ # sélecteurs YAML externalisés (réparer sans coder)
│ ├── sources/ # vostfree, french_manga
│ ├── sources/ # vostfree, french_manga, voiranime
├── services/ # downloads, kitsu, discover, sonarr, torznab, settings
└── templates/ + static/ # UI htmx + Alpine.js, thème sombre
```
app/scrapers/hosters/voiranime.py
+111
View File
@@ -0,0 +1,111 @@
"""Extracteur VoirAnime — endpoint « prepare » → URL lisible (directe ou proxy site).
Chaque source d'un épisode voiranime.xyz se résout via
`/lecteur/prepare/<source_id>?content=..&episode=..` (AJAX : en-têtes Accept JSON et
X-Requested-With obligatoires) qui renvoie :
`{"success": true, "media_type": "mp4"|"hls", "stream_url": "/proxy/media?payload=<b64>"}`.
Le payload base64 est un JSON `{"url": ..., "referer": ..., "kind": ...}` où `url` est
l'URL directe chez l'hébergeur d'origine. Deux cas (vérifiés en live) :
- mp4 : l'URL d'origine est libre d'accès avec son Referer (Sibnet…) → on la retourne
directement (téléchargement/Range natifs, sans double saut).
- hls : les CDN utilisés (SmoothPre & co) signent leurs playlists pour le backend du
site — accès direct 403. Le proxy `/proxy/media?payload=…` du site, lui, sert la
playlist ET ses segments (chemins re-hostés) → on retourne l'URL proxy absolue.
"""
import base64
import binascii
import json
import logging
from urllib.parse import parse_qs, urlparse
from app.scrapers.base import HosterExtractor, ScrapeError, VideoLink, register_hoster
from app.scrapers.http import fetch
logger = logging.getLogger(__name__)
@register_hoster
class VoirAnimeExtractor(HosterExtractor):
name = "voiranime"
domains = ("voiranime.xyz",)
async def extract(self, embed_url: str) -> VideoLink:
origin = self._origin_of(embed_url)
raw = await fetch(
embed_url,
referer=self._referer_for(embed_url),
headers={
"Accept": "application/json",
"X-Requested-With": "XMLHttpRequest",
},
)
try:
data = json.loads(raw)
except json.JSONDecodeError as exc:
raise ScrapeError(f"voiranime : réponse prepare invalide : {raw[:120]}") from exc
if not data.get("success"):
raise ScrapeError(f"voiranime : {data.get('error') or 'échec de préparation du lecteur'}")
media_type = str(data.get("media_type") or "").lower()
stream_url = str(data.get("stream_url") or "")
if media_type == "hls":
if not stream_url:
raise ScrapeError(f"voiranime : flux HLS sans stream_url dans {embed_url}")
return VideoLink(
url=f"{origin}/{stream_url.lstrip('/')}",
hoster=self.name,
headers={"Referer": f"{origin}/"},
is_hls=True,
)
payload = self._extract_payload(stream_url)
url = str(payload.get("url") or "")
if not url:
raise ScrapeError(f"voiranime : payload sans URL de flux dans {embed_url}")
referer = str(payload.get("referer") or "")
return VideoLink(
url=url,
hoster=self.name,
headers={"Referer": referer} if referer else {},
is_hls=url.split("?")[0].endswith(".m3u8"),
)
@staticmethod
def _origin_of(embed_url: str) -> str:
parsed = urlparse(embed_url)
return f"{parsed.scheme}://{parsed.netloc}"
@staticmethod
def _referer_for(embed_url: str) -> str:
"""Reconstruit la page lecteur d'où provient la requête prepare."""
origin = VoirAnimeExtractor._origin_of(embed_url)
query = parse_qs(urlparse(embed_url).query)
content = query.get("content", [""])[0]
episode = query.get("episode", [""])[0]
if content and episode:
return f"{origin}/lecteur/{content}/{episode}"
return f"{origin}/"
@staticmethod
def _extract_payload(stream_url: str) -> dict:
payload_b64 = parse_qs(urlparse(stream_url).query).get("payload", [""])[0]
if not payload_b64:
raise ScrapeError(f"voiranime : pas de payload dans stream_url ({stream_url[:80]})")
padded = payload_b64 + "=" * (-len(payload_b64) % 4)
try:
decoded = base64.b64decode(padded)
except (binascii.Error, ValueError):
try:
decoded = base64.urlsafe_b64decode(padded)
except (binascii.Error, ValueError) as exc:
raise ScrapeError(f"voiranime : payload illisible : {payload_b64[:60]}") from exc
try:
data = json.loads(decoded)
except (json.JSONDecodeError, UnicodeDecodeError) as exc:
raise ScrapeError(f"voiranime : payload non-JSON : {payload_b64[:60]}") from exc
if not isinstance(data, dict):
raise ScrapeError("voiranime : payload inattendu (pas un objet JSON)")
return data
app/scrapers/http.py
+5 -2
View File
@@ -41,13 +41,16 @@ async def fetch(
*,
referer: str | None = None,
retries: int = 2,
headers: dict[str, str] | None = None,
) -> str:
"""GET d'une page avec retries ; lève ScrapeError en cas d'échec définitif."""
headers = {"Referer": referer} if referer else {}
request_headers = dict(headers) if headers else {}
if referer:
request_headers.setdefault("Referer", referer)
last_error: Exception | None = None
for attempt in range(retries + 1):
try:
response = await get_client().get(url, headers=headers)
response = await get_client().get(url, headers=request_headers)
response.raise_for_status()
return response.text
except (httpx.HTTPError, httpx.InvalidURL) as exc:
app/scrapers/sources/voiranime.py
+258
View File
@@ -0,0 +1,258 @@
"""Source VoirAnime (voiranime.xyz) — plateforme VODSPHERE, animes VOSTFR.
Faits structurels (vérifiés en live) :
- Recherche : GET /catalogue?q=<q> → `article.catalogue-card` (lien `a.catalogue-poster`,
poster et titre dans `img[src]` / `img[alt]`).
- Fiche : /catalogue/<slug> — titre `h1`, poster `img[alt^="Affiche de"]`, compteur
d'épisodes dans `.media-detail-meta span` (« 1192 épisodes »), synopsis `.media-detail-story p`.
- Épisodes : `a.detail-episode-card` avec `data-season` et libellé `<small>S1 · E1</small>`.
- Lecteur : /lecteur/<content>/<episode> — attributs data-prepare-url (résolution AJAX)
et data-fallback-source-url (source suivante) ; en fin de chaîne le fallback repointe
vers une source déjà vue → dédoublonnage par id de source obligatoire.
- Accueil : carrousels #carousel-new (nouveautés) et #carousel-added (ajouts), 24 cartes.
- La résolution d'un lien direct se fait dans l'extracteur hoster dédié (voir
app/scrapers/hosters/voiranime.py) : cette source ne renvoie que les URLs prepare.
"""
import logging
import re
from copy import deepcopy
from urllib.parse import quote_plus, urljoin
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_soup
logger = logging.getLogger(__name__)
DEFAULT_CONFIG: dict = {
"search": {
"endpoint": "/catalogue?q={query}",
"result": "article.catalogue-card",
"link": "a.catalogue-poster",
},
"details": {
"title": "h1",
"poster": 'img[alt^="Affiche de"]',
"meta_count": ".media-detail-meta span",
"synopsis": ".media-detail-story p",
},
"episodes": {
"card": "a.detail-episode-card",
},
"player": {
"root": "[data-prepare-url]",
"prepare_attr": "data-prepare-url",
"fallback_attr": "data-fallback-source-url",
"max_sources": 4,
},
"latest": {
"path": "/",
"carousels": ("#carousel-new", "#carousel-added"),
"item": "article.content-card",
"link": "a.card-poster",
},
}
_SLUG_RE = re.compile(r"/catalogue/([a-z0-9-]+)", re.IGNORECASE)
_EPISODE_LABEL_RE = re.compile(r"S\s*(\d+)\s*·\s*E\s*(\d+(?:[.,]\d+)?)", re.IGNORECASE)
_EPISODE_COUNT_RE = re.compile(r"(\d+)\s*épisodes", re.IGNORECASE)
_SOURCE_ID_RE = re.compile(r"/lecteur/prepare/(\d+)")
_FALLBACK_SOURCE_RE = re.compile(r"[?&]source=(\d+)")
def _merged_config() -> dict:
merged = deepcopy(DEFAULT_CONFIG)
for key, values in load_scraper_config("voiranime").items():
if isinstance(values, dict) and isinstance(merged.get(key), dict):
merged[key].update(values)
else:
merged[key] = values
return merged
@register_source
class VoirAnimeScraper(SourceScraper):
name = "voiranime"
label = "VoirAnime"
base_url = "https://voiranime.xyz"
media_types = ("anime",)
# ------------------------------------------------------------- helpers
def _title_url(self, source_id: str) -> str:
return f"{self.base_url}/catalogue/{source_id}"
@staticmethod
def _slug_from_url(url: str) -> str | None:
match = _SLUG_RE.search(url)
return match.group(1) if match else None
@staticmethod
def _text(element: Tag | None) -> str:
return element.get_text(" ", strip=True) if element else ""
def _result_from_card(self, link: Tag) -> SearchResult | None:
href = link.get("href")
if not href:
return None
source_id = self._slug_from_url(str(href))
if not source_id:
return None
img = link.select_one("img")
title = (img.get("alt") if img else None) or source_id.replace("-", " ")
image = img.get("src") if img else None
return SearchResult(
source=self.name,
source_id=source_id,
title=str(title).strip(),
url=self._title_url(source_id),
image_url=str(image) if image else None,
)
# ------------------------------------------------------------- search
async def search(self, query: str) -> list[SearchResult]:
config = _merged_config()
search_cfg = config["search"]
url = self.base_url + search_cfg["endpoint"].format(query=quote_plus(query))
soup = await fetch_soup(url)
results: list[SearchResult] = []
seen: set[str] = set()
for card in soup.select(search_cfg["result"]):
link = card.select_one(search_cfg["link"])
if link is None:
logger.warning("voiranime : carte de résultat sans lien, ignorée")
continue
result = self._result_from_card(link)
if result and result.source_id not in seen:
seen.add(result.source_id)
results.append(result)
return results
# ------------------------------------------------------------- détails
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 = self._text(soup.select_one(details_cfg["title"])) or source_id.replace("-", " ")
poster = soup.select_one(details_cfg["poster"])
image = str(poster["src"]) if poster and poster.get("src") else None
synopsis = self._text(soup.select_one(details_cfg["synopsis"])) or None
episode_count: int | None = None
count_el = soup.select_one(details_cfg["meta_count"])
if count_el:
match = _EPISODE_COUNT_RE.search(self._text(count_el))
if match:
episode_count = int(match.group(1))
episodes = self._parse_episodes(soup, config)
return TitleDetails(
source=self.name,
source_id=source_id,
title=title,
url=url,
synopsis=synopsis,
image_url=image,
episode_count=episode_count if episode_count is not None else len(episodes),
episodes=episodes,
)
# ------------------------------------------------------------- épisodes
async def list_episodes(self, source_id: str) -> list[Episode]:
config = _merged_config()
soup = await fetch_soup(self._title_url(source_id))
return self._parse_episodes(soup, config)
def _parse_episodes(self, soup: BeautifulSoup, config: dict) -> list[Episode]:
episodes: list[Episode] = []
for card in soup.select(config["episodes"]["card"]):
href = card.get("href")
if not href:
continue
label = self._text(card)
match = _EPISODE_LABEL_RE.search(label)
if not match:
logger.warning("voiranime : libellé d'épisode illisible : %r", label[:60])
continue
episodes.append(
Episode(
number=float(match.group(2).replace(",", ".")),
title=None,
url=urljoin(self.base_url + "/", str(href)),
season=int(match.group(1)),
)
)
return episodes
# ------------------------------------------------------------- lecteurs
async def extract_embed_links(self, episode_url: str) -> list[str]:
"""Parcourt la chaîne Lecteur 1 → 2 → … et renvoie les URLs prepare absolues."""
config = _merged_config()
player_cfg = config["player"]
prepare_urls: list[str] = []
seen_sources: set[str] = set()
page_url = episode_url
for _ in range(int(player_cfg["max_sources"])):
soup = await fetch_soup(page_url)
root = soup.select_one(player_cfg["root"])
if root is None:
break
prepare = root.get(player_cfg["prepare_attr"])
if not prepare:
break
prepare_url = urljoin(self.base_url + "/", str(prepare))
match = _SOURCE_ID_RE.search(prepare_url)
source_key = match.group(1) if match else prepare_url
if source_key in seen_sources:
break
seen_sources.add(source_key)
prepare_urls.append(prepare_url)
fallback = root.get(player_cfg["fallback_attr"])
if not fallback:
break
page_url = urljoin(self.base_url + "/", str(fallback))
next_source = _FALLBACK_SOURCE_RE.search(page_url)
if next_source and next_source.group(1) in seen_sources:
break
if not prepare_urls:
raise ScrapeError(f"voiranime : aucun lecteur trouvé sur {episode_url}")
return prepare_urls
# ------------------------------------------------------------- découverte
async def latest(self) -> list[SearchResult]:
"""Nouveautés + ajouts récents — carrousels de la page d'accueil."""
config = _merged_config()
latest_cfg = config["latest"]
soup = await fetch_soup(self.base_url + latest_cfg["path"])
results: list[SearchResult] = []
seen: set[str] = set()
for carousel in latest_cfg["carousels"]:
section = soup.select_one(carousel)
if section is None:
logger.debug("voiranime : carrousel %s absent de l'accueil", carousel)
continue
for card in section.select(latest_cfg["item"]):
link = card.select_one(latest_cfg["link"])
if link is None:
continue
result = self._result_from_card(link)
if result and result.source_id not in seen:
seen.add(result.source_id)
results.append(result)
return results
tests/test_voiranime.py
+238
View File
@@ -0,0 +1,238 @@
"""Tests du scraper VoirAnime (source + extracteur prepare) sur fixtures HTML réelles."""
import base64
import json
import pytest
from app.scrapers.base import ScrapeError, get_source, import_all_scrapers, resolve_hoster
from app.scrapers.hosters.voiranime import VoirAnimeExtractor
from app.scrapers.sources.voiranime import VoirAnimeScraper
# ------------------------------------------------------------ fixtures HTML
SEARCH_HTML = """
<div class="catalogue-grid">
<article class="catalogue-card catalogue-media-card">
<a href="/catalogue/one-piece" class="catalogue-poster">
<img src="https://img.example/one-piece.jpg" alt="one piece">
</a>
</article>
<article class="catalogue-card catalogue-media-card">
<a href="/catalogue/one-punch-man" class="catalogue-poster">
<img src="https://img.example/opm.jpg" alt="one punch man">
</a>
</article>
<article class="catalogue-card catalogue-media-card">
<a href="/catalogue/one-outs" class="catalogue-poster">
<img src="/assets/img/placeholders/content-placeholder.svg" alt="one outs">
</a>
</article>
</div>
"""
DETAILS_HTML = """
<h1>solo leveling</h1>
<div class="media-detail-meta"><span>26 épisodes</span></div>
<img src="https://img.example/solo.jpg" alt="Affiche de solo leveling">
<div class="media-detail-story">
<span class="media-detail-section-label">Synopsis</span>
<p> Sung Jinwoo, chasseur le plus faible. </p>
</div>
<a href="/lecteur/965/29054" class="detail-episode-card" data-season="1">
<small>S1 · E1</small><span>Episode 1</span>
</a>
<a href="/lecteur/965/29055" class="detail-episode-card" data-season="1">
<small>S1 · E2</small><span>Episode 2</span>
</a>
<a href="/lecteur/965/29060" class="detail-episode-card" data-season="2">
<small>S2 · E1</small><span>Episode 1</span>
</a>
<a href="/lecteur/965/29061" class="detail-episode-card" data-season="2">
<small>S2 · E2.5</small><span>Episode 2.5</span>
</a>
"""
PLAYER_PAGE_1 = """
<main class="tv-player" data-player
data-content-id="1626" data-episode-id="46416" data-source-id="169261"
data-prepare-url="/lecteur/prepare/169261?content=1626&amp;episode=46416"
data-fallback-source-url="/lecteur/1626/46416?lang=vostfr&amp;source=169262&amp;previous_source=169261">
</main>
"""
# fin de chaîne : le fallback repointe vers la source 1 (aller-retour observé en live)
PLAYER_PAGE_2 = """
<main class="tv-player" data-player
data-content-id="1626" data-episode-id="46416" data-source-id="169262"
data-prepare-url="/lecteur/prepare/169262?content=1626&amp;episode=46416"
data-fallback-source-url="/lecteur/1626/46416?lang=vostfr&amp;source=169261&amp;previous_source=169262">
</main>
"""
HOME_HTML = """
<div class="carousel" id="carousel-new">
<article class="content-card">
<a href="/catalogue/you-and-i-are-polar-opposites" class="card-poster">
<img src="https://img.example/polar.jpg" alt="you and i are polar opposites">
</a>
</article>
</div>
<div class="carousel" id="carousel-added">
<article class="content-card">
<a href="/catalogue/you-and-i-are-polar-opposites" class="card-poster">
<img src="https://img.example/polar.jpg" alt="you and i are polar opposites">
</a>
</article>
<article class="content-card">
<a href="/catalogue/solo-leveling" class="card-poster">
<img src="https://img.example/solo.jpg" alt="solo leveling">
</a>
</article>
</div>
"""
def _soup(html: str):
from bs4 import BeautifulSoup
return BeautifulSoup(html, "lxml")
def _prepare_payload(url: str, referer: str) -> str:
raw = json.dumps({"url": url, "referer": referer, "kind": "file"}).encode()
return base64.b64encode(raw).decode()
# ------------------------------------------------------------ source
async def test_search_parses_catalogue_cards(monkeypatch):
async def fake_fetch_soup(url, **kwargs):
assert "q=one+piece" in url
return _soup(SEARCH_HTML)
monkeypatch.setattr("app.scrapers.sources.voiranime.fetch_soup", fake_fetch_soup)
results = await VoirAnimeScraper().search("one piece")
assert [r.source_id for r in results] == ["one-piece", "one-punch-man", "one-outs"]
assert results[0].title == "one piece"
assert results[0].url == "https://voiranime.xyz/catalogue/one-piece"
assert results[0].image_url == "https://img.example/one-piece.jpg"
assert results[2].image_url == "/assets/img/placeholders/content-placeholder.svg"
async def test_get_details_and_episodes(monkeypatch):
async def fake_fetch_soup(url, **kwargs):
assert url.endswith("/catalogue/solo-leveling")
return _soup(DETAILS_HTML)
monkeypatch.setattr("app.scrapers.sources.voiranime.fetch_soup", fake_fetch_soup)
scraper = VoirAnimeScraper()
details = await scraper.get_details("solo-leveling")
assert details.title == "solo leveling"
assert details.episode_count == 26
assert details.image_url == "https://img.example/solo.jpg"
assert details.synopsis.startswith("Sung Jinwoo")
episodes = details.episodes
assert len(episodes) == 4
assert (episodes[0].season, episodes[0].number) == (1, 1.0)
assert (episodes[2].season, episodes[2].number) == (2, 1.0)
assert (episodes[3].season, episodes[3].number) == (2, 2.5)
assert episodes[0].url == "https://voiranime.xyz/lecteur/965/29054"
standalone = await scraper.list_episodes("solo-leveling")
assert len(standalone) == len(episodes)
async def test_extract_embed_links_follows_fallback_chain(monkeypatch):
pages = {
"https://voiranime.xyz/lecteur/1626/46416": PLAYER_PAGE_1,
"https://voiranime.xyz/lecteur/1626/46416?lang=vostfr&source=169262&previous_source=169261": PLAYER_PAGE_2,
}
async def fake_fetch_soup(url, **kwargs):
return _soup(pages[url])
monkeypatch.setattr("app.scrapers.sources.voiranime.fetch_soup", fake_fetch_soup)
urls = await VoirAnimeScraper().extract_embed_links("https://voiranime.xyz/lecteur/1626/46416")
assert urls == [
"https://voiranime.xyz/lecteur/prepare/169261?content=1626&episode=46416",
"https://voiranime.xyz/lecteur/prepare/169262?content=1626&episode=46416",
]
async def test_extract_embed_links_no_player_raises(monkeypatch):
async def fake_fetch_soup(url, **kwargs):
return _soup("<html><body>pas de lecteur</body></html>")
monkeypatch.setattr("app.scrapers.sources.voiranime.fetch_soup", fake_fetch_soup)
with pytest.raises(ScrapeError):
await VoirAnimeScraper().extract_embed_links("https://voiranime.xyz/lecteur/1/2")
async def test_latest_merges_carousels_dedup(monkeypatch):
async def fake_fetch_soup(url, **kwargs):
assert url == "https://voiranime.xyz/"
return _soup(HOME_HTML)
monkeypatch.setattr("app.scrapers.sources.voiranime.fetch_soup", fake_fetch_soup)
results = await VoirAnimeScraper().latest()
assert [r.source_id for r in results] == ["you-and-i-are-polar-opposites", "solo-leveling"]
# ------------------------------------------------------------ extracteur hoster
async def test_hoster_decodes_prepare_payload(monkeypatch):
payload = _prepare_payload(
"https://video.sibnet.ru/v/abc123/6235096.mp4", "https://video.sibnet.ru/"
)
async def fake_fetch(url, **kwargs):
assert "X-Requested-With" in kwargs.get("headers", {})
assert kwargs.get("referer") == "https://voiranime.xyz/lecteur/1626/46416"
return json.dumps(
{"success": True, "media_type": "mp4", "stream_url": f"/proxy/media?payload={payload}"}
)
monkeypatch.setattr("app.scrapers.hosters.voiranime.fetch", fake_fetch)
link = await VoirAnimeExtractor().extract(
"https://voiranime.xyz/lecteur/prepare/169261?content=1626&episode=46416"
)
assert link.url == "https://video.sibnet.ru/v/abc123/6235096.mp4"
assert link.hoster == "voiranime"
assert link.headers == {"Referer": "https://video.sibnet.ru/"}
assert link.is_hls is False
async def test_hoster_marks_hls(monkeypatch):
payload = _prepare_payload("https://cdn.example/hls/master.m3u8", "https://cdn.example/")
async def fake_fetch(url, **kwargs):
return json.dumps(
{"success": True, "media_type": "hls", "stream_url": f"/proxy/media?payload={payload}"}
)
monkeypatch.setattr("app.scrapers.hosters.voiranime.fetch", fake_fetch)
link = await VoirAnimeExtractor().extract("https://voiranime.xyz/lecteur/prepare/1?content=2&episode=3")
# HLS signé pour le backend du site → on sert l'URL proxy du site, pas le CDN
assert link.url == f"https://voiranime.xyz/proxy/media?payload={payload}"
assert link.is_hls is True
assert link.headers == {"Referer": "https://voiranime.xyz/"}
async def test_hoster_prepare_failure_raises(monkeypatch):
async def fake_fetch(url, **kwargs):
return json.dumps({"success": False, "error": "Flux extrait inaccessible"})
monkeypatch.setattr("app.scrapers.hosters.voiranime.fetch", fake_fetch)
with pytest.raises(ScrapeError, match="Flux extrait inaccessible"):
await VoirAnimeExtractor().extract("https://voiranime.xyz/lecteur/prepare/1?content=2&episode=3")
def test_source_and_hoster_registered():
import_all_scrapers()
assert get_source("voiranime").label == "VoirAnime"
extractor = resolve_hoster("https://voiranime.xyz/lecteur/prepare/169261?content=1&episode=2")
assert extractor is not None and extractor.name == "voiranime"