Files
ohm_streaming/app/downloaders/series_sites/zonetelechargement.py
root dd1365eff9
CI / Test (Python 3.11) (push) Has been cancelled
CI / Test (Python 3.12) (push) Has been cancelled
CI / Lint (push) Has been cancelled
CI / Type Check (push) Has been cancelled
CI / Summary (push) Has been cancelled
fix: episodes loading in dropdown
- Fix ZT get_episodes: find episode links by text pattern instead of URL
- Episodes now load into #player-container instead of tiny dropdown div
- Scroll to player container after episodes load
2026-03-28 01:04:32 +00:00

237 lines
8.4 KiB
Python

"""Zone-Telechargement series site downloader"""
import logging
import re
from typing import List, Dict, Any, Optional, Tuple
from urllib.parse import urljoin, quote
from bs4 import BeautifulSoup
from app.utils import DomainManager
from .base import BaseSeriesSite
logger = logging.getLogger(__name__)
class ZoneTelechargementDownloader(BaseSeriesSite):
"""
Downloader for Zone-Telechargement series site.
Handles dynamic TLD verification.
"""
def __init__(self):
super().__init__()
self.id = "zonetelechargement"
self.provider_id = "zonetelechargement"
self.default_domain = "zone-telechargement.golf"
self.test_tlds = ["golf", "cam", "net", "org", "blue", "lol", "work", "ws"]
self.base_url = f"https://{self.default_domain}"
self._domain_checked = False
self.client.headers.update(
{
"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,application/xml;q=0.9,image/webp,*/*;q=0.8",
"Accept-Language": "fr-FR,fr;q=0.9,en-US;q=0.8,en;q=0.7",
}
)
async def _ensure_base_url(self):
"""Ensure base_url is set to the current active domain"""
if self._domain_checked:
return
self._domain_checked = True
try:
active_domain = await DomainManager.get_active_domain(
self.provider_id, self.default_domain, self.test_tlds, test_path="/"
)
self.base_url = f"https://{active_domain}"
logger.info(f"Using active domain for Zone-Telechargement: {self.base_url}")
except Exception as e:
logger.warning(
f"Domain check failed for Zone-Telechargement, using default: {e}"
)
def can_handle(self, url: str) -> bool:
"""Check if this downloader can handle the given URL"""
return "zone-telechargement" in url.lower() or "zt-za" in url.lower()
async def search_anime(self, query: str, lang: str = "vf") -> List[Dict[str, str]]:
"""Search for series on Zone-Telechargement.
ZT uses server-side rendered search: GET /?p=series&search=QUERY.
Results are in div.cover_global containers with nested cover_infos_title links.
"""
try:
await self._ensure_base_url()
logger.info(f"Searching Zone-Telechargement for: {query}")
search_url = f"{self.base_url}/"
params = {"p": "series", "search": query}
response = await self.client.get(search_url, params=params)
response.raise_for_status()
html = response.text
soup = BeautifulSoup(html, "lxml")
results = []
for cover_div in soup.find_all("div", class_="cover_global")[:24]:
link_in_cover = cover_div.find("a", class_="mainimg")
if not link_in_cover:
link_in_cover = cover_div.find("a")
if not link_in_cover:
continue
url = link_in_cover.get("href", "")
if not url.startswith("http"):
url = urljoin(self.base_url, url)
img = cover_div.find("img")
cover_image = ""
if img:
cover_image = img.get("data-src") or img.get("src") or ""
if cover_image and not cover_image.startswith("http"):
cover_image = urljoin(self.base_url, cover_image)
title = ""
info_div = cover_div.find("div", class_="cover_infos_title")
if info_div:
title_link = info_div.find("a")
if title_link:
title = title_link.get_text(strip=True)
else:
title = info_div.get_text(strip=True)
else:
title = link_in_cover.get("title", "")
if not title:
title = link_in_cover.get_text(strip=True)
if title and len(title) > 2:
results.append(
{
"title": title,
"url": url,
"cover_image": cover_image,
"provider_id": self.provider_id,
}
)
logger.info(
f"Zone-Telechargement found {len(results)} results for '{query}'"
)
return results
except Exception as e:
logger.error(f"Error searching Zone-Telechargement: {e}")
return []
async def get_episodes(
self, anime_url: str, lang: str = "vf"
) -> List[Dict[str, str]]:
"""Extract episodes from a series page"""
try:
await self._ensure_base_url()
html = await self._fetch_page(anime_url)
soup = BeautifulSoup(html, "lxml")
episodes = []
seen_urls = set()
# ZT lists episodes as <a> tags inside <b> inside div.postinfo
# Text matches "Episode X" pattern, URLs go through dl-protect
for link in soup.find_all("a"):
text = link.get_text(strip=True)
ep_match = re.search(r"episode\s*(\d+)", text, re.I)
if not ep_match:
continue
href = link.get("href", "")
if not href or href in seen_urls:
continue
seen_urls.add(href)
ep_number = int(ep_match.group(1))
episodes.append(
{"episode_number": ep_number, "url": href, "title": text}
)
# Sort by episode number
episodes.sort(key=lambda x: x["episode_number"])
logger.info(f"Found {len(episodes)} episodes on {anime_url}")
return episodes
except Exception as e:
logger.error(f"Error getting episodes from Zone-Telechargement: {e}")
return []
async def get_anime_metadata(self, anime_url: str) -> Dict[str, Any]:
"""Extract metadata from a series page"""
try:
await self._ensure_base_url()
html = await self._fetch_page(anime_url)
soup = BeautifulSoup(html, "lxml")
metadata = {
"title": "",
"synopsis": "",
"genres": [],
"poster_image": "",
"status": "Unknown",
}
title_elem = soup.find("h1")
if title_elem:
metadata["title"] = title_elem.get_text(strip=True)
# Synopsis
syn_elem = soup.find("div", class_="shm-description") or soup.find(
"div", class_="movie-desc"
)
if syn_elem:
metadata["synopsis"] = syn_elem.get_text(strip=True)
# Poster
img_elem = (
soup.find("div", class_="shm-img").find("img")
if soup.find("div", class_="shm-img")
else None
)
if img_elem:
metadata["poster_image"] = urljoin(
self.base_url, img_elem.get("src", "")
)
return metadata
except Exception as e:
logger.error(f"Error getting metadata from Zone-Telechargement: {e}")
return {}
async def get_download_link(self, url: str) -> Tuple[str, str]:
"""Extract video player URL from an episode page"""
try:
await self._ensure_base_url()
html = await self._fetch_page(url)
soup = BeautifulSoup(html, "lxml")
# Look for video player links (Uptobox, 1fichier, etc.)
# ZT often has multiple hosts
links = soup.find_all(
"a", href=re.compile(r"uptobox|1fichier|doodstream|vidmoly")
)
if links:
player_url = links[0].get("href", "")
title = (
soup.find("h1").get_text(strip=True)
if soup.find("h1")
else "Episode"
)
return player_url, title
return "", ""
except Exception as e:
logger.error(f"Error getting download link from Zone-Telechargement: {e}")
return "", ""