"""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.provider_id = "zonetelechargement" self.default_domain = "zone-telechargement.cam" self.test_tlds = ["cam", "net", "org", "blue", "lol", "work"] self.base_url = None # Will be set dynamically 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 not self.base_url: 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}") 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""" try: await self._ensure_base_url() logger.info(f"Searching Zone-Telechargement for: {query}") # ZT uses POST or GET for search depending on the version # Most modern versions use: /index.php?do=search search_url = f"{self.base_url}/index.php?do=search" # Form data for search data = { "do": "search", "subaction": "search", "search_start": "0", "full_search": "0", "result_from": "1", "story": query } response = await self.client.post(search_url, data=data) response.raise_for_status() html = response.text soup = BeautifulSoup(html, 'lxml') results = [] # Look for items items = soup.find_all('div', class_='shm-item') or soup.find_all('div', class_='movie-item') for item in items[:24]: link_elem = item.find('a', class_='shm-title') or item.find('a') if not link_elem: continue url = link_elem.get('href', '') if not url.startswith('http'): url = urljoin(self.base_url, url) title = link_elem.get_text(strip=True) img_elem = item.find('img') cover_image = "" if img_elem: cover_image = img_elem.get('data-src') or img_elem.get('src') or "" if cover_image and not cover_image.startswith('http'): cover_image = urljoin(self.base_url, cover_image) if title and len(title) > 2: results.append({ 'title': title, 'url': url, 'cover_image': cover_image, 'provider_id': self.provider_id }) 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 "", ""