starred/shelfmark
1
0
Fork 0
mirror of https://github.com/calibrain/shelfmark.git synced 2026-09-24 19:00:21 +01:00
Code Issues Packages Projects Releases 10 Wiki Activity

Feature: Add AudiobookBay release source (#619)

Browse Source 
## 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
<img width="600" alt="image"
src="https://github.com/user-attachments/assets/2e10a259-5c35-4065-980d-b59a1c961c9f"
/>
This commit is contained in:
bonsai-dreams
2026-02-16 14:44:47 +00:00
committed by GitHub
parent ccb39e674e
commit dd6fd1e199
13 changed files with 2186 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
readme.md
+2 -1
View File
@@ -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
shelfmark/release_sources/__init__.py
+1
View File
@@ -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
shelfmark/release_sources/audiobookbay/__init__.py
+6
View File
@@ -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
shelfmark/release_sources/audiobookbay/handler.py
+164
View File
@@ -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
shelfmark/release_sources/audiobookbay/scraper.py
+271
View File
@@ -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 <td>Info Hash</td> 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 <td> 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
shelfmark/release_sources/audiobookbay/settings.py
+75
View File
@@ -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'.",
),
]
shelfmark/release_sources/audiobookbay/source.py
+253
View File
@@ -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
)
shelfmark/release_sources/audiobookbay/utils.py
+47
View File
@@ -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))
tests/audiobookbay/__init__.py
+1
View File
@@ -0,0 +1 @@
"""Tests for AudiobookBay integration."""
tests/audiobookbay/test_handler.py
+452
View File
@@ -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
tests/audiobookbay/test_scraper.py
+383
View File
@@ -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 = """
<html>
<body>
<div class="post">
<div class="postTitle"><h2><a href="/abss/test-book-title-by-author/" rel="bookmark">Test Book Title - Test Author</a></h2></div>
<div class="postInfo">Category: Genre&nbsp; <br>Language: English<span style="margin-left:100px;">Keywords: Test Keywords&nbsp;</span><br></div>
<div class="postContent">
<div class="center">
<p class="center">Shared by:<a href="/member/users/index?&mode=userinfo&username=testuser">testuser</a></p>
<p class="center"><a href="https://audiobookbay.lu/abss/test-book-title-by-author/"><img src="https://example.com/cover.jpg" alt="Test Cover" width="250"></a></p>
</div>
<p style="text-align:center;">Posted: 01 Jan 2024<br>Format: <span style="color:#a00;">M4B</span> / Bitrate: <span style="color:#a00;">128 Kbps</span><br>File Size: <span style="color:#00f;">500.00</span> MBs</p>
</div>
<div class="postMeta">
<span class="postLink"><a href="https://audiobookbay.lu/abss/test-book-title-by-author/">Audiobook Details</a></span>
<span class="postComments"><a href="/dload-now?ll=test" rel="nofollow">Direct Download</a></span>
</div>
</div>
<div class="post">
<div class="postTitle"><h2><a href="/abss/another-test-book/" rel="bookmark">Another Test Book - Another Author</a></h2></div>
<div class="postInfo">Category: Fiction&nbsp; <br>Language: Spanish<span style="margin-left:100px;">Keywords: Test&nbsp;</span><br></div>
<div class="postContent">
<div class="center">
<p class="center">Shared by:<a href="/member/users/index?&mode=userinfo&username=user2">user2</a></p>
<p class="center"><a href="https://audiobookbay.lu/abss/another-test-book/"><img src="https://example.com/cover2.jpg" alt="Cover 2" width="250"></a></p>
</div>
<p style="text-align:center;">Posted: 15 Nov 2023<br>Format: <span style="color:#a00;">MP3</span> / Bitrate: <span style="color:#a00;">256 Kbps</span><br>File Size: <span style="color:#00f;">1.01</span> GBs</p>
</div>
<div class="postMeta">
<span class="postLink"><a href="https://audiobookbay.lu/abss/another-test-book/">Audiobook Details</a></span>
<span class="postComments"><a href="/dload-now?ll=test2" rel="nofollow">Direct Download</a></span>
</div>
</div>
</body>
</html>
"""
EMPTY_SEARCH_HTML = """
<html>
<body>
</body>
</html>
"""
# Mock HTML for detail page with info hash and trackers
SAMPLE_DETAIL_HTML = """
<html>
<body>
<table>
<tr>
<td>Info Hash</td>
<td>ABC123DEF456GHI789JKL012MNO345PQR678STU</td>
</tr>
<tr>
<td>Tracker 1</td>
<td>udp://tracker.openbittorrent.com:80</td>
</tr>
<tr>
<td>Tracker 2</td>
<td>http://tracker.example.com:8080</td>
</tr>
<tr>
<td>Other Info</td>
<td>Some other data</td>
</tr>
</table>
</body>
</html>
"""
DETAIL_HTML_NO_TRACKERS = """
<html>
<body>
<table>
<tr>
<td>Info Hash</td>
<td>ABC123DEF456GHI789JKL012MNO345PQR678STU</td>
</tr>
</table>
</body>
</html>
"""
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 = """
<div class="post">
<div class="postTitle"><h2><a href="/abss/relative-link/">Test Book</a></h2></div>
<div class="postInfo">Language: English</div>
<div class="postContent">
<p style="text-align:center;">Posted: 01 Jan 2024<br>Format: M4B<br>File Size: 100 MBs</p>
</div>
</div>
"""
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 = "<html><body></body></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 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 = """
<html>
<body>
<table>
<tr>
<td>Info Hash</td>
<td>ABC 123 DEF 456</td>
</tr>
</table>
</body>
</html>
"""
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
tests/audiobookbay/test_source.py
+428
View File
@@ -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"]
tests/audiobookbay/test_utils.py
+103
View File
@@ -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<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"
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<Book>:Author/Title") == "TestBookAuthorTitle"