"""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 tags inside 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 "", ""