"""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 = [] # ZT typically lists episodes in a table or list of links # Links often look like: /telecharger-series/.../saison-X-episode-Y.html links = soup.find_all("a", href=re.compile(r"episode-\d+")) for i, link in enumerate(links): href = link.get("href", "") if not href.startswith("http"): href = urljoin(self.base_url, href) title = link.get_text(strip=True) ep_match = re.search(r"episode\s*(\d+)", title.lower()) ep_number = int(ep_match.group(1)) if ep_match else i + 1 episodes.append( {"episode_number": ep_number, "url": href, "title": title} ) # Sort by episode number episodes.sort(key=lambda x: x["episode_number"]) 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 "", ""