Template based file naming and torrent hardlinking (#385)

- Added alternative file processing mode. Save files directly into a
library folder and set up file names / directories based on user
preference.
- Uses template based naming and directory creation. E.g. {Author} /
{Series} {Title} {Part} etc. Works for saving correctly to libraries
such as Audiobookshelf.
- Use torrent hardlinking directly into library directories.
This commit is contained in:
Alex
2026-01-01 12:35:22 +00:00
committed by GitHub
parent 06e468d043
commit b2887eb4b0
34 changed files with 5422 additions and 578 deletions
+132 -43
View File
@@ -481,61 +481,50 @@ def network_settings():
def download_settings():
"""Configure download behavior and file locations."""
return [
# === BOOKS SECTION ===
HeadingField(
key="books_heading",
title="Books",
description="Configure how ebooks, comics, and magazines are processed.",
),
SelectField(
key="PROCESSING_MODE",
label="Processing Mode",
description="Ingest mode processes and moves files to an ingest folder. Library mode organizes files with custom naming. Note: RAR/ZIP extraction is only supported in Ingest mode.",
options=[
{"value": "ingest", "label": "Ingest"},
{"value": "library", "label": "Library"},
],
default="ingest",
universal_only=True,
),
TextField(
key="INGEST_DIR",
label="Download Directory",
description="Directory where downloaded files are saved.",
description="Directory where downloaded files are saved for processing. Recommended to use your library's ingest directory.",
default="/cwa-book-ingest",
required=True,
show_when={"field": "PROCESSING_MODE", "value": "ingest"},
),
CheckboxField(
key="USE_BOOK_TITLE",
label="Use Book Info as Filename",
description="Save files using Author, Title and Year instead of ID. May cause issues with special characters.",
description="Save files using Author, Title and Year instead of ID.",
default=True,
),
CheckboxField(
key="AUTO_OPEN_DOWNLOADS_SIDEBAR",
label="Auto-Open Downloads Sidebar",
description="Automatically open the downloads sidebar when a new download is queued.",
default=False,
env_supported=False, # UI-only setting
),
CheckboxField(
key="DOWNLOAD_TO_BROWSER",
label="Download to Browser",
description="Automatically download completed files to your browser.",
default=False,
env_supported=False, # UI-only setting
),
NumberField(
key="MAX_CONCURRENT_DOWNLOADS",
label="Max Concurrent Downloads",
description="Maximum number of simultaneous downloads.",
default=3,
min_value=1,
max_value=10,
requires_restart=True,
),
NumberField(
key="STATUS_TIMEOUT",
label="Status Timeout (seconds)",
description="How long to keep completed/failed downloads in the queue display.",
default=3600,
min_value=60,
max_value=86400,
show_when={"field": "PROCESSING_MODE", "value": "ingest"},
),
CheckboxField(
key="USE_CONTENT_TYPE_DIRECTORIES",
label="Configure Content-Type Directories",
description="Show options to specify custom directories for each content type (fiction, non-fiction, comics, etc.). If a directory is set, that content type will be saved there instead of the default download directory.",
label="Use Content-Type Directories",
description="Route content to separate directories based on type (fiction, non-fiction, comics, etc.).",
default=False,
env_supported=False, # UI-only toggle to show/hide directory fields
env_supported=False,
show_when={"field": "PROCESSING_MODE", "value": "ingest"},
),
HeadingField(
key="content_type_directories_heading",
title="Content-Type Directories",
description="Specify custom directories for each content type. Leave empty to use the default download directory.",
description="Leave empty to use the default ingest directory.",
show_when={"field": "USE_CONTENT_TYPE_DIRECTORIES", "value": True},
),
TextField(
@@ -568,12 +557,6 @@ def download_settings():
placeholder="/cwa-book-ingest/comics",
show_when={"field": "USE_CONTENT_TYPE_DIRECTORIES", "value": True},
),
TextField(
key="INGEST_DIR_AUDIOBOOK",
label="Audiobooks",
placeholder="/cwa-book-ingest/audiobooks",
show_when={"field": "USE_CONTENT_TYPE_DIRECTORIES", "value": True},
),
TextField(
key="INGEST_DIR_STANDARDS_DOCUMENT",
label="Standards Documents",
@@ -592,6 +575,112 @@ def download_settings():
placeholder="/cwa-book-ingest/other",
show_when={"field": "USE_CONTENT_TYPE_DIRECTORIES", "value": True},
),
TextField(
key="LIBRARY_PATH",
label="Library Path",
description="Base path for your book library (e.g., /books).",
placeholder="/books",
required=True,
show_when={"field": "PROCESSING_MODE", "value": "library"},
universal_only=True,
),
TextField(
key="LIBRARY_TEMPLATE",
label="Naming Template",
description="Available: {Author}, {Title}, {Subtitle}, {Year}, {Series}, {SeriesPosition}, {PartNumber}. Use {Series/} for conditional folders.",
default="{Author}/{Title}",
placeholder="{Author}/{Series/}{Title}{ - Subtitle} ({Year})",
show_when={"field": "PROCESSING_MODE", "value": "library"},
universal_only=True,
),
# === AUDIOBOOKS SECTION ===
HeadingField(
key="audiobooks_heading",
title="Audiobooks",
description="Configure how audiobooks are processed.",
universal_only=True,
),
SelectField(
key="PROCESSING_MODE_AUDIOBOOK",
label="Processing Mode",
description="Ingest mode moves files to an ingest folder. Library mode organizes files with custom naming.",
options=[
{"value": "ingest", "label": "Ingest"},
{"value": "library", "label": "Library"},
],
default="ingest",
universal_only=True,
),
TextField(
key="INGEST_DIR_AUDIOBOOK",
label="Download Directory",
description="Leave empty to use the Books download directory.",
placeholder="/audiobooks",
show_when={"field": "PROCESSING_MODE_AUDIOBOOK", "value": "ingest"},
universal_only=True,
),
TextField(
key="LIBRARY_PATH_AUDIOBOOK",
label="Library Path",
description="Base path for your audiobook library (e.g., /audiobooks).",
placeholder="/audiobooks",
required=True,
show_when={"field": "PROCESSING_MODE_AUDIOBOOK", "value": "library"},
universal_only=True,
),
TextField(
key="LIBRARY_TEMPLATE_AUDIOBOOK",
label="Naming Template",
description="Available: {Author}, {Title}, {Subtitle}, {Year}, {Series}, {SeriesPosition}, {PartNumber}. Use {PartNumber} for multi-file audiobooks.",
default="{Author}/{Title}",
placeholder="{Author}/{Series/}{Title}{ - Part }{PartNumber}",
show_when={"field": "PROCESSING_MODE_AUDIOBOOK", "value": "library"},
universal_only=True,
),
# === OPTIONS SECTION ===
HeadingField(
key="library_options_heading",
title="Options",
universal_only=True,
),
CheckboxField(
key="TORRENT_HARDLINK",
label="Hardlink Torrents in Library Mode",
description="Create hardlinks for torrent downloads instead of copying. Requires library and download paths on same filesystem.",
default=True,
universal_only=True,
),
CheckboxField(
key="AUTO_OPEN_DOWNLOADS_SIDEBAR",
label="Auto-Open Downloads Sidebar",
description="Automatically open the downloads sidebar when a new download is queued.",
default=False,
env_supported=False, # UI-only setting
),
CheckboxField(
key="DOWNLOAD_TO_BROWSER",
label="Download to Browser",
description="Automatically download completed files to your browser.",
default=False,
env_supported=False, # UI-only setting
),
NumberField(
key="MAX_CONCURRENT_DOWNLOADS",
label="Max Concurrent Downloads",
description="Maximum number of simultaneous downloads.",
default=3,
min_value=1,
max_value=10,
requires_restart=True,
),
NumberField(
key="STATUS_TIMEOUT",
label="Status Timeout (seconds)",
description="How long to keep completed/failed downloads in the queue display.",
default=3600,
min_value=60,
max_value=86400,
),
]
+17 -17
View File
@@ -14,17 +14,6 @@ def build_filename(
year: Optional[str] = None,
fmt: Optional[str] = None,
) -> str:
"""Build sanitized filename: 'Author - Title (Year).format'
Args:
title: Book title (required)
author: Book author
year: Publication year
fmt: File format/extension
Returns:
Sanitized filename safe for filesystem use
"""
parts = []
if author:
parts.append(author)
@@ -54,6 +43,11 @@ class QueueStatus(str, Enum):
CANCELLED = "cancelled"
class SearchMode(str, Enum):
DIRECT = "direct"
UNIVERSAL = "universal"
@dataclass
class QueueItem:
"""Queue item with priority and metadata."""
@@ -70,12 +64,6 @@ class QueueItem:
@dataclass
class DownloadTask:
"""Source-agnostic download task for the queue.
This replaces BookInfo in the queue, providing a unified interface
for both Direct Download and Universal modes. The handler uses task_id
to fetch whatever source-specific data it needs internally.
"""
task_id: str # Unique ID (e.g., AA MD5 hash, Prowlarr GUID)
source: str # Handler name ("direct_download", "prowlarr")
title: str # Display title for queue sidebar
@@ -88,6 +76,18 @@ class DownloadTask:
preview: Optional[str] = None
content_type: Optional[str] = None # "book (fiction)", "audiobook", "magazine", etc.
# Series info (for library naming templates)
series_name: Optional[str] = None
series_position: Optional[float] = None # Float for novellas (e.g., 1.5)
subtitle: Optional[str] = None # Book subtitle for naming templates
# Hardlinking support
original_download_path: Optional[str] = None # Path in download client (for hardlinking)
# Search mode - determines post-download processing behavior
# See SearchMode enum for behavioral differences
search_mode: Optional[SearchMode] = None
# Runtime state
priority: int = 0
added_time: float = field(default_factory=time.time)
+202
View File
@@ -0,0 +1,202 @@
"""Template-based naming for library organization."""
import os
import re
from pathlib import Path
from typing import Dict, Optional, Union
from cwa_book_downloader.core.logger import setup_logger
logger = setup_logger(__name__)
TOKEN_PATTERN = re.compile(
r'\{([- ._/\[(]*)' # prefix: space, dash, dot, underscore, slash, brackets
r'([A-Za-z]+)' # token name
r'([- ._/\])]*)\}' # suffix: space, dash, dot, underscore, slash, brackets
)
# Characters that are invalid in filenames on various filesystems
INVALID_CHARS = re.compile(r'[\\:*?"<>|]')
def _sanitize(name: str, max_length: int = 245) -> str:
"""Sanitize a string for filesystem use."""
if not name:
return ""
sanitized = INVALID_CHARS.sub('_', name)
sanitized = re.sub(r'^[\s.]+|[\s.]+$', '', sanitized) # Strip whitespace and dots
sanitized = re.sub(r'_+', '_', sanitized) # Collapse underscores
return sanitized[:max_length]
def sanitize_filename(name: str, max_length: int = 245) -> str:
"""Sanitize a string for use as a filename."""
return _sanitize(name, max_length)
def sanitize_path_component(name: str, max_length: int = 245) -> str:
"""Sanitize a string for use as a path component."""
return _sanitize(name, max_length)
def format_series_position(position: Optional[Union[int, float]]) -> str:
if position is None:
return ""
# Check if it's effectively an integer
if isinstance(position, float) and position.is_integer():
return str(int(position))
return str(position)
# Pads numbers to 9 digits for natural sorting (e.g., "Part 2" -> "Part 000000002")
PAD_NUMBERS_PATTERN = re.compile(r'\d+')
def natural_sort_key(path: Union[str, Path]) -> str:
"""Generate a sort key with padded numbers for natural sorting."""
filename = Path(path).name.lower()
return PAD_NUMBERS_PATTERN.sub(lambda m: m.group().zfill(9), filename)
def assign_part_numbers(
files: list[Path],
zero_pad_width: int = 2,
) -> list[tuple[Path, str]]:
"""Sort files naturally and assign sequential part numbers (1, 2, 3...)."""
if not files:
return []
sorted_files = sorted(files, key=natural_sort_key)
return [
(file_path, str(part_num).zfill(zero_pad_width))
for part_num, file_path in enumerate(sorted_files, start=1)
]
def parse_naming_template(
template: str,
metadata: Dict[str, Optional[Union[str, int, float]]],
) -> str:
if not template:
return ""
# Normalize metadata keys to lowercase for case-insensitive matching
normalized = {k.lower(): v for k, v in metadata.items()}
def replace_token(match: re.Match) -> str:
prefix = match.group(1)
token_name = match.group(2).lower()
suffix = match.group(3)
# Get the value for this token
value = normalized.get(token_name)
# Special handling for series position
if token_name == 'seriesposition':
value = format_series_position(value)
# Convert to string
if value is None:
value = ""
else:
value = str(value).strip()
# If value is empty, return empty string (no prefix/suffix)
if not value:
return ""
# Sanitize the value
# If suffix contains a slash, this is meant to be a folder component
if '/' in suffix:
value = sanitize_path_component(value)
else:
value = sanitize_filename(value)
return f"{prefix}{value}{suffix}"
# Replace all tokens
result = TOKEN_PATTERN.sub(replace_token, template)
# Clean up any double slashes that might result from empty tokens
result = re.sub(r'/+', '/', result)
# Remove leading/trailing slashes
result = result.strip('/')
# Clean up any orphaned separators (e.g., " - " at start/end, or " - - ")
result = re.sub(r'^[\s\-_.]+', '', result)
result = re.sub(r'[\s\-_.]+$', '', result)
result = re.sub(r'(\s*-\s*){2,}', ' - ', result)
# Clean up empty parentheses/brackets
result = re.sub(r'\(\s*\)', '', result)
result = re.sub(r'\[\s*\]', '', result)
# Final trim of any trailing separators left after cleanup
result = re.sub(r'[\s\-_.]+$', '', result)
return result
def build_library_path(
base_path: str,
template: str,
metadata: Dict[str, Optional[Union[str, int, float]]],
extension: Optional[str] = None,
) -> Path:
relative = parse_naming_template(template, metadata)
if not relative:
# Fallback to title if template produces empty result
title = metadata.get('Title') or metadata.get('title') or 'Unknown'
relative = sanitize_filename(str(title))
# Remove any path traversal attempts
relative = relative.replace('..', '')
base = Path(base_path).resolve()
full_path = (base / relative).resolve()
# Verify the path is within the base directory
try:
full_path.relative_to(base)
except ValueError:
raise ValueError(f"Path traversal detected: template would escape library directory")
if extension:
ext = extension.lstrip('.')
# Don't use with_suffix() - it replaces everything after the first dot
# e.g., "2.5 - Title" would become "2.epub" instead of "2.5 - Title.epub"
full_path = Path(f"{full_path}.{ext}")
return full_path
def same_filesystem(path1: Union[str, Path], path2: Union[str, Path]) -> bool:
"""Check if two paths are on the same filesystem."""
path1 = Path(path1)
path2 = Path(path2)
def get_device(p: Path) -> Optional[int]:
try:
while not p.exists():
p = p.parent
if p == p.parent:
break
return os.stat(p).st_dev
except (OSError, PermissionError) as e:
logger.debug(f"Cannot stat {p}: {e}")
return None
dev1 = get_device(path1)
dev2 = get_device(path2)
if dev1 is None or dev2 is None:
logger.warning(f"Cannot determine filesystem for hardlink check, falling back to copy")
return False
return dev1 == dev2
+9 -126
View File
@@ -27,6 +27,7 @@ class FieldBase:
show_when: Optional[Dict[str, Any]] = None # Conditional visibility: {"field": "key", "value": "expected"} or {"field": "key", "notEmpty": True}
disabled_when: Optional[Dict[str, Any]] = None # Conditional disable: {"field": "key", "value": "expected", "reason": "..."}
requires_restart: bool = False # Whether changing this setting requires a container restart
universal_only: bool = False # Only show in Universal search mode (hide in Direct mode)
def get_env_var_name(self) -> str:
"""Get the environment variable name for this field."""
@@ -83,19 +84,6 @@ class MultiSelectField(FieldBase):
@dataclass
class OrderableListField(FieldBase):
"""
Drag-and-drop reorderable list with enable/disable toggles.
A generic field for any ordered list of items where each item can be
enabled or disabled. Used for source priority, format preference, etc.
Options define the available items:
[{"id": "item1", "label": "Item 1", "description": "...",
"disabledReason": "...", "isLocked": False}, ...]
Value is stored as:
[{"id": "item1", "enabled": True}, {"id": "item2", "enabled": False}, ...]
"""
# Options can be a list or a callable that returns a list (for lazy evaluation)
# Each option: {id, label, description?, disabledReason?, isLocked?}
options: Any = field(default_factory=list)
@@ -105,12 +93,6 @@ class OrderableListField(FieldBase):
@dataclass
class ActionButton:
"""
Button that triggers a callback function.
Used for actions like "Test Connection" that execute code
and return success/error status.
"""
key: str # Action identifier
label: str # Button text
description: str = "" # Help text
@@ -139,6 +121,7 @@ class HeadingField:
link_url: str = "" # Optional URL for a link
link_text: str = "" # Text for the link (defaults to URL if not provided)
show_when: Optional[Dict[str, Any]] = None # Conditional visibility: {"field": "key", "value": "expected"} or {"field": "key", "notEmpty": True}
universal_only: bool = False # Only show in Universal search mode (hide in Direct mode)
def get_field_type(self) -> str:
return "HeadingField"
@@ -180,20 +163,6 @@ def register_group(
icon: Optional[str] = None,
order: int = 100
) -> None:
"""
Register a settings group.
Groups are collapsible containers for related settings tabs.
Args:
name: Internal name for the group (e.g., "direct_download")
display_name: Display name in UI (e.g., "Direct Download")
icon: Optional icon name for the UI
order: Sort order (lower numbers appear first)
Example:
register_group("direct_download", "Direct Download", icon="download", order=20)
"""
with _REGISTRY_LOCK:
group = SettingsGroup(
name=name,
@@ -212,25 +181,6 @@ def register_settings(
order: int = 100,
group: Optional[str] = None
):
"""
Decorator to register settings for a plugin/module.
The decorated function should return a list of SettingsField objects.
Args:
name: Internal name for the settings tab (e.g., "hardcover")
display_name: Display name in UI (e.g., "Hardcover")
icon: Optional icon name for the UI
order: Sort order (lower numbers appear first)
group: Optional group name this tab belongs to
Example:
@register_settings("hardcover", "Hardcover", icon="book", order=20, group="metadata_providers")
def hardcover_settings():
return [
PasswordField(key="HARDCOVER_API_KEY", label="API Key", required=True),
]
"""
def decorator(func: Callable[[], List[SettingsField]]):
with _REGISTRY_LOCK:
fields = func()
@@ -253,28 +203,6 @@ def register_on_save(
tab_name: str,
handler: Callable[[Dict[str, Any]], Dict[str, Any]]
) -> None:
"""
Register a custom on_save handler for a settings tab.
The handler is called before saving settings and can:
- Validate values (return {"error": True, "message": "..."})
- Transform values (e.g., hash passwords)
- Add computed values
Args:
tab_name: The settings tab name to register the handler for.
handler: Callable that takes values dict and returns:
{"error": bool, "message": str (if error), "values": dict}
Example:
def _on_save_security(values: Dict[str, Any]) -> Dict[str, Any]:
password = values.pop("password", "")
if password:
values["password_hash"] = hash_password(password)
return {"error": False, "values": values}
register_on_save("security", _on_save_security)
"""
with _REGISTRY_LOCK:
_ON_SAVE_HANDLERS[tab_name] = handler
logger.debug(f"Registered on_save handler for tab: {tab_name}")
@@ -324,15 +252,6 @@ def _ensure_config_dir(tab_name: str) -> None:
def load_config_file(tab_name: str) -> Dict[str, Any]:
"""
Load settings from a config file.
Args:
tab_name: The settings tab name.
Returns:
Dict of setting key -> value from config file.
"""
config_path = _get_config_file_path(tab_name)
if not config_path.exists():
@@ -347,16 +266,6 @@ def load_config_file(tab_name: str) -> Dict[str, Any]:
def save_config_file(tab_name: str, values: Dict[str, Any]) -> bool:
"""
Save settings to a config file.
Args:
tab_name: The settings tab name.
values: Dict of setting key -> value to save.
Returns:
True if save succeeded, False otherwise.
"""
try:
_ensure_config_dir(tab_name)
config_path = _get_config_file_path(tab_name)
@@ -376,14 +285,6 @@ def save_config_file(tab_name: str, values: Dict[str, Any]) -> bool:
def sync_env_to_config() -> None:
"""
Sync environment variable values to config files.
This ensures that when ENV vars are set, their values are persisted to config.
When ENV vars are later removed, the config file retains the last known values.
Called once during application startup.
"""
for tab in get_all_settings_tabs():
values_to_sync = {}
@@ -412,18 +313,6 @@ def sync_env_to_config() -> None:
def get_setting_value(field: SettingsField, tab_name: str) -> Any:
"""
Get the current value for a settings field.
Priority: env var > config file > default
Args:
field: The settings field.
tab_name: The settings tab name (for config file lookup).
Returns:
The resolved value.
"""
if isinstance(field, (ActionButton, HeadingField)):
return None # Actions and headings don't have values
@@ -505,6 +394,8 @@ def serialize_field(field: SettingsField, tab_name: str, include_value: bool = T
result["linkText"] = field.link_text or field.link_url
if field.show_when:
result["showWhen"] = field.show_when
if field.universal_only:
result["universalOnly"] = True
return result
result = {
@@ -528,6 +419,11 @@ def serialize_field(field: SettingsField, tab_name: str, include_value: bool = T
if disabled_when:
result["disabledWhen"] = disabled_when
# Add universal_only flag if set
universal_only = getattr(field, 'universal_only', False)
if universal_only:
result["universalOnly"] = True
# Add type-specific properties
if isinstance(field, TextField):
result["placeholder"] = field.placeholder
@@ -681,19 +577,6 @@ def _apply_dns_settings(config) -> None:
def update_settings(tab_name: str, values: Dict[str, Any]) -> Dict[str, Any]:
"""
Update settings for a tab.
Only updates values that are not set via environment variables.
Args:
tab_name: The settings tab name.
values: Dict of key -> value to update.
Returns:
Dict with "success" (bool), "message" (str), "updated" (list of keys),
and "requiresRestart" (bool) indicating if any changed setting requires restart.
"""
tab = get_settings_tab(tab_name)
if not tab:
return {"success": False, "message": f"Unknown settings tab: {tab_name}", "updated": [], "requiresRestart": False}
+469 -48
View File
@@ -37,10 +37,11 @@ from cwa_book_downloader.release_sources.direct_download import SearchUnavailabl
from cwa_book_downloader.core.config import config
from cwa_book_downloader.config.env import TMP_DIR
from cwa_book_downloader.core.utils import get_ingest_dir
from cwa_book_downloader.core.naming import build_library_path, same_filesystem, assign_part_numbers
from cwa_book_downloader.download.archive import is_archive, process_archive
from cwa_book_downloader.release_sources import get_handler, get_source_display_name
from cwa_book_downloader.core.logger import setup_logger
from cwa_book_downloader.core.models import BookInfo, DownloadTask, QueueStatus, SearchFilters
from cwa_book_downloader.core.models import BookInfo, DownloadTask, QueueStatus, SearchFilters, SearchMode
from cwa_book_downloader.core.queue import book_queue
logger = setup_logger(__name__)
@@ -250,22 +251,11 @@ def process_directory(
else:
filename = book_file.name
final_path = ingest_dir / filename
# Handle duplicates
if final_path.exists():
base = final_path.stem
ext = final_path.suffix
counter = 1
while final_path.exists():
final_path = ingest_dir / f"{base}_{counter}{ext}"
counter += 1
shutil.move(str(book_file), str(final_path))
dest_path = ingest_dir / filename
final_path = _atomic_move(book_file, dest_path)
final_paths.append(final_path)
logger.debug(f"Moved to ingest: {final_path.name}")
# Clean up the now-empty directory
shutil.rmtree(directory, ignore_errors=True)
return final_paths, None
@@ -280,6 +270,7 @@ def process_directory(
try:
from cwa_book_downloader.api.websocket import ws_manager
except ImportError:
logger.warning("WebSocket unavailable - real-time updates disabled")
ws_manager = None
# Progress update throttling - track last broadcast time per book
@@ -358,6 +349,7 @@ def queue_book(book_id: str, priority: int = 0, source: str = "direct_download")
size=book_info.size,
preview=book_info.preview,
content_type=book_info.content,
search_mode=SearchMode.DIRECT,
priority=priority,
)
@@ -403,6 +395,11 @@ def queue_release(release_data: dict, priority: int = 0) -> bool:
preview = release_data.get('preview') or extra.get('preview')
content_type = release_data.get('content_type') or extra.get('content_type')
# Get series info for library naming templates
series_name = release_data.get('series_name') or extra.get('series_name')
series_position = release_data.get('series_position') or extra.get('series_position')
subtitle = release_data.get('subtitle') or extra.get('subtitle')
# Create a source-agnostic download task from release data
task = DownloadTask(
task_id=release_data['source_id'],
@@ -414,6 +411,10 @@ def queue_release(release_data: dict, priority: int = 0) -> bool:
size=release_data.get('size'),
preview=preview,
content_type=content_type,
series_name=series_name,
series_position=series_position,
subtitle=subtitle,
search_mode=SearchMode.UNIVERSAL,
priority=priority,
)
@@ -607,6 +608,367 @@ def _download_task(task_id: str, cancel_flag: Event) -> Optional[str]:
return None
def _process_library_mode(
temp_file: Path,
task: DownloadTask,
status_callback,
) -> Optional[str]:
"""Process a download in Library Mode.
Library mode organizes files directly into your library with template-based
naming (e.g., "{Author}/{Series}/{Title}"). Files are transferred as-is
with no processing.
Use ingest mode if you need archive extraction or custom scripts.
Returns:
Path to library file if successful, None if library path not configured
"""
# Check if this is an audiobook and use audiobook-specific settings if configured
content_type = task.content_type.lower() if task.content_type else ""
is_audiobook = "audiobook" in content_type
if is_audiobook:
# Use audiobook-specific settings, falling back to main settings
library_path = config.get("LIBRARY_PATH_AUDIOBOOK") or config.get("LIBRARY_PATH")
template = config.get("LIBRARY_TEMPLATE_AUDIOBOOK") or config.get("LIBRARY_TEMPLATE", "{Author}/{Title}")
else:
library_path = config.get("LIBRARY_PATH")
template = config.get("LIBRARY_TEMPLATE", "{Author}/{Title}")
if not library_path:
logger.warning("Library mode enabled but no library path configured, falling back to ingest")
status_callback("resolving", "Library path not configured, using ingest")
return None
# Validate library path
library_path_obj = Path(library_path)
if not library_path_obj.is_absolute():
logger.warning(f"Library path must be absolute: {library_path}, falling back to ingest")
status_callback("resolving", f"Library path must be absolute: {library_path}")
return None
if not library_path_obj.exists():
try:
library_path_obj.mkdir(parents=True, exist_ok=True)
except (OSError, PermissionError) as e:
logger.warning(f"Cannot create library path: {e}, falling back to ingest")
status_callback("resolving", f"Cannot create library path: {e}")
return None
if not os.access(library_path_obj, os.W_OK):
logger.warning(f"Library path not writable: {library_path}, falling back to ingest")
status_callback("resolving", f"Library path not writable: {library_path}")
return None
# Determine if we should use hardlinking (torrents only)
use_hardlink = False
hardlink_source = None
if task.original_download_path:
# Torrent with download client path available
torrent_hardlink_enabled = config.get("TORRENT_HARDLINK", True)
if torrent_hardlink_enabled:
hardlink_source = Path(task.original_download_path)
if hardlink_source.exists():
# Check same filesystem (required for hardlinks)
if same_filesystem(hardlink_source, library_path):
use_hardlink = True
else:
logger.warning(
f"Cannot hardlink: {hardlink_source} and {library_path} are on different filesystems. "
"Falling back to move. To fix: ensure torrent client downloads to same filesystem as library."
)
status_callback("resolving", "Cannot hardlink (different filesystems), using move")
# Build metadata dict for template
metadata = {
"Author": task.author,
"Title": task.title,
"Subtitle": task.subtitle,
"Year": task.year,
"Series": task.series_name,
"SeriesPosition": task.series_position,
}
try:
if use_hardlink:
status_callback("resolving", "Creating library hardlinks")
else:
status_callback("resolving", "Organizing in library")
# Use torrent client path for hardlinks, staging path for moves
source = hardlink_source if use_hardlink else temp_file
if source.is_dir():
return _transfer_directory_to_library(
source, library_path, template, metadata, task, temp_file, status_callback, use_hardlink
)
else:
return _transfer_file_to_library(
source, library_path, template, metadata, task, temp_file, status_callback, use_hardlink
)
except PermissionError as e:
logger.error(f"Permission denied in library mode: {e}")
status_callback("error", f"Permission denied: {e}")
return None
except Exception as e:
logger.error_trace(f"Library mode failed: {e}")
status_callback("error", f"Library mode failed: {e}")
return None
def _is_torrent_source(source_path: Path, task: DownloadTask) -> bool:
"""Check if source is the torrent client path (needs copy to preserve seeding)."""
if not task.original_download_path:
return False
try:
return source_path.resolve() == Path(task.original_download_path).resolve()
except (OSError, ValueError):
return False
def _get_unique_path(dest_path: Path) -> Path:
"""Return a unique path by appending counter if file already exists.
Note: Has TOCTOU race. Use _atomic_hardlink/_atomic_move for concurrent safety.
"""
if not dest_path.exists():
return dest_path
base = dest_path.stem
ext = dest_path.suffix
counter = 1
while dest_path.exists():
dest_path = dest_path.parent / f"{base}_{counter}{ext}"
counter += 1
logger.info(f"File already exists, saving as: {dest_path.name}")
return dest_path
def _atomic_hardlink(source_path: Path, dest_path: Path, max_attempts: int = 100) -> Path:
"""Create a hardlink with atomic collision detection. Retries with counter suffix on collision."""
base = dest_path.stem
ext = dest_path.suffix
for attempt in range(max_attempts):
try_path = dest_path if attempt == 0 else dest_path.parent / f"{base}_{attempt}{ext}"
try:
os.link(str(source_path), str(try_path))
if attempt > 0:
logger.info(f"File collision resolved: {try_path.name}")
return try_path
except FileExistsError:
continue
raise RuntimeError(f"Could not create hardlink after {max_attempts} attempts: {dest_path}")
def _atomic_copy(source_path: Path, dest_path: Path, max_attempts: int = 100) -> Path:
"""Copy a file with atomic collision detection. Retries with counter suffix on collision."""
base = dest_path.stem
ext = dest_path.suffix
for attempt in range(max_attempts):
try_path = dest_path if attempt == 0 else dest_path.parent / f"{base}_{attempt}{ext}"
try:
# Atomically claim the destination by creating an exclusive file
fd = os.open(str(try_path), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
os.close(fd)
try:
# Copy to temp file first, then replace to avoid partial files
temp_path = try_path.parent / f".{try_path.name}.tmp"
shutil.copy2(str(source_path), str(temp_path))
temp_path.replace(try_path)
if attempt > 0:
logger.info(f"File collision resolved: {try_path.name}")
return try_path
except Exception:
try_path.unlink(missing_ok=True)
temp_path.unlink(missing_ok=True) if 'temp_path' in locals() else None
raise
except FileExistsError:
continue
raise RuntimeError(f"Could not copy file after {max_attempts} attempts: {dest_path}")
def _atomic_move(source_path: Path, dest_path: Path, max_attempts: int = 100) -> Path:
"""Move a file with atomic collision detection. Retries with counter suffix on collision."""
import errno
base = dest_path.stem
ext = dest_path.suffix
for attempt in range(max_attempts):
try_path = dest_path if attempt == 0 else dest_path.parent / f"{base}_{attempt}{ext}"
try:
os.link(str(source_path), str(try_path))
source_path.unlink()
if attempt > 0:
logger.info(f"File collision resolved: {try_path.name}")
return try_path
except FileExistsError:
continue
except OSError as e:
# Cross-filesystem - fall back to exclusive create
if e.errno not in (errno.EXDEV, errno.EMLINK):
raise
try:
fd = os.open(str(try_path), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
os.close(fd)
try:
shutil.move(str(source_path), str(try_path))
if attempt > 0:
logger.info(f"File collision resolved: {try_path.name}")
return try_path
except Exception:
try_path.unlink(missing_ok=True)
raise
except FileExistsError:
continue
raise RuntimeError(f"Could not move file after {max_attempts} attempts: {dest_path}")
def _cleanup_staged_files(temp_file: Path, source_dir: Optional[Path] = None) -> None:
"""Remove staged files. Optionally removes source_dir if empty."""
try:
if temp_file.is_dir():
shutil.rmtree(temp_file)
elif temp_file.exists():
temp_file.unlink()
except (OSError, PermissionError) as e:
logger.debug(f"Cleanup failed for {temp_file}: {e}")
if source_dir and source_dir.is_dir():
try:
source_dir.rmdir()
except OSError:
pass # Directory not empty or permission issue
def _transfer_single_file(
source_path: Path,
dest_path: Path,
use_hardlink: bool,
is_torrent: bool,
) -> Tuple[Path, str]:
"""Transfer a file via hardlink, copy, or move. Returns (final_path, operation_name)."""
if use_hardlink:
return _atomic_hardlink(source_path, dest_path), "hardlink"
if is_torrent:
return _atomic_copy(source_path, dest_path), "copy"
return _atomic_move(source_path, dest_path), "move"
def _transfer_file_to_library(
source_path: Path,
library_base: str,
template: str,
metadata: dict,
task: DownloadTask,
temp_file: Optional[Path],
status_callback,
use_hardlink: bool,
) -> Optional[str]:
"""Transfer a single file to the library with template-based naming."""
extension = source_path.suffix.lstrip('.') or task.format
dest_path = build_library_path(library_base, template, metadata, extension)
dest_path.parent.mkdir(parents=True, exist_ok=True)
is_torrent = _is_torrent_source(source_path, task)
final_path, op = _transfer_single_file(source_path, dest_path, use_hardlink, is_torrent)
logger.info(f"Library {op}: {final_path}")
if use_hardlink:
_cleanup_staged_files(temp_file)
status_callback("complete", "Complete (library mode)")
return str(final_path)
def _transfer_directory_to_library(
source_dir: Path,
library_base: str,
template: str,
metadata: dict,
task: DownloadTask,
temp_file: Optional[Path],
status_callback,
use_hardlink: bool,
) -> Optional[str]:
"""Transfer all files from a directory to the library with template-based naming."""
content_type = task.content_type.lower() if task.content_type else None
supported_formats = _get_supported_formats(content_type)
source_files = [
f for f in source_dir.rglob("*")
if f.is_file() and f.suffix.lower().lstrip('.') in supported_formats
]
if not source_files:
logger.warning(f"No supported files in {source_dir.name}")
status_callback("error", "No supported file formats found")
if temp_file:
_cleanup_staged_files(temp_file)
return None
base_library_path = build_library_path(library_base, template, metadata, extension=None)
base_library_path.parent.mkdir(parents=True, exist_ok=True)
# Check if this is a torrent source that needs copy instead of move
is_torrent = _is_torrent_source(source_dir, task)
transferred_paths = []
if len(source_files) == 1:
# Single file - no part numbering needed
source_file = source_files[0]
ext = source_file.suffix.lstrip('.')
dest_path = base_library_path.with_suffix(f'.{ext}')
final_path, op = _transfer_single_file(source_file, dest_path, use_hardlink, is_torrent)
logger.debug(f"Library {op}: {source_file.name} -> {final_path}")
transferred_paths.append(final_path)
else:
# Multi-file: natural sort then sequential numbering
zero_pad_width = max(len(str(len(source_files))), 2)
files_with_parts = assign_part_numbers(source_files, zero_pad_width)
for source_file, part_number in files_with_parts:
ext = source_file.suffix.lstrip('.')
file_metadata = {**metadata, "PartNumber": part_number}
file_path = build_library_path(library_base, template, file_metadata, extension=ext)
file_path.parent.mkdir(parents=True, exist_ok=True)
final_path, op = _transfer_single_file(source_file, file_path, use_hardlink, is_torrent)
logger.debug(f"Library {op}: {source_file.name} -> {final_path}")
transferred_paths.append(final_path)
# Get operation name for summary log
if use_hardlink:
operation = "hardlinks"
elif is_torrent:
operation = "copies"
else:
operation = "files"
logger.info(f"Created {len(transferred_paths)} library {operation} in {base_library_path.parent}")
# Cleanup staging (not torrent source - that stays for seeding)
if use_hardlink:
_cleanup_staged_files(temp_file)
elif not is_torrent:
_cleanup_staged_files(temp_file, source_dir)
count_msg = f" ({len(transferred_paths)} files," if len(transferred_paths) > 1 else " ("
status_callback("complete", f"Complete{count_msg} library mode)")
return str(transferred_paths[0])
def _post_process_download(
temp_file: Path,
task: DownloadTask,
@@ -626,14 +988,71 @@ def _post_process_download(
Returns:
Final path in ingest directory, or None on failure
"""
# Route to content-type-specific ingest directory if configured
content_type = task.content_type.lower() if task.content_type else None
content_type = task.content_type.lower() if task.content_type else ""
is_audiobook = "audiobook" in content_type
# Validate search_mode
if task.search_mode is None:
logger.warning(f"Task {task.task_id} has no search_mode set, defaulting to Direct mode behavior")
elif task.search_mode not in (SearchMode.DIRECT, SearchMode.UNIVERSAL):
logger.warning(f"Task {task.task_id} has invalid search_mode '{task.search_mode}', defaulting to Direct mode behavior")
# Library mode and audiobook-specific directories only apply to Universal mode
is_universal = task.search_mode == SearchMode.UNIVERSAL
if is_universal:
# Determine processing mode based on content type
if is_audiobook:
processing_mode = config.get("PROCESSING_MODE_AUDIOBOOK", "ingest")
else:
processing_mode = config.get("PROCESSING_MODE", "ingest")
if processing_mode == "library":
result = _process_library_mode(temp_file, task, status_callback)
if result is not None:
return result
# If library mode fails, fall through to normal processing
# Determine ingest directory
default_ingest_dir = get_ingest_dir()
ingest_dir = get_ingest_dir(content_type)
if content_type and ingest_dir != default_ingest_dir:
logger.debug(f"Routing content type '{content_type}' to {ingest_dir}")
if is_universal and is_audiobook:
# Universal audiobooks use dedicated setting, falling back to main ingest dir
audiobook_ingest = config.get("INGEST_DIR_AUDIOBOOK", "")
ingest_dir = Path(audiobook_ingest) if audiobook_ingest else default_ingest_dir
else:
# Direct mode and non-audiobook content use content-type routing
ingest_dir = get_ingest_dir(content_type)
if ingest_dir != default_ingest_dir:
logger.debug(f"Routing '{content_type or 'default'}' to {ingest_dir}")
os.makedirs(ingest_dir, exist_ok=True)
# For torrents going to ingest mode, stage first to preserve seeding
# (Torrent handler returns original path, not staged copy)
if _is_torrent_source(temp_file, task):
status_callback("resolving", "Staging torrent files")
staging_dir = get_staging_dir()
if temp_file.is_dir():
staged_path = staging_dir / temp_file.name
counter = 1
while staged_path.exists():
staged_path = staging_dir / f"{temp_file.name}_{counter}"
counter += 1
shutil.copytree(str(temp_file), str(staged_path))
logger.debug(f"Staged torrent directory: {staged_path.name}")
else:
staged_path = staging_dir / temp_file.name
counter = 1
while staged_path.exists():
staged_path = staging_dir / f"{temp_file.stem}_{counter}{temp_file.suffix}"
counter += 1
shutil.copy2(str(temp_file), str(staged_path))
logger.debug(f"Staged torrent file: {staged_path.name}")
temp_file = staged_path
# Handle archive extraction (RAR/ZIP)
if is_archive(temp_file):
logger.info(f"Archive detected, extracting: {temp_file.name}")
@@ -683,7 +1102,33 @@ def _post_process_download(
# Non-archive: run custom script if configured, then move to ingest
if config.CUSTOM_SCRIPT:
logger.info(f"Running custom script: {config.CUSTOM_SCRIPT}")
subprocess.run([config.CUSTOM_SCRIPT, str(temp_file)])
try:
result = subprocess.run(
[config.CUSTOM_SCRIPT, str(temp_file)],
check=True,
timeout=300, # 5 minute timeout
capture_output=True,
text=True,
)
if result.stdout:
logger.debug(f"Custom script stdout: {result.stdout.strip()}")
except FileNotFoundError:
logger.error(f"Custom script not found: {config.CUSTOM_SCRIPT}")
status_callback("error", f"Custom script not found: {config.CUSTOM_SCRIPT}")
return None
except PermissionError:
logger.error(f"Custom script not executable: {config.CUSTOM_SCRIPT}")
status_callback("error", f"Custom script not executable: {config.CUSTOM_SCRIPT}")
return None
except subprocess.TimeoutExpired:
logger.error(f"Custom script timed out after 300s: {config.CUSTOM_SCRIPT}")
status_callback("error", "Custom script timed out")
return None
except subprocess.CalledProcessError as e:
stderr = e.stderr.strip() if e.stderr else "No error output"
logger.error(f"Custom script failed (exit code {e.returncode}): {stderr}")
status_callback("error", f"Custom script failed: {stderr[:100]}")
return None
# Check cancellation before final move
if cancel_flag.is_set():
@@ -699,39 +1144,15 @@ def _post_process_download(
else:
filename = temp_file.name
final_path = ingest_dir / filename
# Handle duplicate filenames
if final_path.exists():
base = final_path.stem
ext = final_path.suffix
counter = 1
while final_path.exists():
final_path = ingest_dir / f"{base}_{counter}{ext}"
counter += 1
logger.info(f"File already exists, saving as: {final_path.name}")
# Use intermediate .crdownload file for atomic move
intermediate_path = ingest_dir / f"{temp_file.stem}.crdownload"
dest_path = ingest_dir / filename
try:
shutil.move(str(temp_file), str(intermediate_path))
final_path = _atomic_move(temp_file, dest_path)
except Exception as e:
logger.debug(f"Error moving file: {e}, trying copy instead")
try:
shutil.copyfile(str(temp_file), str(intermediate_path))
temp_file.unlink(missing_ok=True)
except Exception as e2:
logger.error(f"Failed to move/copy file to ingest: {e2}")
return None
# Final cancellation check
if cancel_flag.is_set():
logger.info(f"Download cancelled before final rename: {task.task_id}")
intermediate_path.unlink(missing_ok=True)
logger.error(f"Failed to move file to ingest: {e}")
status_callback("error", f"Failed to move file: {e}")
return None
os.rename(str(intermediate_path), str(final_path))
logger.info(f"Download completed: {final_path.name}")
status_callback("complete", "Complete")
@@ -163,6 +163,7 @@ class BookMetadata:
language: Optional[str] = None
genres: List[str] = field(default_factory=list)
source_url: Optional[str] = None # Link to book on provider's site
subtitle: Optional[str] = None # Book subtitle, if any
# Provider-specific display fields for cards/lists
display_fields: List[DisplayField] = field(default_factory=list)
@@ -375,6 +375,7 @@ class HardcoverProvider(MetadataProvider):
books(where: {id: {_eq: $id}}, limit: 1) {
id
title
subtitle
slug
release_date
headline
@@ -465,6 +466,7 @@ class HardcoverProvider(MetadataProvider):
book {
id
title
subtitle
slug
release_date
headline
@@ -606,10 +608,14 @@ class HardcoverProvider(MetadataProvider):
description = item.get("description")
full_description = _combine_headline_description(headline, description)
# Extract subtitle if available in search results
subtitle = item.get("subtitle")
return BookMetadata(
provider="hardcover",
provider_id=str(book_id),
title=title,
subtitle=subtitle,
provider_display_name="Hardcover",
authors=authors,
cover_url=cover_url,
@@ -735,6 +741,7 @@ class HardcoverProvider(MetadataProvider):
provider="hardcover",
provider_id=str(book["id"]),
title=book["title"],
subtitle=book.get("subtitle"),
provider_display_name="Hardcover",
authors=authors,
isbn_10=isbn_10,
@@ -34,6 +34,7 @@ class Release:
indexer: Optional[str] = None # Source name for display
seeders: Optional[int] = None # For torrents
peers: Optional[str] = None # For torrents: "seeders/leechers" display string
content_type: Optional[str] = None # "ebook" or "audiobook" - preserved from search
extra: Dict = field(default_factory=dict) # Source-specific metadata
@@ -921,12 +921,12 @@ def _book_info_to_release(book_info: BookInfo) -> Release:
info_url=f"{network.get_aa_base_url()}/md5/{book_info.id}",
protocol=ReleaseProtocol.HTTP,
indexer="Anna's Archive",
content_type=book_info.content, # Preserve content type from source
extra={
"author": book_info.author,
"publisher": book_info.publisher,
"year": book_info.year,
"language": book_info.language,
"content": book_info.content,
"preview": book_info.preview,
"description": book_info.description,
"download_urls": book_info.download_urls,
@@ -14,13 +14,18 @@ from cwa_book_downloader.core.logger import setup_logger
logger = setup_logger(__name__)
# All recognized ebook formats for parsing IRC result lines.
# All recognized formats for parsing IRC result lines.
# This comprehensive list is used to identify file extensions in results.
# User's configured formats are used separately for filtering.
ALL_EBOOK_FORMATS = {
# Note: IRC source currently only supports ebooks, but audiobook formats
# are included for future-proofing and format detection consistency.
ALL_RECOGNIZED_FORMATS = {
# Ebook formats
'epub', 'mobi', 'azw3', 'azw', 'pdf', 'doc', 'docx',
'html', 'htm', 'rtf', 'txt', 'lit', 'fb2', 'djvu',
'cbr', 'cbz', 'cdr', 'jpg', 'rar', 'zip'
'cbr', 'cbz', 'cdr', 'jpg', 'rar', 'zip',
# Audiobook formats
'm4b', 'mp3', 'm4a', 'flac', 'ogg', 'wma', 'aac', 'wav', 'opus'
}
@@ -104,7 +109,7 @@ def parse_result_line(line: str) -> Optional[SearchResult]:
# Try to extract format from the line
fmt = None
for known_fmt in ALL_EBOOK_FORMATS:
for known_fmt in ALL_RECOGNIZED_FORMATS:
if f'.{known_fmt}' in rest.lower():
fmt = known_fmt
break
@@ -126,7 +131,7 @@ def parse_result_line(line: str) -> Optional[SearchResult]:
# Clean up title (remove extension)
title = title_part
for known_fmt in ALL_EBOOK_FORMATS:
for known_fmt in ALL_RECOGNIZED_FORMATS:
title = re.sub(rf'\.{known_fmt}\b', '', title, flags=re.IGNORECASE)
return SearchResult(
@@ -1,11 +1,7 @@
"""
qBittorrent download client for Prowlarr integration.
Uses the qbittorrent-api library to communicate with qBittorrent's Web API.
"""
"""qBittorrent download client for Prowlarr integration."""
import time
from typing import Optional, Tuple
from typing import List, Optional, Tuple
from cwa_book_downloader.core.config import config
from cwa_book_downloader.core.logger import setup_logger
@@ -37,6 +33,7 @@ class QBittorrentClient(DownloadClient):
if not url:
raise ValueError("QBITTORRENT_URL is required")
self._base_url = url.rstrip("/")
self._client = Client(
host=url,
username=config.get("QBITTORRENT_USERNAME", ""),
@@ -44,6 +41,39 @@ class QBittorrentClient(DownloadClient):
)
self._category = config.get("QBITTORRENT_CATEGORY", "cwabd")
def _get_torrents_info(self, torrent_hash: Optional[str] = None) -> List:
"""Get torrent info using GET (per API spec for read operations)."""
import requests
try:
params = {"hashes": torrent_hash} if torrent_hash else {}
response = self._client._session.get(
f"{self._base_url}/api/v2/torrents/info",
params=params,
timeout=10,
)
response.raise_for_status()
torrents = response.json()
class TorrentInfo:
def __init__(self, data):
for key, value in data.items():
setattr(self, key, value)
return [TorrentInfo(t) for t in torrents]
except requests.exceptions.HTTPError as e:
if e.response is not None and e.response.status_code == 403:
logger.warning("qBittorrent auth failed - check credentials")
else:
logger.warning(f"qBittorrent API error: {e}")
return []
except requests.exceptions.ConnectionError:
logger.warning(f"Cannot connect to qBittorrent at {self._base_url}")
return []
except Exception as e:
logger.debug(f"Failed to get torrents info: {e}")
return []
@staticmethod
def is_configured() -> bool:
"""Check if qBittorrent is configured and selected as the torrent client."""
@@ -111,29 +141,21 @@ class QBittorrentClient(DownloadClient):
logger.debug(f"qBittorrent add result: {result}")
if result == "Ok.":
if expected_hash:
# We know the hash - verify it was added
for attempt in range(10):
torrents = self._client.torrents_info(
torrent_hashes=expected_hash
)
if torrents:
logger.info(f"Added torrent to qBittorrent: {expected_hash}")
return expected_hash
time.sleep(0.5)
if not expected_hash:
raise Exception("Could not determine torrent hash from URL")
# qBittorrent said Ok, trust it even if we can't find it yet
logger.warning(
f"qBittorrent returned Ok but torrent not yet visible. "
f"Returning expected hash: {expected_hash}"
)
return expected_hash
# Wait for torrent to appear in client
for _ in range(10):
torrents = self._get_torrents_info(expected_hash)
for t in torrents:
if t.hash.lower() == expected_hash.lower():
logger.info(f"Added torrent: {t.hash}")
return t.hash.lower()
time.sleep(0.5)
# No hash available - this shouldn't happen often
raise Exception(
"Could not determine torrent hash. "
"Try using a magnet link instead."
)
# Client said Ok, trust it
logger.warning(f"Torrent not yet visible, returning expected hash")
return expected_hash
raise Exception(f"Failed to add torrent: {result}")
except Exception as e:
@@ -151,12 +173,11 @@ class QBittorrentClient(DownloadClient):
Current download status.
"""
try:
torrents = self._client.torrents_info(torrent_hashes=download_id)
if not torrents:
torrents = self._get_torrents_info(download_id)
torrent = next((t for t in torrents if t.hash.lower() == download_id.lower()), None)
if not torrent:
return DownloadStatus.error("Torrent not found")
torrent = torrents[0]
# Map qBittorrent states to our states and user-friendly messages
state_info = {
"downloading": ("downloading", None), # None = use default progress message
@@ -188,15 +209,26 @@ class QBittorrentClient(DownloadClient):
if complete:
message = "Complete"
# Only include ETA if it's reasonable (less than 1 week)
eta = torrent.eta if 0 < torrent.eta < 604800 else None
# Get file path for completed downloads
file_path = None
if complete:
if getattr(torrent, 'content_path', ''):
file_path = torrent.content_path
else:
# Fallback for Amarr which doesn't populate content_path
save_path = getattr(torrent, 'save_path', '')
name = getattr(torrent, 'name', '')
if save_path and name:
file_path = f"{save_path}/{name}"
return DownloadStatus(
progress=torrent.progress * 100,
state="complete" if complete else state,
message=message,
complete=complete,
file_path=torrent.content_path if complete else None,
file_path=file_path,
download_speed=torrent.dlspeed,
eta=eta,
)
@@ -231,20 +263,18 @@ class QBittorrentClient(DownloadClient):
return False
def get_download_path(self, download_id: str) -> Optional[str]:
"""
Get the path where torrent files are located.
Args:
download_id: Torrent info_hash
Returns:
Content path (file or directory), or None.
"""
"""Get the path where torrent files are located."""
try:
torrents = self._client.torrents_info(torrent_hashes=download_id)
if torrents:
return torrents[0].content_path
return None
torrents = self._get_torrents_info(download_id)
torrent = next((t for t in torrents if t.hash.lower() == download_id.lower()), None)
if not torrent:
return None
# Prefer content_path, fall back to save_path/name (for Amarr compatibility)
if getattr(torrent, 'content_path', ''):
return torrent.content_path
save_path = getattr(torrent, 'save_path', '')
name = getattr(torrent, 'name', '')
return f"{save_path}/{name}" if save_path and name else None
except Exception as e:
error_type = type(e).__name__
logger.debug(f"qBittorrent get_download_path failed ({error_type}): {e}")
@@ -257,10 +287,10 @@ class QBittorrentClient(DownloadClient):
if not torrent_info.info_hash:
return None
torrents = self._client.torrents_info(torrent_hashes=torrent_info.info_hash)
if torrents:
status = self.get_status(torrent_info.info_hash)
return (torrent_info.info_hash, status)
torrents = self._get_torrents_info(torrent_info.info_hash)
torrent = next((t for t in torrents if t.hash.lower() == torrent_info.info_hash.lower()), None)
if torrent:
return (torrent.hash.lower(), self.get_status(torrent.hash.lower()))
return None
except Exception as e:
@@ -1,14 +1,4 @@
"""
Shared utilities for torrent clients.
Provides:
- Bencode encoding/decoding for .torrent files
- Info hash extraction from torrent files and magnet links
- URL parsing utilities for torrent clients
Bencode is the encoding used by BitTorrent for .torrent files.
See BEP-3: http://bittorrent.org/beps/bep_0003.html
"""
"""Shared utilities for torrent clients."""
import base64
import hashlib
@@ -29,7 +19,7 @@ class TorrentInfo:
"""Parsed information from a torrent URL."""
info_hash: Optional[str]
"""40-character lowercase hex info_hash, or None if extraction failed."""
"""Lowercase hex info_hash (32 or 40 chars), or None if extraction failed."""
torrent_data: Optional[bytes]
"""Raw .torrent file content, only populated for .torrent URLs."""
@@ -42,24 +32,7 @@ class TorrentInfo:
def extract_torrent_info(url: str, fetch_torrent: bool = True) -> TorrentInfo:
"""
Extract torrent info_hash and optionally fetch torrent file content.
Handles both magnet links and .torrent file URLs. For magnet links,
extracts the hash directly. For .torrent URLs, optionally fetches
the file and parses it to extract the hash.
Also handles the case where a download URL redirects to a magnet link
or returns a magnet link in the response body.
Args:
url: Magnet link or .torrent URL
fetch_torrent: If True, fetch .torrent URLs to extract hash.
Set to False if you only need the hash from magnets.
Returns:
TorrentInfo with extracted hash and optional torrent data.
"""
"""Extract info_hash from magnet link or .torrent URL."""
is_magnet = url.startswith("magnet:")
# Try to extract hash from magnet URL
@@ -120,21 +93,7 @@ def extract_torrent_info(url: str, fetch_torrent: bool = True) -> TorrentInfo:
def parse_transmission_url(url: str) -> Tuple[str, int, str]:
"""
Parse a Transmission URL into host, port, and RPC path.
Handles various URL formats and ensures the path ends with /rpc.
Args:
url: Transmission URL (e.g., "http://transmission:9091" or
"http://localhost:9091/transmission/rpc")
Returns:
Tuple of (host, port, path) for transmission-rpc Client.
- host: Hostname (defaults to "localhost" if not specified)
- port: Port number (defaults to 9091 if not specified)
- path: RPC path (ensures it ends with "/rpc")
"""
"""Parse Transmission URL into (host, port, path)."""
parsed = urlparse(url)
host = parsed.hostname or "localhost"
port = parsed.port or 9091
@@ -148,24 +107,7 @@ def parse_transmission_url(url: str) -> Tuple[str, int, str]:
def bencode_decode(data: bytes) -> tuple:
"""
Decode bencoded data.
Bencode format:
- 'i<integer>e' for integers
- '<length>:<bytes>' for byte strings
- 'l<elements>e' for lists
- 'd<key><value>e' for dicts (keys must be byte strings, sorted)
Args:
data: Bencoded bytes
Returns:
Tuple of (decoded_value, remaining_bytes)
Raises:
ValueError: If data is not valid bencode
"""
"""Decode bencoded data. Returns (value, remaining_bytes)."""
if data[0:1] == b'd':
# Dictionary
result = {}
@@ -202,18 +144,7 @@ def bencode_decode(data: bytes) -> tuple:
def bencode_encode(data) -> bytes:
"""
Encode data to bencode format.
Args:
data: Python object (dict, list, int, bytes, or str)
Returns:
Bencoded bytes
Raises:
ValueError: If data type cannot be bencoded
"""
"""Encode data to bencode format."""
if isinstance(data, dict):
# Keys must be sorted (bencode spec requirement)
result = b'd'
@@ -243,28 +174,13 @@ def bencode_encode(data) -> bytes:
def extract_info_hash_from_torrent(torrent_data: bytes) -> Optional[str]:
"""
Extract info_hash from raw .torrent file data.
The info_hash is the SHA1 hash of the bencoded 'info' dictionary,
which uniquely identifies a torrent in the BitTorrent network.
Args:
torrent_data: Raw bytes of a .torrent file
Returns:
40-character lowercase hex string of the info_hash, or None if extraction fails
"""
"""Extract info_hash from .torrent file data."""
try:
decoded, _ = bencode_decode(torrent_data)
if b'info' not in decoded:
return None
# Re-encode info dict to get canonical bytes for hashing
info_dict = decoded[b'info']
info_bencoded = bencode_encode(info_dict)
# SHA1 hash is required by BitTorrent spec (BEP-3)
info_bencoded = bencode_encode(decoded[b'info'])
return hashlib.sha1(info_bencoded).hexdigest().lower()
except Exception as e:
logger.debug(f"Failed to parse torrent file: {e}")
@@ -272,41 +188,31 @@ def extract_info_hash_from_torrent(torrent_data: bytes) -> Optional[str]:
def extract_hash_from_magnet(magnet_url: str) -> Optional[str]:
"""
Extract info_hash from a magnet URL.
Magnet URIs contain the info_hash in the 'xt' (exact topic) parameter
as either a 40-character hex string or 32-character base32 string.
Args:
magnet_url: Magnet URI string
Returns:
40-character lowercase hex string of the info_hash, or None if extraction fails
"""
"""Extract info_hash from a magnet URL."""
if not magnet_url.startswith("magnet:"):
return None
parsed = urlparse(magnet_url)
params = parse_qs(parsed.query)
# Get the xt (exact topic) parameter
xt_list = params.get("xt", [])
for xt in xt_list:
# Format: urn:btih:<hash>
# Hash can be 40 hex chars or 32 base32 chars
match = re.match(r"urn:btih:([a-fA-F0-9]{40}|[a-zA-Z2-7]{32})", xt)
for xt in params.get("xt", []):
# Format: urn:btih:<hash> (32 or 40 chars)
match = re.match(r"urn:btih:([a-fA-F0-9]{40}|[a-zA-Z0-9]{32})", xt)
if match:
hash_value = match.group(1)
# Convert base32 to hex if needed (32 chars = base32, 40 chars = hex)
if len(hash_value) == 32:
# 40-char hex or 32-char hex (ED2K) - return as-is
if len(hash_value) == 40 or re.match(r'^[a-fA-F0-9]{32}$', hash_value):
return hash_value.lower()
# 32-char base32 - decode to hex
if re.match(r'^[A-Z2-7]{32}$', hash_value.upper()):
try:
decoded = base64.b32decode(hash_value.upper())
return decoded.hex().lower()
except Exception as e:
logger.debug(
f"Base32 decode failed for hash '{hash_value}': {e}, "
f"treating as hex value"
)
return base64.b32decode(hash_value.upper()).hex().lower()
except Exception:
pass
# Fallback: return as-is
return hash_value.lower()
return None
@@ -56,6 +56,7 @@ class ProwlarrHandler(DownloadHandler):
# Look up the cached release
prowlarr_result = get_release(task.task_id)
if not prowlarr_result:
logger.warning(f"Release cache miss: {task.task_id}")
status_callback("error", "Release not found in cache (may have expired)")
return None
@@ -241,37 +242,38 @@ class ProwlarrHandler(DownloadHandler):
task: DownloadTask,
status_callback: Callable[[str, Optional[str]], None],
) -> Optional[str]:
"""Stage completed download for orchestrator post-processing.
"""Handle completed download - stage or return path for orchestrator.
For directories (multi-file torrents), copies the entire directory.
The orchestrator will find and filter book files.
For torrents: returns original path directly (no copy). Orchestrator
will hardlink or copy as needed, avoiding unnecessary staging of
large files like audiobooks.
For usenet: stages to temp directory based on config.
"""
try:
status_callback("resolving", "Staging file")
# Torrents: copy to preserve seeding. Usenet: configurable.
# For torrents, skip staging - return original path directly
# Orchestrator will hardlink (library mode) or copy (ingest mode) as needed
if protocol == "torrent":
use_copy = True
else:
use_copy = config.get("PROWLARR_USENET_ACTION", "move") == "copy"
task.original_download_path = str(source_path)
logger.debug(f"Torrent complete, returning original path: {source_path}")
return str(source_path)
# Usenet: stage based on config
status_callback("resolving", "Staging file")
use_copy = config.get("PROWLARR_USENET_ACTION", "move") == "copy"
from cwa_book_downloader.download.orchestrator import get_staging_dir
staging_dir = get_staging_dir()
if source_path.is_dir():
# Multi-file download: stage entire directory
# Orchestrator will extract book files
staged_path = get_unique_path(staging_dir, source_path.name)
if use_copy:
shutil.copytree(str(source_path), str(staged_path))
else:
shutil.move(str(source_path), str(staged_path))
logger.debug(f"Staged directory: {staged_path.name}")
else:
# Single file download
staged_path = get_unique_path(staging_dir, source_path.stem, source_path.suffix)
if use_copy:
shutil.copy2(str(source_path), str(staged_path))
else:
@@ -52,37 +52,44 @@ def _parse_size(size_bytes: Optional[int]) -> Optional[str]:
# Common ebook formats in priority order
EBOOK_FORMATS = ["epub", "mobi", "azw3", "azw", "pdf", "cbz", "cbr", "fb2", "djvu", "lit", "pdb", "txt"]
# Common audiobook formats
AUDIOBOOK_FORMATS = ["m4b", "mp3", "m4a", "flac", "ogg", "wma", "aac", "wav", "opus"]
# Combined list for format detection (audiobook formats first for priority)
ALL_BOOK_FORMATS = AUDIOBOOK_FORMATS + EBOOK_FORMATS
def _extract_format(title: str) -> Optional[str]:
"""
Extract format from release title with smart parsing.
Supports both ebook formats (epub, mobi, etc.) and audiobook formats (m4b, mp3, etc.).
Priority:
1. File extension at end of title or in quotes (e.g., ".azw3")
2. File extension anywhere in title
3. Format keyword in brackets/parentheses (e.g., "[EPUB]", "(PDF)")
4. Fallback to first format keyword found
1. File extension at end of title or in quotes (e.g., ".azw3", ".m4b")
2. Format keyword in brackets/parentheses (e.g., "[EPUB]", "(MP3)")
3. Format as standalone word (not part of another word)
"""
title_lower = title.lower()
# 1. Look for file extensions (most reliable) - pattern: .format at word boundary or end
# This catches ".azw3", ".epub", etc.
for fmt in EBOOK_FORMATS:
# This catches ".azw3", ".epub", ".m4b", ".mp3", etc.
for fmt in ALL_BOOK_FORMATS:
# Match .format at end of string or followed by non-alphanumeric
pattern = rf'\.{fmt}(?:["\'\s\]\)]|$)'
if re.search(pattern, title_lower):
return fmt
# 2. Look for format in brackets/parentheses (common in release names)
# e.g., "[EPUB]", "(PDF)", "{mobi}"
for fmt in EBOOK_FORMATS:
# e.g., "[EPUB]", "(PDF)", "{mobi}", "[M4B]", "(MP3)"
for fmt in ALL_BOOK_FORMATS:
pattern = rf'[\[\(\{{]{fmt}[\]\)\}}]'
if re.search(pattern, title_lower):
return fmt
# 3. Look for format as standalone word (not part of another word)
# e.g., "epub" but not "republic"
for fmt in EBOOK_FORMATS:
# e.g., "epub" but not "republic", "m4b" but not "m4b123"
for fmt in ALL_BOOK_FORMATS:
# Match format as whole word
pattern = rf'\b{fmt}\b'
if re.search(pattern, title_lower):
@@ -125,15 +132,69 @@ def _extract_language(title: str) -> Optional[str]:
return None
def _prowlarr_result_to_release(result: dict) -> Release:
# Prowlarr category IDs for content type detection
# See: https://wiki.servarr.com/prowlarr/cardigann-yml-definition#categories
AUDIOBOOK_CATEGORY_IDS = {3000, 3030} # 3000 = Audio, 3030 = Audio/Audiobook
EBOOK_CATEGORY_IDS = {7000, 7020} # 7000 = Books, 7020 = Books/Ebook
def _detect_content_type_from_categories(categories: list, fallback: str = "book") -> str:
"""
Detect content type from Prowlarr category IDs.
Prowlarr returns categories as a list of dicts with 'id' and 'name' keys,
or sometimes just category IDs directly.
Args:
categories: List of category objects from Prowlarr result
fallback: Content type to use if categories don't indicate a specific type
Returns:
"audiobook" if audiobook categories detected, "book" otherwise
"""
# Normalize fallback - convert "ebook" to "book" for display consistency
if fallback == "ebook":
fallback = "book"
if not categories:
return fallback
# Extract category IDs from the nested structure
cat_ids = set()
for cat in categories:
if isinstance(cat, dict):
cat_id = cat.get("id")
if cat_id is not None:
cat_ids.add(cat_id)
elif isinstance(cat, int):
cat_ids.add(cat)
# Check for audiobook categories first (more specific)
if cat_ids & AUDIOBOOK_CATEGORY_IDS:
return "audiobook"
# Check for ebook categories
if cat_ids & EBOOK_CATEGORY_IDS:
return "book"
# Fall back to normalized content_type if no recognized categories
return fallback
def _prowlarr_result_to_release(result: dict, search_content_type: str = "ebook") -> Release:
"""
Convert a Prowlarr search result to a Release object.
Uses structured fields from Prowlarr when available:
- protocol: Direct from Prowlarr
- fileName: For format detection (more reliable than title)
- categories: To confirm ebook type
- categories: To detect content type (audiobook vs ebook)
- grabs: Download count
Args:
result: Raw Prowlarr API result
search_content_type: Content type from search, used as fallback if
categories don't indicate a specific type
"""
title = result.get("title", "Unknown")
size_bytes = result.get("size")
@@ -156,6 +217,10 @@ def _prowlarr_result_to_release(result: dict) -> Release:
# Extract language from title (Prowlarr doesn't provide this structured)
language = _extract_language(title)
# Detect content type from categories (per-result), with search type as fallback
categories = result.get("categories", [])
content_type = _detect_content_type_from_categories(categories, search_content_type)
# Build the source_id from GUID or generate from indexer + title
source_id = result.get("guid") or f"{indexer}:{hash(title)}"
@@ -176,9 +241,10 @@ def _prowlarr_result_to_release(result: dict) -> Release:
indexer=indexer,
seeders=seeders if protocol == "torrent" else None,
peers=peers_display if protocol == "torrent" else None,
content_type=content_type, # Detected per-result from categories
extra={
"publish_date": result.get("publishDate"),
"categories": result.get("categories", []),
"categories": categories,
"indexer_id": result.get("indexerId"),
"files": result.get("files"),
"grabs": grabs,
@@ -192,10 +258,12 @@ class ProwlarrSource(ReleaseSource):
Prowlarr release source.
Searches Prowlarr indexers for book releases (torrents and usenet).
Supports both ebooks (category 7000) and audiobooks (category 3030).
"""
name = "prowlarr"
display_name = "Prowlarr"
supported_content_types = ["ebook", "audiobook"] # Explicitly declare support for both
def __init__(self):
self.last_search_type: Optional[str] = None
@@ -233,14 +301,15 @@ class ProwlarrSource(ReleaseSource):
fallback="-",
),
ColumnSchema(
key="format",
label="Format",
key="content_type",
label="Type",
render_type=ColumnRenderType.BADGE,
align=ColumnAlign.CENTER,
width="70px",
width="90px",
hide_mobile=False,
color_hint=ColumnColorHint(type="map", value="format"),
color_hint=ColumnColorHint(type="map", value="content_type"),
uppercase=True,
fallback="-",
),
ColumnSchema(
key="size",
@@ -251,9 +320,9 @@ class ProwlarrSource(ReleaseSource):
hide_mobile=False,
),
],
grid_template="minmax(0,2fr) minmax(80px,1fr) 60px 70px 70px 80px",
grid_template="minmax(0,2fr) minmax(80px,1fr) 60px 70px 90px 80px",
leading_cell=LeadingCellConfig(type=LeadingCellType.NONE), # No leading cell for Prowlarr
supported_filters=["format"], # Prowlarr has unreliable language metadata
supported_filters=[], # Prowlarr has unreliable format/language metadata; content_type is auto-detected
)
def _get_client(self) -> Optional[ProwlarrClient]:
@@ -387,7 +456,7 @@ class ProwlarrSource(ReleaseSource):
logger.warning(f"Expanded search failed for indexer {indexer_id}: {e}")
self.last_search_type = "expanded"
results = [_prowlarr_result_to_release(r) for r in all_results]
results = [_prowlarr_result_to_release(r, content_type) for r in all_results]
if results:
torrent_count = sum(1 for r in results if r.protocol == "torrent")
+59
View File
@@ -8,11 +8,14 @@
#
# Web UIs:
# - cwabd: http://localhost:8084
# - Prowlarr: http://localhost:9696 (no auth by default)
# - qBittorrent: http://localhost:8080 (admin / adminadmin - check logs for temp password)
# - Transmission: http://localhost:9091 (admin / admin)
# - Deluge: http://localhost:8112 (password: deluge)
# - NZBGet: http://localhost:6789 (nzbget / tegbzn6789)
# - SABnzbd: http://localhost:8085 (complete setup wizard for API key)
# - aMule: http://localhost:4711 (password: amule)
# - Amarr: http://localhost:8086 (torznab indexer + qBittorrent emulation for amule)
#
# Hot-reload: Python code changes are picked up automatically (source mounted)
# Rebuild needed only for: requirements changes, frontend changes, Dockerfile changes
@@ -62,6 +65,21 @@ services:
- deluge
restart: unless-stopped
# ============ PROWLARR (INDEXER MANAGER) ============
prowlarr:
image: lscr.io/linuxserver/prowlarr:latest
container_name: test-prowlarr
environment:
- PUID=1000
- PGID=1000
- TZ=UTC
volumes:
- ./.local/test-clients/prowlarr/config:/config
ports:
- "9696:9696"
restart: unless-stopped
# ============ USENET CLIENTS ============
nzbget:
@@ -149,4 +167,45 @@ services:
- "6881:6881/udp"
restart: unless-stopped
# ============ AMULE + AMARR BRIDGE ============
amule:
image: ngosang/amule:latest
container_name: test-amule
environment:
- PUID=1000
- PGID=1000
- TZ=UTC
- GUI_PWD=amule
- WEBUI_PWD=amule
volumes:
- ./.local/test-clients/amule/config:/home/amule/.aMule
- ./.local/test-clients/downloads:/incoming
- ./.local/test-clients/amule/temp:/temp
ports:
- "4711:4711" # Web UI
- "4712:4712" # EC (External Connections) for Amarr
- "4662:4662" # ED2K TCP
- "4665:4665/udp" # ED2K global search
- "4672:4672/udp" # ED2K UDP
restart: unless-stopped
amarr:
image: vexdev/amarr:latest
container_name: test-amarr
environment:
- AMULE_HOST=amule
- AMULE_PORT=4712
- AMULE_PASSWORD=amule
- AMULE_FINISHED_PATH=/downloads
- AMARR_LOG_LEVEL=DEBUG
volumes:
- ./.local/test-clients/amarr/config:/config
- ./.local/test-clients/downloads:/downloads
ports:
- "8086:8080" # Amarr web/API (torznab indexer + qBittorrent emulation)
depends_on:
- amule
restart: unless-stopped
# All services automatically on same network (test-clients_default)
+2 -2
View File
@@ -149,11 +149,11 @@ volumes:
## Health Monitoring
The application exposes a health endpoint at `/api/status`. Add a health check to your compose:
The application exposes a health endpoint at `/api/health` (no authentication required). Add a health check to your compose:
```yaml
healthcheck:
test: ["CMD", "curl", "-sf", "http://localhost:8084/api/status"]
test: ["CMD", "curl", "-sf", "http://localhost:8084/api/health"]
interval: 30s
timeout: 30s
retries: 3
+4 -1
View File
@@ -475,6 +475,9 @@ function App() {
extra: release.extra,
preview: book.preview, // Pass book cover from metadata
content_type: releaseContentType, // For audiobook directory routing
series_name: book.series_name,
series_position: book.series_position,
subtitle: book.subtitle,
});
await fetchStatus();
} catch (error) {
@@ -554,7 +557,6 @@ function App() {
isLoading={isSearching}
onShowToast={showToast}
onRemoveToast={removeToast}
searchMode={searchMode}
contentType={contentType}
onContentTypeChange={setContentType}
/>
@@ -601,6 +603,7 @@ function App() {
searchFieldValues={searchFieldValues}
onSearchFieldChange={updateSearchFieldValue}
contentType={contentType}
onContentTypeChange={setContentType}
/>
{isInitialState && !featureNoticeDismissed && (
+46 -19
View File
@@ -1,4 +1,27 @@
import { ReactNode, useEffect, useLayoutEffect, useRef, useState } from 'react';
import { ReactNode, useCallback, useEffect, useLayoutEffect, useRef, useState } from 'react';
// Simple throttle function to limit how often a function can be called
function throttle<T extends (...args: unknown[]) => void>(fn: T, delay: number): T {
let lastCall = 0;
let timeoutId: ReturnType<typeof setTimeout> | null = null;
return ((...args: unknown[]) => {
const now = Date.now();
const timeSinceLastCall = now - lastCall;
if (timeSinceLastCall >= delay) {
lastCall = now;
fn(...args);
} else if (!timeoutId) {
// Schedule a trailing call
timeoutId = setTimeout(() => {
lastCall = Date.now();
timeoutId = null;
fn(...args);
}, delay - timeSinceLastCall);
}
}) as T;
}
interface DropdownProps {
label?: string;
@@ -62,32 +85,36 @@ export const Dropdown = ({
};
}, [isOpen]);
// Memoize the panel direction calculation
const updatePanelDirection = useCallback(() => {
if (!containerRef.current || !panelRef.current) {
return;
}
const rect = containerRef.current.getBoundingClientRect();
const panelHeight = panelRef.current.offsetHeight || panelRef.current.scrollHeight;
const spaceBelow = window.innerHeight - rect.bottom - 8;
const spaceAbove = rect.top - 8;
const shouldOpenUp = spaceBelow < panelHeight && spaceAbove >= panelHeight;
setPanelDirection(shouldOpenUp ? 'up' : 'down');
}, []);
useLayoutEffect(() => {
if (!isOpen) return;
const updatePanelDirection = () => {
if (!containerRef.current || !panelRef.current) {
return;
}
const rect = containerRef.current.getBoundingClientRect();
const panelHeight = panelRef.current.offsetHeight || panelRef.current.scrollHeight;
const spaceBelow = window.innerHeight - rect.bottom - 8;
const spaceAbove = rect.top - 8;
const shouldOpenUp = spaceBelow < panelHeight && spaceAbove >= panelHeight;
setPanelDirection(shouldOpenUp ? 'up' : 'down');
};
// Throttle scroll/resize handlers to reduce layout thrashing
const throttledUpdate = throttle(updatePanelDirection, 100);
updatePanelDirection();
window.addEventListener('resize', updatePanelDirection);
window.addEventListener('scroll', updatePanelDirection, true);
window.addEventListener('resize', throttledUpdate);
window.addEventListener('scroll', throttledUpdate, true);
return () => {
window.removeEventListener('resize', updatePanelDirection);
window.removeEventListener('scroll', updatePanelDirection, true);
window.removeEventListener('resize', throttledUpdate);
window.removeEventListener('scroll', throttledUpdate, true);
};
}, [isOpen]);
}, [isOpen, updatePanelDirection]);
return (
<div className={`relative ${widthClassName}`} ref={containerRef}>
+3 -42
View File
@@ -31,7 +31,6 @@ interface HeaderProps {
onLogout?: () => void;
onShowToast?: (message: string, type: 'success' | 'error' | 'info', persistent?: boolean) => string;
onRemoveToast?: (id: string) => void;
searchMode?: string;
contentType?: ContentType;
onContentTypeChange?: (type: ContentType) => void;
}
@@ -55,7 +54,6 @@ export const Header = forwardRef<HeaderHandle, HeaderProps>(({
onLogout,
onShowToast,
onRemoveToast,
searchMode = 'direct',
contentType = 'ebook',
onContentTypeChange,
}, ref) => {
@@ -261,46 +259,7 @@ export const Header = forwardRef<HeaderHandle, HeaderProps>(({
}}
>
<div className="py-1">
{/* Content Type Toggle - only shown in Universal search mode */}
{onContentTypeChange && searchMode === 'universal' && (
<>
<div className="px-4 py-2">
<div className="text-xs font-medium text-slate-500 dark:text-slate-400 mb-2">Search for</div>
<div className="flex flex-col rounded-lg overflow-hidden border border-[var(--border-muted)]">
<button
type="button"
onClick={() => onContentTypeChange('ebook')}
className={`w-full px-3 py-2 text-sm font-medium flex items-center gap-2 transition-colors ${
contentType === 'ebook'
? 'bg-emerald-600 text-white'
: 'bg-transparent hover:bg-[var(--hover-surface)]'
}`}
>
<svg className="w-4 h-4" fill="none" viewBox="0 0 24 24" strokeWidth="1.5" stroke="currentColor">
<path strokeLinecap="round" strokeLinejoin="round" d="M12 6.042A8.967 8.967 0 0 0 6 3.75c-1.052 0-2.062.18-3 .512v14.25A8.987 8.987 0 0 1 6 18c2.305 0 4.408.867 6 2.292m0-14.25a8.966 8.966 0 0 1 6-2.292c1.052 0 2.062.18 3 .512v14.25A8.987 8.987 0 0 0 18 18a8.967 8.967 0 0 0-6 2.292m0-14.25v14.25" />
</svg>
Books
</button>
<button
type="button"
onClick={() => onContentTypeChange('audiobook')}
className={`w-full px-3 py-2 text-sm font-medium flex items-center gap-2 transition-colors border-t border-[var(--border-muted)] ${
contentType === 'audiobook'
? 'bg-emerald-600 text-white'
: 'bg-transparent hover:bg-[var(--hover-surface)]'
}`}
>
<svg className="w-4 h-4" fill="none" viewBox="0 0 24 24" strokeWidth="1.5" stroke="currentColor">
<path strokeLinecap="round" strokeLinejoin="round" d="M19.114 5.636a9 9 0 0 1 0 12.728M16.463 8.288a5.25 5.25 0 0 1 0 7.424M6.75 8.25l4.72-4.72a.75.75 0 0 1 1.28.53v15.88a.75.75 0 0 1-1.28.53l-4.72-4.72H4.51c-.88 0-1.704-.507-1.938-1.354A9.009 9.009 0 0 1 2.25 12c0-.83.112-1.633.322-2.396C2.806 8.756 3.63 8.25 4.51 8.25H6.75Z" />
</svg>
Audiobooks
</button>
</div>
</div>
<div className="border-t border-[var(--border-muted)] my-1"></div>
</>
)}
<a
href="https://github.com/calibrain/calibre-web-automated-book-downloader/issues"
target="_blank"
@@ -493,6 +452,8 @@ export const Header = forwardRef<HeaderHandle, HeaderProps>(({
onSubmit={handleHeaderSearch}
onAdvancedToggle={onAdvancedToggle}
isLoading={isLoading}
contentType={contentType}
onContentTypeChange={onContentTypeChange}
/>
</div>
</div>
+176 -16
View File
@@ -1,5 +1,6 @@
import { KeyboardEvent, InputHTMLAttributes, useRef, forwardRef, useImperativeHandle } from 'react';
import { KeyboardEvent, InputHTMLAttributes, useRef, forwardRef, useImperativeHandle, useState, useEffect } from 'react';
import { useSearchMode } from '../contexts/SearchModeContext';
import { ContentType } from '../types';
interface SearchBarProps {
value: string;
@@ -20,6 +21,9 @@ interface SearchBarProps {
searchButtonTitle?: string;
autoComplete?: string;
enterKeyHint?: InputHTMLAttributes<HTMLInputElement>['enterKeyHint'];
// Content type selector props
contentType?: ContentType;
onContentTypeChange?: (type: ContentType) => void;
}
export interface SearchBarHandle {
@@ -45,12 +49,54 @@ export const SearchBar = forwardRef<SearchBarHandle, SearchBarProps>(({
searchButtonTitle = 'Search',
autoComplete = 'off',
enterKeyHint = 'search',
contentType = 'ebook',
onContentTypeChange,
}, ref) => {
const { searchMode } = useSearchMode();
const { searchMode, isUniversalMode } = useSearchMode();
const inputRef = useRef<HTMLInputElement>(null);
const buttonRef = useRef<HTMLButtonElement>(null);
const dropdownRef = useRef<HTMLDivElement>(null);
const hasSearchQuery = value.trim().length > 0;
// Content type dropdown state
const [isDropdownOpen, setIsDropdownOpen] = useState(false);
const showContentTypeSelector = isUniversalMode && !!onContentTypeChange;
// Dynamic placeholder based on content type
const effectivePlaceholder = showContentTypeSelector
? (contentType === 'ebook' ? 'Search books by title, author, ISBN...' : 'Search audiobooks by title, author...')
: placeholder;
// Close dropdown on click outside or escape
useEffect(() => {
if (!isDropdownOpen) return;
const handleClickOutside = (event: MouseEvent) => {
if (dropdownRef.current && !dropdownRef.current.contains(event.target as Node)) {
setIsDropdownOpen(false);
}
};
const handleEscape = (event: KeyboardEvent) => {
if (event.key === 'Escape') {
setIsDropdownOpen(false);
}
};
document.addEventListener('mousedown', handleClickOutside);
document.addEventListener('keydown', handleEscape as unknown as EventListener);
return () => {
document.removeEventListener('mousedown', handleClickOutside);
document.removeEventListener('keydown', handleEscape as unknown as EventListener);
};
}, [isDropdownOpen]);
const handleContentTypeSelect = (type: ContentType) => {
onContentTypeChange?.(type);
setIsDropdownOpen(false);
};
useImperativeHandle(ref, () => ({
submit: () => {
buttonRef.current?.click();
@@ -71,7 +117,8 @@ export const SearchBar = forwardRef<SearchBarHandle, SearchBarProps>(({
const wrapperClasses = ['relative', className].filter(Boolean).join(' ').trim();
const inputClasses = [
'w-full pl-4 pr-40 py-3 rounded-full border outline-none search-input',
'w-full pr-40 py-3 border outline-none search-input',
showContentTypeSelector ? 'pl-3 rounded-r-full' : 'pl-4 rounded-full',
inputClassName,
]
.filter(Boolean)
@@ -85,25 +132,138 @@ export const SearchBar = forwardRef<SearchBarHandle, SearchBarProps>(({
.join(' ')
.trim();
// Content type icons
const BookIcon = () => (
<svg className="w-5 h-5" fill="none" viewBox="0 0 24 24" strokeWidth="1.5" stroke="currentColor">
<path strokeLinecap="round" strokeLinejoin="round" d="M12 6.042A8.967 8.967 0 0 0 6 3.75c-1.052 0-2.062.18-3 .512v14.25A8.987 8.987 0 0 1 6 18c2.305 0 4.408.867 6 2.292m0-14.25a8.966 8.966 0 0 1 6-2.292c1.052 0 2.062.18 3 .512v14.25A8.987 8.987 0 0 0 18 18a8.967 8.967 0 0 0-6 2.292m0-14.25v14.25" />
</svg>
);
const AudiobookIcon = () => (
<svg className="w-5 h-5" fill="none" viewBox="0 0 24 24" strokeWidth="1.5" stroke="currentColor">
<path strokeLinecap="round" strokeLinejoin="round" d="M19.114 5.636a9 9 0 0 1 0 12.728M16.463 8.288a5.25 5.25 0 0 1 0 7.424M6.75 8.25l4.72-4.72a.75.75 0 0 1 1.28.53v15.88a.75.75 0 0 1-1.28.53l-4.72-4.72H4.51c-.88 0-1.704-.507-1.938-1.354A9.009 9.009 0 0 1 2.25 12c0-.83.112-1.633.322-2.396C2.806 8.756 3.63 8.25 4.51 8.25H6.75Z" />
</svg>
);
return (
<div className={wrapperClasses}>
<input
type="search"
placeholder={placeholder}
aria-label={inputAriaLabel}
autoComplete={autoComplete}
enterKeyHint={enterKeyHint}
className={inputClasses}
<div
className="flex items-stretch rounded-full border"
style={{
background: 'var(--bg-soft)',
color: 'var(--text)',
borderColor: 'var(--border-muted)',
}}
value={value}
onChange={e => onChange(e.target.value)}
onKeyDown={handleKeyDown}
ref={inputRef}
/>
>
{/* Content Type Selector */}
{showContentTypeSelector && (
<div className="relative flex-shrink-0 flex" ref={dropdownRef}>
<button
type="button"
onClick={() => setIsDropdownOpen(!isDropdownOpen)}
className="flex items-center gap-1.5 pl-5 pr-2 rounded-l-full transition-colors hover-action"
style={{ color: 'var(--text)' }}
aria-label={`Searching ${contentType === 'ebook' ? 'books' : 'audiobooks'}. Click to change.`}
aria-expanded={isDropdownOpen}
aria-haspopup="listbox"
>
{contentType === 'ebook' ? <BookIcon /> : <AudiobookIcon />}
<svg
className={`w-3 h-3 opacity-50 transition-transform duration-200 ${isDropdownOpen ? 'rotate-180' : ''}`}
fill="none"
stroke="currentColor"
viewBox="0 0 24 24"
strokeWidth="2.5"
>
<path strokeLinecap="round" strokeLinejoin="round" d="m19.5 8.25-7.5 7.5-7.5-7.5" />
</svg>
</button>
{/* Divider */}
<div
className="absolute right-0 top-1/2 -translate-y-1/2 w-px h-6"
style={{ background: 'var(--border-muted)' }}
/>
{/* Dropdown Menu */}
{isDropdownOpen && (
<div
className="absolute left-0 top-full mt-2 w-40 rounded-lg border shadow-lg z-50 overflow-hidden animate-fade-in-down"
style={{
background: 'var(--bg)',
borderColor: 'var(--border-muted)',
}}
role="listbox"
aria-label="Content type options"
>
<button
type="button"
onClick={() => handleContentTypeSelect('ebook')}
className={`w-full px-3 py-2.5 text-sm font-medium flex items-center gap-2.5 transition-colors ${
contentType === 'ebook'
? 'bg-emerald-600 text-white'
: 'hover-surface'
}`}
style={contentType !== 'ebook' ? { color: 'var(--text)' } : undefined}
role="option"
aria-selected={contentType === 'ebook'}
>
<BookIcon />
Books
{contentType === 'ebook' && (
<svg className="w-4 h-4 ml-auto" fill="none" viewBox="0 0 24 24" strokeWidth="2" stroke="currentColor">
<path strokeLinecap="round" strokeLinejoin="round" d="m4.5 12.75 6 6 9-13.5" />
</svg>
)}
</button>
<button
type="button"
onClick={() => handleContentTypeSelect('audiobook')}
className={`w-full px-3 py-2.5 text-sm font-medium flex items-center gap-2.5 transition-colors border-t ${
contentType === 'audiobook'
? 'bg-emerald-600 text-white'
: 'hover-surface'
}`}
style={{
borderColor: 'var(--border-muted)',
...(contentType !== 'audiobook' ? { color: 'var(--text)' } : {}),
}}
role="option"
aria-selected={contentType === 'audiobook'}
>
<AudiobookIcon />
Audiobooks
{contentType === 'audiobook' && (
<svg className="w-4 h-4 ml-auto" fill="none" viewBox="0 0 24 24" strokeWidth="2" stroke="currentColor">
<path strokeLinecap="round" strokeLinejoin="round" d="m4.5 12.75 6 6 9-13.5" />
</svg>
)}
</button>
</div>
)}
</div>
)}
{/* Search Input */}
<input
type="search"
placeholder={effectivePlaceholder}
aria-label={inputAriaLabel}
autoComplete={autoComplete}
enterKeyHint={enterKeyHint}
className={inputClasses}
style={{
background: 'transparent',
color: 'var(--text)',
border: 'none',
}}
value={value}
onChange={e => onChange(e.target.value)}
onKeyDown={handleKeyDown}
ref={inputRef}
/>
</div>
{/* Right-side controls */}
<div className={controlsClasses}>
{hasSearchQuery && (
<button
@@ -23,6 +23,7 @@ interface SearchSectionProps {
searchFieldValues?: Record<string, string | number | boolean>;
onSearchFieldChange?: (key: string, value: string | number | boolean) => void;
contentType?: ContentType;
onContentTypeChange?: (type: ContentType) => void;
}
export const SearchSection = ({
@@ -43,6 +44,7 @@ export const SearchSection = ({
searchFieldValues,
onSearchFieldChange,
contentType = 'ebook',
onContentTypeChange,
}: SearchSectionProps) => {
const { searchMode } = useSearchMode();
@@ -71,9 +73,7 @@ export const SearchSection = ({
isInitialState ? 'opacity-100 mb-6 sm:mb-8' : 'opacity-0 h-0 mb-0 overflow-hidden'
}`}>
<img src={logoUrl} alt="Logo" className="h-8 w-8" />
<h1 className="text-2xl font-semibold">
{contentType === 'audiobook' ? 'Audiobook Search & Download' : 'Book Search & Download'}
</h1>
<h1 className="text-2xl font-semibold">Book Search & Download</h1>
</div>
<div className={`flex flex-col gap-3 search-wrapper transition-all duration-500 ${
isInitialState ? '' : 'hidden'
@@ -84,6 +84,8 @@ export const SearchSection = ({
onSubmit={handleSearch}
isLoading={isLoading}
onAdvancedToggle={onAdvancedToggle}
contentType={contentType}
onContentTypeChange={onContentTypeChange}
/>
<AdvancedFilters
visible={showAdvanced}
@@ -1,4 +1,4 @@
import { useEffect, useRef } from 'react';
import { useEffect, useMemo, useRef } from 'react';
import {
SettingsTab,
SettingsField,
@@ -35,13 +35,20 @@ interface SettingsContentProps {
onAction: (key: string) => Promise<ActionResult>;
isSaving: boolean;
hasChanges: boolean;
isUniversalMode?: boolean; // Whether app is in Universal search mode
}
// Check if a field should be visible based on showWhen condition
// Check if a field should be visible based on showWhen condition and search mode
function isFieldVisible(
field: SettingsField,
values: Record<string, unknown>
values: Record<string, unknown>,
isUniversalMode: boolean
): boolean {
// Check universalOnly - hide these fields in Direct mode
if ('universalOnly' in field && field.universalOnly && !isUniversalMode) {
return false;
}
const showWhen = field.showWhen;
if (!showWhen) return true;
@@ -193,6 +200,7 @@ export const SettingsContent = ({
onAction,
isSaving,
hasChanges,
isUniversalMode = true,
}: SettingsContentProps) => {
const scrollRef = useRef<HTMLDivElement>(null);
@@ -203,6 +211,12 @@ export const SettingsContent = ({
}
}, [tab.name]);
// Memoize the visible fields to avoid recalculating on every render
const visibleFields = useMemo(
() => tab.fields.filter((field) => isFieldVisible(field, values, isUniversalMode)),
[tab.fields, values, isUniversalMode]
);
return (
<div className="flex-1 flex flex-col min-h-0">
{/* Scrollable content area */}
@@ -212,9 +226,7 @@ export const SettingsContent = ({
style={{ paddingBottom: hasChanges ? 'calc(5rem + env(safe-area-inset-bottom))' : '1.5rem' }}
>
<div className="space-y-5">
{tab.fields
.filter((field) => isFieldVisible(field, values))
.map((field) => {
{visibleFields.map((field) => {
const disabledState = getDisabledState(field, values);
return (
<FieldWrapper
@@ -1,5 +1,6 @@
import { useEffect, useState, useCallback, useRef } from 'react';
import { useEffect, useState, useCallback, useRef, useMemo } from 'react';
import { useSettings } from '../../hooks/useSettings';
import { useSearchMode } from '../../contexts/SearchModeContext';
import { SettingsHeader } from './SettingsHeader';
import { SettingsSidebar } from './SettingsSidebar';
import { SettingsContent } from './SettingsContent';
@@ -27,6 +28,8 @@ export const SettingsModal = ({ isOpen, onClose, onShowToast, onSettingsSaved }:
isSaving,
} = useSettings();
const { isUniversalMode } = useSearchMode();
// Track if we're showing detail view on mobile
const [isMobile, setIsMobile] = useState(false);
const [showMobileDetail, setShowMobileDetail] = useState(false);
@@ -143,6 +146,23 @@ export const SettingsModal = ({ isOpen, onClose, onShowToast, onSettingsSaved }:
[selectedTab, executeAction]
);
// Memoize the field change handler to prevent creating new functions on every render
const handleFieldChange = useCallback(
(key: string, value: unknown) => {
if (selectedTab) {
updateValue(selectedTab, key, value);
}
},
[selectedTab, updateValue]
);
// Memoize hasChanges to avoid expensive JSON.stringify comparisons on every render
// Must be before early returns to satisfy React's rules of hooks
const currentTabHasChanges = useMemo(
() => (selectedTab ? hasChanges(selectedTab) : false),
[selectedTab, hasChanges, values]
);
if (!isOpen && !isClosing) return null;
const currentTab = tabs.find((t) => t.name === selectedTab);
@@ -153,7 +173,8 @@ export const SettingsModal = ({ isOpen, onClose, onShowToast, onSettingsSaved }:
return (
<div className="fixed inset-0 z-50 flex items-center justify-center">
<div
className="absolute inset-0 bg-black/50 backdrop-blur-sm"
className="absolute inset-0 bg-black/50 backdrop-blur-[2px]"
style={{ willChange: 'opacity', contain: 'strict' }}
onClick={handleClose}
/>
<div
@@ -189,7 +210,8 @@ export const SettingsModal = ({ isOpen, onClose, onShowToast, onSettingsSaved }:
return (
<div className="fixed inset-0 z-50 flex items-center justify-center">
<div
className="absolute inset-0 bg-black/50 backdrop-blur-sm"
className="absolute inset-0 bg-black/50 backdrop-blur-[2px]"
style={{ willChange: 'opacity', contain: 'strict' }}
onClick={handleClose}
/>
<div
@@ -261,11 +283,12 @@ export const SettingsModal = ({ isOpen, onClose, onShowToast, onSettingsSaved }:
<SettingsContent
tab={currentTab}
values={values[currentTab.name] || {}}
onChange={(key, value) => updateValue(currentTab.name, key, value)}
onChange={handleFieldChange}
onSave={handleSave}
onAction={handleAction}
isSaving={isSaving}
hasChanges={hasChanges(currentTab.name)}
hasChanges={currentTabHasChanges}
isUniversalMode={isUniversalMode}
/>
)}
</>
@@ -279,8 +302,9 @@ export const SettingsModal = ({ isOpen, onClose, onShowToast, onSettingsSaved }:
<div className="fixed inset-0 z-50 flex items-center justify-center p-4">
{/* Backdrop */}
<div
className={`absolute inset-0 bg-black/50 backdrop-blur-sm transition-opacity duration-150
className={`absolute inset-0 bg-black/50 backdrop-blur-[2px] transition-opacity duration-150
${isClosing ? 'opacity-0' : 'opacity-100'}`}
style={{ willChange: 'opacity', contain: 'strict' }}
onClick={handleClose}
/>
@@ -310,11 +334,12 @@ export const SettingsModal = ({ isOpen, onClose, onShowToast, onSettingsSaved }:
<SettingsContent
tab={currentTab}
values={values[currentTab.name] || {}}
onChange={(key, value) => updateValue(currentTab.name, key, value)}
onChange={handleFieldChange}
onSave={handleSave}
onAction={handleAction}
isSaving={isSaving}
hasChanges={hasChanges(currentTab.name)}
hasChanges={currentTabHasChanges}
isUniversalMode={isUniversalMode}
/>
) : (
<div className="flex-1 flex items-center justify-center text-sm opacity-60">
@@ -1,4 +1,4 @@
import { useState } from 'react';
import { useState, useMemo } from 'react';
import { SettingsTab, SettingsGroup } from '../../types/settings';
interface SettingsSidebarProps {
@@ -129,47 +129,52 @@ export const SettingsSidebar = ({
});
};
// Build grouped tabs map
const groupedTabs = new Map<string, SettingsTab[]>();
tabs.forEach((tab) => {
if (tab.group) {
const existing = groupedTabs.get(tab.group) || [];
existing.push(tab);
groupedTabs.set(tab.group, existing);
}
});
// Build a unified sorted list of tabs and groups
const sidebarItems: SidebarItem[] = [];
// Add ungrouped tabs (with section headers where needed)
tabs.forEach((tab) => {
if (!tab.group) {
// Check if this tab needs a section header before it
const sectionHeader = SECTION_HEADERS.find((s) => s.beforeTab === tab.name);
if (sectionHeader) {
sidebarItems.push({ type: 'section', label: sectionHeader.label, order: tab.order - 0.5 });
// Memoize the sidebar items to avoid rebuilding on every render
const sidebarItems = useMemo(() => {
// Build grouped tabs map
const groupedTabs = new Map<string, SettingsTab[]>();
tabs.forEach((tab) => {
if (tab.group) {
const existing = groupedTabs.get(tab.group) || [];
existing.push(tab);
groupedTabs.set(tab.group, existing);
}
sidebarItems.push({ type: 'tab', tab, order: tab.order });
}
});
});
// Add groups (with their tabs) and section headers
groups.forEach((group) => {
const groupTabList = groupedTabs.get(group.name) || [];
if (groupTabList.length > 0) {
// Check if this group needs a section header before it
const sectionHeader = SECTION_HEADERS.find((s) => s.beforeGroup === group.name);
if (sectionHeader) {
// Insert section header just before this group (order - 0.5 to sort before)
sidebarItems.push({ type: 'section', label: sectionHeader.label, order: group.order - 0.5 });
// Build a unified sorted list of tabs and groups
const items: SidebarItem[] = [];
// Add ungrouped tabs (with section headers where needed)
tabs.forEach((tab) => {
if (!tab.group) {
// Check if this tab needs a section header before it
const sectionHeader = SECTION_HEADERS.find((s) => s.beforeTab === tab.name);
if (sectionHeader) {
items.push({ type: 'section', label: sectionHeader.label, order: tab.order - 0.5 });
}
items.push({ type: 'tab', tab, order: tab.order });
}
sidebarItems.push({ type: 'group', group, tabs: groupTabList, order: group.order });
}
});
});
// Sort by order
sidebarItems.sort((a, b) => a.order - b.order);
// Add groups (with their tabs) and section headers
groups.forEach((group) => {
const groupTabList = groupedTabs.get(group.name) || [];
if (groupTabList.length > 0) {
// Check if this group needs a section header before it
const sectionHeader = SECTION_HEADERS.find((s) => s.beforeGroup === group.name);
if (sectionHeader) {
// Insert section header just before this group (order - 0.5 to sort before)
items.push({ type: 'section', label: sectionHeader.label, order: group.order - 0.5 });
}
items.push({ type: 'group', group, tabs: groupTabList, order: group.order });
}
});
// Sort by order
items.sort((a, b) => a.order - b.order);
return items;
}, [tabs, groups]);
if (mode === 'list') {
// Mobile: Clean list style with inset dividers
+3
View File
@@ -166,6 +166,9 @@ export const downloadRelease = async (release: {
extra?: Record<string, unknown>;
preview?: string; // Book cover from metadata provider
content_type?: string; // "ebook" or "audiobook" - for directory routing
series_name?: string;
series_position?: number;
subtitle?: string;
}): Promise<void> => {
await fetchJSON(`${API_BASE}/releases/download`, {
method: 'POST',
+2
View File
@@ -45,6 +45,7 @@ export interface Book {
series_name?: string; // Name of the series
series_position?: number; // This book's position (e.g., 3, 1.5 for novellas)
series_count?: number; // Total books in the series
subtitle?: string;
}
// Status response types
@@ -266,6 +267,7 @@ export interface ReleasesResponse {
provider: string;
provider_id: string;
title: string;
subtitle?: string;
authors?: string[];
isbn_10?: string;
isbn_13?: string;
+2
View File
@@ -44,6 +44,7 @@ export interface BaseField {
showWhen?: ShowWhenCondition; // Conditional visibility based on another field's value
disabledWhen?: DisabledWhenCondition; // Conditional disable based on another field's value
requiresRestart?: boolean; // True if changing this setting requires a container restart
universalOnly?: boolean; // Only show in Universal search mode (hide in Direct mode)
}
// Specific field interfaces
@@ -120,6 +121,7 @@ export interface HeadingFieldConfig {
linkUrl?: string;
linkText?: string;
showWhen?: ShowWhenCondition; // Conditional visibility based on another field's value
universalOnly?: boolean; // Only show in Universal search mode (hide in Direct mode)
}
// Union type for all fields
@@ -28,6 +28,7 @@ export interface MetadataBookData {
series_name?: string;
series_position?: number;
series_count?: number;
subtitle?: string;
}
/**
@@ -55,6 +56,7 @@ export function transformMetadataToBook(data: MetadataBookData): Book {
series_name: data.series_name,
series_position: data.series_position,
series_count: data.series_count,
subtitle: data.subtitle,
info: {
...(data.isbn_13 && { ISBN: data.isbn_13 }),
...(data.isbn_10 && !data.isbn_13 && { ISBN: data.isbn_10 }),
+14
View File
@@ -45,9 +45,16 @@ const DOWNLOAD_TYPE_COLORS: Record<string, ColorStyle> = {
direct: { bg: 'bg-emerald-500/20', text: 'text-emerald-700 dark:text-emerald-300' },
};
const CONTENT_TYPE_COLORS: Record<string, ColorStyle> = {
book: { bg: 'bg-blue-500/20', text: 'text-blue-700 dark:text-blue-300' },
ebook: { bg: 'bg-blue-500/20', text: 'text-blue-700 dark:text-blue-300' }, // Alias for backwards compatibility
audiobook: { bg: 'bg-violet-500/20', text: 'text-violet-700 dark:text-violet-300' },
};
const DEFAULT_FORMAT_COLOR: ColorStyle = { bg: 'bg-cyan-500/20', text: 'text-cyan-700 dark:text-cyan-300' };
const DEFAULT_LANGUAGE_COLOR: ColorStyle = { bg: 'bg-indigo-500/20', text: 'text-indigo-700 dark:text-indigo-300' };
const DEFAULT_DOWNLOAD_TYPE_COLOR: ColorStyle = { bg: 'bg-violet-500/20', text: 'text-violet-700 dark:text-violet-300' };
const DEFAULT_CONTENT_TYPE_COLOR: ColorStyle = { bg: 'bg-gray-500/20', text: 'text-gray-700 dark:text-gray-300' };
const FALLBACK_COLOR: ColorStyle = { bg: 'bg-gray-500/20', text: 'text-gray-700 dark:text-gray-300' };
export function getFormatColor(format?: string): ColorStyle {
@@ -65,6 +72,11 @@ export function getDownloadTypeColor(downloadType?: string): ColorStyle {
return DOWNLOAD_TYPE_COLORS[downloadType.toLowerCase()] || DEFAULT_DOWNLOAD_TYPE_COLOR;
}
export function getContentTypeColor(contentType?: string): ColorStyle {
if (!contentType || contentType === '-') return FALLBACK_COLOR;
return CONTENT_TYPE_COLORS[contentType.toLowerCase()] || DEFAULT_CONTENT_TYPE_COLOR;
}
/**
* Color hint type for dynamic column coloring.
*/
@@ -92,6 +104,8 @@ export function getColorStyleFromHint(value: string, colorHint?: ColumnColorHint
return getLanguageColor(value);
case 'download_type':
return getDownloadTypeColor(value);
case 'content_type':
return getContentTypeColor(value);
default:
return FALLBACK_COLOR;
}
+1
View File
@@ -0,0 +1 @@
# Core module tests
+871
View File
@@ -0,0 +1,871 @@
"""Tests for hardlinking and staging functionality.
Two approaches to preserve torrent files for seeding:
1. **Ingest mode**: Copy to staging → Move to ingest
- Uses `stage_file(copy=True)` to preserve original
- Less efficient (creates temp copy)
2. **Library mode with hardlink**: Hardlink from torrent path → library
- Uses `_atomic_hardlink()` - same inode, no extra disk space
- More efficient but requires same filesystem
"""
import os
import pytest
import tempfile
from pathlib import Path
from unittest.mock import MagicMock, patch
from cwa_book_downloader.core.naming import same_filesystem
class TestStageFile:
"""Tests for stage_file() - the ingest mode approach for torrents."""
def test_copy_mode_preserves_original(self, tmp_path):
"""copy=True preserves original file (for torrent seeding)."""
from cwa_book_downloader.download.orchestrator import stage_file, get_staging_dir
source = tmp_path / "downloads" / "book.epub"
source.parent.mkdir()
source.write_bytes(b"content")
with patch('cwa_book_downloader.download.orchestrator.TMP_DIR', tmp_path / "staging"):
staged = stage_file(source, "task123", copy=True)
assert staged.exists()
assert source.exists() # Original preserved
assert staged.read_bytes() == b"content"
def test_move_mode_removes_original(self, tmp_path):
"""copy=False moves file (original deleted)."""
from cwa_book_downloader.download.orchestrator import stage_file
source = tmp_path / "downloads" / "book.epub"
source.parent.mkdir()
source.write_bytes(b"content")
with patch('cwa_book_downloader.download.orchestrator.TMP_DIR', tmp_path / "staging"):
staged = stage_file(source, "task123", copy=False)
assert staged.exists()
assert not source.exists() # Original deleted
def test_handles_filename_collision(self, tmp_path):
"""Adds counter suffix on collision."""
from cwa_book_downloader.download.orchestrator import stage_file
staging = tmp_path / "staging"
staging.mkdir()
(staging / "book.epub").touch() # Pre-existing file
source = tmp_path / "downloads" / "book.epub"
source.parent.mkdir()
source.write_bytes(b"new content")
with patch('cwa_book_downloader.download.orchestrator.TMP_DIR', staging):
staged = stage_file(source, "task123", copy=True)
assert staged.name == "book_1.epub"
class TestSameFilesystem:
"""Tests for same_filesystem() detection."""
def test_same_directory(self, tmp_path):
"""Two paths in same temp directory are on same filesystem."""
file1 = tmp_path / "file1.txt"
file2 = tmp_path / "file2.txt"
file1.touch()
file2.touch()
assert same_filesystem(file1, file2) is True
def test_same_filesystem_different_dirs(self, tmp_path):
"""Subdirectories of same temp are on same filesystem."""
dir1 = tmp_path / "dir1"
dir2 = tmp_path / "dir2"
dir1.mkdir()
dir2.mkdir()
assert same_filesystem(dir1, dir2) is True
def test_nonexistent_paths_same_parent(self, tmp_path):
"""Non-existent paths with same parent are on same filesystem."""
path1 = tmp_path / "nonexistent1"
path2 = tmp_path / "nonexistent2"
assert same_filesystem(path1, path2) is True
def test_nonexistent_nested_paths(self, tmp_path):
"""Deeply nested non-existent paths check parent filesystem."""
path1 = tmp_path / "a" / "b" / "c" / "file.txt"
path2 = tmp_path / "x" / "y" / "z" / "file.txt"
assert same_filesystem(path1, path2) is True
def test_string_paths(self, tmp_path):
"""Accepts string paths as well as Path objects."""
file1 = tmp_path / "file1.txt"
file1.touch()
assert same_filesystem(str(file1), str(tmp_path)) is True
def test_permission_error_returns_false(self, tmp_path):
"""Returns False when permission denied (safe fallback)."""
with patch('os.stat', side_effect=PermissionError("denied")):
assert same_filesystem(tmp_path, tmp_path) is False
def test_oserror_returns_false(self, tmp_path):
"""Returns False on OS errors (safe fallback)."""
with patch('os.stat', side_effect=OSError("error")):
assert same_filesystem(tmp_path, tmp_path) is False
class TestAtomicHardlink:
"""Tests for _atomic_hardlink() function."""
def test_creates_hardlink(self, tmp_path):
"""Creates hardlink to source file."""
from cwa_book_downloader.download.orchestrator import _atomic_hardlink
source = tmp_path / "source.txt"
source.write_text("content")
dest = tmp_path / "dest.txt"
result = _atomic_hardlink(source, dest)
assert result == dest
assert result.exists()
assert result.read_text() == "content"
# Verify it's a hardlink (same inode)
assert os.stat(source).st_ino == os.stat(result).st_ino
def test_handles_collision_with_counter(self, tmp_path):
"""Appends counter suffix when destination exists."""
from cwa_book_downloader.download.orchestrator import _atomic_hardlink
source = tmp_path / "source.txt"
source.write_text("new content")
dest = tmp_path / "dest.txt"
dest.write_text("existing")
result = _atomic_hardlink(source, dest)
assert result == tmp_path / "dest_1.txt"
assert result.read_text() == "new content"
assert dest.read_text() == "existing"
def test_multiple_collisions(self, tmp_path):
"""Increments counter until finding free slot."""
from cwa_book_downloader.download.orchestrator import _atomic_hardlink
source = tmp_path / "source.txt"
source.write_text("new")
(tmp_path / "dest.txt").touch()
(tmp_path / "dest_1.txt").touch()
(tmp_path / "dest_2.txt").touch()
result = _atomic_hardlink(source, tmp_path / "dest.txt")
assert result == tmp_path / "dest_3.txt"
def test_preserves_extension(self, tmp_path):
"""Keeps extension when adding counter suffix."""
from cwa_book_downloader.download.orchestrator import _atomic_hardlink
source = tmp_path / "book.epub"
source.write_bytes(b"epub content")
(tmp_path / "book.epub").touch()
result = _atomic_hardlink(source, tmp_path / "book.epub")
assert result.suffix == ".epub"
assert result.name == "book_1.epub"
class TestAtomicMove:
"""Tests for _atomic_move() function."""
def test_moves_file(self, tmp_path):
"""Moves file from source to destination."""
from cwa_book_downloader.download.orchestrator import _atomic_move
source = tmp_path / "source.txt"
source.write_text("content")
dest = tmp_path / "dest.txt"
result = _atomic_move(source, dest)
assert result == dest
assert not source.exists()
assert result.read_text() == "content"
def test_handles_collision(self, tmp_path):
"""Appends counter on collision."""
from cwa_book_downloader.download.orchestrator import _atomic_move
source = tmp_path / "source.txt"
source.write_text("new")
dest = tmp_path / "dest.txt"
dest.write_text("existing")
result = _atomic_move(source, dest)
assert result == tmp_path / "dest_1.txt"
assert not source.exists()
assert dest.read_text() == "existing"
assert result.read_text() == "new"
def test_cross_filesystem_fallback(self):
"""Falls back to copy when cross-filesystem."""
from cwa_book_downloader.download.orchestrator import _atomic_move
import errno
with tempfile.TemporaryDirectory() as dir1, tempfile.TemporaryDirectory() as dir2:
source = Path(dir1) / "source.txt"
source.write_text("content")
dest = Path(dir2) / "dest.txt"
# This should work even if dirs are on different filesystems
# (uses fallback to copy)
result = _atomic_move(source, dest)
assert not source.exists()
assert result.read_text() == "content"
class TestHardlinkWithLibraryMode:
"""Tests for hardlinking in library mode context."""
@pytest.fixture
def mock_config(self):
"""Mock config for library mode."""
with patch('cwa_book_downloader.download.orchestrator.config') as mock:
mock.get = MagicMock(side_effect=lambda key, default=None: {
"LIBRARY_PATH": None,
"LIBRARY_PATH_AUDIOBOOK": None,
"LIBRARY_TEMPLATE": "{Author}/{Title}",
"LIBRARY_TEMPLATE_AUDIOBOOK": "{Author}/{Title}",
"TORRENT_HARDLINK": True,
"PROCESSING_MODE": "library",
"PROCESSING_MODE_AUDIOBOOK": "library",
}.get(key, default))
yield mock
@pytest.fixture
def sample_task(self):
"""Create a sample DownloadTask for testing."""
from cwa_book_downloader.core.models import DownloadTask, SearchMode
return DownloadTask(
task_id="test123",
source="prowlarr",
title="The Way of Kings",
author="Brandon Sanderson",
format="epub",
search_mode=SearchMode.UNIVERSAL,
)
def test_transfer_file_hardlink(self, tmp_path, sample_task):
"""Single file transferred via hardlink."""
from cwa_book_downloader.download.orchestrator import _transfer_file_to_library
library = tmp_path / "library"
library.mkdir()
source = tmp_path / "downloads" / "book.epub"
source.parent.mkdir()
source.write_bytes(b"epub content")
temp_file = tmp_path / "staging" / "book.epub"
temp_file.parent.mkdir()
temp_file.write_bytes(b"staged content")
status_cb = MagicMock()
result = _transfer_file_to_library(
source_path=source,
library_base=str(library),
template="{Author}/{Title}",
metadata={"Author": "Brandon Sanderson", "Title": "Mistborn"},
task=sample_task,
temp_file=temp_file,
status_callback=status_cb,
use_hardlink=True,
)
assert result is not None
result_path = Path(result)
assert result_path.exists()
assert result_path.parent.name == "Brandon Sanderson"
assert result_path.name == "Mistborn.epub"
# Source should still exist (hardlink)
assert source.exists()
# Temp file should be cleaned up
assert not temp_file.exists()
status_cb.assert_called_with("complete", "Complete (library mode)")
def test_transfer_file_move(self, tmp_path, sample_task):
"""Single file transferred via move."""
from cwa_book_downloader.download.orchestrator import _transfer_file_to_library
library = tmp_path / "library"
library.mkdir()
source = tmp_path / "staging" / "book.epub"
source.parent.mkdir()
source.write_bytes(b"epub content")
status_cb = MagicMock()
result = _transfer_file_to_library(
source_path=source,
library_base=str(library),
template="{Author}/{Title}",
metadata={"Author": "Brandon Sanderson", "Title": "Mistborn"},
task=sample_task,
temp_file=source,
status_callback=status_cb,
use_hardlink=False,
)
assert result is not None
result_path = Path(result)
assert result_path.exists()
# Source should NOT exist (moved)
assert not source.exists()
status_cb.assert_called_with("complete", "Complete (library mode)")
def test_transfer_directory_hardlink_multifile(self, tmp_path, sample_task):
"""Directory with multiple files transferred via hardlinks."""
from cwa_book_downloader.download.orchestrator import _transfer_directory_to_library
library = tmp_path / "library"
library.mkdir()
source_dir = tmp_path / "downloads" / "audiobook"
source_dir.mkdir(parents=True)
# Create source audio files
(source_dir / "Part 1.mp3").write_bytes(b"audio1")
(source_dir / "Part 2.mp3").write_bytes(b"audio2")
(source_dir / "Part 10.mp3").write_bytes(b"audio10")
# Create temp staging dir
temp_dir = tmp_path / "staging" / "audiobook"
temp_dir.mkdir(parents=True)
(temp_dir / "Part 1.mp3").write_bytes(b"staged1")
(temp_dir / "Part 2.mp3").write_bytes(b"staged2")
(temp_dir / "Part 10.mp3").write_bytes(b"staged10")
sample_task.content_type = "audiobook"
status_cb = MagicMock()
with patch('cwa_book_downloader.download.orchestrator._get_supported_formats', return_value=["mp3"]):
result = _transfer_directory_to_library(
source_dir=source_dir,
library_base=str(library),
template="{Author}/{Title}{ - PartNumber}", # Correct token format
metadata={"Author": "Brandon Sanderson", "Title": "The Way of Kings"},
task=sample_task,
temp_file=temp_dir,
status_callback=status_cb,
use_hardlink=True,
)
assert result is not None
result_path = Path(result)
assert result_path.parent.name == "Brandon Sanderson"
# Check all 3 files created with sequential part numbers
author_dir = library / "Brandon Sanderson"
files = sorted(author_dir.glob("*.mp3"))
assert len(files) == 3
assert files[0].name == "The Way of Kings - 01.mp3"
assert files[1].name == "The Way of Kings - 02.mp3"
assert files[2].name == "The Way of Kings - 03.mp3"
# Source files should still exist (hardlinks)
assert (source_dir / "Part 1.mp3").exists()
assert (source_dir / "Part 2.mp3").exists()
assert (source_dir / "Part 10.mp3").exists()
# Temp dir should be cleaned up
assert not temp_dir.exists()
def test_transfer_directory_move(self, tmp_path, sample_task):
"""Directory transferred via move (non-torrent)."""
from cwa_book_downloader.download.orchestrator import _transfer_directory_to_library
library = tmp_path / "library"
library.mkdir()
source_dir = tmp_path / "staging" / "audiobook"
source_dir.mkdir(parents=True)
(source_dir / "Chapter 01.mp3").write_bytes(b"audio1")
(source_dir / "Chapter 02.mp3").write_bytes(b"audio2")
sample_task.content_type = "audiobook"
status_cb = MagicMock()
with patch('cwa_book_downloader.download.orchestrator._get_supported_formats', return_value=["mp3"]):
result = _transfer_directory_to_library(
source_dir=source_dir,
library_base=str(library),
template="{Author}/{Title}{ - Part PartNumber}",
metadata={"Author": "Brandon Sanderson", "Title": "The Way of Kings"},
task=sample_task,
temp_file=source_dir,
status_callback=status_cb,
use_hardlink=False,
)
assert result is not None
author_dir = library / "Brandon Sanderson"
files = list(author_dir.glob("*.mp3"))
assert len(files) == 2
# Source dir should be cleaned up
assert not source_dir.exists()
def test_single_file_in_directory_no_part_number(self, tmp_path, sample_task):
"""Single file in directory doesn't get part number."""
from cwa_book_downloader.download.orchestrator import _transfer_directory_to_library
library = tmp_path / "library"
library.mkdir()
source_dir = tmp_path / "downloads" / "book"
source_dir.mkdir(parents=True)
(source_dir / "book.epub").write_bytes(b"content")
temp_dir = tmp_path / "staging" / "book"
temp_dir.mkdir(parents=True)
(temp_dir / "book.epub").write_bytes(b"staged")
status_cb = MagicMock()
with patch('cwa_book_downloader.download.orchestrator._get_supported_formats', return_value=["epub"]):
result = _transfer_directory_to_library(
source_dir=source_dir,
library_base=str(library),
template="{Author}/{Title}{ - PartNumber}", # Correct token format
metadata={"Author": "Brandon Sanderson", "Title": "Mistborn"},
task=sample_task,
temp_file=temp_dir,
status_callback=status_cb,
use_hardlink=True,
)
result_path = Path(result)
# Single file should NOT have part number (conditional prefix stripped)
assert result_path.name == "Mistborn.epub"
class TestHardlinkDecisionLogic:
"""Tests for the decision to use hardlinks vs moves."""
@pytest.fixture
def sample_task(self):
from cwa_book_downloader.core.models import DownloadTask, SearchMode
return DownloadTask(
task_id="test123",
source="prowlarr",
title="Test Book",
author="Test Author",
format="epub",
search_mode=SearchMode.UNIVERSAL,
)
def test_hardlink_enabled_same_filesystem(self, tmp_path, sample_task):
"""Hardlink used when enabled and same filesystem."""
from cwa_book_downloader.download.orchestrator import _process_library_mode
library = tmp_path / "library"
library.mkdir()
staging = tmp_path / "staging"
staging.mkdir()
source = tmp_path / "downloads" / "book.epub"
source.parent.mkdir()
source.write_bytes(b"content")
staged = staging / "book.epub"
staged.write_bytes(b"staged")
# Task has original_download_path (torrent scenario)
sample_task.original_download_path = str(source)
status_cb = MagicMock()
with patch('cwa_book_downloader.download.orchestrator.config') as mock_config:
mock_config.get = MagicMock(side_effect=lambda key, default=None: {
"LIBRARY_PATH": str(library),
"LIBRARY_TEMPLATE": "{Author}/{Title}",
"TORRENT_HARDLINK": True,
"PROCESSING_MODE": "library",
}.get(key, default))
result = _process_library_mode(staged, sample_task, status_cb)
assert result is not None
# Source should still exist (hardlinked)
assert source.exists()
def test_hardlink_disabled_falls_back_to_move(self, tmp_path, sample_task):
"""Move used when hardlink disabled in config."""
from cwa_book_downloader.download.orchestrator import _process_library_mode
library = tmp_path / "library"
library.mkdir()
source = tmp_path / "downloads" / "book.epub"
source.parent.mkdir()
source.write_bytes(b"content")
staged = tmp_path / "staging" / "book.epub"
staged.parent.mkdir()
staged.write_bytes(b"staged")
sample_task.original_download_path = str(source)
status_cb = MagicMock()
with patch('cwa_book_downloader.download.orchestrator.config') as mock_config:
mock_config.get = MagicMock(side_effect=lambda key, default=None: {
"LIBRARY_PATH": str(library),
"LIBRARY_TEMPLATE": "{Author}/{Title}",
"TORRENT_HARDLINK": False, # Disabled
"PROCESSING_MODE": "library",
}.get(key, default))
result = _process_library_mode(staged, sample_task, status_cb)
assert result is not None
# Staged file should be moved (not exist)
assert not staged.exists()
def test_no_original_path_uses_staging(self, tmp_path, sample_task):
"""Without original_download_path, moves from staging."""
from cwa_book_downloader.download.orchestrator import _process_library_mode
library = tmp_path / "library"
library.mkdir()
staged = tmp_path / "staging" / "book.epub"
staged.parent.mkdir()
staged.write_bytes(b"content")
# No original_download_path (direct download scenario)
sample_task.original_download_path = None
status_cb = MagicMock()
with patch('cwa_book_downloader.download.orchestrator.config') as mock_config:
mock_config.get = MagicMock(side_effect=lambda key, default=None: {
"LIBRARY_PATH": str(library),
"LIBRARY_TEMPLATE": "{Author}/{Title}",
"TORRENT_HARDLINK": True,
"PROCESSING_MODE": "library",
}.get(key, default))
result = _process_library_mode(staged, sample_task, status_cb)
assert result is not None
# Staged file should be moved
assert not staged.exists()
class TestHardlinkInodeVerification:
"""Tests that verify hardlinks share the same inode."""
def test_hardlink_shares_inode(self, tmp_path):
"""Hardlinked files share same inode."""
from cwa_book_downloader.download.orchestrator import _atomic_hardlink
source = tmp_path / "source.txt"
source.write_text("shared content")
dest = tmp_path / "dest.txt"
result = _atomic_hardlink(source, dest)
source_inode = os.stat(source).st_ino
dest_inode = os.stat(result).st_ino
assert source_inode == dest_inode
def test_hardlink_reflects_changes(self, tmp_path):
"""Changes to source reflect in hardlink."""
from cwa_book_downloader.download.orchestrator import _atomic_hardlink
source = tmp_path / "source.txt"
source.write_text("original")
dest = tmp_path / "dest.txt"
result = _atomic_hardlink(source, dest)
# Modify source
source.write_text("modified")
# Hardlink should see the change
assert result.read_text() == "modified"
def test_hardlink_count_increases(self, tmp_path):
"""Link count increases with each hardlink."""
from cwa_book_downloader.download.orchestrator import _atomic_hardlink
source = tmp_path / "source.txt"
source.write_text("content")
# Initial link count is 1
assert os.stat(source).st_nlink == 1
dest1 = _atomic_hardlink(source, tmp_path / "link1.txt")
assert os.stat(source).st_nlink == 2
dest2 = _atomic_hardlink(source, tmp_path / "link2.txt")
assert os.stat(source).st_nlink == 3
class TestTorrentOptimization:
"""Tests for optimized torrent handling - skip staging when possible."""
@pytest.fixture
def sample_task(self):
from cwa_book_downloader.core.models import DownloadTask, SearchMode
return DownloadTask(
task_id="test123",
source="prowlarr",
title="Test Book",
author="Test Author",
format="epub",
search_mode=SearchMode.UNIVERSAL,
)
def test_is_torrent_source_true(self, tmp_path, sample_task):
"""Detects when source is the torrent client path."""
from cwa_book_downloader.download.orchestrator import _is_torrent_source
torrent_path = tmp_path / "downloads" / "book.epub"
torrent_path.parent.mkdir()
torrent_path.touch()
sample_task.original_download_path = str(torrent_path)
assert _is_torrent_source(torrent_path, sample_task) is True
def test_is_torrent_source_false_no_original(self, tmp_path, sample_task):
"""Returns False when no original_download_path set."""
from cwa_book_downloader.download.orchestrator import _is_torrent_source
some_path = tmp_path / "staging" / "book.epub"
sample_task.original_download_path = None
assert _is_torrent_source(some_path, sample_task) is False
def test_is_torrent_source_false_different_path(self, tmp_path, sample_task):
"""Returns False when paths don't match."""
from cwa_book_downloader.download.orchestrator import _is_torrent_source
torrent_path = tmp_path / "downloads" / "book.epub"
staging_path = tmp_path / "staging" / "book.epub"
sample_task.original_download_path = str(torrent_path)
assert _is_torrent_source(staging_path, sample_task) is False
def test_library_mode_torrent_no_hardlink_copies(self, tmp_path, sample_task):
"""Library mode copies (not moves) torrent files when hardlink unavailable."""
from cwa_book_downloader.download.orchestrator import _transfer_file_to_library
library = tmp_path / "library"
library.mkdir()
torrent_path = tmp_path / "downloads" / "book.epub"
torrent_path.parent.mkdir()
torrent_path.write_bytes(b"content")
# Set up as torrent source
sample_task.original_download_path = str(torrent_path)
status_cb = MagicMock()
result = _transfer_file_to_library(
source_path=torrent_path,
library_base=str(library),
template="{Author}/{Title}",
metadata={"Author": "Test Author", "Title": "Test Book"},
task=sample_task,
temp_file=torrent_path,
status_callback=status_cb,
use_hardlink=False, # No hardlink
)
assert result is not None
assert Path(result).exists()
# Original should still exist (copied, not moved)
assert torrent_path.exists()
def test_library_mode_non_torrent_moves(self, tmp_path, sample_task):
"""Library mode moves (not copies) non-torrent files."""
from cwa_book_downloader.download.orchestrator import _transfer_file_to_library
library = tmp_path / "library"
library.mkdir()
staging_path = tmp_path / "staging" / "book.epub"
staging_path.parent.mkdir()
staging_path.write_bytes(b"content")
# No original_download_path = not a torrent
sample_task.original_download_path = None
status_cb = MagicMock()
result = _transfer_file_to_library(
source_path=staging_path,
library_base=str(library),
template="{Author}/{Title}",
metadata={"Author": "Test Author", "Title": "Test Book"},
task=sample_task,
temp_file=staging_path,
status_callback=status_cb,
use_hardlink=False,
)
assert result is not None
assert Path(result).exists()
# Original should be gone (moved)
assert not staging_path.exists()
def test_directory_torrent_copies_all_files(self, tmp_path, sample_task):
"""Multi-file torrent directory copies all files to library."""
from cwa_book_downloader.download.orchestrator import _transfer_directory_to_library
library = tmp_path / "library"
library.mkdir()
torrent_dir = tmp_path / "downloads" / "audiobook"
torrent_dir.mkdir(parents=True)
(torrent_dir / "part1.mp3").write_bytes(b"audio1")
(torrent_dir / "part2.mp3").write_bytes(b"audio2")
sample_task.original_download_path = str(torrent_dir)
sample_task.content_type = "audiobook"
status_cb = MagicMock()
with patch('cwa_book_downloader.download.orchestrator._get_supported_formats', return_value=["mp3"]):
result = _transfer_directory_to_library(
source_dir=torrent_dir,
library_base=str(library),
template="{Author}/{Title}{ - PartNumber}",
metadata={"Author": "Test Author", "Title": "Test Book"},
task=sample_task,
temp_file=torrent_dir,
status_callback=status_cb,
use_hardlink=False,
)
assert result is not None
# Original files should still exist
assert (torrent_dir / "part1.mp3").exists()
assert (torrent_dir / "part2.mp3").exists()
# Library files should exist
author_dir = library / "Test Author"
assert len(list(author_dir.glob("*.mp3"))) == 2
class TestEdgeCases:
"""Edge cases and error handling."""
def test_empty_directory_returns_none(self, tmp_path):
"""Empty source directory returns None."""
from cwa_book_downloader.download.orchestrator import _transfer_directory_to_library
from cwa_book_downloader.core.models import DownloadTask, SearchMode
task = DownloadTask(
task_id="test",
source="prowlarr",
title="Test",
author="Author",
format="epub",
search_mode=SearchMode.UNIVERSAL,
)
library = tmp_path / "library"
library.mkdir()
source_dir = tmp_path / "empty"
source_dir.mkdir()
status_cb = MagicMock()
with patch('cwa_book_downloader.download.orchestrator._get_supported_formats', return_value=["epub"]):
result = _transfer_directory_to_library(
source_dir=source_dir,
library_base=str(library),
template="{Title}",
metadata={"Title": "Test"},
task=task,
temp_file=source_dir,
status_callback=status_cb,
use_hardlink=False,
)
assert result is None
def test_nonexistent_source_for_hardlink(self, tmp_path):
"""Missing source file prevents hardlink creation."""
from cwa_book_downloader.download.orchestrator import _process_library_mode
from cwa_book_downloader.core.models import DownloadTask, SearchMode
task = DownloadTask(
task_id="test",
source="prowlarr",
title="Test",
author="Author",
format="epub",
search_mode=SearchMode.UNIVERSAL,
original_download_path=str(tmp_path / "nonexistent.epub"),
)
library = tmp_path / "library"
library.mkdir()
staged = tmp_path / "staging" / "book.epub"
staged.parent.mkdir()
staged.write_bytes(b"content")
status_cb = MagicMock()
with patch('cwa_book_downloader.download.orchestrator.config') as mock_config:
mock_config.get = MagicMock(side_effect=lambda key, default=None: {
"LIBRARY_PATH": str(library),
"LIBRARY_TEMPLATE": "{Title}",
"TORRENT_HARDLINK": True,
"PROCESSING_MODE": "library",
}.get(key, default))
result = _process_library_mode(staged, task, status_cb)
# Should fall back to move since original doesn't exist
assert result is not None
assert not staged.exists()
def test_permission_denied_library_path(self, tmp_path):
"""Handles permission denied on library path."""
from cwa_book_downloader.download.orchestrator import _process_library_mode
from cwa_book_downloader.core.models import DownloadTask, SearchMode
task = DownloadTask(
task_id="test",
source="prowlarr",
title="Test",
author="Author",
format="epub",
search_mode=SearchMode.UNIVERSAL,
)
staged = tmp_path / "staging" / "book.epub"
staged.parent.mkdir()
staged.write_bytes(b"content")
status_cb = MagicMock()
with patch('cwa_book_downloader.download.orchestrator.config') as mock_config:
mock_config.get = MagicMock(side_effect=lambda key, default=None: {
"LIBRARY_PATH": "/nonexistent/protected/path",
"LIBRARY_TEMPLATE": "{Title}",
"PROCESSING_MODE": "library",
}.get(key, default))
result = _process_library_mode(staged, task, status_cb)
# Should return None (fall back to ingest)
assert result is None
File diff suppressed because it is too large Load Diff
+585
View File
@@ -0,0 +1,585 @@
"""
Tests for the naming module - template parsing and library path building.
"""
import pytest
from pathlib import Path
import tempfile
import os
import shutil
from cwa_book_downloader.core.naming import (
natural_sort_key,
assign_part_numbers,
parse_naming_template,
build_library_path,
sanitize_filename,
sanitize_path_component,
format_series_position,
)
class TestNaturalSortAndAssignment:
def test_natural_sort_simple_numbers(self):
files = ["Part 2.mp3", "Part 10.mp3", "Part 1.mp3"]
assert sorted(files, key=natural_sort_key) == ["Part 1.mp3", "Part 2.mp3", "Part 10.mp3"]
def test_natural_sort_cd_track_pattern(self):
files = ["CD2_Track10.mp3", "CD1_Track2.mp3", "CD1_Track10.mp3", "CD2_Track1.mp3"]
assert sorted(files, key=natural_sort_key) == [
"CD1_Track2.mp3", "CD1_Track10.mp3", "CD2_Track1.mp3", "CD2_Track10.mp3"
]
def test_assign_part_numbers_empty(self):
assert assign_part_numbers([]) == []
def test_assign_part_numbers_sorted(self):
files = [Path("Part 3.mp3"), Path("Part 1.mp3"), Path("Part 2.mp3")]
assert assign_part_numbers(files) == [
(Path("Part 1.mp3"), "01"), (Path("Part 2.mp3"), "02"), (Path("Part 3.mp3"), "03")
]
def test_assign_part_numbers_custom_padding(self):
files = [Path("a.mp3"), Path("b.mp3")]
assert assign_part_numbers(files, zero_pad_width=3) == [(Path("a.mp3"), "001"), (Path("b.mp3"), "002")]
def test_no_false_positives_fahrenheit_451(self):
files = [Path("Fahrenheit 451 - Part 2.mp3"), Path("Fahrenheit 451 - Part 1.mp3")]
result = assign_part_numbers(files)
assert result[0] == (Path("Fahrenheit 451 - Part 1.mp3"), "01")
class TestParseNamingTemplate:
"""Tests for template parsing with variable substitution."""
def test_simple_substitution(self):
"""Test basic token replacement."""
result = parse_naming_template(
"{Author}/{Title}",
{"Author": "Brandon Sanderson", "Title": "The Way of Kings"}
)
assert result == "Brandon Sanderson/The Way of Kings"
def test_conditional_suffix(self):
"""Test conditional suffix inclusion."""
template = "{Author}/{Series/}{Title}"
# With series
result = parse_naming_template(template, {
"Author": "Brandon Sanderson",
"Series": "Stormlight Archive",
"Title": "The Way of Kings"
})
assert result == "Brandon Sanderson/Stormlight Archive/The Way of Kings"
# Without series
result = parse_naming_template(template, {
"Author": "Brandon Sanderson",
"Series": None,
"Title": "The Way of Kings"
})
assert result == "Brandon Sanderson/The Way of Kings"
def test_conditional_prefix(self):
"""Test conditional prefix inclusion."""
template = "{Title}{ - Subtitle}"
# With subtitle
result = parse_naming_template(template, {
"Title": "The Way of Kings",
"Subtitle": "Journey Before Destination"
})
assert result == "The Way of Kings - Journey Before Destination"
# Without subtitle
result = parse_naming_template(template, {
"Title": "The Way of Kings",
"Subtitle": None
})
assert result == "The Way of Kings"
def test_subtitle_token(self):
"""Test subtitle in various template positions."""
metadata = {
"Author": "Brandon Sanderson",
"Title": "The Way of Kings",
"Subtitle": "Book One of the Stormlight Archive"
}
# Subtitle after title
result = parse_naming_template("{Author}/{Title} - {Subtitle}", metadata)
assert result == "Brandon Sanderson/The Way of Kings - Book One of the Stormlight Archive"
# Conditional subtitle
result = parse_naming_template("{Author}/{Title}{ - Subtitle}", metadata)
assert result == "Brandon Sanderson/The Way of Kings - Book One of the Stormlight Archive"
def test_part_number_token(self):
"""Test PartNumber in templates."""
metadata = {
"Author": "Brandon Sanderson",
"Title": "The Way of Kings",
"PartNumber": "01"
}
# Literal " - Part " in template
result = parse_naming_template("{Author}/{Title} - Part {PartNumber}", metadata)
assert result == "Brandon Sanderson/The Way of Kings - Part 01"
# Conditional prefix on PartNumber itself
result = parse_naming_template("{Author}/{Title}{ - PartNumber}", metadata)
assert result == "Brandon Sanderson/The Way of Kings - 01"
def test_part_number_without_value(self):
"""Test PartNumber when not provided."""
metadata = {
"Author": "Brandon Sanderson",
"Title": "The Way of Kings",
"PartNumber": None
}
# Conditional prefix: " - " only appears if PartNumber has value
result = parse_naming_template("{Author}/{Title}{ - PartNumber}", metadata)
assert result == "Brandon Sanderson/The Way of Kings"
def test_series_position(self):
"""Test series position formatting."""
template = "{SeriesPosition - }{Title}"
# Integer position
result = parse_naming_template(template, {"SeriesPosition": 1, "Title": "Book"})
assert result == "1 - Book"
# Float position (novella)
result = parse_naming_template(template, {"SeriesPosition": 1.5, "Title": "Book"})
assert result == "1.5 - Book"
# No position
result = parse_naming_template(template, {"SeriesPosition": None, "Title": "Book"})
assert result == "Book"
def test_year_token(self):
"""Test year in templates."""
result = parse_naming_template(
"{Author}/{Title} ({Year})",
{"Author": "Sanderson", "Title": "Book", "Year": 2010}
)
assert result == "Sanderson/Book (2010)"
def test_case_insensitive_tokens(self):
"""Test that token matching is case-insensitive."""
result = parse_naming_template(
"{author}/{TITLE}",
{"Author": "Sanderson", "Title": "Book"}
)
assert result == "Sanderson/Book"
def test_special_characters_sanitized(self):
"""Test that special characters are sanitized."""
result = parse_naming_template(
"{Author}/{Title}",
{"Author": "Author: Name", "Title": "Book: Subtitle?"}
)
assert ":" not in result
assert "?" not in result
def test_empty_template(self):
"""Test empty template."""
assert parse_naming_template("", {"Title": "Book"}) == ""
def test_empty_metadata(self):
"""Test with no metadata values."""
result = parse_naming_template("{Author}/{Title}", {})
assert result == ""
def test_complex_template(self):
"""Test complex template with multiple conditional tokens."""
template = "{Author}/{Series/}{SeriesPosition - }{Title}{ - Subtitle} ({Year})"
# All fields present
result = parse_naming_template(template, {
"Author": "Brandon Sanderson",
"Series": "Stormlight",
"SeriesPosition": 1,
"Title": "The Way of Kings",
"Subtitle": "Epic Fantasy",
"Year": 2010
})
assert result == "Brandon Sanderson/Stormlight/1 - The Way of Kings - Epic Fantasy (2010)"
# Minimal fields
result = parse_naming_template(template, {
"Author": "Brandon Sanderson",
"Title": "The Way of Kings"
})
assert result == "Brandon Sanderson/The Way of Kings"
class TestBuildLibraryPath:
"""Tests for complete library path building."""
def test_basic_path(self):
"""Test basic path building."""
path = build_library_path(
"/books",
"{Author}/{Title}",
{"Author": "Sanderson", "Title": "Book"},
extension="epub"
)
assert path == Path("/books/Sanderson/Book.epub")
def test_path_with_subtitle(self):
"""Test path with subtitle."""
path = build_library_path(
"/books",
"{Author}/{Title}{ - Subtitle}",
{"Author": "Sanderson", "Title": "Book", "Subtitle": "A Novel"},
extension="epub"
)
assert path == Path("/books/Sanderson/Book - A Novel.epub")
def test_path_with_part_number(self):
"""Test path with part number for audiobooks."""
path = build_library_path(
"/audiobooks",
"{Author}/{Title} - Part {PartNumber}",
{"Author": "Sanderson", "Title": "Book", "PartNumber": "01"},
extension="mp3"
)
assert path == Path("/audiobooks/Sanderson/Book - Part 01.mp3")
def test_path_traversal_prevented(self):
"""Test that path traversal is prevented."""
with pytest.raises(ValueError, match="traversal"):
build_library_path(
"/books",
"{Author}/{Title}",
{"Author": "../../../etc", "Title": "passwd"},
extension="txt"
)
def test_fallback_to_title(self):
"""Test fallback when template produces empty result."""
path = build_library_path(
"/books",
"{Series/}{Title}",
{"Title": "Book"},
extension="epub"
)
assert "Book" in str(path)
def test_no_extension(self):
"""Test path without extension."""
path = build_library_path(
"/books",
"{Author}/{Title}",
{"Author": "Sanderson", "Title": "Book"},
extension=None
)
assert path == Path("/books/Sanderson/Book")
class TestSanitizeFilename:
"""Tests for filename sanitization."""
@pytest.mark.parametrize("input_name,expected", [
("normal_file", "normal_file"),
("file:with:colons", "file_with_colons"),
("file*with*stars", "file_with_stars"),
("file?with?questions", "file_with_questions"),
('file"with"quotes', "file_with_quotes"),
("file<with>angles", "file_with_angles"),
("file|with|pipes", "file_with_pipes"),
])
def test_invalid_chars_replaced(self, input_name, expected):
"""Test that invalid characters are replaced."""
assert sanitize_filename(input_name) == expected
def test_leading_trailing_stripped(self):
"""Test that leading/trailing whitespace and dots are stripped."""
assert sanitize_filename(" file ") == "file"
assert sanitize_filename("...file...") == "file"
assert sanitize_filename(". file .") == "file"
def test_multiple_underscores_collapsed(self):
"""Test that multiple underscores are collapsed."""
assert sanitize_filename("file___name") == "file_name"
def test_max_length_enforced(self):
"""Test that max length is enforced."""
long_name = "a" * 300
result = sanitize_filename(long_name, max_length=100)
assert len(result) == 100
def test_empty_string(self):
"""Test empty string handling."""
assert sanitize_filename("") == ""
assert sanitize_filename(None) == ""
class TestFormatSeriesPosition:
"""Tests for series position formatting."""
def test_integer_position(self):
"""Test integer positions."""
assert format_series_position(1) == "1"
assert format_series_position(10) == "10"
def test_float_integer_position(self):
"""Test float that's effectively an integer."""
assert format_series_position(1.0) == "1"
assert format_series_position(5.0) == "5"
def test_float_position(self):
"""Test actual float positions (novellas)."""
assert format_series_position(1.5) == "1.5"
assert format_series_position(2.3) == "2.3"
def test_none_position(self):
"""Test None handling."""
assert format_series_position(None) == ""
class TestIntegration:
"""Integration tests for complete library path workflows."""
def test_audiobook_multi_part_workflow(self):
"""Test complete workflow for multi-part audiobook."""
base_metadata = {
"Author": "Brandon Sanderson",
"Title": "The Way of Kings",
"Subtitle": "Stormlight Archive Book 1",
"Year": 2010,
"Series": "Stormlight Archive",
"SeriesPosition": 1,
}
# Use literal " - Part " text for part numbers
template = "{Author}/{Series/}{Title} - Part {PartNumber}"
# Simulate processing multiple files (unsorted order)
files = [
Path("The Way of Kings - Part 03.mp3"),
Path("The Way of Kings - Part 01.mp3"),
Path("The Way of Kings - Part 02.mp3"),
]
# Use assign_part_numbers to sort and number sequentially
files_with_parts = assign_part_numbers(files)
for file_path, part_num in files_with_parts:
file_metadata = {**base_metadata, "PartNumber": part_num}
path = build_library_path("/audiobooks", template, file_metadata, extension="mp3")
assert "Brandon Sanderson" in str(path)
assert "Stormlight Archive" in str(path)
assert "The Way of Kings" in str(path)
assert f"Part {part_num}" in str(path)
def test_ebook_with_subtitle_workflow(self):
"""Test complete workflow for ebook with subtitle."""
metadata = {
"Author": "Frank Herbert",
"Title": "Dune",
"Subtitle": "Deluxe Edition",
"Year": 1965,
}
template = "{Author}/{Title}{ - Subtitle} ({Year})"
path = build_library_path("/books", template, metadata, extension="epub")
assert path == Path("/books/Frank Herbert/Dune - Deluxe Edition (1965).epub")
def test_ebook_without_subtitle_workflow(self):
"""Test workflow for ebook without subtitle."""
metadata = {
"Author": "Frank Herbert",
"Title": "Dune",
"Year": 1965,
}
template = "{Author}/{Title}{ - Subtitle} ({Year})"
path = build_library_path("/books", template, metadata, extension="epub")
assert path == Path("/books/Frank Herbert/Dune (1965).epub")
class TestFilesystemOperations:
"""Tests for actual filesystem operations - folder creation and file handling."""
@pytest.fixture
def temp_library(self):
"""Create a temporary library directory."""
temp_dir = tempfile.mkdtemp(prefix="test_library_")
yield Path(temp_dir)
shutil.rmtree(temp_dir, ignore_errors=True)
def test_creates_new_folder_structure(self, temp_library):
"""Test that new folder structure is created correctly."""
metadata = {"Author": "Brandon Sanderson", "Title": "Mistborn"}
template = "{Author}/{Title}"
path = build_library_path(str(temp_library), template, metadata, extension="epub")
# Create the directory structure
path.parent.mkdir(parents=True, exist_ok=True)
# Path is: temp_library/Brandon Sanderson/Mistborn.epub
assert path.parent.exists()
assert path.parent.name == "Brandon Sanderson"
assert path.name == "Mistborn.epub"
def test_adds_file_to_existing_folder(self, temp_library):
"""Test that files can be added to existing folders."""
# Create existing author folder with a book
author_dir = temp_library / "Brandon Sanderson"
author_dir.mkdir(parents=True)
existing_book = author_dir / "Elantris.epub"
existing_book.write_text("existing book content")
# Add a new book to the same author folder
metadata = {"Author": "Brandon Sanderson", "Title": "Mistborn"}
template = "{Author}/{Title}"
path = build_library_path(str(temp_library), template, metadata, extension="epub")
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("new book content")
# Both books should exist
assert existing_book.exists()
assert path.exists()
assert len(list(author_dir.iterdir())) == 2
def test_adds_multiple_parts_to_same_folder(self, temp_library):
"""Test adding multiple audiobook parts to the same folder."""
base_metadata = {
"Author": "Brandon Sanderson",
"Title": "The Way of Kings",
"Series": "Stormlight Archive",
}
template = "{Author}/{Series}/{Title} - Part {PartNumber}"
parts = ["01", "02", "03"]
created_files = []
for part_num in parts:
file_metadata = {**base_metadata, "PartNumber": part_num}
path = build_library_path(str(temp_library), template, file_metadata, extension="mp3")
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(f"part {part_num} content")
created_files.append(path)
# All files should exist in the same folder
for f in created_files:
assert f.exists()
# All files should be in the same directory
parent_dir = created_files[0].parent
assert all(f.parent == parent_dir for f in created_files)
assert len(list(parent_dir.iterdir())) == 3
def test_nested_series_folder_structure(self, temp_library):
"""Test creating deeply nested folder structures for series."""
metadata = {
"Author": "Brandon Sanderson",
"Series": "Cosmere/Stormlight Archive",
"Title": "The Way of Kings",
}
template = "{Author}/{Series/}{Title}"
path = build_library_path(str(temp_library), template, metadata, extension="epub")
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("content")
assert path.exists()
# Check the nested structure
assert "Brandon Sanderson" in str(path)
assert "Cosmere" in str(path)
assert "Stormlight Archive" in str(path)
def test_file_collision_detection(self, temp_library):
"""Test behavior when a file with the same name already exists."""
metadata = {"Author": "Brandon Sanderson", "Title": "Mistborn"}
template = "{Author}/{Title}"
path = build_library_path(str(temp_library), template, metadata, extension="epub")
path.parent.mkdir(parents=True, exist_ok=True)
# Create first file
path.write_text("original content")
assert path.exists()
# Build same path again - path building should still work
path2 = build_library_path(str(temp_library), template, metadata, extension="epub")
assert path == path2
# Note: The actual collision handling (overwrite, rename, skip)
# is done in the orchestrator, not in build_library_path
def test_special_characters_in_folder_names(self, temp_library):
"""Test that special characters are sanitized in folder names."""
metadata = {
"Author": "Author: With Colons",
"Title": "Book? With <Special> Characters*"
}
template = "{Author}/{Title}"
path = build_library_path(str(temp_library), template, metadata, extension="epub")
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("content")
assert path.exists()
# Verify no invalid characters in path
path_str = str(path)
for char in ':*?"<>|':
assert char not in path_str
def test_empty_series_skips_folder(self, temp_library):
"""Test that empty series doesn't create empty folder level."""
metadata = {
"Author": "Brandon Sanderson",
"Title": "Elantris",
"Series": None,
}
template = "{Author}/{Series/}{Title}"
path = build_library_path(str(temp_library), template, metadata, extension="epub")
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("content")
# Should be Author/Title.epub, not Author//Title.epub or Author/None/Title.epub
assert path.exists()
assert path.parent.name == "Brandon Sanderson"
assert path.name == "Elantris.epub"
def test_unicode_in_folder_names(self, temp_library):
"""Test that unicode characters work in folder names."""
metadata = {
"Author": "Андрей Сапковский", # Cyrillic
"Title": "Ведьмак", # Cyrillic
}
template = "{Author}/{Title}"
path = build_library_path(str(temp_library), template, metadata, extension="epub")
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("content")
assert path.exists()
assert "Андрей Сапковский" in str(path)
def test_very_long_path_components(self, temp_library):
"""Test that very long names are truncated."""
long_title = "A" * 300 # Longer than typical filesystem limits
metadata = {
"Author": "Author",
"Title": long_title,
}
template = "{Author}/{Title}"
path = build_library_path(str(temp_library), template, metadata, extension="epub")
# Path should be buildable without error
assert path is not None
# Title component should be truncated
assert len(path.stem) < 250
+169
View File
@@ -0,0 +1,169 @@
"""Tests for natural sort and sequential part number assignment."""
import pytest
from pathlib import Path
from cwa_book_downloader.core.naming import natural_sort_key, assign_part_numbers
class TestNaturalSortKey:
def test_simple_numbers(self):
files = ["Part 2.mp3", "Part 10.mp3", "Part 1.mp3"]
assert sorted(files, key=natural_sort_key) == ["Part 1.mp3", "Part 2.mp3", "Part 10.mp3"]
def test_leading_zeros(self):
files = ["Track 01.mp3", "Track 10.mp3", "Track 02.mp3"]
assert sorted(files, key=natural_sort_key) == ["Track 01.mp3", "Track 02.mp3", "Track 10.mp3"]
def test_case_insensitive(self):
files = ["PART 2.mp3", "part 1.mp3", "Part 3.mp3"]
assert sorted(files, key=natural_sort_key) == ["part 1.mp3", "PART 2.mp3", "Part 3.mp3"]
def test_multiple_numbers_in_filename(self):
files = ["CD2_Track10.mp3", "CD1_Track2.mp3", "CD1_Track10.mp3", "CD2_Track1.mp3"]
assert sorted(files, key=natural_sort_key) == [
"CD1_Track2.mp3", "CD1_Track10.mp3", "CD2_Track1.mp3", "CD2_Track10.mp3"
]
def test_no_numbers(self):
files = ["charlie.mp3", "alpha.mp3", "bravo.mp3"]
assert sorted(files, key=natural_sort_key) == ["alpha.mp3", "bravo.mp3", "charlie.mp3"]
def test_path_objects(self):
files = [Path("file10.mp3"), Path("file2.mp3"), Path("file1.mp3")]
assert [f.name for f in sorted(files, key=natural_sort_key)] == ["file1.mp3", "file2.mp3", "file10.mp3"]
def test_uses_filename_only(self):
files = [Path("/z/dir/file1.mp3"), Path("/a/dir/file2.mp3")]
sorted_files = sorted(files, key=natural_sort_key)
assert sorted_files[0].name == "file1.mp3"
class TestAssignPartNumbers:
def test_empty_list(self):
assert assign_part_numbers([]) == []
def test_single_file(self):
assert assign_part_numbers([Path("book.mp3")]) == [(Path("book.mp3"), "01")]
def test_multiple_files_sorted(self):
files = [Path("Part 3.mp3"), Path("Part 1.mp3"), Path("Part 2.mp3")]
assert assign_part_numbers(files) == [
(Path("Part 1.mp3"), "01"),
(Path("Part 2.mp3"), "02"),
(Path("Part 3.mp3"), "03"),
]
def test_natural_sort_applied(self):
files = [Path("Chapter 10.mp3"), Path("Chapter 2.mp3"), Path("Chapter 1.mp3")]
assert assign_part_numbers(files) == [
(Path("Chapter 1.mp3"), "01"),
(Path("Chapter 2.mp3"), "02"),
(Path("Chapter 10.mp3"), "03"),
]
def test_custom_zero_padding(self):
files = [Path("a.mp3"), Path("b.mp3")]
assert assign_part_numbers(files, zero_pad_width=3) == [(Path("a.mp3"), "001"), (Path("b.mp3"), "002")]
def test_many_files_padding(self):
files = [Path(f"track_{i}.mp3") for i in range(100, 0, -1)]
result = assign_part_numbers(files, zero_pad_width=3)
assert result[0] == (Path("track_1.mp3"), "001")
assert result[-1] == (Path("track_100.mp3"), "100")
class TestRealWorldScenarios:
def test_standard_part_naming(self):
files = [
Path("The Way of Kings - Part 02.mp3"),
Path("The Way of Kings - Part 01.mp3"),
Path("The Way of Kings - Part 10.mp3"),
Path("The Way of Kings - Part 03.mp3"),
]
result = assign_part_numbers(files)
assert [r[0].name for r in result] == [
"The Way of Kings - Part 01.mp3",
"The Way of Kings - Part 02.mp3",
"The Way of Kings - Part 03.mp3",
"The Way of Kings - Part 10.mp3",
]
def test_cd_track_naming(self):
files = [Path("CD02_Track01.mp3"), Path("CD01_Track02.mp3"), Path("CD01_Track01.mp3"), Path("CD02_Track02.mp3")]
result = assign_part_numbers(files)
assert [r[0].name for r in result] == [
"CD01_Track01.mp3", "CD01_Track02.mp3", "CD02_Track01.mp3", "CD02_Track02.mp3"
]
def test_disc_track_naming(self):
files = [Path("Disc 1 - Track 10.mp3"), Path("Disc 1 - Track 2.mp3"), Path("Disc 2 - Track 1.mp3")]
result = assign_part_numbers(files)
assert [r[0].name for r in result] == [
"Disc 1 - Track 2.mp3", "Disc 1 - Track 10.mp3", "Disc 2 - Track 1.mp3"
]
def test_simple_numbered_files(self):
files = [Path("02 Chapter Two.mp3"), Path("01 Chapter One.mp3"), Path("10 Chapter Ten.mp3")]
result = assign_part_numbers(files)
assert [r[0].name for r in result] == ["01 Chapter One.mp3", "02 Chapter Two.mp3", "10 Chapter Ten.mp3"]
def test_bracketed_numbers(self):
files = [Path("Book Title [03].mp3"), Path("Book Title [01].mp3"), Path("Book Title [02].mp3")]
result = assign_part_numbers(files)
assert [r[0].name for r in result] == ["Book Title [01].mp3", "Book Title [02].mp3", "Book Title [03].mp3"]
class TestNoFalsePositives:
"""Titles with numbers (451, 1984, etc.) don't cause issues with sequential assignment."""
def test_fahrenheit_451(self):
files = [Path("Fahrenheit 451 - Part 2.mp3"), Path("Fahrenheit 451 - Part 1.mp3")]
result = assign_part_numbers(files)
assert result[0] == (Path("Fahrenheit 451 - Part 1.mp3"), "01")
assert result[1] == (Path("Fahrenheit 451 - Part 2.mp3"), "02")
def test_1984(self):
files = [Path("1984 - Chapter 03.mp3"), Path("1984 - Chapter 01.mp3"), Path("1984 - Chapter 02.mp3")]
result = assign_part_numbers(files)
assert [r[0].name for r in result] == ["1984 - Chapter 01.mp3", "1984 - Chapter 02.mp3", "1984 - Chapter 03.mp3"]
def test_catch_22(self):
files = [Path("Catch-22 Part 2.mp3"), Path("Catch-22 Part 1.mp3")]
result = assign_part_numbers(files)
assert result[0] == (Path("Catch-22 Part 1.mp3"), "01")
def test_2001_a_space_odyssey(self):
files = [Path("2001 A Space Odyssey - 02.mp3"), Path("2001 A Space Odyssey - 01.mp3")]
result = assign_part_numbers(files)
assert result[0][0].name == "2001 A Space Odyssey - 01.mp3"
class TestEdgeCases:
def test_identical_filenames_different_dirs(self):
files = [Path("/dir2/track.mp3"), Path("/dir1/track.mp3")]
result = assign_part_numbers(files)
assert len(result) == 2
def test_unicode_filenames(self):
files = [Path("日本語タイトル 02.mp3"), Path("日本語タイトル 01.mp3")]
result = assign_part_numbers(files)
assert result[0][0].name == "日本語タイトル 01.mp3"
def test_very_large_numbers(self):
files = [Path("track_1000.mp3"), Path("track_100.mp3"), Path("track_10.mp3")]
result = assign_part_numbers(files)
assert [r[0].name for r in result] == ["track_10.mp3", "track_100.mp3", "track_1000.mp3"]
def test_mixed_extensions(self):
files = [Path("track_2.m4b"), Path("track_1.mp3"), Path("track_3.flac")]
result = assign_part_numbers(files)
assert [r[0].name for r in result] == ["track_1.mp3", "track_2.m4b", "track_3.flac"]
def test_no_numbers_alphabetical(self):
files = [Path("zebra.mp3"), Path("apple.mp3"), Path("mango.mp3")]
result = assign_part_numbers(files)
assert [r[0].name for r in result] == ["apple.mp3", "mango.mp3", "zebra.mp3"]