root
3b405f2a42
- Implemented DomainManager in app/utils.py for TLD rotation and caching. - Created ZoneTelechargementDownloader in app/downloaders/series_sites/zonetelechargement.py. - Integrated Zone-Telechargement into series search and provider list. - Updated .gitignore to exclude domain_cache.json.
211 lines
7.7 KiB
Python
211 lines
7.7 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.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 "", ""
|