From dd6fd1e199ff6a396444acaa1731d15967f55d74 Mon Sep 17 00:00:00 2001 From: bonsai-dreams Date: Mon, 16 Feb 2026 06:44:47 -0800 Subject: [PATCH] Feature: Add AudiobookBay release source (#619) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary Adds AudiobookBay as a web-scraping release source for audiobook torrents. Once enabled, a new tab shows up in the Find Releases modal. ## What's New - **AudiobookBay source** – Search AudiobookBay for audiobook torrents from the Shelfmark UI - **Torrent downloads** – Extract magnet links from detail pages and add them to the configured torrent client - **Audiobook-only** – Source is limited to audiobooks - **Download clients:** Currently uses the torrent client configured under **Prowlarr > Download Clients**. - Audiobook-specific categories (e.g. `QBITTORRENT_CATEGORY_AUDIOBOOK`) are applied when set. - **Settings → AudiobookBay**: - Enable toggle - Hostname - Max pages to search (default 5) - Rate limit delay in seconds (default 1) ## How It Works 1. User searches for an audiobook; AudiobookBay is queried if enabled. 2. Results show title, language, format, and size 3. User selects a release; the handler fetches the detail page and extracts the magnet link. 4. Magnet link is sent to the configured torrent client. ## Testing - Unit tests for source, handler, scraper, and utils - Mocked HTTP requests and torrent client calls - Coverage for search, relevance filtering, language mapping, size parsing, and download flow ### Screenshots image --- readme.md | 3 +- shelfmark/release_sources/__init__.py | 1 + .../release_sources/audiobookbay/__init__.py | 6 + .../release_sources/audiobookbay/handler.py | 164 +++++++ .../release_sources/audiobookbay/scraper.py | 271 +++++++++++ .../release_sources/audiobookbay/settings.py | 75 +++ .../release_sources/audiobookbay/source.py | 253 ++++++++++ .../release_sources/audiobookbay/utils.py | 47 ++ tests/audiobookbay/__init__.py | 1 + tests/audiobookbay/test_handler.py | 452 ++++++++++++++++++ tests/audiobookbay/test_scraper.py | 383 +++++++++++++++ tests/audiobookbay/test_source.py | 428 +++++++++++++++++ tests/audiobookbay/test_utils.py | 103 ++++ 13 files changed, 2186 insertions(+), 1 deletion(-) create mode 100644 shelfmark/release_sources/audiobookbay/__init__.py create mode 100644 shelfmark/release_sources/audiobookbay/handler.py create mode 100644 shelfmark/release_sources/audiobookbay/scraper.py create mode 100644 shelfmark/release_sources/audiobookbay/settings.py create mode 100644 shelfmark/release_sources/audiobookbay/source.py create mode 100644 shelfmark/release_sources/audiobookbay/utils.py create mode 100644 tests/audiobookbay/__init__.py create mode 100644 tests/audiobookbay/test_handler.py create mode 100644 tests/audiobookbay/test_scraper.py create mode 100644 tests/audiobookbay/test_source.py create mode 100644 tests/audiobookbay/test_utils.py diff --git a/readme.md b/readme.md index 317228e..d137adf 100644 --- a/readme.md +++ b/readme.md @@ -106,6 +106,7 @@ See the full [Environment Variables Reference](docs/environment-variables.md) fo Some of the additional options available in Settings: - **Fast Download Key** - Use your paid account to skip Cloudflare challenges entirely and use faster, direct downloads - **Prowlarr** - Configure indexers and download clients to download books and audiobooks +- **AudiobookBay** - Web scraping source for audiobook torrents (audiobooks only) - **IRC** - Add details for IRC book sources and download directly from the UI - **Library Link** - Add a link to your Calibre-Web or Booklore instance in the UI header - **File processing** - Customiseable download paths, file renaming and directory creation with template-based renaming @@ -139,7 +140,7 @@ A smaller image without the built-in Cloudflare bypasser. Ideal for: - **External bypassers** - Already running FlareSolverr or ByParr for other services - **Fast downloads** - Using fast download sources -- **Alternative sources only** - Exclusively using Prowlarr, IRC, or other sources +- **Alternative sources only** - Exclusively using Prowlarr, AudiobookBay, IRC, or other sources - **Audiobooks** - Using Shelfmark exclusively for audiobooks ```bash diff --git a/shelfmark/release_sources/__init__.py b/shelfmark/release_sources/__init__.py index 7189782..ac17034 100644 --- a/shelfmark/release_sources/__init__.py +++ b/shelfmark/release_sources/__init__.py @@ -366,3 +366,4 @@ def get_source_display_name(name: str) -> str: from shelfmark.release_sources import direct_download # noqa: F401, E402 from shelfmark.release_sources import prowlarr # noqa: F401, E402 from shelfmark.release_sources import irc # noqa: F401, E402 +from shelfmark.release_sources import audiobookbay # noqa: F401, E402 diff --git a/shelfmark/release_sources/audiobookbay/__init__.py b/shelfmark/release_sources/audiobookbay/__init__.py new file mode 100644 index 0000000..748caf9 --- /dev/null +++ b/shelfmark/release_sources/audiobookbay/__init__.py @@ -0,0 +1,6 @@ +"""AudiobookBay release source - web scraping for audiobook torrents.""" + +# Import to trigger registration +from shelfmark.release_sources.audiobookbay import source # noqa: F401, E402 +from shelfmark.release_sources.audiobookbay import handler # noqa: F401, E402 +from shelfmark.release_sources.audiobookbay import settings # noqa: F401, E402 diff --git a/shelfmark/release_sources/audiobookbay/handler.py b/shelfmark/release_sources/audiobookbay/handler.py new file mode 100644 index 0000000..4406590 --- /dev/null +++ b/shelfmark/release_sources/audiobookbay/handler.py @@ -0,0 +1,164 @@ +"""AudiobookBay download handler - extracts magnet links and sends to torrent clients.""" + +from threading import Event +from typing import Callable, Optional + +from shelfmark.core.config import config +from shelfmark.core.logger import setup_logger +from shelfmark.core.models import DownloadTask +from shelfmark.core.utils import is_audiobook +from shelfmark.release_sources import DownloadHandler, register_handler +from shelfmark.release_sources.audiobookbay import scraper +from shelfmark.release_sources.prowlarr.clients import ( + DownloadClient, + get_client, + list_configured_clients, +) + +logger = setup_logger(__name__) + + +@register_handler("audiobookbay") +class AudiobookBayHandler(DownloadHandler): + """Handler for AudiobookBay downloads via configured torrent client.""" + + def _get_category_for_task(self, client: DownloadClient, task: DownloadTask) -> Optional[str]: + """Get audiobook category if configured and applicable, else None for default.""" + if not is_audiobook(task.content_type): + return None + + # Client-specific audiobook category config keys + audiobook_keys = { + "qbittorrent": "QBITTORRENT_CATEGORY_AUDIOBOOK", + "transmission": "TRANSMISSION_CATEGORY_AUDIOBOOK", + "deluge": "DELUGE_CATEGORY_AUDIOBOOK", + } + audiobook_key = audiobook_keys.get(client.name) + if audiobook_key: + category = config.get(audiobook_key, "") + if category: + return category + + # Fallback to general category + general_keys = { + "qbittorrent": "QBITTORRENT_CATEGORY", + "transmission": "TRANSMISSION_CATEGORY", + "deluge": "DELUGE_CATEGORY", + } + general_key = general_keys.get(client.name) + if general_key: + return config.get(general_key, "") or None + + return None + + def download( + self, + task: DownloadTask, + cancel_flag: Event, + progress_callback: Callable[[float], None], + status_callback: Callable[[str, Optional[str]], None], + ) -> Optional[str]: + """Execute download by extracting magnet link and sending to torrent client. + + Args: + task: Download task with task_id containing detail URL + cancel_flag: Event to check for cancellation + progress_callback: Called with progress percentage (0-100) + status_callback: Called with (status, message) for status updates + + Returns: + None (torrents don't return file path immediately) + """ + try: + # Check for cancellation before starting + if cancel_flag.is_set(): + logger.info(f"Download cancelled before starting: {task.task_id}") + status_callback("cancelled", "Cancelled") + return None + + # task.task_id contains the detail page URL + detail_url = task.task_id + hostname = config.get("ABB_HOSTNAME", "audiobookbay.lu") + + # Extract magnet link from detail page + status_callback("resolving", "Extracting magnet link") + magnet_link = scraper.extract_magnet_link(detail_url, hostname) + + if not magnet_link: + status_callback("error", "Failed to extract magnet link from detail page") + return None + + logger.info(f"Extracted magnet link: {magnet_link[:100]}...") + + # Get torrent client + client = get_client("torrent") + if not client: + configured = list_configured_clients() + if not configured: + status_callback("error", "No torrent clients configured. Configure qBittorrent or Transmission in settings.") + else: + status_callback("error", "No torrent client configured") + return None + + # Check if this download already exists in the client + status_callback("resolving", f"Checking {client.name}") + category = self._get_category_for_task(client, task) + existing = client.find_existing(magnet_link, category=category) + + if existing: + download_id, existing_status = existing + logger.info(f"Found existing download in {client.name}: {download_id}") + + if existing_status.complete: + logger.info("Existing download is complete") + status_callback("resolving", "Found existing download") + # Return the path from the existing download + file_path = client.get_download_path(download_id) + if file_path: + return file_path + else: + status_callback("error", "Could not locate existing download path") + return None + else: + logger.info("Existing download in progress") + status_callback("downloading", "Resuming existing download") + # Poll for completion (simplified - could reuse Prowlarr's polling logic) + # For now, just return None and let the orchestrator handle it + return None + + # Add new download + status_callback("resolving", f"Sending to {client.name}") + try: + release_name = task.title or "Unknown" + category = self._get_category_for_task(client, task) + download_id = client.add_download( + url=magnet_link, + name=release_name, + category=category, + expected_hash=None, # Extract from magnet if needed + ) + logger.info(f"Added to {client.name}: {download_id} for '{release_name}'") + status_callback("downloading", "Download started") + except Exception as e: + logger.error(f"Failed to add to {client.name}: {e}") + status_callback("error", f"Failed to add to {client.name}: {e}") + return None + + # Torrents don't return file path immediately + # The orchestrator will handle polling via the download client + return None + + except Exception as e: + logger.error(f"AudiobookBay download error: {e}") + status_callback("error", str(e)) + return None + + def cancel(self, task_id: str) -> bool: + """Cancel an in-progress download. + + Torrents can't be cancelled from Shelfmark side. + User must cancel in torrent client. + """ + logger.debug(f"Cancel requested for AudiobookBay task: {task_id}") + # Torrents are managed by the client, we can't cancel them here + return False diff --git a/shelfmark/release_sources/audiobookbay/scraper.py b/shelfmark/release_sources/audiobookbay/scraper.py new file mode 100644 index 0000000..c901ae5 --- /dev/null +++ b/shelfmark/release_sources/audiobookbay/scraper.py @@ -0,0 +1,271 @@ +"""Web scraping functions for AudiobookBay.""" + +import re +import time +from typing import List, Optional, Dict +from urllib.parse import quote + +import requests +from bs4 import BeautifulSoup + +from shelfmark.core.config import config +from shelfmark.core.logger import setup_logger +from shelfmark.download import network + +logger = setup_logger(__name__) + +# Default trackers if none found on page +DEFAULT_TRACKERS = [ + "udp://tracker.openbittorrent.com:80", + "udp://opentor.org:2710", + "udp://tracker.ccc.de:80", + "udp://tracker.blackunicorn.xyz:6969", + "udp://tracker.coppersurfer.tk:6969", + "udp://tracker.leechers-paradise.org:6969", +] + +# Required headers to avoid blocking +REQUEST_HEADERS = { + 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/115.0.0.0 Safari/537.36' +} + + +def search_audiobookbay( + query: str, + max_pages: int = 5, + hostname: str = "audiobookbay.lu" +) -> List[Dict[str, str]]: + """Search AudiobookBay for audiobooks matching the query. + + Args: + query: Search query string + max_pages: Maximum number of pages to fetch + hostname: AudiobookBay hostname (e.g., "audiobookbay.lu") + + Returns: + List of dicts with keys: title, link, cover, language, format, bitrate, size, posted_date + """ + results = [] + rate_limit_delay = config.get("ABB_RATE_LIMIT_DELAY", 1.0) + + # Iterate through pages + for page in range(1, max_pages + 1): + # Construct URL - use + for spaces (matching audiobookbay-automated implementation) + # This avoids aggressive encoding that PHP-based sites may reject + query_encoded = query.replace(' ', '+') + url = f"https://{hostname}/page/{page}/?s={query_encoded}&cat=undefined%2Cundefined" + + try: + # Make request with proxy support + response = requests.get( + url, + headers=REQUEST_HEADERS, + proxies=network.get_proxies(url), + timeout=30, + allow_redirects=True + ) + + if response.status_code != 200: + logger.warning(f"Failed to fetch page {page}. Status Code: {response.status_code}") + break + + # Check if we were redirected to the homepage (search was rejected/blocked) + final_url = response.url.rstrip('/') + base_url = f"https://{hostname}".rstrip('/') + if final_url == base_url or final_url == f"{base_url}/": + # Search was redirected to homepage - this means the search failed + # This can happen due to geo-blocking, rate limiting, or invalid query format + if page == 1: + logger.warning(f"Search query '{query}' was redirected to homepage - search may be blocked or invalid") + break + + # Parse HTML + soup = BeautifulSoup(response.text, 'html.parser') + + # Extract book entries + posts = soup.select('.post') + if not posts: + # No more results + break + + for post in posts: + try: + # Extract title + title_elem = post.select_one('.postTitle > h2 > a') + if not title_elem: + continue + + title = title_elem.text.strip() + + # Extract link (relative, needs hostname prefix) + href = title_elem.get('href', '') + if not href: + continue + + if href.startswith('http'): + link = href + else: + link = f"https://{hostname}{href}" + + # Extract cover image (try .postContent .center img first, then fallback to any img) + cover = None + cover_elem = post.select_one('.postContent .center img') or post.select_one('img') + if cover_elem: + cover = cover_elem.get('src', '') + if cover and not cover.startswith('http'): + cover = f"https://{hostname}{cover}" + + # Extract language from .postInfo + language = None + post_info = post.select_one('.postInfo') + if post_info: + info_text = post_info.get_text(separator=' ', strip=True).replace('\xa0', ' ') + lang_match = re.search(r'Language:\s*([A-Za-z]+)', info_text) + if lang_match: + language = lang_match.group(1).strip() + + # Extract format, bitrate, size, and posted date from .postContent + posted_date = None + format_type = None + bitrate = None + size_str = None + + post_content = post.select_one('.postContent') + if post_content: + content_text = post_content.get_text(separator=' ', strip=True).replace('\xa0', ' ') + + # Extract posted date + posted_match = re.search(r'Posted:\s*(\d+\s+[A-Za-z]+\s+\d{4})', content_text) + if posted_match: + posted_date = posted_match.group(1).strip() + + # Extract format (e.g., "M4B", "MP3") + format_match = re.search(r'Format:\s*([A-Za-z0-9]+)', content_text) + if format_match: + format_type = format_match.group(1).strip() + + # Extract bitrate (e.g., "256 Kbps") + bitrate_match = re.search(r'Bitrate:\s*([\d]+\s*[A-Za-z/]+)', content_text) + if bitrate_match: + bitrate = bitrate_match.group(1).strip() + + # Extract file size (e.g., "11.68 GBs") + size_match = re.search(r'File Size:\s*([\d.]+)\s*([A-Za-z]+)', content_text) + if size_match: + size_str = f"{size_match.group(1)} {size_match.group(2)}" + + results.append({ + 'title': title, + 'link': link, + 'cover': cover or None, + 'language': language, + 'format': format_type, + 'bitrate': bitrate, + 'size': size_str, + 'posted_date': posted_date, + }) + except Exception as e: + logger.debug(f"Skipping post due to error: {e}") + continue + + # Rate limiting delay between pages + if page < max_pages and rate_limit_delay > 0: + time.sleep(rate_limit_delay) + + except requests.exceptions.RequestException as e: + logger.warning(f"Request error on page {page}: {e}") + break + except Exception as e: + logger.error(f"Unexpected error on page {page}: {e}") + break + + logger.info(f"Found {len(results)} results for query '{query}'") + return results + + +def extract_magnet_link( + details_url: str, + hostname: str = "audiobookbay.lu" +) -> Optional[str]: + """Extract info hash and trackers from book detail page, then construct magnet link. + + Args: + details_url: URL of the book's detail page + hostname: AudiobookBay hostname (for logging) + + Returns: + Magnet link, or None if extraction fails + """ + try: + # Fetch detail page + response = requests.get( + details_url, + headers=REQUEST_HEADERS, + proxies=network.get_proxies(details_url), + timeout=30 + ) + + if response.status_code != 200: + logger.warning(f"Failed to fetch details page. Status Code: {response.status_code}") + return None + + soup = BeautifulSoup(response.text, 'html.parser') + + # 1. Extract Info Hash + # Look for Info Hash and get next sibling value + info_hash = None + info_hash_rows = soup.find_all('td') + for td in info_hash_rows: + if td.text.strip().lower() == 'info hash': + next_td = td.find_next_sibling('td') + if next_td: + info_hash = next_td.text.strip() + break + + # Alternative: search for text containing "Info Hash" and get next element + if not info_hash: + for elem in soup.find_all(string=re.compile(r'Info Hash', re.IGNORECASE)): + parent = elem.parent + if parent and parent.name == 'td': + next_td = parent.find_next_sibling('td') + if next_td: + info_hash = next_td.text.strip() + break + + if not info_hash: + logger.warning("Info Hash not found on the page.") + return None + + # Clean up info hash (remove whitespace, ensure uppercase) + info_hash = re.sub(r'\s+', '', info_hash).upper() + + # 2. Extract Trackers + # Find all containing udp:// or http:// + trackers = [] + for td in soup.find_all('td'): + text = td.text.strip() + if text.startswith(('udp://', 'http://', 'https://')): + trackers.append(text) + + # 3. Use default trackers if none found + if not trackers: + logger.debug("No trackers found on the page. Using default trackers.") + trackers = DEFAULT_TRACKERS + + # 4. Construct Magnet Link + # Format: magnet:?xt=urn:btih:{INFO_HASH}&tr={TRACKER1}&tr={TRACKER2}... + tracker_params = "&".join( + f"tr={quote(tracker)}" + for tracker in trackers + ) + magnet_link = f"magnet:?xt=urn:btih:{info_hash}&{tracker_params}" + + logger.debug(f"Generated Magnet Link: {magnet_link[:100]}...") + return magnet_link + + except requests.exceptions.RequestException as e: + logger.error(f"Request error extracting magnet link: {e}") + return None + except Exception as e: + logger.error(f"Failed to extract magnet link: {e}") + return None diff --git a/shelfmark/release_sources/audiobookbay/settings.py b/shelfmark/release_sources/audiobookbay/settings.py new file mode 100644 index 0000000..0173cc3 --- /dev/null +++ b/shelfmark/release_sources/audiobookbay/settings.py @@ -0,0 +1,75 @@ +"""AudiobookBay settings registration.""" + +from shelfmark.core.settings_registry import ( + register_group, + register_settings, + CheckboxField, + TextField, + NumberField, + HeadingField, +) + + +# ==================== Register Group ==================== + +register_group( + name="audiobookbay", + display_name="AudiobookBay", + icon="download", + order=45, # After Prowlarr (order 40) +) + + +# ==================== Register Settings ==================== + +@register_settings("audiobookbay_config", "Configuration", group="audiobookbay", order=1) +def audiobookbay_config_settings(): + """AudiobookBay configuration settings.""" + return [ + CheckboxField( + key="ABB_ENABLED", + label="Enable AudiobookBay", + description="Enable AudiobookBay as a release source for audiobooks.", + default=False, + ), + TextField( + key="ABB_HOSTNAME", + label="Hostname", + description="AudiobookBay domain (e.g., audiobookbay.lu, audiobookbay.is)", + placeholder="audiobookbay.lu", + default="audiobookbay.lu", + show_when={"field": "ABB_ENABLED", "value": True}, + ), + NumberField( + key="ABB_PAGE_LIMIT", + label="Max Pages to Search", + description="Maximum number of search result pages to fetch (1-10).", + default=5, + min_value=1, + max_value=10, + show_when={"field": "ABB_ENABLED", "value": True}, + ), + NumberField( + key="ABB_RATE_LIMIT_DELAY", + label="Rate Limit Delay (seconds)", + description="Delay between requests in seconds to avoid rate limiting (0-10).", + default=1.0, + min_value=0.0, + max_value=10.0, + show_when={"field": "ABB_ENABLED", "value": True}, + ), + ] + + +# ==================== Download Clients Tab ==================== + +@register_settings("audiobookbay_clients", "Download Clients", group="audiobookbay", order=2) +def audiobookbay_clients_settings(): + """AudiobookBay download client settings.""" + return [ + HeadingField( + key="abb_torrent_heading", + title="Torrent Client", + description="The AudiobookBay integration uses the torrent client that is configured under 'Prowlarr' > 'Download Clients'.", + ), + ] diff --git a/shelfmark/release_sources/audiobookbay/source.py b/shelfmark/release_sources/audiobookbay/source.py new file mode 100644 index 0000000..dd50e81 --- /dev/null +++ b/shelfmark/release_sources/audiobookbay/source.py @@ -0,0 +1,253 @@ +"""AudiobookBay release source - searches AudiobookBay for audiobook torrents.""" + +import hashlib +from typing import List, Optional, TYPE_CHECKING + +if TYPE_CHECKING: + from shelfmark.core.search_plan import ReleaseSearchPlan + +from shelfmark.core.config import config +from shelfmark.core.logger import setup_logger +from shelfmark.metadata_providers import BookMetadata +from shelfmark.release_sources import ( + Release, + ReleaseProtocol, + ReleaseSource, + register_source, + ReleaseColumnConfig, + ColumnSchema, + ColumnRenderType, + ColumnAlign, + ColumnColorHint, +) +from shelfmark.release_sources.audiobookbay import scraper +from shelfmark.release_sources.audiobookbay.utils import parse_size + +logger = setup_logger(__name__) + + +# Map language names to ISO 639-1 codes (matching frontend color maps) +LANGUAGE_MAP = { + "english": "en", + "spanish": "es", + "french": "fr", + "german": "de", + "italian": "it", + "portuguese": "pt", + "russian": "ru", + "japanese": "ja", + "chinese": "zh", + "dutch": "nl", + "swedish": "sv", + "norwegian": "no", + "danish": "da", + "finnish": "fi", + "polish": "pl", + "czech": "cs", + "hungarian": "hu", + "korean": "ko", + "arabic": "ar", + "hebrew": "he", + "turkish": "tr", + "greek": "el", + "hindi": "hi", + "thai": "th", + "vietnamese": "vi", + "indonesian": "id", + "ukrainian": "uk", + "romanian": "ro", + "bulgarian": "bg", + "catalan": "ca", + "croatian": "hr", + "slovenian": "sl", + "serbian": "sr", +} + + +def _map_language(language: str) -> Optional[str]: + """Map language name to ISO 639-1 code. + + Args: + language: Language name (e.g., "English") + + Returns: + ISO 639-1 code (e.g., "en"), or original string if no mapping found, or None if input is empty + """ + if not language: + return None + + lang_lower = language.lower().strip() + return LANGUAGE_MAP.get(lang_lower, lang_lower) + + +def _generate_source_id(detail_url: str) -> str: + """Generate a unique source ID from detail URL.""" + return hashlib.md5(detail_url.encode()).hexdigest() + + +@register_source("audiobookbay") +class AudiobookBaySource(ReleaseSource): + """Release source for AudiobookBay audiobook torrents.""" + + name = "audiobookbay" + display_name = "AudiobookBay" + supported_content_types = ["audiobook"] # ONLY audiobooks + + def search( + self, + book: BookMetadata, + plan: "ReleaseSearchPlan", + expand_search: bool = False, + content_type: str = "ebook" + ) -> List[Release]: + """Search AudiobookBay for audiobook releases. + + Args: + book: Book metadata + plan: Search plan with query variants + expand_search: Ignored (always searches) + content_type: Must be "audiobook" for this source + + Returns: + List of Release objects + """ + # Only search for audiobooks + if content_type != "audiobook": + return [] + + hostname = config.get("ABB_HOSTNAME", "audiobookbay.lu") + max_pages = config.get("ABB_PAGE_LIMIT", 5) + + # Build search query from plan + if plan.manual_query: + query = plan.manual_query + elif plan.title_variants: + # Use first title variant with author + variant = plan.title_variants[0] + query = f"{variant.title} {variant.author}".strip() + else: + query = book.title or "" + + if not query: + logger.debug("No search query available") + return [] + + # Convert to lowercase (matching audiobookbay-automated implementation) + query_lower = query.lower() + logger.info(f"Searching AudiobookBay for: {query_lower}") + + try: + # Search AudiobookBay + results = scraper.search_audiobookbay( + query=query_lower, + max_pages=max_pages, + hostname=hostname + ) + + # Extract query words for relevance checking + query_words = set(word.lower() for word in query_lower.split() if len(word) > 2) + + releases = [] + for result in results: + try: + title = result['title'] + + # Basic relevance check: ensure title contains at least one query word + # This filters out homepage "Latest" feed items that may leak through + if query_words: + title_lower = title.lower() + if not any(word in title_lower for word in query_words): + logger.debug(f"Filtering out irrelevant result: {title}") + continue + + # Generate unique source ID + source_id = _generate_source_id(result['link']) + + # Extract and parse metadata + format_type = result.get('format') + size_str = result.get('size') + size_bytes = parse_size(size_str) if size_str else None + language_raw = result.get('language') + language_code = _map_language(language_raw) if language_raw else None + + # Create Release object + release = Release( + source="audiobookbay", + source_id=source_id, + title=title, + format=format_type.lower() if format_type else None, + language=language_code, + size=size_str, + size_bytes=size_bytes, + download_url=result['link'], # Detail page URL (used by handler) + info_url=result['link'], # Make title clickable + protocol=ReleaseProtocol.TORRENT, + indexer="AudiobookBay", + seeders=None, # Not available on search page + peers=None, + content_type="audiobook", + extra={ + "preview": result.get('cover'), + "detail_url": result['link'], + "bitrate": result.get('bitrate'), + "posted_date": result.get('posted_date'), + "language_raw": language_raw, # Keep original for reference + } + ) + releases.append(release) + except Exception as e: + logger.warning(f"Failed to create release from result: {e}") + continue + + logger.info(f"Found {len(releases)} releases from AudiobookBay") + return releases + + except Exception as e: + logger.error(f"AudiobookBay search error: {e}") + return [] + + def is_available(self) -> bool: + """Check if AudiobookBay source is enabled.""" + return config.get("ABB_ENABLED", False) is True + + def get_column_config(self) -> ReleaseColumnConfig: + """Get column configuration for AudiobookBay releases. + + Shows title, language, format, and size columns. + No seeders/peers since ABB doesn't show this on search page. + """ + return ReleaseColumnConfig( + columns=[ + ColumnSchema( + key="language", + label="Lang", + render_type=ColumnRenderType.BADGE, + align=ColumnAlign.CENTER, + width="60px", + hide_mobile=True, + color_hint=ColumnColorHint(type="map", value="language"), + uppercase=True, + fallback="", + ), + ColumnSchema( + key="format", + label="Format", + render_type=ColumnRenderType.BADGE, + align=ColumnAlign.CENTER, + width="80px", + hide_mobile=False, + color_hint=ColumnColorHint(type="map", value="format"), + uppercase=True, + ), + ColumnSchema( + key="size", + label="Size", + render_type=ColumnRenderType.SIZE, + align=ColumnAlign.CENTER, + width="80px", + hide_mobile=False, + ), + ], + grid_template="minmax(0,2fr) 60px 80px 80px", + supported_filters=["format", "language"], # Enable format and language filters + ) diff --git a/shelfmark/release_sources/audiobookbay/utils.py b/shelfmark/release_sources/audiobookbay/utils.py new file mode 100644 index 0000000..b90f954 --- /dev/null +++ b/shelfmark/release_sources/audiobookbay/utils.py @@ -0,0 +1,47 @@ +"""Utility functions for AudiobookBay integration.""" + +import re +from typing import Optional + + +def sanitize_title(title: str) -> str: + """Remove characters that are invalid in filenames. + + Args: + title: Book title + + Returns: + Sanitized title + """ + return re.sub(r'[<>:"/\\|?*]', '', title).strip() + + +def parse_size(size_str: Optional[str]) -> Optional[int]: + """Parse size string to bytes. + + Args: + size_str: Size string (e.g., "1.5 GB", "500 MB", "11.68 GBs") + + Returns: + Size in bytes, or None if parsing fails + """ + if not size_str: + return None + + # Match number and unit, handling "GBs" as well as "GB" (case-insensitive) + match = re.search(r'([\d.]+)\s*([BKMGT]B?)S?', size_str.upper()) + if not match: + return None + + value = float(match.group(1)) + unit = match.group(2) + + multipliers = { + 'B': 1, + 'KB': 1024, + 'MB': 1024 ** 2, + 'GB': 1024 ** 3, + 'TB': 1024 ** 4, + } + + return int(value * multipliers.get(unit, 1)) diff --git a/tests/audiobookbay/__init__.py b/tests/audiobookbay/__init__.py new file mode 100644 index 0000000..39149b5 --- /dev/null +++ b/tests/audiobookbay/__init__.py @@ -0,0 +1 @@ +"""Tests for AudiobookBay integration.""" diff --git a/tests/audiobookbay/test_handler.py b/tests/audiobookbay/test_handler.py new file mode 100644 index 0000000..3fc1907 --- /dev/null +++ b/tests/audiobookbay/test_handler.py @@ -0,0 +1,452 @@ +""" +Tests for AudiobookBay download handler. +""" + +from threading import Event +from unittest.mock import Mock, patch, MagicMock +import pytest + +from shelfmark.core.models import DownloadTask +from shelfmark.release_sources.audiobookbay.handler import AudiobookBayHandler +from shelfmark.release_sources.prowlarr.clients import ( + DownloadStatus, + DownloadState, +) + + +class ProgressRecorder: + """Records progress and status updates during download.""" + + def __init__(self): + self.progress_values = [] + self.status_updates = [] + + def progress_callback(self, progress: float): + self.progress_values.append(progress) + + def status_callback(self, status: str, message=None): + self.status_updates.append((status, message)) + + @property + def last_status(self): + return self.status_updates[-1][0] if self.status_updates else None + + @property + def last_message(self): + return self.status_updates[-1][1] if self.status_updates else None + + @property + def statuses(self): + return [s[0] for s in self.status_updates] + + +class TestAudiobookBayHandlerDownload: + """Tests for AudiobookBayHandler.download().""" + + @patch('shelfmark.release_sources.audiobookbay.handler.scraper.extract_magnet_link') + @patch('shelfmark.release_sources.audiobookbay.handler.get_client') + def test_download_success(self, mock_get_client, mock_extract_magnet): + """Test successful download initiation.""" + mock_extract_magnet.return_value = "magnet:?xt=urn:btih:abc123" + + mock_client = MagicMock() + mock_client.name = "qbittorrent" + mock_client.find_existing.return_value = None + mock_client.add_download.return_value = "download_id_123" + mock_get_client.return_value = mock_client + + handler = AudiobookBayHandler() + task = DownloadTask( + task_id="https://audiobookbay.lu/abss/test-book/", + source="audiobookbay", + title="Test Book", + content_type="audiobook", + ) + cancel_flag = Event() + recorder = ProgressRecorder() + + result = handler.download( + task=task, + cancel_flag=cancel_flag, + progress_callback=recorder.progress_callback, + status_callback=recorder.status_callback, + ) + + assert result is None # Torrents don't return path immediately + mock_extract_magnet.assert_called_once_with( + "https://audiobookbay.lu/abss/test-book/", + "audiobookbay.lu" + ) + mock_client.add_download.assert_called_once() + assert "downloading" in recorder.statuses + + @patch('shelfmark.release_sources.audiobookbay.handler.scraper.extract_magnet_link') + @patch('shelfmark.release_sources.audiobookbay.handler.get_client') + def test_download_existing_complete(self, mock_get_client, mock_extract_magnet): + """Test handling existing complete download.""" + mock_extract_magnet.return_value = "magnet:?xt=urn:btih:abc123" + + mock_client = MagicMock() + mock_client.name = "qbittorrent" + mock_client.find_existing.return_value = ( + "existing_id", + DownloadStatus( + progress=100, + state=DownloadState.COMPLETE, + message="Complete", + complete=True, + file_path="/path/to/book.m4b", + ), + ) + mock_client.get_download_path.return_value = "/path/to/book.m4b" + mock_get_client.return_value = mock_client + + handler = AudiobookBayHandler() + task = DownloadTask( + task_id="https://audiobookbay.lu/abss/test-book/", + source="audiobookbay", + title="Test Book", + content_type="audiobook", + ) + cancel_flag = Event() + recorder = ProgressRecorder() + + result = handler.download( + task=task, + cancel_flag=cancel_flag, + progress_callback=recorder.progress_callback, + status_callback=recorder.status_callback, + ) + + assert result == "/path/to/book.m4b" + mock_client.add_download.assert_not_called() + + @patch('shelfmark.release_sources.audiobookbay.handler.scraper.extract_magnet_link') + @patch('shelfmark.release_sources.audiobookbay.handler.get_client') + def test_download_existing_in_progress(self, mock_get_client, mock_extract_magnet): + """Test handling existing in-progress download.""" + mock_extract_magnet.return_value = "magnet:?xt=urn:btih:abc123" + + mock_client = MagicMock() + mock_client.name = "qbittorrent" + mock_client.find_existing.return_value = ( + "existing_id", + DownloadStatus( + progress=50, + state=DownloadState.DOWNLOADING, + message="Downloading", + complete=False, + file_path=None, + ), + ) + mock_get_client.return_value = mock_client + + handler = AudiobookBayHandler() + task = DownloadTask( + task_id="https://audiobookbay.lu/abss/test-book/", + source="audiobookbay", + title="Test Book", + content_type="audiobook", + ) + cancel_flag = Event() + recorder = ProgressRecorder() + + result = handler.download( + task=task, + cancel_flag=cancel_flag, + progress_callback=recorder.progress_callback, + status_callback=recorder.status_callback, + ) + + assert result is None + assert "downloading" in recorder.statuses + mock_client.add_download.assert_not_called() + + @patch('shelfmark.release_sources.audiobookbay.handler.scraper.extract_magnet_link') + @patch('shelfmark.release_sources.audiobookbay.handler.get_client') + def test_download_cancellation(self, mock_get_client, mock_extract_magnet): + """Test that cancellation is respected.""" + handler = AudiobookBayHandler() + task = DownloadTask( + task_id="https://audiobookbay.lu/abss/test-book/", + source="audiobookbay", + title="Test Book", + content_type="audiobook", + ) + cancel_flag = Event() + cancel_flag.set() # Set immediately + recorder = ProgressRecorder() + + result = handler.download( + task=task, + cancel_flag=cancel_flag, + progress_callback=recorder.progress_callback, + status_callback=recorder.status_callback, + ) + + assert result is None + assert "cancelled" in recorder.statuses + mock_extract_magnet.assert_not_called() + + @patch('shelfmark.release_sources.audiobookbay.handler.scraper.extract_magnet_link') + @patch('shelfmark.release_sources.audiobookbay.handler.get_client') + def test_download_no_magnet_link(self, mock_get_client, mock_extract_magnet): + """Test handling when magnet link extraction fails.""" + mock_extract_magnet.return_value = None + + handler = AudiobookBayHandler() + task = DownloadTask( + task_id="https://audiobookbay.lu/abss/test-book/", + source="audiobookbay", + title="Test Book", + content_type="audiobook", + ) + cancel_flag = Event() + recorder = ProgressRecorder() + + result = handler.download( + task=task, + cancel_flag=cancel_flag, + progress_callback=recorder.progress_callback, + status_callback=recorder.status_callback, + ) + + assert result is None + assert recorder.last_status == "error" + assert "magnet link" in recorder.last_message.lower() + + @patch('shelfmark.release_sources.audiobookbay.handler.scraper.extract_magnet_link') + @patch('shelfmark.release_sources.audiobookbay.handler.get_client') + @patch('shelfmark.release_sources.audiobookbay.handler.list_configured_clients') + def test_download_no_client_configured(self, mock_list_clients, mock_get_client, mock_extract_magnet): + """Test handling when no torrent client is configured.""" + mock_extract_magnet.return_value = "magnet:?xt=urn:btih:abc123" + mock_get_client.return_value = None + mock_list_clients.return_value = [] + + handler = AudiobookBayHandler() + task = DownloadTask( + task_id="https://audiobookbay.lu/abss/test-book/", + source="audiobookbay", + title="Test Book", + content_type="audiobook", + ) + cancel_flag = Event() + recorder = ProgressRecorder() + + result = handler.download( + task=task, + cancel_flag=cancel_flag, + progress_callback=recorder.progress_callback, + status_callback=recorder.status_callback, + ) + + assert result is None + assert recorder.last_status == "error" + assert "client" in recorder.last_message.lower() + + @patch('shelfmark.release_sources.audiobookbay.handler.scraper.extract_magnet_link') + @patch('shelfmark.release_sources.audiobookbay.handler.get_client') + def test_download_client_add_failure(self, mock_get_client, mock_extract_magnet): + """Test handling when client.add_download fails.""" + mock_extract_magnet.return_value = "magnet:?xt=urn:btih:abc123" + + mock_client = MagicMock() + mock_client.name = "qbittorrent" + mock_client.find_existing.return_value = None + mock_client.add_download.side_effect = Exception("Client error") + mock_get_client.return_value = mock_client + + handler = AudiobookBayHandler() + task = DownloadTask( + task_id="https://audiobookbay.lu/abss/test-book/", + source="audiobookbay", + title="Test Book", + content_type="audiobook", + ) + cancel_flag = Event() + recorder = ProgressRecorder() + + result = handler.download( + task=task, + cancel_flag=cancel_flag, + progress_callback=recorder.progress_callback, + status_callback=recorder.status_callback, + ) + + assert result is None + assert recorder.last_status == "error" + + @patch('shelfmark.release_sources.audiobookbay.handler.scraper.extract_magnet_link') + @patch('shelfmark.release_sources.audiobookbay.handler.get_client') + def test_download_existing_no_path(self, mock_get_client, mock_extract_magnet): + """Test handling when existing download has no path.""" + mock_extract_magnet.return_value = "magnet:?xt=urn:btih:abc123" + + mock_client = MagicMock() + mock_client.name = "qbittorrent" + mock_client.find_existing.return_value = ( + "existing_id", + DownloadStatus( + progress=100, + state=DownloadState.COMPLETE, + message="Complete", + complete=True, + file_path=None, + ), + ) + mock_client.get_download_path.return_value = None + mock_get_client.return_value = mock_client + + handler = AudiobookBayHandler() + task = DownloadTask( + task_id="https://audiobookbay.lu/abss/test-book/", + source="audiobookbay", + title="Test Book", + content_type="audiobook", + ) + cancel_flag = Event() + recorder = ProgressRecorder() + + result = handler.download( + task=task, + cancel_flag=cancel_flag, + progress_callback=recorder.progress_callback, + status_callback=recorder.status_callback, + ) + + assert result is None + assert recorder.last_status == "error" + assert "path" in recorder.last_message.lower() + + +class TestAudiobookBayHandlerCategory: + """Tests for category selection.""" + + @patch('shelfmark.release_sources.audiobookbay.handler.scraper.extract_magnet_link') + @patch('shelfmark.release_sources.audiobookbay.handler.get_client') + @patch('shelfmark.release_sources.audiobookbay.handler.config.get') + def test_category_selection_qbittorrent_audiobook(self, mock_config_get, mock_get_client, mock_extract_magnet): + """Test audiobook category selection for qBittorrent.""" + mock_extract_magnet.return_value = "magnet:?xt=urn:btih:abc123" + + def config_get(key, default=""): + if key == "QBITTORRENT_CATEGORY_AUDIOBOOK": + return "audiobooks" + return default + + mock_config_get.side_effect = config_get + + mock_client = MagicMock() + mock_client.name = "qbittorrent" + mock_client.find_existing.return_value = None + mock_client.add_download.return_value = "download_id" + mock_get_client.return_value = mock_client + + handler = AudiobookBayHandler() + task = DownloadTask( + task_id="https://audiobookbay.lu/abss/test-book/", + source="audiobookbay", + title="Test Book", + content_type="audiobook", + ) + cancel_flag = Event() + recorder = ProgressRecorder() + + handler.download( + task=task, + cancel_flag=cancel_flag, + progress_callback=recorder.progress_callback, + status_callback=recorder.status_callback, + ) + + # Verify category was passed + call_kwargs = mock_client.add_download.call_args.kwargs + assert call_kwargs['category'] == "audiobooks" + + @patch('shelfmark.release_sources.audiobookbay.handler.scraper.extract_magnet_link') + @patch('shelfmark.release_sources.audiobookbay.handler.get_client') + @patch('shelfmark.release_sources.audiobookbay.handler.config.get') + def test_category_selection_transmission_general(self, mock_config_get, mock_get_client, mock_extract_magnet): + """Test fallback to general category for Transmission.""" + mock_extract_magnet.return_value = "magnet:?xt=urn:btih:abc123" + + def config_get(key, default=""): + if key == "TRANSMISSION_CATEGORY": + return "books" + return default + + mock_config_get.side_effect = config_get + + mock_client = MagicMock() + mock_client.name = "transmission" + mock_client.find_existing.return_value = None + mock_client.add_download.return_value = "download_id" + mock_get_client.return_value = mock_client + + handler = AudiobookBayHandler() + task = DownloadTask( + task_id="https://audiobookbay.lu/abss/test-book/", + source="audiobookbay", + title="Test Book", + content_type="audiobook", + ) + cancel_flag = Event() + recorder = ProgressRecorder() + + handler.download( + task=task, + cancel_flag=cancel_flag, + progress_callback=recorder.progress_callback, + status_callback=recorder.status_callback, + ) + + # Verify general category was used + call_kwargs = mock_client.add_download.call_args.kwargs + assert call_kwargs['category'] == "books" + + @patch('shelfmark.release_sources.audiobookbay.handler.scraper.extract_magnet_link') + @patch('shelfmark.release_sources.audiobookbay.handler.get_client') + @patch('shelfmark.release_sources.audiobookbay.handler.config.get') + def test_category_selection_non_audiobook(self, mock_config_get, mock_get_client, mock_extract_magnet): + """Test that non-audiobook content types don't get category.""" + mock_extract_magnet.return_value = "magnet:?xt=urn:btih:abc123" + + mock_config_get.return_value = "" + + mock_client = MagicMock() + mock_client.name = "qbittorrent" + mock_client.find_existing.return_value = None + mock_client.add_download.return_value = "download_id" + mock_get_client.return_value = mock_client + + handler = AudiobookBayHandler() + task = DownloadTask( + task_id="https://audiobookbay.lu/abss/test-book/", + source="audiobookbay", + title="Test Book", + content_type="ebook", # Not audiobook + ) + cancel_flag = Event() + recorder = ProgressRecorder() + + handler.download( + task=task, + cancel_flag=cancel_flag, + progress_callback=recorder.progress_callback, + status_callback=recorder.status_callback, + ) + + # Verify no category was passed + call_kwargs = mock_client.add_download.call_args.kwargs + assert call_kwargs['category'] is None + + +class TestAudiobookBayHandlerCancel: + """Tests for AudiobookBayHandler.cancel().""" + + def test_cancel_returns_false(self): + """Test that cancel always returns False (torrents can't be cancelled).""" + handler = AudiobookBayHandler() + result = handler.cancel("test-task-id") + assert result is False diff --git a/tests/audiobookbay/test_scraper.py b/tests/audiobookbay/test_scraper.py new file mode 100644 index 0000000..2352804 --- /dev/null +++ b/tests/audiobookbay/test_scraper.py @@ -0,0 +1,383 @@ +""" +Tests for AudiobookBay scraper functions. +""" + +from unittest.mock import Mock, patch +import pytest +import requests + +from shelfmark.release_sources.audiobookbay import scraper + + +# Mock HTML based on real ABB structure +SAMPLE_SEARCH_HTML = """ + + +
+

Test Book Title - Test Author

+
Category: Genre 
Language: EnglishKeywords: Test Keywords 
+
+
+

Shared by:testuser

+

Test Cover

+
+

Posted: 01 Jan 2024
Format: M4B / Bitrate: 128 Kbps
File Size: 500.00 MBs

+
+
+ Audiobook Details + Direct Download +
+
+
+

Another Test Book - Another Author

+
Category: Fiction 
Language: SpanishKeywords: Test 
+
+
+

Shared by:user2

+

Cover 2

+
+

Posted: 15 Nov 2023
Format: MP3 / Bitrate: 256 Kbps
File Size: 1.01 GBs

+
+
+ Audiobook Details + Direct Download +
+
+ + +""" + +EMPTY_SEARCH_HTML = """ + + + + +""" + +# Mock HTML for detail page with info hash and trackers +SAMPLE_DETAIL_HTML = """ + + + + + + + + + + + + + + + + + + + +
Info HashABC123DEF456GHI789JKL012MNO345PQR678STU
Tracker 1udp://tracker.openbittorrent.com:80
Tracker 2http://tracker.example.com:8080
Other InfoSome other data
+ + +""" + +DETAIL_HTML_NO_TRACKERS = """ + + + + + + + +
Info HashABC123DEF456GHI789JKL012MNO345PQR678STU
+ + +""" + + +class TestSearchAudiobookbay: + """Tests for the search_audiobookbay function.""" + + @patch('shelfmark.release_sources.audiobookbay.scraper.requests.get') + @patch('shelfmark.release_sources.audiobookbay.scraper.network.get_proxies') + @patch('shelfmark.release_sources.audiobookbay.scraper.config.get') + def test_search_audiobookbay_success(self, mock_config_get, mock_get_proxies, mock_get): + """Test successful search with results.""" + mock_config_get.return_value = 1.0 # rate_limit_delay + mock_get_proxies.return_value = {} + + # Mock response + mock_response = Mock() + mock_response.status_code = 200 + mock_response.url = "https://audiobookbay.lu/page/1/?s=test+query&cat=undefined%2Cundefined" + mock_response.text = SAMPLE_SEARCH_HTML + mock_get.return_value = mock_response + + results = scraper.search_audiobookbay("test query", max_pages=1, hostname="audiobookbay.lu") + + assert len(results) == 2 + assert results[0]['title'] == "Test Book Title - Test Author" + assert results[0]['link'] == "https://audiobookbay.lu/abss/test-book-title-by-author/" + assert results[0]['language'] == "English" + assert results[0]['format'] == "M4B" + assert results[0]['bitrate'] == "128 Kbps" + assert results[0]['size'] == "500.00 MBs" + assert results[0]['posted_date'] == "01 Jan 2024" + assert results[0]['cover'] == "https://example.com/cover.jpg" + + assert results[1]['title'] == "Another Test Book - Another Author" + assert results[1]['language'] == "Spanish" + assert results[1]['format'] == "MP3" + assert results[1]['size'] == "1.01 GBs" + + @patch('shelfmark.release_sources.audiobookbay.scraper.requests.get') + @patch('shelfmark.release_sources.audiobookbay.scraper.network.get_proxies') + @patch('shelfmark.release_sources.audiobookbay.scraper.config.get') + def test_search_audiobookbay_pagination(self, mock_config_get, mock_get_proxies, mock_get): + """Test pagination through multiple pages.""" + mock_config_get.return_value = 0.0 # No delay for faster tests + mock_get_proxies.return_value = {} + + # First page response + mock_response_page1 = Mock() + mock_response_page1.status_code = 200 + mock_response_page1.url = "https://audiobookbay.lu/page/1/?s=test&cat=undefined%2Cundefined" + mock_response_page1.text = SAMPLE_SEARCH_HTML + + # Second page response (empty) + mock_response_page2 = Mock() + mock_response_page2.status_code = 200 + mock_response_page2.url = "https://audiobookbay.lu/page/2/?s=test&cat=undefined%2Cundefined" + mock_response_page2.text = EMPTY_SEARCH_HTML + + mock_get.side_effect = [mock_response_page1, mock_response_page2] + + results = scraper.search_audiobookbay("test", max_pages=2, hostname="audiobookbay.lu") + + assert len(results) == 2 # Only from first page + assert mock_get.call_count == 2 + + @patch('shelfmark.release_sources.audiobookbay.scraper.requests.get') + @patch('shelfmark.release_sources.audiobookbay.scraper.network.get_proxies') + @patch('shelfmark.release_sources.audiobookbay.scraper.config.get') + def test_search_audiobookbay_empty(self, mock_config_get, mock_get_proxies, mock_get): + """Test search with no results.""" + mock_config_get.return_value = 1.0 + mock_get_proxies.return_value = {} + + mock_response = Mock() + mock_response.status_code = 200 + mock_response.url = "https://audiobookbay.lu/page/1/?s=test&cat=undefined%2Cundefined" + mock_response.text = EMPTY_SEARCH_HTML + mock_get.return_value = mock_response + + results = scraper.search_audiobookbay("test", max_pages=1, hostname="audiobookbay.lu") + + assert len(results) == 0 + + @patch('shelfmark.release_sources.audiobookbay.scraper.requests.get') + @patch('shelfmark.release_sources.audiobookbay.scraper.network.get_proxies') + @patch('shelfmark.release_sources.audiobookbay.scraper.config.get') + def test_search_audiobookbay_error_non_200(self, mock_config_get, mock_get_proxies, mock_get): + """Test error handling for non-200 status code.""" + mock_config_get.return_value = 1.0 + mock_get_proxies.return_value = {} + + mock_response = Mock() + mock_response.status_code = 404 + mock_get.return_value = mock_response + + results = scraper.search_audiobookbay("test", max_pages=1, hostname="audiobookbay.lu") + + assert len(results) == 0 + + @patch('shelfmark.release_sources.audiobookbay.scraper.requests.get') + @patch('shelfmark.release_sources.audiobookbay.scraper.network.get_proxies') + @patch('shelfmark.release_sources.audiobookbay.scraper.config.get') + def test_search_audiobookbay_redirect_to_homepage(self, mock_config_get, mock_get_proxies, mock_get): + """Test handling redirect to homepage (blocked/invalid search).""" + mock_config_get.return_value = 1.0 + mock_get_proxies.return_value = {} + + mock_response = Mock() + mock_response.status_code = 200 + mock_response.url = "https://audiobookbay.lu" # Redirected to homepage + mock_response.text = EMPTY_SEARCH_HTML + mock_get.return_value = mock_response + + results = scraper.search_audiobookbay("test", max_pages=1, hostname="audiobookbay.lu") + + assert len(results) == 0 + + @patch('shelfmark.release_sources.audiobookbay.scraper.requests.get') + @patch('shelfmark.release_sources.audiobookbay.scraper.network.get_proxies') + @patch('shelfmark.release_sources.audiobookbay.scraper.config.get') + def test_search_audiobookbay_request_exception(self, mock_config_get, mock_get_proxies, mock_get): + """Test handling request exceptions.""" + mock_config_get.return_value = 1.0 + mock_get_proxies.return_value = {} + + mock_get.side_effect = requests.exceptions.RequestException("Connection error") + + results = scraper.search_audiobookbay("test", max_pages=1, hostname="audiobookbay.lu") + + assert len(results) == 0 + + @patch('shelfmark.release_sources.audiobookbay.scraper.requests.get') + @patch('shelfmark.release_sources.audiobookbay.scraper.network.get_proxies') + @patch('shelfmark.release_sources.audiobookbay.scraper.config.get') + def test_search_audiobookbay_relative_link(self, mock_config_get, mock_get_proxies, mock_get): + """Test handling relative links in results.""" + mock_config_get.return_value = 1.0 + mock_get_proxies.return_value = {} + + html_with_relative_link = """ +
+

Test Book

+
Language: English
+
+

Posted: 01 Jan 2024
Format: M4B
File Size: 100 MBs

+
+
+ """ + + mock_response = Mock() + mock_response.status_code = 200 + mock_response.url = "https://audiobookbay.lu/page/1/?s=test&cat=undefined%2Cundefined" + mock_response.text = html_with_relative_link + mock_get.return_value = mock_response + + results = scraper.search_audiobookbay("test", max_pages=1, hostname="audiobookbay.lu") + + assert len(results) == 1 + assert results[0]['link'] == "https://audiobookbay.lu/abss/relative-link/" + + +class TestExtractMagnetLink: + """Tests for the extract_magnet_link function.""" + + @patch('shelfmark.release_sources.audiobookbay.scraper.requests.get') + @patch('shelfmark.release_sources.audiobookbay.scraper.network.get_proxies') + def test_extract_magnet_link_success(self, mock_get_proxies, mock_get): + """Test successful magnet link extraction.""" + mock_get_proxies.return_value = {} + + mock_response = Mock() + mock_response.status_code = 200 + mock_response.text = SAMPLE_DETAIL_HTML + mock_get.return_value = mock_response + + magnet_link = scraper.extract_magnet_link( + "https://audiobookbay.lu/abss/test-book/", + hostname="audiobookbay.lu" + ) + + assert magnet_link is not None + assert magnet_link.startswith("magnet:?xt=urn:btih:") + assert "ABC123DEF456GHI789JKL012MNO345PQR678STU" in magnet_link + assert "udp%3A//tracker.openbittorrent.com%3A80" in magnet_link + assert "http%3A//tracker.example.com%3A8080" in magnet_link + + @patch('shelfmark.release_sources.audiobookbay.scraper.requests.get') + @patch('shelfmark.release_sources.audiobookbay.scraper.network.get_proxies') + def test_extract_magnet_link_fallback(self, mock_get_proxies, mock_get): + """Test fallback to default trackers when none found.""" + mock_get_proxies.return_value = {} + + mock_response = Mock() + mock_response.status_code = 200 + mock_response.text = DETAIL_HTML_NO_TRACKERS + mock_get.return_value = mock_response + + magnet_link = scraper.extract_magnet_link( + "https://audiobookbay.lu/abss/test-book/", + hostname="audiobookbay.lu" + ) + + assert magnet_link is not None + assert magnet_link.startswith("magnet:?xt=urn:btih:") + assert "ABC123DEF456GHI789JKL012MNO345PQR678STU" in magnet_link + # Should contain default trackers + assert "udp%3A//tracker.openbittorrent.com%3A80" in magnet_link + + @patch('shelfmark.release_sources.audiobookbay.scraper.requests.get') + @patch('shelfmark.release_sources.audiobookbay.scraper.network.get_proxies') + def test_extract_magnet_link_no_info_hash(self, mock_get_proxies, mock_get): + """Test handling missing info hash.""" + mock_get_proxies.return_value = {} + + mock_response = Mock() + mock_response.status_code = 200 + mock_response.text = "" + mock_get.return_value = mock_response + + magnet_link = scraper.extract_magnet_link( + "https://audiobookbay.lu/abss/test-book/", + hostname="audiobookbay.lu" + ) + + assert magnet_link is None + + @patch('shelfmark.release_sources.audiobookbay.scraper.requests.get') + @patch('shelfmark.release_sources.audiobookbay.scraper.network.get_proxies') + def test_extract_magnet_link_non_200(self, mock_get_proxies, mock_get): + """Test handling non-200 status code.""" + mock_get_proxies.return_value = {} + + mock_response = Mock() + mock_response.status_code = 404 + mock_get.return_value = mock_response + + magnet_link = scraper.extract_magnet_link( + "https://audiobookbay.lu/abss/test-book/", + hostname="audiobookbay.lu" + ) + + assert magnet_link is None + + @patch('shelfmark.release_sources.audiobookbay.scraper.requests.get') + @patch('shelfmark.release_sources.audiobookbay.scraper.network.get_proxies') + def test_extract_magnet_link_request_exception(self, mock_get_proxies, mock_get): + """Test handling request exceptions.""" + mock_get_proxies.return_value = {} + + mock_get.side_effect = requests.exceptions.RequestException("Connection error") + + magnet_link = scraper.extract_magnet_link( + "https://audiobookbay.lu/abss/test-book/", + hostname="audiobookbay.lu" + ) + + assert magnet_link is None + + @patch('shelfmark.release_sources.audiobookbay.scraper.requests.get') + @patch('shelfmark.release_sources.audiobookbay.scraper.network.get_proxies') + def test_extract_magnet_link_cleans_info_hash(self, mock_get_proxies, mock_get): + """Test that info hash whitespace is cleaned.""" + mock_get_proxies.return_value = {} + + html_with_whitespace = """ + + + + + + + +
Info HashABC 123 DEF 456
+ + + """ + + mock_response = Mock() + mock_response.status_code = 200 + mock_response.text = html_with_whitespace + mock_get.return_value = mock_response + + magnet_link = scraper.extract_magnet_link( + "https://audiobookbay.lu/abss/test-book/", + hostname="audiobookbay.lu" + ) + + assert magnet_link is not None + # Info hash should be cleaned (no spaces, uppercase) + assert "ABC123DEF456" in magnet_link diff --git a/tests/audiobookbay/test_source.py b/tests/audiobookbay/test_source.py new file mode 100644 index 0000000..5810cc1 --- /dev/null +++ b/tests/audiobookbay/test_source.py @@ -0,0 +1,428 @@ +""" +Tests for AudiobookBay release source. +""" + +from unittest.mock import Mock, patch +import pytest + +from shelfmark.metadata_providers import BookMetadata +from shelfmark.core.search_plan import ReleaseSearchPlan, ReleaseSearchVariant +from shelfmark.release_sources.audiobookbay.source import ( + AudiobookBaySource, + _map_language, + _generate_source_id, +) + + +class TestMapLanguage: + """Tests for the _map_language function.""" + + def test_map_language_english(self): + """Test mapping English language.""" + assert _map_language("English") == "en" + assert _map_language("english") == "en" + assert _map_language("ENGLISH") == "en" + + def test_map_language_spanish(self): + """Test mapping Spanish language.""" + assert _map_language("Spanish") == "es" + assert _map_language("spanish") == "es" + + def test_map_language_french(self): + """Test mapping French language.""" + assert _map_language("French") == "fr" + assert _map_language("french") == "fr" + + def test_map_language_german(self): + """Test mapping German language.""" + assert _map_language("German") == "de" + assert _map_language("german") == "de" + + def test_map_language_unknown(self): + """Test mapping unknown language returns lowercase.""" + assert _map_language("UnknownLanguage") == "unknownlanguage" + assert _map_language("Klingon") == "klingon" + + def test_map_language_empty(self): + """Test empty language returns None.""" + assert _map_language("") is None + assert _map_language(None) is None + + def test_map_language_with_whitespace(self): + """Test language with whitespace is trimmed.""" + assert _map_language(" English ") == "en" + assert _map_language("\tFrench\n") == "fr" + + +class TestGenerateSourceId: + """Tests for the _generate_source_id function.""" + + def test_generate_source_id_consistent(self): + """Test that same URL generates same ID.""" + url = "https://audiobookbay.lu/abss/test-book/" + id1 = _generate_source_id(url) + id2 = _generate_source_id(url) + assert id1 == id2 + assert len(id1) == 32 # MD5 hex digest length + + def test_generate_source_id_different_urls(self): + """Test that different URLs generate different IDs.""" + url1 = "https://audiobookbay.lu/abss/book1/" + url2 = "https://audiobookbay.lu/abss/book2/" + id1 = _generate_source_id(url1) + id2 = _generate_source_id(url2) + assert id1 != id2 + + +class TestAudiobookBaySource: + """Tests for the AudiobookBaySource class.""" + + def test_is_available_enabled(self, monkeypatch): + """Test is_available when enabled.""" + def mock_get(key, default=False): + if key == "ABB_ENABLED": + return True + return default + + import shelfmark.release_sources.audiobookbay.source as source_module + monkeypatch.setattr(source_module.config, "get", mock_get) + + source = AudiobookBaySource() + assert source.is_available() is True + + def test_is_available_disabled(self, monkeypatch): + """Test is_available when disabled.""" + def mock_get(key, default=False): + if key == "ABB_ENABLED": + return False + return default + + import shelfmark.release_sources.audiobookbay.source as source_module + monkeypatch.setattr(source_module.config, "get", mock_get) + + source = AudiobookBaySource() + assert source.is_available() is False + + def test_search_non_audiobook_content_type(self): + """Test that non-audiobook content types return empty.""" + source = AudiobookBaySource() + book = BookMetadata( + provider="test", + provider_id="123", + title="Test Book", + authors=["Test Author"], + ) + plan = ReleaseSearchPlan( + languages=["en"], + isbn_candidates=[], + author="Test Author", + title_variants=[ReleaseSearchVariant(title="Test Book", author="Test Author")], + grouped_title_variants=[], + ) + + results = source.search(book, plan, content_type="ebook") + assert results == [] + + @patch('shelfmark.release_sources.audiobookbay.source.scraper.search_audiobookbay') + def test_search_query_generation_manual(self, mock_search): + """Test search with manual query.""" + mock_search.return_value = [] + + source = AudiobookBaySource() + book = BookMetadata( + provider="test", + provider_id="123", + title="Test Book", + authors=["Test Author"], + ) + plan = ReleaseSearchPlan( + languages=["en"], + isbn_candidates=[], + author="Test Author", + title_variants=[ReleaseSearchVariant(title="Test Book", author="Test Author")], + grouped_title_variants=[], + manual_query="custom search query", + ) + + source.search(book, plan, content_type="audiobook") + + mock_search.assert_called_once() + call_args = mock_search.call_args + assert call_args.kwargs['query'] == "custom search query" + + @patch('shelfmark.release_sources.audiobookbay.source.scraper.search_audiobookbay') + def test_search_query_generation_from_variants(self, mock_search): + """Test search query generation from title variants.""" + mock_search.return_value = [] + + source = AudiobookBaySource() + book = BookMetadata( + provider="test", + provider_id="123", + title="Test Book", + authors=["Test Author"], + ) + plan = ReleaseSearchPlan( + languages=["en"], + isbn_candidates=[], + author="Test Author", + title_variants=[ReleaseSearchVariant(title="Test Book", author="Test Author")], + grouped_title_variants=[], + ) + + source.search(book, plan, content_type="audiobook") + + mock_search.assert_called_once() + call_args = mock_search.call_args + assert call_args.kwargs['query'] == "test book test author" + + @patch('shelfmark.release_sources.audiobookbay.source.scraper.search_audiobookbay') + def test_search_query_generation_from_title_only(self, mock_search): + """Test search query generation when only title available.""" + mock_search.return_value = [] + + source = AudiobookBaySource() + book = BookMetadata( + provider="test", + provider_id="123", + title="Test Book", + authors=[], + ) + plan = ReleaseSearchPlan( + languages=["en"], + isbn_candidates=[], + author="", + title_variants=[], + grouped_title_variants=[], + ) + + source.search(book, plan, content_type="audiobook") + + mock_search.assert_called_once() + call_args = mock_search.call_args + assert call_args.kwargs['query'] == "test book" + + @patch('shelfmark.release_sources.audiobookbay.source.scraper.search_audiobookbay') + def test_search_empty_query(self, mock_search): + """Test search with empty query returns empty.""" + source = AudiobookBaySource() + book = BookMetadata( + provider="test", + provider_id="123", + title="", + authors=[], + ) + plan = ReleaseSearchPlan( + languages=["en"], + isbn_candidates=[], + author="", + title_variants=[], + grouped_title_variants=[], + ) + + results = source.search(book, plan, content_type="audiobook") + + assert results == [] + mock_search.assert_not_called() + + @patch('shelfmark.release_sources.audiobookbay.source.scraper.search_audiobookbay') + def test_search_relevance_filtering(self, mock_search): + """Test that irrelevant results are filtered out.""" + mock_search.return_value = [ + { + 'title': 'Test Book by Test Author', + 'link': 'https://audiobookbay.lu/abss/test-book/', + 'format': 'M4B', + 'size': '500 MB', + 'language': 'English', + }, + { + 'title': 'Something Completely Different', + 'link': 'https://audiobookbay.lu/abss/unrelated/', + 'format': 'MP3', + 'size': '1 GB', + 'language': 'English', + }, + ] + + source = AudiobookBaySource() + book = BookMetadata( + provider="test", + provider_id="123", + title="Test Book", + authors=["Test Author"], + ) + plan = ReleaseSearchPlan( + languages=["en"], + isbn_candidates=[], + author="Test Author", + title_variants=[ReleaseSearchVariant(title="Test Book", author="Test Author")], + grouped_title_variants=[], + ) + + results = source.search(book, plan, content_type="audiobook") + + # Should filter out "Unrelated Book Title" as it doesn't contain query words + assert len(results) == 1 + assert results[0].title == 'Test Book by Test Author' + + @patch('shelfmark.release_sources.audiobookbay.source.scraper.search_audiobookbay') + def test_search_result_mapping(self, mock_search): + """Test conversion of scraper results to Release objects.""" + mock_search.return_value = [ + { + 'title': 'Test Book - Test Author', + 'link': 'https://audiobookbay.lu/abss/test-book/', + 'format': 'M4B', + 'size': '500.00 MBs', + 'language': 'English', + 'bitrate': '128 Kbps', + 'posted_date': '01 Jan 2024', + 'cover': 'https://example.com/cover.jpg', + }, + ] + + source = AudiobookBaySource() + book = BookMetadata( + provider="test", + provider_id="123", + title="Test Book", + authors=["Test Author"], + ) + plan = ReleaseSearchPlan( + languages=["en"], + isbn_candidates=[], + author="Test Author", + title_variants=[ReleaseSearchVariant(title="Test Book", author="Test Author")], + grouped_title_variants=[], + ) + + results = source.search(book, plan, content_type="audiobook") + + assert len(results) == 1 + release = results[0] + assert release.source == "audiobookbay" + assert release.title == "Test Book - Test Author" + assert release.format == "m4b" + assert release.language == "en" + assert release.size == "500.00 MBs" + assert release.download_url == "https://audiobookbay.lu/abss/test-book/" + assert release.info_url == "https://audiobookbay.lu/abss/test-book/" + assert release.protocol.value == "torrent" + assert release.indexer == "AudiobookBay" + assert release.content_type == "audiobook" + assert release.extra['preview'] == "https://example.com/cover.jpg" + assert release.extra['detail_url'] == "https://audiobookbay.lu/abss/test-book/" + assert release.extra['bitrate'] == "128 Kbps" + assert release.extra['posted_date'] == "01 Jan 2024" + assert release.extra['language_raw'] == "English" + + @patch('shelfmark.release_sources.audiobookbay.source.scraper.search_audiobookbay') + def test_search_handles_scraper_exception(self, mock_search): + """Test that scraper exceptions are handled gracefully.""" + mock_search.side_effect = Exception("Scraper error") + + source = AudiobookBaySource() + book = BookMetadata( + provider="test", + provider_id="123", + title="Test Book", + authors=["Test Author"], + ) + plan = ReleaseSearchPlan( + languages=["en"], + isbn_candidates=[], + author="Test Author", + title_variants=[ReleaseSearchVariant(title="Test Book", author="Test Author")], + grouped_title_variants=[], + ) + + results = source.search(book, plan, content_type="audiobook") + + assert results == [] + + @patch('shelfmark.release_sources.audiobookbay.source.scraper.search_audiobookbay') + def test_search_handles_invalid_result(self, mock_search): + """Test that invalid results are skipped.""" + mock_search.return_value = [ + { + 'title': 'Relevant Book', + 'link': 'https://audiobookbay.lu/abss/valid/', + 'format': 'M4B', + }, + { + # Missing required fields + 'title': 'Relevant But Invalid', + }, + ] + + source = AudiobookBaySource() + book = BookMetadata( + provider="test", + provider_id="123", + title="Relevant", + authors=["Author"], + ) + plan = ReleaseSearchPlan( + languages=["en"], + isbn_candidates=[], + author="Author", + title_variants=[ReleaseSearchVariant(title="Relevant", author="Author")], + grouped_title_variants=[], + ) + + results = source.search(book, plan, content_type="audiobook") + + # Should only include valid result + assert len(results) == 1 + assert results[0].title == "Relevant Book" + + @patch('shelfmark.release_sources.audiobookbay.source.scraper.search_audiobookbay') + def test_search_config_hostname(self, mock_search, monkeypatch): + """Test that custom hostname is used from config.""" + mock_search.return_value = [] + + def mock_get(key, default=None): + if key == "ABB_HOSTNAME": + return "audiobookbay.is" + if key == "ABB_PAGE_LIMIT": + return 3 + return default + + import shelfmark.release_sources.audiobookbay.source as source_module + monkeypatch.setattr(source_module.config, "get", mock_get) + + source = AudiobookBaySource() + book = BookMetadata( + provider="test", + provider_id="123", + title="Test Book", + authors=["Test Author"], + ) + plan = ReleaseSearchPlan( + languages=["en"], + isbn_candidates=[], + author="Test Author", + title_variants=[ReleaseSearchVariant(title="Test Book", author="Test Author")], + grouped_title_variants=[], + ) + + source.search(book, plan, content_type="audiobook") + + call_args = mock_search.call_args + assert call_args.kwargs['hostname'] == "audiobookbay.is" + assert call_args.kwargs['max_pages'] == 3 + + def test_get_column_config(self): + """Test column configuration.""" + source = AudiobookBaySource() + config = source.get_column_config() + + assert config is not None + assert len(config.columns) == 3 + column_keys = [col.key for col in config.columns] + assert "language" in column_keys + assert "format" in column_keys + assert "size" in column_keys + assert "seeders" not in column_keys # ABB doesn't show seeders + assert config.supported_filters == ["format", "language"] diff --git a/tests/audiobookbay/test_utils.py b/tests/audiobookbay/test_utils.py new file mode 100644 index 0000000..1b58ed8 --- /dev/null +++ b/tests/audiobookbay/test_utils.py @@ -0,0 +1,103 @@ +""" +Tests for AudiobookBay utility functions. +""" + +import pytest + +from shelfmark.release_sources.audiobookbay.utils import parse_size, sanitize_title + + +class TestParseSize: + """Tests for the parse_size function.""" + + def test_parse_size_bytes(self): + """Test parsing byte sizes.""" + assert parse_size("100 B") == 100 + assert parse_size("512 B") == 512 + assert parse_size("100B") == 100 # No space + + def test_parse_size_kilobytes(self): + """Test parsing kilobyte sizes.""" + assert parse_size("1 KB") == 1024 + assert parse_size("2 KB") == 2048 + assert parse_size("1.5 KB") == int(1.5 * 1024) + assert parse_size("1KBs") == 1024 # Handles "KBs" suffix + + def test_parse_size_megabytes(self): + """Test parsing megabyte sizes.""" + assert parse_size("1 MB") == 1024 ** 2 + assert parse_size("500 MB") == 500 * (1024 ** 2) + assert parse_size("1.5 MB") == int(1.5 * (1024 ** 2)) + assert parse_size("500.00 MBs") == int(500.00 * (1024 ** 2)) # Handles "MBs" suffix + + def test_parse_size_gigabytes(self): + """Test parsing gigabyte sizes.""" + assert parse_size("1 GB") == 1024 ** 3 + assert parse_size("11.68 GB") == int(11.68 * (1024 ** 3)) + assert parse_size("1.01 GBs") == int(1.01 * (1024 ** 3)) # Handles "GBs" suffix + + def test_parse_size_terabytes(self): + """Test parsing terabyte sizes.""" + assert parse_size("1 TB") == 1024 ** 4 + assert parse_size("2.5 TB") == int(2.5 * (1024 ** 4)) + + def test_parse_size_case_insensitive(self): + """Test that size parsing is case insensitive.""" + assert parse_size("1 gb") == 1024 ** 3 + assert parse_size("1 Gb") == 1024 ** 3 + assert parse_size("1 GB") == 1024 ** 3 + assert parse_size("1 gbs") == 1024 ** 3 + + def test_parse_size_none(self): + """Test that None returns None.""" + assert parse_size(None) is None + + def test_parse_size_empty_string(self): + """Test that empty string returns None.""" + assert parse_size("") is None + + def test_parse_size_invalid_format(self): + """Test that invalid formats return None.""" + assert parse_size("invalid") is None + assert parse_size("123") is None # No unit + assert parse_size("abc MB") is None # Invalid number + + def test_parse_size_with_whitespace(self): + """Test parsing with various whitespace.""" + assert parse_size(" 1 GB ") == 1024 ** 3 + assert parse_size("1.5\tMB") == int(1.5 * (1024 ** 2)) + + +class TestSanitizeTitle: + """Tests for the sanitize_title function.""" + + def test_sanitize_title_removes_invalid_chars(self): + """Test removing invalid filename characters.""" + assert sanitize_title("Test") == "TestBook" + assert sanitize_title("Test:Book") == "TestBook" + assert sanitize_title("Test/Book") == "TestBook" + assert sanitize_title("Test\\Book") == "TestBook" + assert sanitize_title("Test|Book") == "TestBook" + assert sanitize_title("Test?Book") == "TestBook" + assert sanitize_title("Test*Book") == "TestBook" + assert sanitize_title('Test"Book') == "TestBook" + + def test_sanitize_title_preserves_valid_chars(self): + """Test that valid characters are preserved.""" + assert sanitize_title("Test Book - Author") == "Test Book - Author" + assert sanitize_title("Test Book (2024)") == "Test Book (2024)" + assert sanitize_title("Test Book [Special]") == "Test Book [Special]" + assert sanitize_title("Test Book's Title") == "Test Book's Title" + + def test_sanitize_title_strips_whitespace(self): + """Test that leading/trailing whitespace is stripped.""" + assert sanitize_title(" Test Book ") == "Test Book" + assert sanitize_title("\tTest Book\n") == "Test Book" + + def test_sanitize_title_empty(self): + """Test that empty string returns empty.""" + assert sanitize_title("") == "" + + def test_sanitize_title_multiple_invalid_chars(self): + """Test removing multiple invalid characters.""" + assert sanitize_title("Test:Author/Title") == "TestBookAuthorTitle"