From 6cb9d24265306e03c66812eac1cf12f62d8ec57a Mon Sep 17 00:00:00 2001 From: ewgsta <159681870+ewgsta@users.noreply.github.com> Date: Mon, 13 Jul 2026 02:57:57 +0300 Subject: [PATCH 1/4] fix(animecix) repair provider and add movie support --- weeb_cli/config.py | 2 +- weeb_cli/providers/tr/animecix.py | 829 +++++++++++++++++----- weeb_cli/providers/tr/animecix_helpers.py | 358 ++++++++++ 3 files changed, 991 insertions(+), 198 deletions(-) create mode 100644 weeb_cli/providers/tr/animecix_helpers.py diff --git a/weeb_cli/config.py b/weeb_cli/config.py index c90e200..176362b 100644 --- a/weeb_cli/config.py +++ b/weeb_cli/config.py @@ -64,7 +64,7 @@ def get_default_download_dir() -> str: "ytdlp_format": "bestvideo+bestaudio/best", "scraping_source": None, "show_description": True, - "debug_mode": False, + "debug_mode": True, "download_max_retries": 3, "download_retry_delay": 10, "discord_rpc_enabled": True, diff --git a/weeb_cli/providers/tr/animecix.py b/weeb_cli/providers/tr/animecix.py index 4464b62..54381b0 100644 --- a/weeb_cli/providers/tr/animecix.py +++ b/weeb_cli/providers/tr/animecix.py @@ -1,9 +1,24 @@ +""" +Animecix provider for weeb-cli. + +This provider supports animecix.tv, a Turkish anime streaming site. +Uses modern API endpoints with dynamic header management for stable operation. + +Features: +- Dynamic header fetching with Playwright (cached for 1 hour) +- Comprehensive error handling and logging +- Support for series and movies +- Multiple season/episode handling +""" + import json -import re import time -import urllib.request -from urllib.parse import urlparse, parse_qs, quote, urlsplit, urlunsplit -from typing import List, Optional +from typing import List, Optional, Dict, Any +from urllib.parse import urlparse, parse_qs, quote + +import requests +from requests.adapters import HTTPAdapter +from urllib3.util.retry import Retry from weeb_cli.providers.base import ( BaseProvider, @@ -13,279 +28,699 @@ StreamLink ) from weeb_cli.providers.registry import register_provider +from weeb_cli.providers.tr.animecix_helpers import ( + get_dynamic_headers, + build_headers, + clean_title_name, + parse_episode_number, + parse_season_number, + retry_with_backoff +) +from weeb_cli.exceptions import ProviderError +from weeb_cli.services.logger import debug -BASE_URL = "https://animecix.tv/" -ALT_URL = "https://mangacix.net/" -VIDEO_PLAYERS = ["tau-video.xyz", "sibnet"] +# API endpoints +BASE_URL = "https://animecix.tv" +SEARCH_URL = f"{BASE_URL}/secure/search" +TITLES_URL = f"{BASE_URL}/secure/titles" +EPISODE_VIDEOS_URL = f"{BASE_URL}/secure/episode-videos-points" -HEADERS = { - "Accept": "application/json", - "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" -} +# Video player domains +VIDEO_PLAYERS = ["tau-video.xyz", "sibnet.ru"] -def _http_get(url: str, timeout: int = 15) -> bytes: - sp = urlsplit(url) - safe_path = quote(sp.path, safe="/:%@") - safe_url = urlunsplit((sp.scheme, sp.netloc, safe_path, sp.query, sp.fragment)) +def create_session() -> requests.Session: + """ + Create a requests session with retry logic. - req = urllib.request.Request(safe_url, headers=HEADERS) - with urllib.request.urlopen(req, timeout=timeout) as resp: - return resp.read() - - -def _get_json(url: str, timeout: int = 15): - try: - data = _http_get(url, timeout) - return json.loads(data) - except Exception: - return None + Returns: + Configured requests.Session with automatic retries. + """ + session = requests.Session() + + # Configure retry strategy + retry_strategy = Retry( + total=3, + backoff_factor=1, + status_forcelist=[429, 500, 502, 503, 504], + allowed_methods=["HEAD", "GET", "OPTIONS"] + ) + + adapter = HTTPAdapter(max_retries=retry_strategy) + session.mount("http://", adapter) + session.mount("https://", adapter) + + return session @register_provider("animecix", lang="tr", region="TR") class AnimeCixProvider(BaseProvider): + """ + Animecix.tv provider implementation. + + Supports searching, fetching details, and extracting streams from animecix.tv. + Uses dynamic headers and modern API endpoints for stability. + """ def __init__(self): super().__init__() + self.session = create_session() + self.dynamic_headers = None + debug("[AnimeCix] Provider initialized") + + def _get_headers(self, force_refresh: bool = False) -> Dict[str, str]: + """ + Get headers for API requests. + Args: + force_refresh: Force refresh of dynamic headers. + + Returns: + Complete headers dictionary. + """ + if self.dynamic_headers is None or force_refresh: + self.dynamic_headers = get_dynamic_headers(force_refresh) + debug(f"[AnimeCix] Dynamic headers refreshed: {list(self.dynamic_headers.keys())}") + + return build_headers(self.dynamic_headers) + + def _request_json(self, url: str, params: Optional[Dict[str, Any]] = None, + retry_count: int = 0) -> Optional[Dict[str, Any]]: + """ + Make JSON API request with error handling. + + Args: + url: Request URL. + params: Query parameters. + retry_count: Current retry attempt (internal). + + Returns: + Parsed JSON response or None on failure. + """ + try: + force_refresh = (retry_count > 1) + headers = self._get_headers(force_refresh=force_refresh) + + debug(f"[AnimeCix] GET {url} (retry={retry_count})") + if params: + debug(f"[AnimeCix] Params: {params}") + + resp = self.session.get(url, headers=headers, params=params, timeout=20) + + debug(f"[AnimeCix] Response: {resp.status_code}") + + # Handle authentication errors by refreshing headers + if resp.status_code in (401, 403) and retry_count < 3: + debug(f"[AnimeCix] Auth error, retrying (attempt {retry_count + 1}/3)") + time.sleep(1) + return self._request_json(url, params, retry_count + 1) + + resp.raise_for_status() + + data = resp.json() + return data + + except requests.exceptions.JSONDecodeError as e: + debug(f"[AnimeCix] JSON decode error: {e}") + return None + except requests.exceptions.RequestException as e: + debug(f"[AnimeCix] Request error: {e}") + return None + except Exception as e: + debug(f"[AnimeCix] Unexpected error: {e}") + return None + def search(self, query: str) -> List[AnimeResult]: - q = (query or "").strip().replace(" ", "-") - q_enc = quote(q, safe="-") - url = f"{BASE_URL}secure/search/{q_enc}?type=&limit=20" + """ + Search for anime by query string. + + Uses /secure/search/{query} endpoint. + + Args: + query: Search query (anime title). + + Returns: + List of anime search results. + + Raises: + ProviderError: If search fails. + """ + if not query or not query.strip(): + debug("[AnimeCix] Empty search query") + return [] + + debug(f"[AnimeCix] Searching for: {query}") + + # Clean and encode query + query_clean = clean_title_name(query) + url = f"{SEARCH_URL}/{quote(query_clean, safe='-')}" + params = {"type": "", "limit": 20} + + data = self._request_json(url, params) - data = _get_json(url) if not data or "results" not in data: + debug("[AnimeCix] No results found") return [] results = [] for item in data["results"]: + anime_id = item.get("id") name = item.get("name") - _id = item.get("id") - if name and _id: - results.append(AnimeResult( - id=str(_id), - title=str(name), - type=self._parse_type(item.get("title_type", "")) - )) + + if not anime_id or not name: + continue + + result = AnimeResult( + id=str(anime_id), + title=name, + type=self._parse_type(item.get("title_type", "")), + cover=item.get("poster"), + year=item.get("year") + ) + results.append(result) + debug(f"[AnimeCix] Found {len(results)} results") return results def get_details(self, anime_id: str) -> Optional[AnimeDetails]: + """ + Get detailed information for an anime. + + Uses /secure/titles/{id} endpoint to fetch full anime details. + + Args: + anime_id: Unique anime identifier. + + Returns: + Complete anime details with episodes. + """ try: - safe_id = int(anime_id) + title_id = int(anime_id) except (ValueError, TypeError): + debug(f"[AnimeCix] Invalid anime_id: {anime_id}") return None - url = f"{ALT_URL}secure/related-videos?episode=1&season=1&titleId={safe_id}&videoId=637113" - data = _get_json(url) + debug(f"[AnimeCix] Fetching details for anime_id={title_id}") - title_data = None - if data and "videos" in data: - videos = data.get("videos") or [] - if videos: - title_data = videos[0].get("title") + url = f"{TITLES_URL}/{title_id}" + data = self._request_json(url, params={"titleId": title_id}) + + if not data or "title" not in data: + debug(f"[AnimeCix] No data returned for anime_id={title_id}") + return None - episodes = self.get_episodes(anime_id) + title = data["title"] - if not episodes: - movie_url = self._get_movie_url(safe_id) - if movie_url: - title_name = title_data.get("name", "Film") if title_data else "Film" - episodes = [Episode( - id=movie_url, - number=1, - title=title_name, - url=movie_url - )] - - if not title_data: - return AnimeDetails( - id=anime_id, - title=anime_id, - episodes=episodes, - total_episodes=len(episodes) - ) + # Extract basic info + anime_name = title.get("name", "Unknown") + description = title.get("description") + poster = title.get("poster") + year = title.get("year") + anime_type = self._parse_type(title.get("type", "")) + + # Extract genres + genres = [g.get("name", "") for g in title.get("genres", []) if g.get("name")] + + # Get episodes from seasons + episodes = self._extract_episodes_from_title(title, title_id) + + debug(f"[AnimeCix] {anime_name}: {len(episodes)} episodes found") return AnimeDetails( id=anime_id, - title=title_data.get("name", ""), - description=title_data.get("description"), - cover=title_data.get("poster"), - genres=[g.get("name", "") for g in title_data.get("genres", [])], - year=title_data.get("year"), + title=anime_name, + description=description, + cover=poster, + genres=genres, + year=year, + status=title.get("status"), episodes=episodes, total_episodes=len(episodes) ) - def _get_movie_url(self, title_id: int) -> Optional[str]: - url = f"{ALT_URL}secure/titles/{title_id}" - data = _get_json(url) + def _extract_episodes_from_title(self, title: Dict[str, Any], title_id: int) -> List[Episode]: + """ + Extract episode list from title data. - if not data or "title" not in data: - return None + Since /secure/titles endpoint doesn't include episode data, + we need to fetch each season separately with seasonNumber parameter. - title = data["title"] - videos = title.get("videos") or [] + Args: + title: Title data from API. + title_id: Title ID for fetching seasons. - if videos: - return videos[0].get("url") + Returns: + List of Episode objects with composite IDs including season/episode info. + """ + episodes = [] - return None - - def get_episodes(self, anime_id: str) -> List[Episode]: - try: - safe_id = int(anime_id) - except (ValueError, TypeError): - return [] + # Check anime type - use title_type as fallback since type can be null + anime_type = title.get("type") or title.get("title_type") or "" + anime_type = anime_type.lower() if anime_type else "" + is_movie = "movie" in anime_type or "film" in anime_type + + # For movies, try videos endpoint + if is_movie: + videos = title.get("videos", []) + if videos: + debug(f"[AnimeCix] Movie detected, {len(videos)} video sources available") + # For movies, create a single episode that represents the movie + # Store all video sources in a special format for later stream extraction + # Use title_id as episode ID with special marker + episodes.append(Episode( + id=f"movie:{title_id}", # Special format for movies + number=1, + title=title.get("name", "Movie"), + season=1, + url=None + )) + return episodes + else: + debug(f"[AnimeCix] Movie detected but no videos found") + return [] + + # For series, fetch each season with seasonNumber parameter + seasons = title.get("seasons", []) + season_count = title.get("season_count", len(seasons)) + + if not seasons and season_count: + debug(f"[AnimeCix] No seasons array, using season_count={season_count}") + seasons = [{"number": i} for i in range(1, season_count + 1)] - seasons = self._get_seasons(safe_id) if not seasons: - seasons = [0] + debug(f"[AnimeCix] No seasons found for title_id={title_id}") + return [] - episodes = [] - seen = set() + debug(f"[AnimeCix] Fetching {len(seasons)} seasons") + + # Fetch each season's episodes + title_name = clean_title_name(title.get("name_romanji") or title.get("name", "")) - for sidx in seasons: - url = f"{ALT_URL}secure/related-videos?episode=1&season={sidx+1}&titleId={safe_id}&videoId=637113" - data = _get_json(url) + for season in seasons: + season_number = parse_season_number(season, 1) - if not data or "videos" not in data: + # Fetch season details with seasonNumber parameter + season_episodes = self._fetch_season_episodes(title_id, season_number, title_name) + episodes.extend(season_episodes) + + return episodes + + def _fetch_season_episodes(self, title_id: int, season_number: int, title_name: str) -> List[Episode]: + """ + Fetch episodes for a specific season. + + Uses /secure/titles/{id}?seasonNumber={num}&titleName={name} endpoint. + + Args: + title_id: Anime ID. + season_number: Season number. + title_name: URL-safe anime title name. + + Returns: + List of Episode objects for this season with composite IDs. + """ + params = { + "titleId": title_id, + "seasonNumber": season_number + } + + if title_name: + params["titleName"] = title_name + + url = f"{TITLES_URL}/{title_id}" + data = self._request_json(url, params) + + if not data or "title" not in data: + debug(f"[AnimeCix] Failed to fetch season {season_number} for title_id={title_id}") + return [] + + title = data["title"] + episodes = [] + + # Try to get episodes from season data + seasons = title.get("seasons", []) + + for season in seasons: + if parse_season_number(season, 0) != season_number: continue - for v in data["videos"]: - name = v.get("name") - ep_url = v.get("url") - - if not name or not ep_url: - continue - if name in seen: - continue - - seen.add(name) - ep_num = self._parse_episode_number(name, len(episodes) + 1) + # Check episodePagination + episode_pagination = season.get("episodePagination", {}) + episode_list = episode_pagination.get("data", []) + + debug(f"[AnimeCix] Season {season_number}: found {len(episode_list)} episodes") + + for ep_data in episode_list: + ep_id = ep_data.get("id") + ep_number = ep_data.get("episode_number") + ep_name = ep_data.get("name", f"Episode {ep_number}") - episodes.append(Episode( - id=ep_url, - number=ep_num, - title=name, - season=self._parse_season_from_url(ep_url, sidx + 1), - url=ep_url - )) + if ep_id and ep_number is not None: + # Create composite ID: "id:season:episode" for faster stream lookup + composite_id = f"{ep_id}:{season_number}:{ep_number}" + + episodes.append(Episode( + id=composite_id, + number=ep_number, + title=ep_name, + season=season_number, + url=None + )) return episodes + def get_episodes(self, anime_id: str) -> List[Episode]: + """ + Get list of available episodes for an anime. + + This method fetches full details and extracts episodes. + Consider using get_details() directly for better performance. + + Args: + anime_id: Unique anime identifier. + + Returns: + List of Episode objects. + """ + debug(f"[AnimeCix] Getting episodes for anime_id={anime_id}") + + details = self.get_details(anime_id) + if details: + return details.episodes + + return [] + def get_streams(self, anime_id: str, episode_id: str) -> List[StreamLink]: - embed_path = episode_id.lstrip("/") + """ + Get stream URLs for a specific episode. + + Uses /secure/episode-videos-points endpoint to get video sources. + + Note: This method expects episode_id to be in format "id:season:episode" + or will attempt to fetch details if plain ID is provided (slower). + + Args: + anime_id: Unique anime identifier. + episode_id: Episode identifier from get_episodes(). - if embed_path.startswith("http"): - full_url = embed_path - else: - full_url = f"{BASE_URL}{quote(embed_path, safe='/:?=&')}" + Returns: + List of StreamLink objects with different quality options. + """ + debug(f"[AnimeCix] get_streams START: anime_id={anime_id}, episode_id={episode_id}") try: - req = urllib.request.Request(full_url, headers=HEADERS) - resp = urllib.request.urlopen(req, timeout=15) - final_url = resp.geturl() + title_id = int(anime_id) + except (ValueError, TypeError): + debug(f"[AnimeCix] Invalid anime_id: {anime_id}") + return [] + + debug(f"[AnimeCix] Parsed title_id={title_id}") + + # Check if this is a movie (special format: "movie:{id}") + if str(episode_id).startswith("movie:"): + debug(f"[AnimeCix] Movie detected, fetching video sources directly") + return self._get_movie_streams(title_id) + + # Try to parse composite ID first (format: "id:season:episode") + ep_number = None + season_number = None + + debug(f"[AnimeCix] Checking for composite ID...") + + if ":" in str(episode_id): + parts = str(episode_id).split(":") + debug(f"[AnimeCix] Composite ID parts: {parts}") - time.sleep(1) + if len(parts) >= 3: + try: + ep_id, season_number, ep_number = int(parts[0]), int(parts[1]), int(parts[2]) + debug(f"[AnimeCix] Parsed composite: ep_id={ep_id}, season={season_number}, episode={ep_number}") + # SUCCESS: We have season and episode numbers, skip to video fetch + except ValueError as e: + debug(f"[AnimeCix] Failed to parse composite ID: {e}") + + if ep_number is None or season_number is None: + debug(f"[AnimeCix] No composite ID, fetching details (SLOW PATH)...") + details = self.get_details(anime_id) - p = urlparse(final_url) - parts = p.path.strip("/").split("/") + if not details or not details.episodes: + debug(f"[AnimeCix] No details/episodes found") + return [] - embed_id = None - if len(parts) >= 2: - if parts[0] == "embed": - embed_id = parts[1] - else: - embed_id = parts[0] - elif len(parts) == 1 and parts[0]: - embed_id = parts[0] + debug(f"[AnimeCix] Got {len(details.episodes)} episodes, searching for {episode_id}...") - qs = parse_qs(p.query) - vid = (qs.get("vid") or [None])[0] + target_episode = None + for ep in details.episodes: + # Try exact match first + if ep.id == episode_id: + target_episode = ep + break + if ":" in ep.id: + ep_parts = ep.id.split(":") + if len(ep_parts) >= 1 and ep_parts[0] == str(episode_id): + target_episode = ep + break - if not embed_id or not vid: + if not target_episode: + debug(f"[AnimeCix] Episode {episode_id} not found in {len(details.episodes)} episodes") return [] - api_url = f"https://{VIDEO_PLAYERS[0]}/api/video/{embed_id}?vid={vid}" - video_data = _get_json(api_url) + ep_number = target_episode.number + season_number = target_episode.season + debug(f"[AnimeCix] Found episode: S{season_number}E{ep_number}") + + debug(f"[AnimeCix] Resolved to S{season_number}E{ep_number}, fetching videos...") + + params = { + "titleId": title_id, + "episode": ep_number, + "season": season_number + } + + debug(f"[AnimeCix] Calling episode-videos-points API with params: {params}") + data = self._request_json(EPISODE_VIDEOS_URL, params) + + if not data: + debug(f"[AnimeCix] No data returned for episode videos") + return [] + + # Extract video sources + videos = data.get("videos", []) + if not videos: + debug(f"[AnimeCix] No videos found in response") + return [] + + debug(f"[AnimeCix] Found {len(videos)} video sources") + + streams = [] + sources_tried = 0 + max_sources = 10 + + for idx, video in enumerate(videos, 1): + if sources_tried >= max_sources: + debug(f"[AnimeCix] Reached max sources limit ({max_sources})") + break + + video_url = video.get("url") - if not video_data or "urls" not in video_data: - return [] + if not video_url: + debug(f"[AnimeCix] Video {idx}: no URL, skipping") + continue - streams = [] - for u in video_data["urls"]: - label = u.get("label") - url = u.get("url") - if url: - streams.append(StreamLink( - url=url, - quality=label or "auto", - server="tau-video" - )) - - def _quality_sort_key(stream): - quality = stream.quality.lower().replace('p', '') - try: - return -int(quality) - except ValueError: - return 1 - - streams.sort(key=_quality_sort_key) - return streams + sources_tried += 1 + debug(f"[AnimeCix] Video {idx}: extracting from {video_url}") - except Exception: - return [] + # Extract embed info from URL + try: + embed_streams = self._extract_embed_streams(video_url) + debug(f"[AnimeCix] Video {idx}: got {len(embed_streams)} streams") + + if embed_streams: + streams.extend(embed_streams) + # Stop after first successful source (usually has all qualities) + debug(f"[AnimeCix] Got {len(embed_streams)} streams from source {idx}, stopping") + break + + except Exception as e: + debug(f"[AnimeCix] Video {idx}: error - {e}") + continue + + # Sort by quality (highest first) + streams.sort(key=self._quality_sort_key) + + debug(f"[AnimeCix] get_streams COMPLETE: {len(streams)} streams total") + return streams - def _get_seasons(self, title_id: int) -> List[int]: - try: - safe_id = int(title_id) - except (ValueError, TypeError): - return [0] + def _get_movie_streams(self, title_id: int) -> List[StreamLink]: + """ + Get stream URLs for a movie. + + Movies have video sources directly in the title endpoint. - url = f"{ALT_URL}secure/related-videos?episode=1&season=1&titleId={safe_id}&videoId=637113" - data = _get_json(url) + Args: + title_id: Movie title ID. - if not data or "videos" not in data: - return [0] + Returns: + List of StreamLink objects. + """ + debug(f"[AnimeCix] Fetching movie streams for title_id={title_id}") + + url = f"{TITLES_URL}/{title_id}" + data = self._request_json(url, params={"titleId": title_id}) + + if not data or "title" not in data: + debug(f"[AnimeCix] No data returned for movie") + return [] + + title = data["title"] + videos = title.get("videos", []) - videos = data.get("videos") or [] if not videos: - return [0] + debug(f"[AnimeCix] No videos found for movie") + return [] + + debug(f"[AnimeCix] Found {len(videos)} video sources for movie") + + # For movies, try first working video source + streams = [] + sources_tried = 0 + max_sources = 10 + + for idx, video in enumerate(videos, 1): + if sources_tried >= max_sources: + debug(f"[AnimeCix] Reached max sources limit ({max_sources})") + break + + video_url = video.get("url") + + if not video_url: + debug(f"[AnimeCix] Video {idx}: no URL, skipping") + continue + + sources_tried += 1 + debug(f"[AnimeCix] Video {idx}: extracting from {video_url}") + + try: + embed_streams = self._extract_embed_streams(video_url) + debug(f"[AnimeCix] Video {idx}: got {len(embed_streams)} streams") + + if embed_streams: + streams.extend(embed_streams) + # Stop after first successful source + debug(f"[AnimeCix] Got {len(embed_streams)} streams from source {idx}, stopping") + break + + except Exception as e: + debug(f"[AnimeCix] Video {idx}: error - {e}") + continue + + # Sort by quality + streams.sort(key=self._quality_sort_key) + + debug(f"[AnimeCix] Movie streams complete: {len(streams)} streams total") + return streams + + def _extract_embed_streams(self, embed_url: str) -> List[StreamLink]: + """ + Extract stream URLs from embed page. + + Directly extracts embed ID from URL and calls video API. + No redirect following or vid parameter needed. + + Args: + embed_url: Embed page URL (e.g., https://tau-video.xyz/embed/{id}). + + Returns: + List of StreamLink objects. + """ + # Parse embed URL to extract embed ID + parsed = urlparse(embed_url) + path_parts = parsed.path.strip("/").split("/") + + # Extract embed ID from path + embed_id = None + if len(path_parts) >= 2 and path_parts[0] == "embed": + embed_id = path_parts[1] + elif len(path_parts) == 1: + embed_id = path_parts[0] - title = (videos[0] or {}).get("title") or {} - seasons = title.get("seasons") or [] + if not embed_id: + debug(f"[AnimeCix] Could not extract embed_id from: {embed_url}") + return [] + + debug(f"[AnimeCix] Extracted embed_id: {embed_id}") + + api_url = f"https://{VIDEO_PLAYERS[0]}/api/video/{embed_id}" + debug(f"[AnimeCix] Fetching video API: {api_url}") + + video_data = self._request_json(api_url) + + if not video_data: + debug(f"[AnimeCix] No response from video API") + return [] + + if "urls" not in video_data: + debug(f"[AnimeCix] No 'urls' key in video API response") + return [] + + # Extract streams + streams = [] + for url_info in video_data["urls"]: + stream_url = url_info.get("url") + quality = url_info.get("label", "auto") + + if stream_url: + streams.append(StreamLink( + url=stream_url, + quality=quality, + server="tau-video", + headers={"Referer": embed_url} + )) + + debug(f"[AnimeCix] Extracted {len(streams)} streams from embed") + return streams + + def _quality_sort_key(self, stream: StreamLink) -> int: + """ + Sort key for stream quality (highest first). + + Args: + stream: StreamLink object. + + Returns: + Integer sort key (negative for descending order). + """ + quality = stream.quality.lower().replace('p', '').strip() - if seasons: - return list(range(len(seasons))) - return [0] + # Special cases + if quality == "4k" or "2160" in quality: + return -2160 + if quality == "auto": + return 1 # Auto quality goes to end + + # Try to parse resolution + try: + return -int(quality) + except ValueError: + return 0 def _parse_type(self, title_type: str) -> str: + """ + Parse anime type from title_type string. + + Args: + title_type: Type string from API. + + Returns: + Normalized type: "movie", "ova", or "series". + """ title_type = (title_type or "").lower() + if "movie" in title_type or "film" in title_type: return "movie" if "ova" in title_type: return "ova" + if "special" in title_type: + return "special" + return "series" - - def _parse_episode_number(self, name: str, fallback: int) -> int: - patterns = [ - r'(?:bölüm|episode|ep)\s*(\d+)', - r'(\d+)\.\s*(?:bölüm|episode)', - r'^(\d+)$' - ] - - name_lower = name.lower() - for pattern in patterns: - match = re.search(pattern, name_lower) - if match: - return int(match.group(1)) - - return fallback - - def _parse_season_from_url(self, url: str, fallback: int) -> int: - try: - qs = parse_qs(url.split("?", 1)[1]) if "?" in url else {} - return int(qs["season"][0]) - except (KeyError, IndexError, ValueError): - return fallback diff --git a/weeb_cli/providers/tr/animecix_helpers.py b/weeb_cli/providers/tr/animecix_helpers.py new file mode 100644 index 0000000..3c196a7 --- /dev/null +++ b/weeb_cli/providers/tr/animecix_helpers.py @@ -0,0 +1,358 @@ +""" +Helper utilities for Animecix provider. + +This module provides dynamic header fetching, HTTP utilities, and helper functions +specifically for interacting with animecix.tv API. +""" + +import json +import re +import time +from typing import Dict, Optional, Any +from pathlib import Path +from datetime import datetime, timedelta + +try: + from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeoutError + PLAYWRIGHT_AVAILABLE = True +except ImportError: + PLAYWRIGHT_AVAILABLE = False + +from weeb_cli.services.logger import debug +from weeb_cli.config import CONFIG_DIR + +STATIC_HEADERS = { + "accept": "application/json, text/plain, */*", + "accept-language": "tr-TR,tr;q=0.9,en-US;q=0.8,en;q=0.7", + "priority": "u=1, i", + "sec-ch-ua": '"Chromium";v="131", "Not_A Brand";v="24"', + "sec-ch-ua-mobile": "?0", + "sec-ch-ua-platform": '"Linux"', + "sec-fetch-dest": "empty", + "sec-fetch-mode": "cors", + "sec-fetch-site": "same-origin", + "sec-gpc": "1", + "user-agent": ( + "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36" + ), +} + +EXCLUDED_HEADERS = { + "accept", "accept-language", "priority", "sec-ch-ua", + "sec-ch-ua-mobile", "sec-ch-ua-platform", "sec-fetch-dest", + "sec-fetch-mode", "sec-fetch-site", "sec-gpc", "user-agent", + "host", "connection", "content-length" +} + +HEADERS_CACHE_FILE = CONFIG_DIR / "animecix_headers.json" +CACHE_DURATION = timedelta(hours=1) + + +def get_cached_headers() -> Optional[Dict[str, str]]: + """ + Load cached dynamic headers if they exist and are still valid. + + Returns: + Dictionary of cached headers, or None if cache is invalid/missing. + """ + if not HEADERS_CACHE_FILE.exists(): + return None + + try: + with open(HEADERS_CACHE_FILE, "r", encoding="utf-8") as f: + cache_data = json.load(f) + + cached_time = datetime.fromisoformat(cache_data.get("timestamp", "")) + if datetime.now() - cached_time > CACHE_DURATION: + debug("[AnimeCix] Header cache expired") + return None + + headers = cache_data.get("headers", {}) + if headers: + debug("[AnimeCix] Using cached headers") + return headers + + return None + except Exception as e: + debug(f"[AnimeCix] Failed to load header cache: {e}") + return None + + +def save_headers_cache(headers: Dict[str, str]) -> None: + """ + Save dynamic headers to cache file. + + Args: + headers: Dictionary of dynamic headers to cache. + """ + try: + CONFIG_DIR.mkdir(parents=True, exist_ok=True) + cache_data = { + "timestamp": datetime.now().isoformat(), + "headers": headers + } + with open(HEADERS_CACHE_FILE, "w", encoding="utf-8") as f: + json.dump(cache_data, f, indent=2) + debug(f"[AnimeCix] Saved headers to cache") + except Exception as e: + debug(f"[AnimeCix] Failed to save header cache: {e}") + + +def get_dynamic_headers(force_refresh: bool = False) -> Dict[str, str]: + """ + Get dynamic headers required for animecix.tv API requests. + + Uses Playwright to extract headers from a real browser session. + Falls back to cached headers if Playwright is not available or fails. + + Args: + force_refresh: If True, bypass cache and fetch fresh headers. + + Returns: + Dictionary of dynamic headers (e.g., x-e-h, cookies). + """ + # Try cached headers first + if not force_refresh: + cached = get_cached_headers() + if cached: + return cached + + if not PLAYWRIGHT_AVAILABLE: + debug("[AnimeCix] Playwright not available, cannot fetch dynamic headers") + return {} + + debug("[AnimeCix] Fetching fresh dynamic headers with Playwright...") + dynamic_headers = {} + + try: + with sync_playwright() as p: + browser = None + + browsers = [ + ("chrome", p.chromium), + ("msedge", p.chromium), + ("firefox", p.firefox), + ] + + for channel, launcher in browsers: + try: + if channel == "firefox": + browser = launcher.launch(headless=True) + else: + browser = launcher.launch(channel=channel, headless=True) + debug(f"[AnimeCix] Launched browser: {channel}") + break + except Exception as e: + debug(f"[AnimeCix] Failed to launch {channel}: {e}") + continue + + if not browser: + debug("[AnimeCix] Falling back to Playwright Chromium") + browser = p.chromium.launch(headless=True) + + page = browser.new_page() + + def handle_request(request): + nonlocal dynamic_headers + if "secure/titles" in request.url or "secure/episode" in request.url: + all_headers = request.headers + dynamic_headers = { + k: v for k, v in all_headers.items() + if k.lower() not in EXCLUDED_HEADERS + } + debug(f"[AnimeCix] Captured {len(dynamic_headers)} dynamic headers") + + page.on("request", handle_request) + + try: + page.goto("https://animecix.tv/browse", + wait_until="networkidle", + timeout=30000) + except PlaywrightTimeoutError: + debug("[AnimeCix] Page load timeout, continuing anyway") + + if not dynamic_headers: + debug("[AnimeCix] Waiting additional time for requests...") + page.wait_for_timeout(3000) + + browser.close() + + if dynamic_headers: + save_headers_cache(dynamic_headers) + + return dynamic_headers + + except Exception as e: + debug(f"[AnimeCix] Error fetching dynamic headers: {e}") + return {} + + +def build_headers(dynamic_headers: Optional[Dict[str, str]] = None) -> Dict[str, str]: + """ + Build complete headers by merging static and dynamic headers. + + Args: + dynamic_headers: Optional dynamic headers to merge. If None, fetches them. + + Returns: + Complete headers dictionary ready for HTTP requests. + """ + headers = dict(STATIC_HEADERS) + + if dynamic_headers is None: + dynamic_headers = get_dynamic_headers() + + headers.update(dynamic_headers) + + return headers + + +def clean_title_name(title: str) -> str: + """ + Convert anime title to URL-safe slug format. + + Removes/replaces special characters to match animecix.tv URL format. + + Args: + title: Original anime title. + + Returns: + URL-safe slug string. + + Example: + >>> clean_title_name("One Piece: Episode 1000!") + "one-piece-episode-1000" + """ + if not title: + return "" + + # Convert to lowercase + title = title.lower() + + # Replace/remove special characters + title = title.replace("&", "and") + title = title.replace("'", "") + title = title.replace("!", "") + title = title.replace("?", "") + title = title.replace(":", "") + title = title.replace(".", "") + + title = re.sub(r'[^a-z0-9\s-]', '', title) + + title = title.replace(" ", "-") + + title = re.sub(r'-+', '-', title) + + title = title.strip('-') + + return title + + +def parse_episode_number(name: str, fallback: int = 1) -> int: + """ + Extract episode number from episode name. + + Supports various naming patterns: + - "Bölüm 5" + - "Episode 12" + - "5. Bölüm" + - "12" + + Args: + name: Episode name string. + fallback: Default number if parsing fails. + + Returns: + Extracted episode number or fallback. + + Example: + >>> parse_episode_number("Bölüm 42") + 42 + >>> parse_episode_number("Unknown Name", 10) + 10 + """ + if not name: + return fallback + + patterns = [ + r'(?:bölüm|episode|ep)\s*[:\-]?\s*(\d+)', + r'(\d+)\.\s*(?:bölüm|episode)', + r'^(\d+)$', + r'[sS](\d+)[eE](\d+)', + ] + + name_lower = name.lower() + for pattern in patterns: + match = re.search(pattern, name_lower) + if match: + if len(match.groups()) > 1: + return int(match.group(2)) + return int(match.group(1)) + + debug(f"[AnimeCix] Could not parse episode number from '{name}', using fallback {fallback}") + return fallback + + +def parse_season_number(data: Any, fallback: int = 1) -> int: + """ + Extract season number from various data formats. + + Args: + data: Season data (dict with 'number' key, int, or URL string). + fallback: Default season number if parsing fails. + + Returns: + Extracted season number or fallback. + """ + if isinstance(data, dict): + return data.get("number", fallback) + + if isinstance(data, int): + return data + + if isinstance(data, str): + try: + from urllib.parse import parse_qs + if "?" in data: + qs = parse_qs(data.split("?", 1)[1]) + if "season" in qs: + return int(qs["season"][0]) + except (ValueError, KeyError, IndexError): + pass + + return fallback + + +def retry_with_backoff(func, max_retries: int = 3, initial_delay: float = 1.0): + """ + Retry a function with exponential backoff. + + Args: + func: Function to retry (should raise exception on failure). + max_retries: Maximum number of retry attempts. + initial_delay: Initial delay in seconds between retries. + + Returns: + Function result if successful. + + Raises: + Last exception if all retries fail. + """ + last_exception = None + delay = initial_delay + + for attempt in range(max_retries): + try: + return func() + except Exception as e: + last_exception = e + debug(f"[AnimeCix] Attempt {attempt + 1}/{max_retries} failed: {e}") + + if attempt < max_retries - 1: + debug(f"[AnimeCix] Retrying in {delay:.1f}s...") + time.sleep(delay) + delay *= 2 + + # All retries failed + raise last_exception From 54b3604b6135f141e7e45bad707332f30a3ac0f4 Mon Sep 17 00:00:00 2001 From: ewgsta <159681870+ewgsta@users.noreply.github.com> Date: Mon, 13 Jul 2026 03:07:00 +0300 Subject: [PATCH 2/4] fix(i18n): fix cache translation keys --- weeb_cli/commands/settings/settings_cache.py | 30 ++++++++++---------- weeb_cli/locales/de.json | 1 - weeb_cli/locales/tr.json | 3 +- 3 files changed, 16 insertions(+), 18 deletions(-) diff --git a/weeb_cli/commands/settings/settings_cache.py b/weeb_cli/commands/settings/settings_cache.py index 9ffb402..00e4901 100644 --- a/weeb_cli/commands/settings/settings_cache.py +++ b/weeb_cli/commands/settings/settings_cache.py @@ -17,24 +17,24 @@ def cache_settings_menu(): while True: console.clear() - show_header(i18n.t("settings.cache.title")) + show_header(i18n.t("settings.cache_title")) cache = get_cache() stats = cache.get_stats() - console.print(f"[dim]{i18n.t('settings.cache.memory_entries')}: {stats['memory_entries']}[/dim]") - console.print(f"[dim]{i18n.t('settings.cache.file_entries')}: {stats['file_entries']}[/dim]") - console.print(f"[dim]{i18n.t('settings.cache.total_size')}: {stats['total_size_mb']} MB[/dim]\n") + console.print(f"[dim]{i18n.t('settings.cache_memory_entries')}: {stats['memory_entries']}[/dim]") + console.print(f"[dim]{i18n.t('settings.cache_file_entries')}: {stats['file_entries']}[/dim]") + console.print(f"[dim]{i18n.t('settings.cache_total_size')}: {stats['total_size_mb']} MB[/dim]\n") choices = [ - i18n.t("settings.cache.clear_all"), - i18n.t("settings.cache.clear_provider"), - i18n.t("settings.cache.cleanup_old"), + i18n.t("settings.cache_clear_all"), + i18n.t("settings.cache_clear_provider"), + i18n.t("settings.cache_cleanup_old"), ] try: answer = questionary.select( - i18n.t("settings.cache.prompt"), + i18n.t("settings.cache_prompt"), choices=choices, pointer=">", use_shortcuts=False, @@ -46,24 +46,24 @@ def cache_settings_menu(): if answer is None: return - if answer == i18n.t("settings.cache.clear_all"): + if answer == i18n.t("settings.cache_clear_all"): confirm = questionary.confirm( - i18n.t("settings.cache.confirm_clear_all"), + i18n.t("settings.cache_confirm_clear_all"), default=False ).ask() if confirm: cache.clear() - console.print(f"[green]{i18n.t('settings.cache.cleared')}[/green]") + console.print(f"[green]{i18n.t('settings.cache_cleared')}[/green]") time.sleep(1) - elif answer == i18n.t("settings.cache.clear_provider"): + elif answer == i18n.t("settings.cache_clear_provider"): provider = config.get("scraping_source", "None") removed = cache.invalidate_provider(provider) - console.print(f"[green]{i18n.t('settings.cache.provider_cleared')}: {removed} {i18n.t('common.items')}[/green]") + console.print(f"[green]{i18n.t('settings.cache_provider_cleared')}: {removed} {i18n.t('common.items')}[/green]") time.sleep(1) - elif answer == i18n.t("settings.cache.cleanup_old"): + elif answer == i18n.t("settings.cache_cleanup_old"): removed = cache.cleanup(max_age=86400) - console.print(f"[green]{i18n.t('settings.cache.cleaned')}: {removed} {i18n.t('common.items')}[/green]") + console.print(f"[green]{i18n.t('settings.cache_cleaned')}: {removed} {i18n.t('common.items')}[/green]") time.sleep(1) diff --git a/weeb_cli/locales/de.json b/weeb_cli/locales/de.json index c2d96e3..f6e0066 100644 --- a/weeb_cli/locales/de.json +++ b/weeb_cli/locales/de.json @@ -139,7 +139,6 @@ "restore_success": "Wiederherstellung erfolgreich!", "restore_failed": "Wiederherstellung fehlgeschlagen.", "cache_title": "Cache-Verwaltung", - "cache_title": "Cache-Verwaltung", "cache_memory_entries": "Speichereinträge", "cache_file_entries": "Dateieinträge", "cache_total_size": "Gesamtgröße", diff --git a/weeb_cli/locales/tr.json b/weeb_cli/locales/tr.json index 317e419..ab92b6a 100644 --- a/weeb_cli/locales/tr.json +++ b/weeb_cli/locales/tr.json @@ -139,13 +139,12 @@ "restore_success": "Geri yükleme başarılı!", "restore_failed": "Geri yükleme başarısız.", "cache_title": "Önbellek Yönetimi", - "cache_title": "Önbellek Yönetimi", "cache_memory_entries": "Bellekteki girişler", "cache_file_entries": "Dosyadaki girişler", "cache_total_size": "Toplam boyut", "cache_clear_all": "Tüm önbelleği temizle", "cache_clear_provider": "Mevcut kaynak önbelleğini temizle", - "cache_cleanup_old": "Eski önbelleği temizle (>24s)", + "cache_cleanup_old": "Eski önbelleği temizle (>24h)", "cache_prompt": "İşlem seçin", "cache_confirm_clear_all": "Tüm önbellek temizlensin mi?", "cache_cleared": "Önbellek temizlendi", From b1b221b83ddc795870c4cabb304cc77d938f6479 Mon Sep 17 00:00:00 2001 From: ewgsta <159681870+ewgsta@users.noreply.github.com> Date: Mon, 13 Jul 2026 03:38:46 +0300 Subject: [PATCH 3/4] feat(turkanime): provider curl-cffi and improved caching --- weeb_cli/providers/tr/turkanime.py | 436 +++++++++++++++++++++++++---- 1 file changed, 382 insertions(+), 54 deletions(-) diff --git a/weeb_cli/providers/tr/turkanime.py b/weeb_cli/providers/tr/turkanime.py index 5eb471b..0eb263b 100644 --- a/weeb_cli/providers/tr/turkanime.py +++ b/weeb_cli/providers/tr/turkanime.py @@ -1,9 +1,24 @@ +""" +Turkanime provider for weeb-cli. + +This provider supports turkanime.tv, a Turkish anime streaming site. +Uses curl-cffi for Firefox impersonation to bypass Cloudflare protection. + +Features: +- Cloudflare bypass with curl-cffi (Firefox TLS fingerprint) +- Dynamic key extraction and caching +- CSRF token management +- Support for multiple fansubs and video servers +- AES decryption for video URLs +""" + import os import re import json from typing import List, Optional, Dict from hashlib import md5 from base64 import b64decode +from html import unescape from weeb_cli.providers.base import ( BaseProvider, @@ -13,12 +28,15 @@ StreamLink ) from weeb_cli.providers.registry import register_provider +from weeb_cli.config import CONFIG_DIR +from weeb_cli.services.logger import debug try: from curl_cffi import requests as curl_requests HAS_CURL_CFFI = True except ImportError: HAS_CURL_CFFI = False + import requests as fallback_requests try: from Crypto.Cipher import AES @@ -26,18 +44,16 @@ except ImportError: HAS_CRYPTO = False -try: - from appdirs import user_cache_dir - HAS_APPDIRS = True -except ImportError: - HAS_APPDIRS = False - BASE_URL = "https://turkanime.tv" +KEY_CACHE_FILE = CONFIG_DIR / "turkanime_key.cache" +CSRF_CACHE_FILE = CONFIG_DIR / "turkanime_csrf.cache" + _session = None _base_url = None _key_cache = None _csrf_cache = None +# Supported video players by priority SUPPORTED_PLAYERS = [ "YADISK", "MAIL", "ALUCARD(BETA)", "PIXELDRAIN", "AMATERASU(BETA)", "HDVID", "ODNOKLASSNIKI", "GDRIVE", "MP4UPLOAD", "DAILYMOTION", @@ -46,31 +62,55 @@ def _init_session(): + """Initialize HTTP session with Cloudflare bypass. + + Uses curl-cffi with Firefox impersonation for TLS fingerprint spoofing. + Falls back to regular requests if curl-cffi is not available. + + Returns: + Configured session object. + """ global _session, _base_url if _session is not None: return _session + debug("[Turkanime] Initializing session") + if HAS_CURL_CFFI: _session = curl_requests.Session(impersonate="firefox", allow_redirects=True) + debug("[Turkanime] Using curl-cffi with Firefox impersonation") else: - import requests - _session = requests.Session() + _session = fallback_requests.Session() + debug("[Turkanime] Warning: curl-cffi not available, using fallback (may fail with Cloudflare)") _base_url = BASE_URL + # Verify connection and handle redirects try: res = _session.get(BASE_URL + "/", timeout=30) if res.status_code == 200: final_url = res.url if hasattr(res, 'url') else BASE_URL _base_url = final_url.rstrip('/') - except Exception: + debug(f"[Turkanime] Base URL set to: {_base_url}") + except Exception as e: + debug(f"[Turkanime] Failed to verify connection: {e}") _base_url = BASE_URL return _session -def _fetch(path: str, headers: Dict[str, str] = None) -> str: +def _fetch(path: str, headers: Dict[str, str] = None, data: Dict = None) -> str: + """Make HTTP request with proper headers. + + Args: + path: URL path (absolute or relative). + headers: Additional headers to merge. + data: POST data (if provided, makes POST request). + + Returns: + Response text content. + """ global _base_url session = _init_session() @@ -80,8 +120,12 @@ def _fetch(path: str, headers: Dict[str, str] = None) -> str: if _base_url is None: _base_url = BASE_URL - path = path if path.startswith("/") else "/" + path - url = _base_url + path + # Build full URL + if not path.startswith("http"): + path = path if path.startswith("/") else "/" + path + url = _base_url + path + else: + url = path default_headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0", @@ -93,43 +137,68 @@ def _fetch(path: str, headers: Dict[str, str] = None) -> str: default_headers.update(headers) try: - response = session.get(url, headers=default_headers, timeout=30) + if data: + debug(f"[Turkanime] POST {url}") + response = session.post(url, headers=default_headers, data=data, timeout=30) + else: + debug(f"[Turkanime] GET {url}") + response = session.get(url, headers=default_headers, timeout=30) return response.text - except Exception: + except Exception as e: + debug(f"[Turkanime] Request failed: {e}") return "" def _obtain_key() -> bytes: + """Extract and cache the AES decryption key. + + The key is dynamically embedded in JavaScript files and changes periodically. + This function reverse engineers the obfuscated code to extract it. + + Process: + 1. Fetch /embed/#/url/ and find imported JS files + 2. Find the JS file containing 'decrypt' + 3. Extract obfuscated list and find the longest element (the key) + + Returns: + AES encryption key as bytes, or empty bytes on failure. + """ global _key_cache if _key_cache: + debug("[Turkanime] Using cached key") return _key_cache - if HAS_APPDIRS: - try: - cache_file = os.path.join(user_cache_dir(), "turkanimu_key.cache") - if os.path.isfile(cache_file): - with open(cache_file, "r", encoding="utf-8") as f: - cached = f.read().strip().encode() - if cached: - _key_cache = cached - return _key_cache - except Exception: - pass + # Try to load from cache file + try: + if KEY_CACHE_FILE.exists(): + with open(KEY_CACHE_FILE, "r", encoding="utf-8") as f: + cached = f.read().strip().encode() + if cached: + _key_cache = cached + debug("[Turkanime] Loaded key from cache file") + return _key_cache + except Exception as e: + debug(f"[Turkanime] Failed to load key cache: {e}") + + debug("[Turkanime] Extracting fresh key from embed JS") try: embed_html = _fetch("/embed/#/url/") js_files = re.findall(r"/embed/js/embeds\..*?\.js", embed_html) if len(js_files) < 2: + debug("[Turkanime] Not enough JS files found") return b"" js1 = _fetch(js_files[1]) js1_imports = re.findall("[a-z0-9]{16}", js1) if not js1_imports: + debug("[Turkanime] No imports found in first JS") return b"" + # Find JS file containing 'decrypt' j2 = _fetch(f'/embed/js/embeds.{js1_imports[0]}.js') if "'decrypt'" not in j2 and len(js1_imports) > 1: j2 = _fetch(f'/embed/js/embeds.{js1_imports[1]}.js') @@ -138,6 +207,7 @@ def _obtain_key() -> bytes: r'function a\d_0x[\w]{1,4}\(\)\{var _0x\w{3,8}=\[(.*?)\];', j2 ) if not match: + debug("[Turkanime] Obfuscated list not found") return b"" obfuscate_list = match.group(1) @@ -146,27 +216,50 @@ def _obtain_key() -> bytes: key=lambda i: len(re.sub(r"\\x\d\d", "?", i)) ).encode() - if HAS_APPDIRS and _key_cache: + debug(f"[Turkanime] Extracted key: {len(_key_cache)} bytes") + + # Save to cache + if _key_cache: try: - cache_dir = user_cache_dir() - os.makedirs(cache_dir, exist_ok=True) - cache_file = os.path.join(cache_dir, "turkanimu_key.cache") - with open(cache_file, "w", encoding="utf-8") as f: + KEY_CACHE_FILE.parent.mkdir(parents=True, exist_ok=True) + with open(KEY_CACHE_FILE, "w", encoding="utf-8") as f: f.write(_key_cache.decode("utf-8")) - except Exception: - pass + debug("[Turkanime] Saved key to cache file") + except Exception as e: + debug(f"[Turkanime] Failed to save key cache: {e}") return _key_cache - except Exception: + except Exception as e: + debug(f"[Turkanime] Key extraction failed: {e}") return b"" def _decrypt_cipher(key: bytes, data: bytes) -> str: + """Decrypt CryptoJS.AES encrypted data (Python implementation). + + Implements the same algorithm as CryptoJS.AES.decrypt with: + - Salted key derivation (OpenSSL compatible) + - CBC mode decryption + - PKCS7 padding removal + + Args: + key: AES encryption key. + data: Base64 encoded encrypted data. + + Returns: + Decrypted plaintext string, or empty string on failure. + + References: + - https://stackoverflow.com/a/36780727 + - https://gist.github.com/ysfchn/e96304fb41375bad0fdf9a5e837da631 + """ if not HAS_CRYPTO: + debug("[Turkanime] PyCryptodome not available") return "" def salted_key(data: bytes, salt: bytes, output: int = 48): + """Generate key and IV from password and salt (OpenSSL compatible).""" data += salt k = md5(data).digest() final_key = k @@ -176,6 +269,7 @@ def salted_key(data: bytes, salt: bytes, output: int = 48): return final_key[:output] def unpad(data: bytes) -> bytes: + """Remove PKCS7 padding.""" return data[:-(data[-1] if isinstance(data[-1], int) else ord(data[-1]))] try: @@ -187,50 +281,85 @@ def unpad(data: bytes) -> bytes: crypt = AES.new(salted_key(key, salt, output=32), iv=iv, mode=AES.MODE_CBC) return unpad(crypt.decrypt(cipher_text)).decode("utf-8") - except Exception: + except Exception as e: + debug(f"[Turkanime] Decryption failed: {e}") return "" def _get_real_url(url_cipher: str) -> str: - if HAS_APPDIRS: - try: - cache_file = os.path.join(user_cache_dir(), "turkanimu_key.cache") - if os.path.isfile(cache_file): - with open(cache_file, "r", encoding="utf-8") as f: - cached_key = f.read().strip().encode() - plaintext = _decrypt_cipher(cached_key, url_cipher.encode()) - if plaintext: - return "https:" + json.loads(plaintext) - except Exception: - pass + """Decrypt video URL from cipher text. + Args: + url_cipher: Encrypted URL string. + + Returns: + Decrypted video URL with https: prefix. + """ + # Try cached key first + if KEY_CACHE_FILE.exists(): + try: + with open(KEY_CACHE_FILE, "r", encoding="utf-8") as f: + cached_key = f.read().strip().encode() + plaintext = _decrypt_cipher(cached_key, url_cipher.encode()) + if plaintext: + debug("[Turkanime] Decrypted URL with cached key") + return "https:" + json.loads(plaintext) + except Exception as e: + debug(f"[Turkanime] Cached key failed: {e}") + + # Get fresh key key = _obtain_key() if not key: + debug("[Turkanime] Failed to obtain key") return "" plaintext = _decrypt_cipher(key, url_cipher.encode()) if not plaintext: + debug("[Turkanime] Decryption failed") return "" try: - return "https:" + json.loads(plaintext) - except Exception: + url = "https:" + json.loads(plaintext) + debug(f"[Turkanime] Decrypted URL successfully") + return url + except Exception as e: + debug(f"[Turkanime] URL parsing failed: {e}") return "" def _decrypt_jsjiamiv7(ciphertext: str, key: str) -> str: + """Decrypt jsjiamiv7 obfuscated data. + + Implements reverse of jsjiamiv7 obfuscator: + 1. Translate custom base64 alphabet to standard + 2. Base64 decode + 3. RC4 decryption (KSA + PRGA algorithm) + + Args: + ciphertext: Obfuscated ciphertext. + key: RC4 encryption key. + + Returns: + Decrypted plaintext string. + + Reference: + - https://en.wikipedia.org/wiki/RC4 + """ _CUSTOM = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789+/" _STD = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/" _TRANSLATE = str.maketrans(_CUSTOM, _STD) + # Translate and pad t = ciphertext.translate(_TRANSLATE) t += "=" * (-len(t) % 4) try: data = b64decode(t).decode("utf-8") - except Exception: + except Exception as e: + debug(f"[Turkanime] jsjiamiv7 decode failed: {e}") return "" + # RC4 KSA (Key Scheduling Algorithm) S = list(range(256)) j = 0 klen = len(key) @@ -239,6 +368,7 @@ def _decrypt_jsjiamiv7(ciphertext: str, key: str) -> str: j = (j + S[i] + ord(key[i % klen])) & 0xff S[i], S[j] = S[j], S[i] + # RC4 PRGA (Pseudo-Random Generation Algorithm) i = j = 0 out = [] for ch in data: @@ -251,11 +381,33 @@ def _decrypt_jsjiamiv7(ciphertext: str, key: str) -> str: def _obtain_csrf() -> Optional[str]: + """Extract and cache CSRF token from player.js. + + The CSRF token is encrypted with jsjiamiv7 and embedded in player.js. + + Returns: + CSRF token string, or None on failure. + """ global _csrf_cache if _csrf_cache: + debug("[Turkanime] Using cached CSRF token") return _csrf_cache + # Try to load from cache file + try: + if CSRF_CACHE_FILE.exists(): + with open(CSRF_CACHE_FILE, "r", encoding="utf-8") as f: + cached = f.read().strip() + if cached: + _csrf_cache = cached + debug("[Turkanime] Loaded CSRF from cache file") + return _csrf_cache + except Exception as e: + debug(f"[Turkanime] Failed to load CSRF cache: {e}") + + debug("[Turkanime] Extracting fresh CSRF token") + try: res = _fetch("/js/player.js") @@ -263,6 +415,7 @@ def _obtain_csrf() -> Optional[str]: candidates = re.findall(r"'([a-zA-Z\d\+\/]{96,156})',", res) if not key_match or not candidates: + debug("[Turkanime] CSRF extraction failed: key or candidates not found") return None key = key_match[0] @@ -271,19 +424,44 @@ def _obtain_csrf() -> Optional[str]: decrypted = _decrypt_jsjiamiv7(ct, key) if re.search(r"^[a-zA-Z/\+]+$", decrypted): _csrf_cache = decrypted + debug(f"[Turkanime] Extracted CSRF token") + + # Save to cache + try: + CSRF_CACHE_FILE.parent.mkdir(parents=True, exist_ok=True) + with open(CSRF_CACHE_FILE, "w", encoding="utf-8") as f: + f.write(_csrf_cache) + debug("[Turkanime] Saved CSRF to cache file") + except Exception as e: + debug(f"[Turkanime] Failed to save CSRF cache: {e}") + return _csrf_cache + debug("[Turkanime] No valid CSRF found in candidates") return None - except Exception: + except Exception as e: + debug(f"[Turkanime] CSRF extraction failed: {e}") return None def _unmask_real_url(url_mask: str) -> str: + """Unmask real URL for Turkanime's custom players. + + Players like Alucard, Amaterasu, Bankai, HDVID use masked URLs. + This function uses CSRF token to unmask them. + + Args: + url_mask: Masked URL from Turkanime player. + + Returns: + Real video URL, or original mask if unmask fails. + """ if "turkanime" not in url_mask: return url_mask csrf = _obtain_csrf() if not csrf: + debug("[Turkanime] Cannot unmask: CSRF not available") return url_mask try: @@ -297,18 +475,68 @@ def _unmask_real_url(url_mask: str) -> str: if url.startswith("//"): url = "https:" + url + debug(f"[Turkanime] Unmasked URL successfully") return url - except Exception: + except Exception as e: + debug(f"[Turkanime] Unmask failed: {e}") return url_mask @register_provider("turkanime", lang="tr", region="TR") class TurkAnimeProvider(BaseProvider): + """Turkanime.tv provider implementation. + + Supports searching, fetching details, and extracting streams from turkanime.tv. + Uses curl-cffi with Firefox impersonation for Cloudflare bypass. + """ def __init__(self): super().__init__() + debug("[Turkanime] Provider initialized") def search(self, query: str) -> List[AnimeResult]: + """Search for anime by query. + + Uses /arama endpoint with POST request for search. + Falls back to /ajax/tamliste if search fails. + + Args: + query: Search query string. + + Returns: + List of anime results. + """ + debug(f"[Turkanime] Searching for: {query}") + + # Try search endpoint first + html = _fetch("/arama", data={"arama": query}) + if html: + matches = re.findall(r'/anime/([^"\'>]+)["\'] [^>]*?title=["\']([^"]+?) izle', html) + if matches: + results = [] + for slug, title in matches[:20]: + title_clean = unescape(title) + results.append(AnimeResult( + id=slug, + title=title_clean + )) + debug(f"[Turkanime] Found {len(results)} results via search") + return results + + # Check if redirected to single result + redirect_match = re.findall('window.location ?= ?"anime/(.*?)"', html) + if redirect_match: + slug = redirect_match[0] + # Fetch title + anime_html = _fetch(f'/anime/{slug}') + title_match = re.findall(r'(.*?)', anime_html) + if title_match: + title = title_match[0].split(" izle")[0].strip() + debug(f"[Turkanime] Single result redirect to: {slug}") + return [AnimeResult(id=slug, title=title)] + + # Fallback: filter full anime list + debug("[Turkanime] Falling back to full list search") html = _fetch("/ajax/tamliste") if not html: return [] @@ -326,22 +554,36 @@ def search(self, query: str) -> List[AnimeResult]: title=title_clean )) + debug(f"[Turkanime] Found {len(results)} results via full list") return results[:20] def get_details(self, anime_id: str) -> Optional[AnimeDetails]: + """Get detailed information for an anime. + + Args: + anime_id: Anime slug. + + Returns: + AnimeDetails object with all information. + """ + debug(f"[Turkanime] Getting details for: {anime_id}") html = _fetch(f'/anime/{anime_id}') if not html: return None + # Extract title title_match = re.findall(r'(.*?)', html) - title = title_match[0] if title_match else anime_id + title = title_match[0].split(" izle")[0].strip() if title_match else anime_id - img_match = re.findall(r'twitter.image" content="(.*?)"', html) + # Extract cover image + img_match = re.findall(r'twitter:image" content="(.*?)"', html) cover = img_match[0] if img_match else None + # Extract internal ID for episodes anime_id_match = re.findall(r'serilerb/(.*?)\.jpg', html) internal_id = anime_id_match[0] if anime_id_match else "" + # Extract description description = None desc_match = re.search(r'twitter:description"\s+content="([^"]+)"', html) if not desc_match: @@ -350,6 +592,7 @@ def get_details(self, anime_id: str) -> Optional[AnimeDetails]: import html as html_module description = html_module.unescape(desc_match.group(1)).strip() + # Parse info table info = {} info_table = re.findall(r'
()', html, re.DOTALL) if info_table: @@ -358,12 +601,16 @@ def get_details(self, anime_id: str) -> Optional[AnimeDetails]: val = re.sub(r"<[^>]*>", "", val).strip() info[key] = val + # Extract genres genres = [] if "Anime Türü" in info: genres = [g.strip() for g in info["Anime Türü"].split(" ") if g.strip()] + # Get episodes episodes = self._get_episodes_internal(internal_id) if internal_id else [] + debug(f"[Turkanime] Found {len(episodes)} episodes") + return AnimeDetails( id=anime_id, title=title, @@ -376,6 +623,15 @@ def get_details(self, anime_id: str) -> Optional[AnimeDetails]: ) def get_episodes(self, anime_id: str) -> List[Episode]: + """Get list of episodes for an anime. + + Args: + anime_id: Anime slug. + + Returns: + List of Episode objects. + """ + debug(f"[Turkanime] Getting episodes for: {anime_id}") html = _fetch(f'/anime/{anime_id}') if not html: return [] @@ -384,16 +640,25 @@ def get_episodes(self, anime_id: str) -> List[Episode]: internal_id = anime_id_match[0] if anime_id_match else "" if not internal_id: + debug("[Turkanime] Could not find internal ID") return [] return self._get_episodes_internal(internal_id) def _get_episodes_internal(self, internal_id: str) -> List[Episode]: + """Fetch episodes using internal anime ID. + + Args: + internal_id: Internal anime identifier. + + Returns: + List of Episode objects. + """ html = _fetch(f'/ajax/bolumler&animeId={internal_id}') if not html: return [] - matches = re.findall(r'/video/(.*?)\\?".*?title=.*?"(.*?)\\?"', html) + matches = re.findall(r'/video/(.*?)\\?".*?title=\\?"(.*?)\\?"', html) episodes = [] for i, (slug, title) in enumerate(matches, 1): @@ -405,16 +670,32 @@ def _get_episodes_internal(self, internal_id: str) -> List[Episode]: title=title_clean )) + debug(f"[Turkanime] Found {len(episodes)} episodes") return episodes def get_streams(self, anime_id: str, episode_id: str) -> List[StreamLink]: + """Get stream URLs for an episode. + + Extracts video URLs from multiple fansubs and players. + Handles both single fansub and multi-fansub episodes. + + Args: + anime_id: Anime slug. + episode_id: Episode slug. + + Returns: + List of StreamLink objects. + """ + debug(f"[Turkanime] Getting streams for episode: {episode_id}") html = _fetch(f'/video/{episode_id}') if not html: return [] streams = [] - if not re.search(r".*birden fazla grup", html): + # Check if single fansub or multiple + if "birden fazla grup" not in html: + # Single fansub fansub_match = re.findall(r" ([^\\<>]*).*?iframe", html) fansub = fansub_match[0] if fansub_match else "Unknown" @@ -427,13 +708,18 @@ def get_streams(self, anime_id: str, episode_id: str) -> List[StreamLink]: html ) + debug(f"[Turkanime] Found {len(video_matches)} videos for fansub: {fansub}") + for cipher_or_path, player in video_matches: stream = self._process_video(cipher_or_path, player, fansub) if stream: streams.append(stream) else: + # Multiple fansubs fansub_matches = re.findall(r"(ajax/videosec&.*?)'.*? ?(.*?)", html) + debug(f"[Turkanime] Found {len(fansub_matches)} fansubs") + for path, fansub in fansub_matches: src = _fetch(path) @@ -451,32 +737,65 @@ def get_streams(self, anime_id: str, episode_id: str) -> List[StreamLink]: if stream: streams.append(stream) + debug(f"[Turkanime] Total streams found: {len(streams)}") return streams def _process_video(self, cipher_or_path: str, player: str, fansub: str) -> Optional[StreamLink]: + """Process a video URL (decrypt if needed). + + Args: + cipher_or_path: Either encrypted URL or path to fetch it. + player: Player name. + fansub: Fansub name. + + Returns: + StreamLink object, or None if processing fails. + """ + # Filter unsupported players if player.upper() not in SUPPORTED_PLAYERS: + debug(f"[Turkanime] Unsupported player: {player}") return None + # If path, fetch the cipher if "/" in cipher_or_path: src = _fetch(cipher_or_path) cipher_match = re.findall(r'/embed/#/url/(.*?)\?status', src) if not cipher_match: + # Check for direct iframe (some videos are not encrypted) + tmp = re.findall('