starred/shelfmark
1
0
Fork 0
mirror of https://github.com/calibrain/shelfmark.git synced 2026-10-05 12:31:11 +01:00
Code Issues Packages Projects Releases 10 Wiki Activity

refactor: make direct download provider-driven (#1337)

Browse Source 
This is the refactor for the download handler
This commit is contained in:
Tom
2026-09-14 01:16:53 -04:00
committed by GitHub
parent af21d1da1f
commit a5cd9f0bfb
15 changed files with 1261 additions and 621 deletions
Download Patch File Download Diff File Expand all files Collapse all files
shelfmark/release_sources/direct_download/__init__.py
+20
View File
@@ -0,0 +1,20 @@
"""Direct Download release source and public entry points.
Importing the source and handler classes registers them with Shelfmark.
"""
from shelfmark.release_sources.direct_download.annas_archive import search_books
from shelfmark.release_sources.direct_download.common import DirectDownloadUnavailableError
from shelfmark.release_sources.direct_download.handler import DirectDownloadHandler
from shelfmark.release_sources.direct_download.source import DirectDownloadSource
__all__ = [
"DirectDownloadUnavailableError",
"DirectDownloadHandler",
"DirectDownloadSource",
"SearchUnavailableError",
"search_books",
]
# Compatibility alias for integrations that imported the old module-level name.
SearchUnavailableError = DirectDownloadUnavailableError
shelfmark/release_sources/direct_download.py → shelfmark/release_sources/direct_download/annas_archive.py
+130 -435
View File
@@ -1,9 +1,8 @@
"""Direct download source - Anna's Archive/Libgen with fallback cascade."""
"""Anna's Archive search, metadata parsing, and MD5 mirror download cascade."""
import itertools
import json
import re
import threading
import time
import unicodedata
from contextlib import contextmanager
@@ -11,7 +10,7 @@ from contextvars import ContextVar
from dataclasses import replace
from http import HTTPStatus
from pathlib import Path
from typing import TYPE_CHECKING, ClassVar, NoReturn, TypedDict
from typing import TYPE_CHECKING, NoReturn, TypedDict
from urllib.parse import quote, urlparse
import requests
@@ -19,30 +18,33 @@ from bs4 import BeautifulSoup, Tag
from bs4.element import NavigableString
from shelfmark.bypass.challenge import MAX_CHALLENGE_HTML_CHARS, challenge_marker
from shelfmark.config.env import DEBUG_SKIP_SOURCES, TMP_DIR
from shelfmark.config.env import DEBUG_SKIP_SOURCES
from shelfmark.core import search_deadline
from shelfmark.core.config import config
from shelfmark.core.languages import language_alias_map
from shelfmark.core.logger import setup_logger
from shelfmark.core.models import DownloadTask, SearchFilters, build_filename
from shelfmark.core.utils import CONTENT_TYPES, get_aa_content_type_dir
from shelfmark.core.utils import is_audiobook as check_audiobook
from shelfmark.core.models import SearchFilters
from shelfmark.core.utils import CONTENT_TYPES
from shelfmark.download import http as downloader
from shelfmark.download import network
from shelfmark.release_sources import (
BrowseRecord,
ColumnAlign,
ColumnColorHint,
ColumnRenderType,
ColumnSchema,
DownloadHandler,
Release,
ReleaseColumnConfig,
ReleaseProtocol,
ReleaseSource,
SourceUnavailableError,
register_handler,
register_source,
from shelfmark.release_sources import BrowseRecord
from shelfmark.release_sources.direct_download.common import (
MIN_VALID_FILE_SIZE as _MIN_VALID_FILE_SIZE,
)
from shelfmark.release_sources.direct_download.common import (
DirectDownloadUnavailableError,
ParsedSearchResult,
get_attr,
get_supported_formats,
html_response_text,
language_alias_to_code,
normalize_language_token,
normalize_requested_languages,
normalize_size,
parse_search_items,
parse_search_page,
)
from shelfmark.release_sources.direct_download.common import (
book_matches_requested_languages as _book_matches_requested_languages,
)
if TYPE_CHECKING:
@@ -67,18 +69,6 @@ def _raise_runtime_error(message: str) -> NoReturn:
raise RuntimeError(message)
def _coerce_str_list(value: object) -> list[str]:
"""Return only string items from a config value."""
if not isinstance(value, list | tuple):
return []
return [item for item in value if isinstance(item, str)]
def _get_supported_formats() -> list[str]:
"""Return configured supported formats as a clean string list."""
return _coerce_str_list(config.SUPPORTED_FORMATS)
def _parse_source_priority_entries(
value: object,
*,
@@ -108,13 +98,6 @@ def _parse_source_priority_entries(
return entries
def _html_response_text(response: str | tuple[str, str]) -> str:
"""Extract the HTML body from downloader responses."""
if isinstance(response, tuple):
return response[0]
return response
def _html_response_url(response: str | tuple[str, str]) -> str | None:
"""The URL that actually answered, when the downloader was asked to report it.
@@ -125,22 +108,6 @@ def _html_response_url(response: str | tuple[str, str]) -> str | None:
return None
def _attr_to_str(value: object) -> str | None:
"""Convert a BeautifulSoup attribute value to a plain string."""
if isinstance(value, str):
return value
if isinstance(value, list):
for item in value:
if isinstance(item, str):
return item
return None
def _get_attr(tag: Tag, attr: str) -> str | None:
"""Safely fetch a tag attribute as a string."""
return _attr_to_str(tag.get(attr))
def _first_stripped_text(tag: Tag | None) -> str | None:
"""Return the first non-empty stripped string from a tag."""
if tag is None:
@@ -212,9 +179,9 @@ _DOWNLOAD_SOURCES = [
]
_SOURCE_FAILURE_THRESHOLD = 4
_MIN_VALID_FILE_SIZE = 10 * 1024
_AA_COUNTDOWN_MAX_SECONDS = 300
# --- Distant-path language detection ---
_DISTANT_PATH_EXTENSIONS = (
@@ -252,8 +219,6 @@ _LANGUAGE_CODE_TOKEN_PATTERN = re.compile(
r"(?:^|[\s_./\\\-\[(])([A-Za-z]{2,3})(?=$|[\s_./\\\-)\]])"
)
_LANGUAGE_NAME_TOKEN_PATTERN = re.compile(r"[a-z]{4,}(?:-[a-z0-9]+)?")
_LANGUAGE_ALIAS_TO_CODE: dict[str, str] | None = None
_LANGUAGE_ALIAS_LOCK = threading.Lock()
_LANGUAGE_PLACEHOLDERS = frozenset({"", "-", "--", "unknown", "unk", "n/a", "na"})
# Short codes that appear in common words — require bracket/key context to accept
_AMBIGUOUS_SHORT_LANGUAGE_CODES = frozenset({"de", "en", "it", "la", "no", "or", "is", "in"})
@@ -269,34 +234,11 @@ def _is_language_from_path_enabled() -> bool:
return bool(config.get("DIRECT_DOWNLOAD_LANGUAGE_FROM_PATH", False))
def _normalize_language_token(value: str) -> str:
normalized = value.strip().lower()
for dash in ("‑", "–", "—", "−"):
normalized = normalized.replace(dash, "-")
return normalized
def _fold_text(value: str) -> str:
normalized = unicodedata.normalize("NFKD", value)
return "".join(c for c in normalized if not unicodedata.combining(c)).lower()
def _language_alias_to_code() -> dict[str, str]:
"""Alias to code map, delegating to the shared language data."""
global _LANGUAGE_ALIAS_TO_CODE
cached = _LANGUAGE_ALIAS_TO_CODE
if cached is not None:
return cached
with _LANGUAGE_ALIAS_LOCK:
cached = _LANGUAGE_ALIAS_TO_CODE
if cached is not None:
return cached
_LANGUAGE_ALIAS_TO_CODE = language_alias_map()
return _LANGUAGE_ALIAS_TO_CODE
def _extract_distant_path(row: Tag, *, enabled: bool) -> str | None:
"""Extract the Windows-style file path from an AA search result row."""
if not enabled:
@@ -342,7 +284,7 @@ def _detect_language_from_distant_path(path: str | None) -> str | None:
if not path:
return None
aliases = _language_alias_to_code()
aliases = language_alias_to_code()
if not aliases:
return None
@@ -350,12 +292,12 @@ def _detect_language_from_distant_path(path: str | None) -> str | None:
strong_candidates: list[str] = []
for code in _BRACKETED_LANGUAGE_CODE_PATTERN.findall(path):
normalized = _normalize_language_token(code)
normalized = normalize_language_token(code)
if normalized in aliases:
strong_candidates.append(aliases[normalized])
for code in _KEYED_LANGUAGE_CODE_PATTERN.findall(path):
normalized = _normalize_language_token(code)
normalized = normalize_language_token(code)
if normalized in aliases:
strong_candidates.append(aliases[normalized])
@@ -364,7 +306,7 @@ def _detect_language_from_distant_path(path: str | None) -> str | None:
return non_ambiguous[0]
for token in _LANGUAGE_NAME_TOKEN_PATTERN.findall(folded_path):
normalized = _normalize_language_token(token)
normalized = normalize_language_token(token)
if normalized in aliases:
candidate = aliases[normalized]
if candidate not in _AMBIGUOUS_SHORT_LANGUAGE_CODES:
@@ -374,7 +316,7 @@ def _detect_language_from_distant_path(path: str | None) -> str | None:
return strong_candidates[0]
for code in _LANGUAGE_CODE_TOKEN_PATTERN.findall(path):
normalized = _normalize_language_token(code)
normalized = normalize_language_token(code)
if normalized in _AMBIGUOUS_SHORT_LANGUAGE_CODES:
continue
if normalized in aliases:
@@ -386,38 +328,7 @@ def _detect_language_from_distant_path(path: str | None) -> str | None:
def _is_missing_or_placeholder_language(language: str | None) -> bool:
if language is None:
return True
return _normalize_language_token(language) in _LANGUAGE_PLACEHOLDERS
def _normalize_requested_languages(languages: list[str] | None) -> set[str]:
if not languages:
return set()
aliases = _language_alias_to_code()
normalized: set[str] = set()
for value in languages:
token = _normalize_language_token(str(value))
if not token or token == "all": # noqa: S105 - "all" is a language sentinel
continue
normalized.add(aliases.get(token, token))
return normalized
def _book_matches_requested_languages(book_language: str | None, requested: set[str]) -> bool:
"""Return True when a book's language matches the requested filter.
Books with unknown/missing language always pass — the server-side &lang= filter
already narrowed the result set, so dropping unlabelled rows hides valid results.
"""
if not requested:
return True
if not book_language:
return True
aliases = _language_alias_to_code()
normalized_book = aliases.get(
_normalize_language_token(book_language),
_normalize_language_token(book_language),
)
return normalized_book in requested
return normalize_language_token(language) in _LANGUAGE_PLACEHOLDERS
def _is_configured_zlib_link(url: str) -> bool:
@@ -516,7 +427,7 @@ def _is_source_enabled(source_id: str) -> bool:
return False
def _get_direct_download_unavailable_reason() -> str | None:
def get_unavailable_reason() -> str | None:
"""Return a user-facing reason when Direct Download cannot be used."""
from shelfmark.core import mirrors
@@ -534,23 +445,14 @@ def _get_direct_download_unavailable_reason() -> str | None:
return None
def _ensure_direct_download_available() -> None:
def ensure_available() -> None:
"""Raise a source-unavailable error when Direct Download is disabled or unconfigured."""
reason = _get_direct_download_unavailable_reason()
reason = get_unavailable_reason()
if reason:
raise SearchUnavailableError(reason)
_SIZE_UNIT_PATTERN = re.compile(r"(kb|mb|gb|tb)", re.IGNORECASE)
def _normalize_size(size_str: str) -> str:
"""Normalize size string by uppercasing units (e.g., '5.2 mb' -> '5.2 MB')."""
return _SIZE_UNIT_PATTERN.sub(lambda m: m.group(1).upper(), size_str.strip())
class SearchUnavailableError(SourceUnavailableError):
"""Raised when Anna's Archive cannot be reached via any mirror/DNS."""
SearchUnavailableError = DirectDownloadUnavailableError
# Markers that prove a 200 really came from Anna's Archive, and markers that mean we
@@ -599,7 +501,7 @@ _search_page_cache: ContextVar[dict[str, tuple[str, Tag | None]] | None] = Conte
@contextmanager
def _search_page_reuse() -> Iterator[None]:
def search_page_reuse() -> Iterator[None]:
"""Fetch each distinct AA search URL at most once per search.
One search asks AA for the same URL more than once. The language-filter retry in
@@ -717,7 +619,7 @@ def _fetch_search_table_uncached(
allow_bypasser_fallback=True,
include_response_url=True,
)
html = _html_response_text(response)
html = html_response_text(response)
# Checked on the body, not on `response`: with include_response_url the give-up
# shape is the tuple ("", url), and a tuple is truthy.
if not html:
@@ -808,7 +710,7 @@ def search_books(query: str, filters: SearchFilters) -> list[BrowseRecord]:
filters_query = ""
path_language_enabled = _is_language_from_path_enabled()
requested_langs = _normalize_requested_languages(filters.lang)
requested_langs = normalize_requested_languages(filters.lang)
# When path-language inference is on and a language is requested, skip the
# server-side &lang= filter: lgli files often have no AA language metadata
@@ -826,7 +728,7 @@ def search_books(query: str, filters: SearchFilters) -> list[BrowseRecord]:
for value in filters.content:
filters_query += f"&content={quote(value)}"
formats_to_use = filters.format or _get_supported_formats()
formats_to_use = filters.format or get_supported_formats()
index = 1
for filter_type, filter_values in vars(filters).items():
@@ -860,16 +762,18 @@ def search_books(query: str, filters: SearchFilters) -> list[BrowseRecord]:
msg = f"Expected results table tag, got {type(tbody).__name__}"
raise TypeError(msg)
books = []
for line_tr in tbody.find_all("tr"):
book = _parse_search_result_row(line_tr)
if book:
books.append(book)
books = parse_search_page(
tbody,
filters,
provider_id="annas_archive",
item_selector="tr",
extract_item=_extract_aa_search_result,
)
if path_language_enabled and requested_langs:
books = [b for b in books if _book_matches_requested_languages(b.language, requested_langs)]
supported_formats = _get_supported_formats()
supported_formats = get_supported_formats()
books.sort(
key=lambda x: (
@@ -905,13 +809,13 @@ def get_book_info(book_id: str, *, fetch_download_count: bool = True) -> BrowseR
)
raise SearchUnavailableError(f"Unable to reach download source. {detail}")
soup = BeautifulSoup(_html_response_text(html), "html.parser")
soup = BeautifulSoup(html_response_text(html), "html.parser")
return _parse_book_info_page(soup, book_id, fetch_download_count=fetch_download_count)
def _parse_search_result_row(row: Tag) -> BrowseRecord | None:
"""Parse a single search result row into a browse record."""
def _extract_aa_search_result(row: Tag) -> ParsedSearchResult | None:
"""Extract Anna's Archive table fields for the shared parser."""
try:
if row.text.strip().lower().startswith("your ad here"):
return None
@@ -921,7 +825,7 @@ def _parse_search_result_row(row: Tag) -> BrowseRecord | None:
if len(cells) < 11 or not anchors:
return None
record_id = (_get_attr(anchors[0], "href") or "").split("/")[-1]
record_id = (get_attr(anchors[0], "href") or "").split("/")[-1]
if not record_id:
return None
@@ -929,7 +833,7 @@ def _parse_search_result_row(row: Tag) -> BrowseRecord | None:
distant_path = _extract_distant_path(row, enabled=path_language_enabled)
preview_img = cells[0].find("img")
preview = _get_attr(preview_img, "src") if isinstance(preview_img, Tag) else None
preview = get_attr(preview_img, "src") if isinstance(preview_img, Tag) else None
title_span = cells[1].find("span")
if isinstance(title_span, Tag):
@@ -963,25 +867,37 @@ def _parse_search_result_row(row: Tag) -> BrowseRecord | None:
detected = _detect_language_from_distant_path(distant_path)
language = detected or "unknown"
return BrowseRecord(
id=record_id,
return ParsedSearchResult(
key=record_id,
record_id=record_id,
title=title,
source="direct_download",
formats=(file_format.lower(),),
preview=preview,
author=author,
publisher=publisher,
year=year,
language=language,
content=content.lower() if content else None,
format=file_format.lower() if file_format else None,
size=size,
download_path=distant_path,
source_url=f"{network.get_aa_base_url()}/md5/{record_id}",
)
except (AttributeError, IndexError, KeyError, TypeError) as e:
logger.error_trace(f"Error parsing search result row: {e}")
return None
def _parse_search_result_row(row: Tag) -> BrowseRecord | None:
"""Compatibility wrapper for parsing one Anna's Archive result row."""
records = parse_search_items(
[row],
None,
provider_id="annas_archive",
extract_item=_extract_aa_search_result,
)
return records[0] if records else None
def _parse_book_info_page(
soup: BeautifulSoup,
book_id: str,
@@ -999,7 +915,7 @@ def _parse_book_info_page(
node = data.select_one("div:nth-of-type(1) > img")
if isinstance(node, Tag):
preview = _get_attr(node, "src") or ""
preview = get_attr(node, "src") or ""
main_inner = next(
(tag for tag in soup.find_all("div", {"class": "main-inner"}) if isinstance(tag, Tag)),
@@ -1023,7 +939,7 @@ def _parse_book_info_page(
for anchor in soup.find_all("a"):
try:
text = anchor.text.strip().lower()
href = _get_attr(anchor, "href")
href = get_attr(anchor, "href")
if not href:
continue
@@ -1075,7 +991,7 @@ def _parse_book_info_page(
file_format = ""
size = ""
content = ""
supported_formats = _get_supported_formats()
supported_formats = get_supported_formats()
for _details in all_details:
_details = _details.split(" · ")
@@ -1084,7 +1000,7 @@ def _parse_book_info_page(
if file_format == "" and stripped_lower in supported_formats:
file_format = f.strip().lower()
if size == "" and any(u in f.strip().lower() for u in ("mb", "kb", "gb")):
size = _normalize_size(f)
size = normalize_size(f)
if content == "":
for ct in CONTENT_TYPES:
if ct in f.strip().lower():
@@ -1096,7 +1012,7 @@ def _parse_book_info_page(
if file_format == "" and stripped and " " not in stripped:
file_format = stripped
if size == "" and "." in stripped:
size = _normalize_size(f)
size = normalize_size(f)
book_title = (_find_in_divs(divs, "🔍") or [""])[0].strip("🔍").strip()
@@ -1134,7 +1050,7 @@ def _parse_book_info_page(
summary_url, selector=network.AAMirrorSelector(), allow_bypasser_fallback=False
)
if summary_response:
summary_data = json.loads(_html_response_text(summary_response))
summary_data = json.loads(html_response_text(summary_response))
if "downloads_total" in summary_data:
info["Downloads"] = [str(summary_data["downloads_total"])]
except (
@@ -1456,11 +1372,11 @@ def _get_download_urls_from_welib(
logger.warning("Welib page empty for %s", book_id)
return []
soup = BeautifulSoup(_html_response_text(html), "html.parser")
soup = BeautifulSoup(html_response_text(html), "html.parser")
links = [
downloader.get_absolute_url(url, href)
for a in soup.find_all("a", href=True)
if (href := _get_attr(a, "href")) and "/slow_download/" in href
if (href := get_attr(a, "href")) and "/slow_download/" in href
]
return list(dict.fromkeys(links)) # Dedupe while preserving order
@@ -1524,7 +1440,7 @@ def _extract_libgen_download_url(link: str, cancel_flag: Event | None = None) ->
return download_url
def _download_book(
def download_book(
book_info: BrowseRecord,
book_path: Path,
progress_callback: Callable[[float], None] | None = None,
@@ -1650,7 +1566,7 @@ def _get_download_url(
page = downloader.html_get_page(
link, selector=sel, cancel_flag=cancel_flag, status_callback=status_callback
)
page_data = json.loads(_html_response_text(page))
page_data = json.loads(html_response_text(page))
download_url = page_data.get("download_url", "")
return (
downloader.get_absolute_url(link, download_url) if isinstance(download_url, str) else ""
@@ -1665,7 +1581,7 @@ def _get_download_url(
if not html:
return ""
soup = BeautifulSoup(_html_response_text(html), "html.parser")
soup = BeautifulSoup(html_response_text(html), "html.parser")
url = ""
# Z-Library
@@ -1678,9 +1594,9 @@ def _get_download_url(
link, selector=sel, cancel_flag=cancel_flag, status_callback=status_callback
)
if html:
soup = BeautifulSoup(_html_response_text(html), "html.parser")
soup = BeautifulSoup(html_response_text(html), "html.parser")
dl = soup.find("a", href=True, class_="addDownloadedBook")
url = (_get_attr(dl, "href") or "") if isinstance(dl, Tag) else ""
url = (get_attr(dl, "href") or "") if isinstance(dl, Tag) else ""
# AA slow download / partner servers
elif "/slow_download/" in link:
@@ -1693,7 +1609,7 @@ def _get_download_url(
soup, "Download"
)
if get_btn:
url = _get_attr(get_btn, "href") or ""
url = get_attr(get_btn, "href") or ""
else:
logger.warning("Unknown source type, couldn't find download link: %s", link)
url = ""
@@ -1727,11 +1643,11 @@ def _extract_slow_download_url(
soup, "Download now", contains=True
)
if dl_link:
return _get_attr(dl_link, "href") or ""
return get_attr(dl_link, "href") or ""
for a_tag in soup.find_all("a", href=True):
if a_tag.has_attr("download"):
href = _get_attr(a_tag, "href")
href = get_attr(a_tag, "href")
if not href:
continue
if href.startswith("http") and "/slow_download/" not in href:
@@ -1762,7 +1678,7 @@ def _extract_slow_download_url(
parent = copy_text.parent
next_link = parent.find_next("a", href=True)
if isinstance(next_link, Tag):
next_href = _get_attr(next_link, "href")
next_href = get_attr(next_link, "href")
if next_href:
return next_href
code_elem = parent.find_next("code")
@@ -1825,7 +1741,7 @@ def _extract_slow_download_url(
)
if not html:
return ""
new_soup = BeautifulSoup(_html_response_text(html), "html.parser")
new_soup = BeautifulSoup(html_response_text(html), "html.parser")
return _extract_slow_download_url(
new_soup,
link,
@@ -1915,119 +1831,48 @@ def _parse_countdown_seconds_from_element(element: Tag) -> int | None:
return None
def _browse_record_to_release(record: BrowseRecord) -> Release:
"""Convert a browse record to a Release object.
class AnnasArchiveProvider:
"""Anna's Archive provider, including its specialized MD5 mirror cascade."""
This bridges the direct source's browse data to the generic release model.
"""
return Release(
source=record.source,
source_id=record.id,
title=record.title,
format=record.format,
language=record.language, # Top-level language for filtering
size=record.size,
download_url=record.download_urls[0] if record.download_urls else None,
info_url=f"{network.get_aa_base_url()}/md5/{record.id}",
protocol=ReleaseProtocol.HTTP,
indexer="Direct Download",
content_type=record.content, # Preserve content type from source
extra={
"author": record.author,
"publisher": record.publisher,
"year": record.year,
"language": record.language,
"preview": record.preview,
"description": record.description,
"download_urls": record.download_urls,
"info": record.info,
},
)
@register_source("direct_download")
class DirectDownloadSource(ReleaseSource):
"""Direct download source - searches web sources for books.
This wraps the search_books() functionality to provide releases
via the plugin interface.
"""
name = "direct_download"
display_name = "Direct Download"
supported_content_types: ClassVar[list[str]] = ["ebook"] # Direct downloads only support ebooks
id = "annas_archive"
display_name = "Anna's Archive"
def __init__(self) -> None:
"""Initialize per-instance search state for direct downloads."""
# Tracks which search method was used in the last search() call
# "isbn" = ISBN search returned results, "title_author" = title+author was used
self._last_search_type: str = "title_author"
self._last_search_type = "title_author"
@property
def last_search_type(self) -> str:
"""Returns the search type used in the last search() call."""
return self._last_search_type
def get_column_config(self) -> ReleaseColumnConfig:
"""Column configuration for Direct Download source.
def is_enabled(self) -> bool:
from shelfmark.core import mirrors
Shows language, format, and size badges for each release.
Language is hidden on mobile; format and size are shown.
"""
return ReleaseColumnConfig(
columns=[
ColumnSchema(
key="extra.language",
label="Language",
render_type=ColumnRenderType.BADGE,
align=ColumnAlign.CENTER,
width="60px",
hide_mobile=False, # Language shown on mobile
color_hint=ColumnColorHint(type="map", value="language"),
uppercase=True,
),
ColumnSchema(
key="format",
label="Format",
render_type=ColumnRenderType.BADGE,
align=ColumnAlign.CENTER,
width="80px",
hide_mobile=False, # Format shown on mobile
color_hint=ColumnColorHint(type="map", value="format"),
uppercase=True,
),
ColumnSchema(
key="size",
label="Size",
render_type=ColumnRenderType.SIZE,
align=ColumnAlign.CENTER,
width="80px",
hide_mobile=False, # Size shown on mobile
),
],
grid_template="minmax(0,2fr) 60px 80px 80px",
supported_filters=["format", "language"], # AA has reliable language metadata
return mirrors.has_aa_mirror_configuration()
def handles(self, url: str) -> bool:
from shelfmark.core import mirrors
hostname = (urlparse(url).hostname or "").lower().rstrip(".")
if not hostname:
return False
return any(
hostname == (urlparse(base_url).hostname or "").lower().rstrip(".")
for base_url in mirrors.get_aa_mirrors()
)
def get_record(
self,
record_id: str,
*,
fetch_download_count: bool = True,
) -> BrowseRecord | None:
"""Resolve a direct-download record for direct-mode info/download flows."""
_ensure_direct_download_available()
def get_record(self, record_id: str, *, fetch_download_count: bool = True) -> BrowseRecord:
ensure_available()
return get_book_info(record_id, fetch_download_count=fetch_download_count)
def search_results_are_releases(self) -> bool:
"""Direct search results already represent concrete downloadable releases."""
return True
def get_destination_override(self, task: DownloadTask) -> Path | None:
"""Apply Anna's Archive content-type routing when configured."""
if check_audiobook(task.content_type):
return None
return get_aa_content_type_dir(task.content_type)
def download(
self,
book_info: BrowseRecord,
book_path: Path,
progress_callback: Callable[[float], None] | None,
cancel_flag: Event | None,
status_callback: Callable[[str, str | None], None] | None,
) -> str | None:
return download_book(book_info, book_path, progress_callback, cancel_flag, status_callback)
def _search_books_with_language_fallback(
self,
@@ -2055,14 +1900,15 @@ class DirectDownloadSource(ReleaseSource):
*,
expand_search: bool = False,
content_type: str = "ebook",
) -> list[Release]:
"""Search for releases using the book's metadata.
The whole fan-out runs under one page cache, so a URL built twice by different
passes is fetched once. See `_search_page_reuse`.
"""
with _search_page_reuse():
return self._search(book, plan, expand_search=expand_search, content_type=content_type)
) -> list[BrowseRecord]:
"""Search for releases using the book's metadata with request-local page reuse."""
with search_page_reuse():
return self._search(
book,
plan,
expand_search=expand_search,
content_type=content_type,
)
def _search(
self,
@@ -2071,8 +1917,8 @@ class DirectDownloadSource(ReleaseSource):
*,
expand_search: bool = False,
content_type: str = "ebook",
) -> list[Release]:
"""Search for releases using the book's metadata.
) -> list[BrowseRecord]:
"""Run Anna's Archive's ISBN-first and localized-title search strategy.
Priority: ISBN search first (most precise), then title+author fallback.
For non-English languages, uses localized titles from book.titles_by_language.
@@ -2085,7 +1931,7 @@ class DirectDownloadSource(ReleaseSource):
content_type: Ignored - Direct download uses format filtering instead
"""
_ensure_direct_download_available()
ensure_available()
lang_filter = plan.languages
# Reset search type tracking
@@ -2102,7 +1948,7 @@ class DirectDownloadSource(ReleaseSource):
query, filters, search_label="manual"
)
self._last_search_type = "manual" if query else "title_author"
return [_browse_record_to_release(record) for record in results]
return results
# ISBN search first (unless expand_search requested)
if plan.manual_query:
@@ -2119,7 +1965,7 @@ class DirectDownloadSource(ReleaseSource):
if results:
logger.info("Found %s releases via ISBN", len(results))
self._last_search_type = "isbn"
return [_browse_record_to_release(record) for record in results]
return results
logger.debug("No ISBN results, falling back to title+author")
except SearchUnavailableError:
raise
@@ -2184,155 +2030,4 @@ class DirectDownloadSource(ReleaseSource):
except Exception:
logger.exception("Search error")
return [_browse_record_to_release(record) for record in all_results]
def is_available(self) -> bool:
"""Check if Direct Download has been explicitly enabled and configured."""
return _get_direct_download_unavailable_reason() is None
@register_handler("direct_download")
class DirectDownloadHandler(DownloadHandler):
"""Handler for direct HTTP downloads from Anna's Archive, Libgen, etc.
Receives a DownloadTask with task_id (AA MD5 hash) and cascades through
sources in priority order. The AA page is only fetched if AA slow sources
are enabled in the user's source priority configuration.
"""
def download(
self,
task: DownloadTask,
cancel_flag: Event,
progress_callback: Callable[[float], None],
status_callback: Callable[[str, str | None], None],
) -> str | None:
"""Execute a direct HTTP download.
Uses task.task_id (AA MD5 hash) to cascade through sources in priority
order. The AA page is only fetched if AA slow sources are enabled.
Args:
task: Download task with task_id (AA MD5 hash)
cancel_flag: Event to check for cancellation
progress_callback: Called with progress percentage (0-100)
status_callback: Called with (status, message) for status updates
Returns:
Path to downloaded file if successful, None otherwise
"""
try:
# Check for cancellation before starting
if cancel_flag.is_set():
logger.info("Download cancelled before starting: %s", task.task_id)
status_callback("cancelled", "Cancelled")
return None
# Create browse record from task data - NO AA page fetch here
# AA page is fetched lazily by _fetch_aa_page_urls only when
# we actually reach an AA slow source in the priority order
book_info = BrowseRecord(
id=task.task_id,
title=task.title,
source="direct_download",
author=task.author,
year=task.year,
format=task.format,
size=task.size,
preview=task.preview,
)
return self._execute_download(
book_info, cancel_flag, progress_callback, status_callback
)
except Exception as e:
if cancel_flag.is_set():
logger.info("Download cancelled during error handling: %s", task.task_id)
status_callback("cancelled", "Cancelled")
else:
logger.exception("Error downloading book")
status_callback("error", str(e))
return None
def _execute_download(
self,
book_info: BrowseRecord,
cancel_flag: Event,
progress_callback: Callable[[float], None],
status_callback: Callable[[str, str | None], None],
) -> str | None:
"""Execute the direct-download flow with a fetched browse record.
This contains the core download logic: cascade through sources,
handle bypass, move to final location.
"""
try:
logger.debug("Starting download: %s", book_info.title)
# Prepare paths - use descriptive staging filename, orchestrator will rename
# based on FILE_ORGANIZATION setting
file_org = config.get("FILE_ORGANIZATION", "rename")
if file_org == "none":
book_name = f"{book_info.id}.{book_info.format or 'bin'}"
else:
book_name = build_filename(
book_info.title,
book_info.author,
book_info.year,
book_info.format,
)
book_path = TMP_DIR / book_name
# Check cancellation before download
if cancel_flag.is_set():
logger.info("Download cancelled before download call: %s", book_info.id)
status_callback("cancelled", "Cancelled")
return None
# Execute download via _download_book (handles cascade and bypass)
status_callback("resolving", "Finding download source")
success_url = _download_book(
book_info, book_path, progress_callback, cancel_flag, status_callback
)
# Check for cancellation after download
if cancel_flag.is_set():
logger.info("Download cancelled during download: %s", book_info.id)
if book_path.exists():
book_path.unlink()
status_callback("cancelled", "Cancelled")
return None
if not success_url:
if network.dns_interference_detected():
status_callback(
"error",
"All sources failed - your network/ISP appears to be blocking "
"Anna's Archive. Enable DNS-over-HTTPS in settings.",
)
else:
status_callback("error", "All download sources failed")
return None
# Return temp path - orchestrator handles post-processing (archive extraction, ingest)
return str(book_path)
except Exception:
if cancel_flag.is_set():
logger.info("Download cancelled during error handling: %s", book_info.id)
status_callback("cancelled", "Cancelled")
else:
logger.exception("Error downloading book")
return None
def cancel(self, task_id: str) -> bool:
"""Cancel an in-progress download.
Cancellation is handled via the cancel_flag passed to download().
This method exists for the interface but actual cancellation
happens through the Event flag mechanism.
"""
# Cancellation is handled by the orchestrator via cancel_flag
return False
return all_results
shelfmark/release_sources/direct_download/common.py
+264
View File
@@ -0,0 +1,264 @@
"""Shared contracts and result normalization for Direct Download websites."""
import hashlib
import re
import threading
from dataclasses import dataclass
from typing import TYPE_CHECKING, Protocol, runtime_checkable
from bs4 import BeautifulSoup, Tag
from shelfmark.core.config import config
from shelfmark.core.languages import language_alias_map
from shelfmark.release_sources import BrowseRecord, SourceUnavailableError
if TYPE_CHECKING:
from collections.abc import Callable, Iterable
from pathlib import Path
from threading import Event
from shelfmark.core.models import SearchFilters
from shelfmark.core.search_plan import ReleaseSearchPlan
from shelfmark.metadata_providers import BookMetadata
_LANGUAGE_ALIAS_TO_CODE: dict[str, str] | None = None
_LANGUAGE_ALIAS_LOCK = threading.Lock()
_SIZE_UNIT_PATTERN = re.compile(r"(kb|mb|gb|tb)", re.IGNORECASE)
MIN_VALID_FILE_SIZE = 10 * 1024
class DirectDownloadUnavailableError(SourceUnavailableError):
"""Raised when the composite Direct Download source cannot be reached."""
def coerce_str_list(value: object) -> list[str]:
"""Return only string items from a config value."""
if not isinstance(value, list | tuple):
return []
return [item for item in value if isinstance(item, str)]
def get_supported_formats() -> list[str]:
"""Return configured supported formats as a clean string list."""
return coerce_str_list(config.SUPPORTED_FORMATS)
def html_response_text(response: str | tuple[str, str]) -> str:
"""Extract the HTML body from downloader responses."""
if isinstance(response, tuple):
return response[0]
return response
def attr_to_str(value: object) -> str | None:
"""Convert a BeautifulSoup attribute value to a plain string."""
if isinstance(value, str):
return value
if isinstance(value, list):
for item in value:
if isinstance(item, str):
return item
return None
def get_attr(tag: Tag, attr: str) -> str | None:
"""Safely fetch a tag attribute as a string."""
return attr_to_str(tag.get(attr))
@dataclass(frozen=True)
class ParsedSearchResult:
"""Provider-neutral fields extracted from one search-result element."""
key: str
title: str
formats: tuple[str, ...]
record_id: str | None = None
author: str | None = None
publisher: str | None = None
year: str | None = None
language: str | None = None
content: str | None = None
size: str | None = None
preview: str | None = None
source_url: str | None = None
download_path: str | None = None
class DirectDownloadProvider(Protocol):
"""Provider lifecycle used by the composite Direct Download source."""
id: str
display_name: str
def is_enabled(self) -> bool: ...
def handles(self, url: str) -> bool: ...
def search(
self,
book: BookMetadata,
plan: ReleaseSearchPlan,
*,
expand_search: bool = False,
content_type: str = "ebook",
) -> list[BrowseRecord]: ...
def download(
self,
book_info: BrowseRecord,
book_path: Path,
progress_callback: Callable[[float], None] | None,
cancel_flag: Event | None,
status_callback: Callable[[str, str | None], None] | None,
) -> str | None: ...
@runtime_checkable
class RecordLookupProvider(Protocol):
"""Optional capability for providers that can reopen source-native records."""
def get_record(
self, record_id: str, *, fetch_download_count: bool = True
) -> BrowseRecord | None: ...
def normalize_language_token(value: str) -> str:
normalized = value.strip().lower()
for dash in ("‑", "–", "—", "−"):
normalized = normalized.replace(dash, "-")
return normalized
def language_alias_to_code() -> dict[str, str]:
"""Alias to code map, delegating to the shared language data."""
global _LANGUAGE_ALIAS_TO_CODE
cached = _LANGUAGE_ALIAS_TO_CODE
if cached is not None:
return cached
with _LANGUAGE_ALIAS_LOCK:
cached = _LANGUAGE_ALIAS_TO_CODE
if cached is not None:
return cached
_LANGUAGE_ALIAS_TO_CODE = language_alias_map()
return _LANGUAGE_ALIAS_TO_CODE
def normalize_requested_languages(languages: list[str] | None) -> set[str]:
if not languages:
return set()
aliases = language_alias_to_code()
normalized: set[str] = set()
for value in languages:
token = normalize_language_token(str(value))
if not token or token == "all": # noqa: S105 - "all" is a language sentinel
continue
normalized.add(aliases.get(token, token))
return normalized
def book_matches_requested_languages(book_language: str | None, requested: set[str]) -> bool:
"""Return True when a book's language matches the requested filter.
Books with unknown/missing language always pass — the server-side &lang= filter
already narrowed the result set, so dropping unlabelled rows hides valid results.
"""
if not requested:
return True
if not book_language:
return True
aliases = language_alias_to_code()
normalized_book = aliases.get(
normalize_language_token(book_language),
normalize_language_token(book_language),
)
return normalized_book in requested
def normalize_size(size_str: str) -> str:
"""Normalize size string by uppercasing units (e.g., '5.2 mb' -> '5.2 MB')."""
return _SIZE_UNIT_PATTERN.sub(lambda m: m.group(1).upper(), size_str.strip())
def parse_search_items(
items: Iterable[Tag],
filters: SearchFilters | None,
*,
provider_id: str,
extract_item: Callable[[Tag], ParsedSearchResult | None],
) -> list[BrowseRecord]:
"""Normalize provider-specific HTML elements into Direct Download records.
Providers only describe how fields are extracted from their DOM. Language and
format filtering, stable IDs, and BrowseRecord construction stay shared.
"""
requested_languages = normalize_requested_languages(filters.lang) if filters else set()
requested_formats = (
{value.casefold() for value in (filters.format or get_supported_formats())}
if filters
else set()
)
records: list[BrowseRecord] = []
for item in items:
parsed = extract_item(item)
if parsed is None:
continue
normalized_language = normalize_language_token(parsed.language) if parsed.language else ""
language = language_alias_to_code().get(normalized_language, normalized_language) or None
if not book_matches_requested_languages(language, requested_languages):
continue
formats = parsed.formats or ("",)
for book_format in formats:
normalized_format = book_format.casefold()
if (
normalized_format
and requested_formats
and normalized_format not in requested_formats
):
continue
record_id = parsed.record_id
if not record_id or len(formats) > 1:
source_key = f"{parsed.key}#{normalized_format}"
digest = hashlib.blake2b(source_key.encode(), digest_size=16).hexdigest()
record_id = f"{provider_id}:{digest}"
records.append(
BrowseRecord(
id=record_id,
title=parsed.title,
source="direct_download",
author=parsed.author,
publisher=parsed.publisher,
year=parsed.year,
language=language,
format=normalized_format or None,
size=parsed.size,
preview=parsed.preview,
content=parsed.content,
source_url=parsed.source_url,
download_path=parsed.download_path,
)
)
return records
def parse_search_page(
page: str | BeautifulSoup | Tag,
filters: SearchFilters | None,
*,
provider_id: str,
item_selector: str,
extract_item: Callable[[Tag], ParsedSearchResult | None],
) -> list[BrowseRecord]:
"""Parse a result page using provider-specific selectors and extraction."""
root = BeautifulSoup(page, "html.parser") if isinstance(page, str) else page
return parse_search_items(
(item for item in root.select(item_selector) if isinstance(item, Tag)),
filters,
provider_id=provider_id,
extract_item=extract_item,
)
shelfmark/release_sources/direct_download/handler.py
+177
View File
@@ -0,0 +1,177 @@
"""Direct Download routing, staging, and cancellation."""
from pathlib import Path
from typing import TYPE_CHECKING
from shelfmark.config.env import TMP_DIR
from shelfmark.core.config import config
from shelfmark.core.logger import setup_logger
from shelfmark.core.models import DownloadTask, build_filename
from shelfmark.download import network
from shelfmark.release_sources import (
BrowseRecord,
DownloadHandler,
register_handler,
)
from shelfmark.release_sources.direct_download import registry
if TYPE_CHECKING:
from collections.abc import Callable
from pathlib import Path
from threading import Event
logger = setup_logger(__name__)
def _download_book(
book_info: BrowseRecord,
book_path: Path,
progress_callback: Callable[[float], None] | None = None,
cancel_flag: Event | None = None,
status_callback: Callable[[str, str | None], None] | None = None,
) -> str | None:
"""Route a website record or an Anna's Archive MD5 to its download flow."""
provider = registry.provider_for_record(book_info)
if provider is None:
msg = f"No Direct Download provider owns record {book_info.id!r}"
raise RuntimeError(msg)
return provider.download(book_info, book_path, progress_callback, cancel_flag, status_callback)
@register_handler("direct_download")
class DirectDownloadHandler(DownloadHandler):
"""Route and stage downloads from registered Direct Download providers."""
def download(
self,
task: DownloadTask,
cancel_flag: Event,
progress_callback: Callable[[float], None],
status_callback: Callable[[str, str | None], None],
) -> str | None:
"""Execute a provider-owned direct HTTP download.
Args:
task: Download task with a provider-owned source ID
cancel_flag: Event to check for cancellation
progress_callback: Called with progress percentage (0-100)
status_callback: Called with (status, message) for status updates
Returns:
Path to downloaded file if successful, None otherwise
"""
try:
# Check for cancellation before starting
if cancel_flag.is_set():
logger.info("Download cancelled before starting: %s", task.task_id)
status_callback("cancelled", "Cancelled")
return None
# Reconstruct the provider-owned record without resolving it again.
book_info = BrowseRecord(
id=task.task_id,
title=task.title,
source="direct_download",
author=task.author,
year=task.year,
format=task.format,
size=task.size,
preview=task.preview,
source_url=task.source_url,
)
return self._execute_download(
book_info, cancel_flag, progress_callback, status_callback
)
except Exception as e:
if cancel_flag.is_set():
logger.info("Download cancelled during error handling: %s", task.task_id)
status_callback("cancelled", "Cancelled")
else:
logger.exception("Error downloading book")
status_callback("error", str(e))
return None
def _execute_download(
self,
book_info: BrowseRecord,
cancel_flag: Event,
progress_callback: Callable[[float], None],
status_callback: Callable[[str, str | None], None],
) -> str | None:
"""Execute the direct-download flow with a fetched browse record.
This contains the core download logic: cascade through sources,
handle bypass, move to final location.
"""
try:
logger.debug("Starting download: %s", book_info.title)
# Prepare paths - use descriptive staging filename, orchestrator will rename
# based on FILE_ORGANIZATION setting
file_org = config.get("FILE_ORGANIZATION", "rename")
if file_org == "none":
book_name = f"{book_info.id}.{book_info.format or 'bin'}"
else:
book_name = build_filename(
book_info.title,
book_info.author,
book_info.year,
book_info.format,
)
book_path = TMP_DIR / book_name
# Check cancellation before download
if cancel_flag.is_set():
logger.info("Download cancelled before download call: %s", book_info.id)
status_callback("cancelled", "Cancelled")
return None
# Execute download via _download_book (handles cascade and bypass)
status_callback("resolving", "Finding download source")
success_url = _download_book(
book_info, book_path, progress_callback, cancel_flag, status_callback
)
# Check for cancellation after download
if cancel_flag.is_set():
logger.info("Download cancelled during download: %s", book_info.id)
if book_path.exists():
book_path.unlink()
status_callback("cancelled", "Cancelled")
return None
if not success_url:
if network.dns_interference_detected():
status_callback(
"error",
"All sources failed - your network/ISP appears to be blocking "
"Anna's Archive. Enable DNS-over-HTTPS in settings.",
)
else:
status_callback("error", "All download sources failed")
return None
# Return temp path - orchestrator handles post-processing (archive extraction, ingest)
return str(book_path)
except Exception:
if cancel_flag.is_set():
logger.info("Download cancelled during error handling: %s", book_info.id)
status_callback("cancelled", "Cancelled")
else:
logger.exception("Error downloading book")
return None
def cancel(self, task_id: str) -> bool:
"""Cancel an in-progress download.
Cancellation is handled via the cancel_flag passed to download().
This method exists for the interface but actual cancellation
happens through the Event flag mechanism.
"""
# Cancellation is handled by the orchestrator via cancel_flag
return False
shelfmark/release_sources/direct_download/registry.py
+93
View File
@@ -0,0 +1,93 @@
"""Provider composition and dispatch for the Direct Download release source."""
import re
from typing import TYPE_CHECKING
from shelfmark.core.config import config
from shelfmark.release_sources.direct_download.annas_archive import AnnasArchiveProvider
if TYPE_CHECKING:
from collections.abc import Sequence
from shelfmark.release_sources import BrowseRecord
from shelfmark.release_sources.direct_download.common import DirectDownloadProvider
PROVIDER_TYPES = (AnnasArchiveProvider,)
_AA_MD5_PATTERN = re.compile(r"^[0-9a-f]{32}$", re.IGNORECASE)
def create_providers() -> tuple[DirectDownloadProvider, ...]:
"""Create request-local providers so mutable search state is not shared."""
return tuple(provider_type() for provider_type in PROVIDER_TYPES)
def enabled_providers(
providers: Sequence[DirectDownloadProvider] | None = None,
) -> tuple[DirectDownloadProvider, ...]:
if not config.get("DIRECT_DOWNLOAD_ENABLED", False):
return ()
candidates = providers if providers is not None else create_providers()
return tuple(provider for provider in candidates if provider.is_enabled())
def get_unavailable_reason(
providers: Sequence[DirectDownloadProvider] | None = None,
) -> str | None:
if not config.get("DIRECT_DOWNLOAD_ENABLED", False):
return "Direct Download is disabled. Enable the source in Settings."
if not enabled_providers(providers):
return (
"Direct Download is not configured. Enable and configure at least one "
"download provider in Settings."
)
return None
def provider_by_id(
provider_id: str | None,
providers: Sequence[DirectDownloadProvider] | None = None,
) -> DirectDownloadProvider | None:
if not provider_id:
return None
candidates = providers if providers is not None else create_providers()
return next((provider for provider in candidates if provider.id == provider_id), None)
def provider_for_record(
record: BrowseRecord,
providers: Sequence[DirectDownloadProvider] | None = None,
) -> DirectDownloadProvider | None:
"""Resolve a record explicitly, retaining safe compatibility with legacy tasks."""
candidates = providers if providers is not None else create_providers()
prefix, separator, _remainder = record.id.partition(":")
if separator:
provider = provider_by_id(prefix, candidates)
if provider is not None:
return provider
if record.source_url:
provider = next(
(provider for provider in candidates if provider.handles(record.source_url)),
None,
)
if provider is not None:
return provider
# Anna's Archive records historically carried only their raw MD5. Preserve those
# persisted tasks without treating arbitrary unknown URLs as Anna's Archive.
if _AA_MD5_PATTERN.fullmatch(record.id):
return provider_by_id("annas_archive", candidates)
return None
def provider_for_record_id(
record_id: str,
providers: Sequence[DirectDownloadProvider] | None = None,
) -> DirectDownloadProvider | None:
candidates = providers if providers is not None else create_providers()
prefix, separator, _remainder = record_id.partition(":")
if separator:
return provider_by_id(prefix, candidates)
# Record lookup predates provider-qualified IDs, so unqualified IDs are AA IDs.
return provider_by_id("annas_archive", candidates)
shelfmark/release_sources/direct_download/source.py
+204
View File
@@ -0,0 +1,204 @@
"""Direct Download search and release-source integration."""
from pathlib import Path
from typing import TYPE_CHECKING, ClassVar
import requests
from shelfmark.core.logger import setup_logger
from shelfmark.core.utils import get_aa_content_type_dir
from shelfmark.core.utils import is_audiobook as check_audiobook
from shelfmark.release_sources import (
BrowseRecord,
ColumnAlign,
ColumnColorHint,
ColumnRenderType,
ColumnSchema,
Release,
ReleaseColumnConfig,
ReleaseProtocol,
ReleaseSource,
SourceUnavailableError,
register_source,
)
from shelfmark.release_sources.direct_download import registry
from shelfmark.release_sources.direct_download.common import (
DirectDownloadUnavailableError,
RecordLookupProvider,
)
if TYPE_CHECKING:
from pathlib import Path
from shelfmark.core.models import DownloadTask
from shelfmark.core.search_plan import ReleaseSearchPlan
from shelfmark.metadata_providers import BookMetadata
logger = setup_logger(__name__)
def _browse_record_to_release(record: BrowseRecord) -> Release:
"""Convert a browse record to a Release object.
This bridges the direct source's browse data to the generic release model.
"""
provider = registry.provider_for_record(record)
provider_id = provider.id if provider is not None else None
return Release(
source=record.source,
source_id=record.id,
title=record.title,
format=record.format,
language=record.language, # Top-level language for filtering
size=record.size,
download_url=record.source_url
or (record.download_urls[0] if record.download_urls else None),
info_url=record.source_url,
protocol=ReleaseProtocol.HTTP,
indexer="Direct Download",
content_type=record.content, # Preserve content type from source
extra={
"author": record.author,
"publisher": record.publisher,
"year": record.year,
"language": record.language,
"preview": record.preview,
"description": record.description,
"download_urls": record.download_urls,
"info": record.info,
"direct_download_provider": provider_id,
# Kept for older frontends and persisted request payloads.
"web_provider": provider_id if provider_id != "annas_archive" else None,
},
)
@register_source("direct_download")
class DirectDownloadSource(ReleaseSource):
"""Direct download source - searches web sources for books.
This wraps the search_books() functionality to provide releases
via the plugin interface.
"""
name = "direct_download"
display_name = "Direct Download"
supported_content_types: ClassVar[list[str]] = ["ebook"] # Direct downloads only support ebooks
def __init__(self) -> None:
"""Initialize per-instance search state for direct downloads."""
self._providers = registry.create_providers()
@property
def last_search_type(self) -> str:
"""Returns the search type used in the last search() call."""
provider = registry.provider_by_id("annas_archive", self._providers)
return str(getattr(provider, "last_search_type", "title_author"))
def get_column_config(self) -> ReleaseColumnConfig:
"""Column configuration for Direct Download source.
Shows language, format, and size badges for each release.
Language is hidden on mobile; format and size are shown.
"""
return ReleaseColumnConfig(
columns=[
ColumnSchema(
key="extra.language",
label="Language",
render_type=ColumnRenderType.BADGE,
align=ColumnAlign.CENTER,
width="60px",
hide_mobile=False, # Language shown on mobile
color_hint=ColumnColorHint(type="map", value="language"),
uppercase=True,
),
ColumnSchema(
key="format",
label="Format",
render_type=ColumnRenderType.BADGE,
align=ColumnAlign.CENTER,
width="80px",
hide_mobile=False, # Format shown on mobile
color_hint=ColumnColorHint(type="map", value="format"),
uppercase=True,
),
ColumnSchema(
key="size",
label="Size",
render_type=ColumnRenderType.SIZE,
align=ColumnAlign.CENTER,
width="80px",
hide_mobile=False, # Size shown on mobile
),
],
grid_template="minmax(0,2fr) 60px 80px 80px",
supported_filters=["format", "language"], # AA has reliable language metadata
)
def get_record(
self,
record_id: str,
*,
fetch_download_count: bool = True,
) -> BrowseRecord | None:
"""Resolve a direct-download record for direct-mode info/download flows."""
provider = registry.provider_for_record_id(record_id, self._providers)
if provider is None or not isinstance(provider, RecordLookupProvider):
return None
native_id = record_id.partition(":")[2] or record_id
return provider.get_record(native_id, fetch_download_count=fetch_download_count)
def search_results_are_releases(self) -> bool:
"""Direct search results already represent concrete downloadable releases."""
return True
def get_destination_override(self, task: DownloadTask) -> Path | None:
"""Apply Anna's Archive content-type routing when configured."""
if check_audiobook(task.content_type):
return None
return get_aa_content_type_dir(task.content_type)
def search(
self,
book: BookMetadata,
plan: ReleaseSearchPlan,
*,
expand_search: bool = False,
content_type: str = "ebook",
) -> list[Release]:
"""Search every enabled provider through the shared provider lifecycle."""
unavailable_reason = registry.get_unavailable_reason(self._providers)
if unavailable_reason:
raise DirectDownloadUnavailableError(unavailable_reason)
releases: list[Release] = []
unavailable_errors: list[SourceUnavailableError] = []
for provider in registry.enabled_providers(self._providers):
try:
records = provider.search(
book,
plan,
expand_search=expand_search,
content_type=content_type,
)
except SourceUnavailableError as exc:
unavailable_errors.append(exc)
continue
except (
RuntimeError,
TypeError,
ValueError,
requests.exceptions.RequestException,
) as exc:
logger.warning("%s search failed: %s", provider.display_name, exc)
continue
releases.extend(_browse_record_to_release(record) for record in records)
if unavailable_errors and not releases:
raise unavailable_errors[0]
return releases
def is_available(self) -> bool:
"""Check if Direct Download has been explicitly enabled and configured."""
return registry.get_unavailable_reason(self._providers) is None
tests/direct_download/test_handler.py
+8 -7
View File
@@ -1,5 +1,6 @@
from threading import Event
from shelfmark.core.config import config
from shelfmark.core.models import DownloadTask
from shelfmark.release_sources.direct_download import DirectDownloadHandler
@@ -13,11 +14,11 @@ def test_direct_download_handler_builds_staging_filename_from_browse_record(monk
captured["path"] = book_path
return "https://example.com/file.epub"
import shelfmark.release_sources.direct_download as dd
from shelfmark.release_sources.direct_download import handler as dd
monkeypatch.setattr(dd, "_download_book", fake_download_book)
monkeypatch.setattr(
dd.config,
config,
"get",
lambda key, default=None: "rename" if key == "FILE_ORGANIZATION" else default,
)
@@ -47,11 +48,11 @@ def test_direct_download_handler_uses_source_id_filename_when_organization_disab
captured["path"] = book_path
return "https://example.com/file.epub"
import shelfmark.release_sources.direct_download as dd
from shelfmark.release_sources.direct_download import handler as dd
monkeypatch.setattr(dd, "_download_book", fake_download_book)
monkeypatch.setattr(
dd.config,
config,
"get",
lambda key, default=None: "none" if key == "FILE_ORGANIZATION" else default,
)
@@ -78,7 +79,7 @@ def test_direct_download_handler_skips_download_when_cancelled_before_start(monk
def unexpected_download(*_args, **_kwargs):
raise AssertionError("_download_book should not run when the task is already cancelled")
import shelfmark.release_sources.direct_download as dd
from shelfmark.release_sources.direct_download import handler as dd
monkeypatch.setattr(dd, "_download_book", unexpected_download)
@@ -113,12 +114,12 @@ def test_direct_download_handler_removes_partial_file_when_cancelled_after_downl
cancel_flag.set()
return "https://example.com/file.epub"
import shelfmark.release_sources.direct_download as dd
from shelfmark.release_sources.direct_download import handler as dd
monkeypatch.setattr(dd, "_download_book", fake_download_book)
monkeypatch.setattr(dd, "TMP_DIR", tmp_path)
monkeypatch.setattr(
dd.config,
config,
"get",
lambda key, default=None: "rename" if key == "FILE_ORGANIZATION" else default,
)
tests/direct_download/test_search_budget.py
+18 -15
View File
@@ -10,13 +10,16 @@ a challenge failure became a gateway timeout (issue #1276).
import pytest
import shelfmark.release_sources.direct_download as dd
from shelfmark.core import search_deadline
from shelfmark.release_sources.direct_download import annas_archive as aa
from shelfmark.release_sources.direct_download import source as dd
@pytest.fixture(autouse=True)
def _no_ambient_deadline():
def _no_ambient_deadline(monkeypatch):
token = search_deadline._current.set(None)
monkeypatch.setattr(dd.registry, "get_unavailable_reason", lambda _providers: None)
monkeypatch.setattr(dd.registry, "enabled_providers", lambda providers: (providers[0],))
yield
search_deadline._current.reset(token)
@@ -35,13 +38,13 @@ class _Selector:
def test_fetch_search_table_gives_up_when_the_budget_is_spent(monkeypatch):
"""Every mirror shares the protection, so another mirror is another wasted solve."""
fetches: list[str] = []
monkeypatch.setattr(dd.downloader, "html_get_page", lambda url, **_k: fetches.append(url) or "")
monkeypatch.setattr(dd.network, "get_available_aa_urls", lambda: ["https://annas-archive.gl"])
monkeypatch.setattr(aa.downloader, "html_get_page", lambda url, **_k: fetches.append(url) or "")
monkeypatch.setattr(aa.network, "get_available_aa_urls", lambda: ["https://annas-archive.gl"])
with search_deadline.search_deadline(60) as deadline:
deadline.event.set()
with pytest.raises(dd.SearchUnavailableError) as excinfo:
dd._fetch_search_table("https://annas-archive.gl/search?q=dune", _Selector())
with pytest.raises(aa.SearchUnavailableError) as excinfo:
aa._fetch_search_table("https://annas-archive.gl/search?q=dune", _Selector())
assert fetches == [], "no fetch should have been attempted"
assert "ran out of time" in str(excinfo.value)
@@ -49,11 +52,11 @@ def test_fetch_search_table_gives_up_when_the_budget_is_spent(monkeypatch):
def test_fetch_search_table_runs_normally_within_budget(monkeypatch):
page = "<html><body><main><table><tbody></tbody></table></main></body></html>"
monkeypatch.setattr(dd.downloader, "html_get_page", lambda _url, **_k: page)
monkeypatch.setattr(dd.network, "get_available_aa_urls", lambda: ["https://annas-archive.gl"])
monkeypatch.setattr(aa.downloader, "html_get_page", lambda _url, **_k: page)
monkeypatch.setattr(aa.network, "get_available_aa_urls", lambda: ["https://annas-archive.gl"])
with search_deadline.search_deadline(60):
html, table = dd._fetch_search_table("https://annas-archive.gl/search?q=dune", _Selector())
html, table = aa._fetch_search_table("https://annas-archive.gl/search?q=dune", _Selector())
assert table is not None
assert html == page
@@ -98,8 +101,8 @@ def test_title_variants_stop_once_the_budget_is_spent(monkeypatch):
deadline.event.set()
return []
monkeypatch.setattr(dd, "search_books", fake_search_books)
monkeypatch.setattr(dd, "_ensure_direct_download_available", lambda: None)
monkeypatch.setattr(aa, "search_books", fake_search_books)
monkeypatch.setattr(aa, "ensure_available", lambda: None)
source = dd.DirectDownloadSource()
with search_deadline.search_deadline(60):
@@ -111,8 +114,8 @@ def test_title_variants_stop_once_the_budget_is_spent(monkeypatch):
def test_all_title_variants_run_within_budget(monkeypatch):
queries: list[str] = []
monkeypatch.setattr(dd, "search_books", lambda q, _f: queries.append(q) or [])
monkeypatch.setattr(dd, "_ensure_direct_download_available", lambda: None)
monkeypatch.setattr(aa, "search_books", lambda q, _f: queries.append(q) or [])
monkeypatch.setattr(aa, "ensure_available", lambda: None)
source = dd.DirectDownloadSource()
with search_deadline.search_deadline(60):
@@ -134,8 +137,8 @@ def test_language_filter_retry_is_skipped_on_a_spent_budget(monkeypatch):
deadline.event.set()
return []
monkeypatch.setattr(dd, "search_books", fake_search_books)
monkeypatch.setattr(dd, "_ensure_direct_download_available", lambda: None)
monkeypatch.setattr(aa, "search_books", fake_search_books)
monkeypatch.setattr(aa, "ensure_available", lambda: None)
source = dd.DirectDownloadSource()
with search_deadline.search_deadline(60):
tests/direct_download/test_search_challenge_false_positive.py
+31 -32
View File
@@ -10,6 +10,8 @@ unsolved challenge, telling users to go fix a bypasser that had just succeeded.
import pytest
from bs4 import Tag
from shelfmark.release_sources.direct_download import annas_archive as aa
# Verbatim from a live annas-archive.pk 403, trimmed of nothing that matters: this is
# what an interstitial actually looks like, and it is under a kilobyte.
DDOS_GUARD_INTERSTITIAL = (
@@ -58,7 +60,7 @@ class _Selector:
def _patch_pages(monkeypatch, pages: list[str]):
import shelfmark.release_sources.direct_download as dd
from shelfmark.release_sources.direct_download import source as dd
calls: list[str] = []
@@ -66,8 +68,8 @@ def _patch_pages(monkeypatch, pages: list[str]):
calls.append(url)
return pages[len(calls) - 1] if len(calls) <= len(pages) else ""
monkeypatch.setattr(dd.downloader, "html_get_page", fake_get)
monkeypatch.setattr(dd.network, "get_available_aa_urls", lambda: ["a", "b"])
monkeypatch.setattr(aa.downloader, "html_get_page", fake_get)
monkeypatch.setattr(aa.network, "get_available_aa_urls", lambda: ["a", "b"])
return dd, calls
@@ -80,8 +82,6 @@ def search_logs():
"""
import logging
import shelfmark.release_sources.direct_download as dd
messages: list[str] = []
class _Capture(logging.Handler):
@@ -89,38 +89,37 @@ def search_logs():
messages.append(record.getMessage())
handler = _Capture()
dd.logger.addHandler(handler)
previous = dd.logger.level
dd.logger.setLevel(logging.DEBUG)
aa.logger.addHandler(handler)
previous = aa.logger.level
aa.logger.setLevel(logging.DEBUG)
# Logger.setLevel only invalidates the is-enabled cache through the manager, which
# these loggers are not registered with; without this the DEBUG line stays filtered.
dd.logger._cache.clear()
aa.logger._cache.clear()
try:
yield messages
finally:
dd.logger.removeHandler(handler)
dd.logger.setLevel(previous)
aa.logger.removeHandler(handler)
aa.logger.setLevel(previous)
def test_the_size_guard_is_what_separates_a_real_page_from_an_interstitial():
"""The two inputs this bug turned on, checked directly."""
import shelfmark.release_sources.direct_download as dd
from shelfmark.bypass.challenge import MAX_CHALLENGE_HTML_CHARS
assert len(DDOS_GUARD_INTERSTITIAL) < MAX_CHALLENGE_HTML_CHARS
assert len(AA_PAGE_WITHOUT_TABLE) > MAX_CHALLENGE_HTML_CHARS
# Both contain "ddos-guard"; only one is a challenge.
assert "ddos-guard" in AA_PAGE_WITHOUT_TABLE.lower()
assert dd._looks_like_challenge_page(DDOS_GUARD_INTERSTITIAL)
assert not dd._looks_like_challenge_page(AA_PAGE_WITHOUT_TABLE)
assert aa._looks_like_challenge_page(DDOS_GUARD_INTERSTITIAL)
assert not aa._looks_like_challenge_page(AA_PAGE_WITHOUT_TABLE)
def test_real_aa_page_without_a_table_is_not_reported_as_a_challenge(monkeypatch):
"""The #1289 failure: a served AA page raised "unsolved protection challenge"."""
dd, calls = _patch_pages(monkeypatch, [AA_PAGE_WITHOUT_TABLE])
_dd, calls = _patch_pages(monkeypatch, [AA_PAGE_WITHOUT_TABLE])
selector = _Selector(["https://real.test", "https://other.test"])
html, table = dd._fetch_search_table("https://real.test/search?q=malice", selector)
html, table = aa._fetch_search_table("https://real.test/search?q=malice", selector)
assert table is None
assert html == AA_PAGE_WITHOUT_TABLE
@@ -131,22 +130,22 @@ def test_real_aa_page_without_a_table_is_not_reported_as_a_challenge(monkeypatch
def test_aa_markers_win_over_challenge_markers_on_the_same_page(monkeypatch):
"""Ordering, not just the size guard, keeps a marker-carrying AA page readable."""
dd, _calls = _patch_pages(monkeypatch, [AA_PAGE_WITHOUT_TABLE])
monkeypatch.setattr(dd, "_looks_like_challenge_page", lambda _html: True)
_dd, _calls = _patch_pages(monkeypatch, [AA_PAGE_WITHOUT_TABLE])
monkeypatch.setattr(aa, "_looks_like_challenge_page", lambda _html: True)
selector = _Selector(["https://real.test", "https://other.test"])
_html, table = dd._fetch_search_table("https://real.test/search?q=malice", selector)
_html, table = aa._fetch_search_table("https://real.test/search?q=malice", selector)
assert table is None
def test_genuine_interstitial_still_raises(monkeypatch):
"""The behaviour the check exists for is untouched."""
dd, _calls = _patch_pages(monkeypatch, [DDOS_GUARD_INTERSTITIAL])
_dd, _calls = _patch_pages(monkeypatch, [DDOS_GUARD_INTERSTITIAL])
selector = _Selector(["https://real.test", "https://other.test"])
with pytest.raises(dd.SearchUnavailableError, match="protection challenge"):
dd._fetch_search_table("https://real.test/search?q=malice", selector)
with pytest.raises(aa.SearchUnavailableError, match="protection challenge"):
aa._fetch_search_table("https://real.test/search?q=malice", selector)
assert selector.quarantined == []
@@ -156,10 +155,10 @@ def test_results_table_is_still_returned(monkeypatch):
page = AA_PAGE_WITHOUT_TABLE.replace(
"<main>", "<main><table><tbody><tr><td>Malice</td></tr></tbody></table>"
)
dd, _calls = _patch_pages(monkeypatch, [page])
_dd, _calls = _patch_pages(monkeypatch, [page])
selector = _Selector(["https://real.test"])
_html, table = dd._fetch_search_table("https://real.test/search?q=malice", selector)
_html, table = aa._fetch_search_table("https://real.test/search?q=malice", selector)
assert isinstance(table, Tag)
@@ -171,10 +170,10 @@ def test_untabled_page_is_fingerprinted_in_the_log(monkeypatch, search_logs):
rather than reverse-engineered: size, whether the size guard applied, the AA markers
found, and the challenge marker (or its absence).
"""
dd, _calls = _patch_pages(monkeypatch, [AA_PAGE_WITHOUT_TABLE])
_dd, _calls = _patch_pages(monkeypatch, [AA_PAGE_WITHOUT_TABLE])
selector = _Selector(["https://real.test", "https://other.test"])
dd._fetch_search_table("https://real.test/search?q=malice", selector)
aa._fetch_search_table("https://real.test/search?q=malice", selector)
verdict = next(m for m in search_logs if "no results table" in m)
assert f"bytes={len(AA_PAGE_WITHOUT_TABLE)}" in verdict
@@ -190,11 +189,11 @@ def test_untabled_page_is_fingerprinted_in_the_log(monkeypatch, search_logs):
def test_interstitial_fingerprint_names_the_marker_that_proved_it(monkeypatch, search_logs):
"""The same line must also settle the opposite case, without needing the body."""
dd, _calls = _patch_pages(monkeypatch, [DDOS_GUARD_INTERSTITIAL])
_dd, _calls = _patch_pages(monkeypatch, [DDOS_GUARD_INTERSTITIAL])
selector = _Selector(["https://real.test", "https://other.test"])
with pytest.raises(dd.SearchUnavailableError):
dd._fetch_search_table("https://real.test/search?q=malice", selector)
with pytest.raises(aa.SearchUnavailableError):
aa._fetch_search_table("https://real.test/search?q=malice", selector)
verdict = next(m for m in search_logs if "no results table" in m)
assert "over_challenge_size_cap=False" in verdict
@@ -204,14 +203,14 @@ def test_interstitial_fingerprint_names_the_marker_that_proved_it(monkeypatch, s
def test_fingerprint_failure_never_breaks_a_search(monkeypatch):
"""Diagnostics are best-effort; a bug in them must not cost the user their search."""
dd, _calls = _patch_pages(monkeypatch, [AA_PAGE_WITHOUT_TABLE])
_dd, _calls = _patch_pages(monkeypatch, [AA_PAGE_WITHOUT_TABLE])
def boom(_html):
raise RuntimeError("marker scan blew up")
monkeypatch.setattr(dd, "challenge_marker", boom)
monkeypatch.setattr(aa, "challenge_marker", boom)
selector = _Selector(["https://real.test", "https://other.test"])
_html, table = dd._fetch_search_table("https://real.test/search?q=malice", selector)
_html, table = aa._fetch_search_table("https://real.test/search?q=malice", selector)
assert table is None
tests/direct_download/test_search_page_reuse.py
+44 -42
View File
@@ -11,8 +11,10 @@ solve, tens of seconds for nothing. See issue #1285.
import pytest
from bs4 import BeautifulSoup
import shelfmark.release_sources.direct_download as dd
from shelfmark.core import search_deadline
from shelfmark.core.config import config
from shelfmark.core.models import SearchFilters
from shelfmark.release_sources.direct_download import annas_archive as aa
@pytest.fixture(autouse=True)
@@ -39,9 +41,9 @@ _PAGE = "<html><body><main><table><tbody></tbody></table></main></body></html>"
def _count_fetches(monkeypatch) -> list[str]:
fetched: list[str] = []
monkeypatch.setattr(
dd.downloader, "html_get_page", lambda url, **_k: fetched.append(url) or _PAGE
aa.downloader, "html_get_page", lambda url, **_k: fetched.append(url) or _PAGE
)
monkeypatch.setattr(dd.network, "get_available_aa_urls", lambda: ["https://annas-archive.gl"])
monkeypatch.setattr(aa.network, "get_available_aa_urls", lambda: ["https://annas-archive.gl"])
return fetched
@@ -49,9 +51,9 @@ def test_repeated_url_is_fetched_once_within_one_search(monkeypatch):
fetched = _count_fetches(monkeypatch)
url = "https://annas-archive.gl/search?q=dune"
with dd._search_page_reuse():
first_html, first_table = dd._fetch_search_table(url, _Selector())
second_html, second_table = dd._fetch_search_table(url, _Selector())
with aa.search_page_reuse():
first_html, first_table = aa._fetch_search_table(url, _Selector())
second_html, second_table = aa._fetch_search_table(url, _Selector())
assert fetched == [url], "the second ask should have been served from the search's cache"
assert first_html == second_html
@@ -61,9 +63,9 @@ def test_repeated_url_is_fetched_once_within_one_search(monkeypatch):
def test_distinct_urls_are_still_fetched_separately(monkeypatch):
fetched = _count_fetches(monkeypatch)
with dd._search_page_reuse():
dd._fetch_search_table("https://annas-archive.gl/search?q=dune", _Selector())
dd._fetch_search_table("https://annas-archive.gl/search?q=dune&lang=en", _Selector())
with aa.search_page_reuse():
aa._fetch_search_table("https://annas-archive.gl/search?q=dune", _Selector())
aa._fetch_search_table("https://annas-archive.gl/search?q=dune&lang=en", _Selector())
assert len(fetched) == 2
@@ -73,10 +75,10 @@ def test_cache_does_not_leak_between_searches(monkeypatch):
fetched = _count_fetches(monkeypatch)
url = "https://annas-archive.gl/search?q=dune"
with dd._search_page_reuse():
dd._fetch_search_table(url, _Selector())
with dd._search_page_reuse():
dd._fetch_search_table(url, _Selector())
with aa.search_page_reuse():
aa._fetch_search_table(url, _Selector())
with aa.search_page_reuse():
aa._fetch_search_table(url, _Selector())
assert fetched == [url, url]
@@ -86,8 +88,8 @@ def test_without_the_context_every_fetch_still_goes_out(monkeypatch):
fetched = _count_fetches(monkeypatch)
url = "https://annas-archive.gl/search?q=dune"
dd._fetch_search_table(url, _Selector())
dd._fetch_search_table(url, _Selector())
aa._fetch_search_table(url, _Selector())
aa._fetch_search_table(url, _Selector())
assert fetched == [url, url]
@@ -97,12 +99,12 @@ def test_a_failure_is_not_cached(monkeypatch):
fetched = _count_fetches(monkeypatch)
url = "https://annas-archive.gl/search?q=dune"
with dd._search_page_reuse():
with aa.search_page_reuse():
with search_deadline.search_deadline(60) as deadline:
deadline.event.set()
with pytest.raises(dd.SearchUnavailableError):
dd._fetch_search_table(url, _Selector())
dd._fetch_search_table(url, _Selector())
with pytest.raises(aa.SearchUnavailableError):
aa._fetch_search_table(url, _Selector())
aa._fetch_search_table(url, _Selector())
assert fetched == [url], "the successful retry should be the only fetch"
@@ -119,14 +121,14 @@ def test_a_give_up_page_is_not_cached(monkeypatch):
parked = "<html><body>This domain is for sale.</body></html>"
fetched: list[str] = []
monkeypatch.setattr(
dd.downloader, "html_get_page", lambda url, **_k: fetched.append(url) or parked
aa.downloader, "html_get_page", lambda url, **_k: fetched.append(url) or parked
)
monkeypatch.setattr(dd.network, "get_available_aa_urls", lambda: ["https://annas-archive.gl"])
monkeypatch.setattr(aa.network, "get_available_aa_urls", lambda: ["https://annas-archive.gl"])
url = "https://annas-archive.gl/search?q=dune"
with dd._search_page_reuse():
assert dd._fetch_search_table(url, _Selector()) == (parked, None)
assert dd._fetch_search_table(url, _Selector()) == (parked, None)
with aa.search_page_reuse():
assert aa._fetch_search_table(url, _Selector()) == (parked, None)
assert aa._fetch_search_table(url, _Selector()) == (parked, None)
assert fetched == [url, url], "the second pass must not inherit the first's give-up"
@@ -136,14 +138,14 @@ def test_a_genuinely_empty_result_is_still_cached(monkeypatch):
empty = "<html><body><main>No files found. <a href='/md5/x'>x</a></main></body></html>"
fetched: list[str] = []
monkeypatch.setattr(
dd.downloader, "html_get_page", lambda url, **_k: fetched.append(url) or empty
aa.downloader, "html_get_page", lambda url, **_k: fetched.append(url) or empty
)
monkeypatch.setattr(dd.network, "get_available_aa_urls", lambda: ["https://annas-archive.gl"])
monkeypatch.setattr(aa.network, "get_available_aa_urls", lambda: ["https://annas-archive.gl"])
url = "https://annas-archive.gl/search?q=nothing"
with dd._search_page_reuse():
assert dd._fetch_search_table(url, _Selector()) == (empty, None)
assert dd._fetch_search_table(url, _Selector()) == (empty, None)
with aa.search_page_reuse():
assert aa._fetch_search_table(url, _Selector()) == (empty, None)
assert aa._fetch_search_table(url, _Selector()) == (empty, None)
assert fetched == [url], "an empty answer is an answer; re-solving for it buys nothing"
@@ -152,7 +154,7 @@ def test_language_retry_reuses_the_page_it_already_fetched(monkeypatch):
"""The end-to-end shape: language-from-path makes both passes build the same URL."""
fetched = _count_fetches(monkeypatch)
original_get = dd.config.get
original_get = config.get
def _fake_get(key: str, default=None, user_id=None):
del user_id
@@ -160,15 +162,15 @@ def test_language_retry_reuses_the_page_it_already_fetched(monkeypatch):
return True
return original_get(key, default)
monkeypatch.setattr(dd.config, "get", _fake_get)
monkeypatch.setattr(dd.network, "get_aa_base_url", lambda: "https://annas-archive.gl")
monkeypatch.setattr(config, "get", _fake_get)
monkeypatch.setattr(aa.network, "get_aa_base_url", lambda: "https://annas-archive.gl")
filters_with_lang = dd.SearchFilters(lang=["en"])
filters_without = dd.SearchFilters()
filters_with_lang = SearchFilters(lang=["en"])
filters_without = SearchFilters()
with dd._search_page_reuse():
dd.search_books("dune", filters_with_lang)
dd.search_books("dune", filters_without)
with aa.search_page_reuse():
aa.search_books("dune", filters_with_lang)
aa.search_books("dune", filters_without)
assert len(fetched) == 1, f"both passes build the same URL, got {fetched}"
assert "lang=" not in fetched[0]
@@ -179,12 +181,12 @@ def test_soup_reuse_is_safe_for_repeated_parsing(monkeypatch):
page = (
"<html><body><main><table><tbody><tr><td>row</td></tr></tbody></table></main></body></html>"
)
monkeypatch.setattr(dd.downloader, "html_get_page", lambda _url, **_k: page)
monkeypatch.setattr(dd.network, "get_available_aa_urls", lambda: ["https://annas-archive.gl"])
monkeypatch.setattr(aa.downloader, "html_get_page", lambda _url, **_k: page)
monkeypatch.setattr(aa.network, "get_available_aa_urls", lambda: ["https://annas-archive.gl"])
with dd._search_page_reuse():
_, first = dd._fetch_search_table("https://annas-archive.gl/search?q=dune", _Selector())
_, second = dd._fetch_search_table("https://annas-archive.gl/search?q=dune", _Selector())
with aa.search_page_reuse():
_, first = aa._fetch_search_table("https://annas-archive.gl/search?q=dune", _Selector())
_, second = aa._fetch_search_table("https://annas-archive.gl/search?q=dune", _Selector())
assert first is not None
assert second is not None
tests/direct_download/test_search_parked_mirror.py
+19 -18
View File
@@ -8,6 +8,8 @@ rotation and every later search pays for it again.
import pytest
from bs4 import Tag
from shelfmark.release_sources.direct_download import annas_archive as aa
PARKED_PAGE = """<!doctype html><html><head><title>annas-archive.li</title></head>
<body><h1>This domain is for sale</h1><p>Inquire now. Buy this domain.</p></body></html>"""
@@ -48,7 +50,7 @@ class _Selector:
def _patch_pages(monkeypatch, pages: list[str]):
"""Serve `pages` in order, recording the URL each call was made against."""
import shelfmark.release_sources.direct_download as dd
from shelfmark.release_sources.direct_download import source as dd
calls: list[str] = []
@@ -56,16 +58,16 @@ def _patch_pages(monkeypatch, pages: list[str]):
calls.append(url)
return pages[len(calls) - 1] if len(calls) <= len(pages) else ""
monkeypatch.setattr(dd.downloader, "html_get_page", fake_get)
monkeypatch.setattr(dd.network, "get_available_aa_urls", lambda: ["a", "b", "c"])
monkeypatch.setattr(aa.downloader, "html_get_page", fake_get)
monkeypatch.setattr(aa.network, "get_available_aa_urls", lambda: ["a", "b", "c"])
return dd, calls
def test_parked_mirror_is_quarantined_and_search_retries_next_mirror(monkeypatch):
dd, calls = _patch_pages(monkeypatch, [PARKED_PAGE, AA_RESULTS_PAGE])
_dd, calls = _patch_pages(monkeypatch, [PARKED_PAGE, AA_RESULTS_PAGE])
selector = _Selector(["https://parked.test", "https://real.test"])
html, table = dd._fetch_search_table("https://parked.test/search?q=dune", selector)
html, table = aa._fetch_search_table("https://parked.test/search?q=dune", selector)
assert selector.quarantined == ["https://parked.test"]
assert isinstance(table, Tag)
@@ -76,10 +78,10 @@ def test_parked_mirror_is_quarantined_and_search_retries_next_mirror(monkeypatch
def test_genuinely_empty_aa_result_does_not_quarantine(monkeypatch):
"""'No files found.' is a real answer from a healthy mirror."""
dd, _calls = _patch_pages(monkeypatch, [AA_EMPTY_PAGE])
_dd, _calls = _patch_pages(monkeypatch, [AA_EMPTY_PAGE])
selector = _Selector(["https://real.test", "https://other.test"])
html, table = dd._fetch_search_table("https://real.test/search?q=zzz", selector)
html, table = aa._fetch_search_table("https://real.test/search?q=zzz", selector)
assert selector.quarantined == []
assert table is None
@@ -92,29 +94,28 @@ def test_challenge_page_is_reported_not_passed_off_as_an_empty_result(monkeypatc
The mirror is alive and holds our clearance, so it must not be quarantined - but
returning it as "no table" made the caller tell the user their query found nothing.
"""
dd, _calls = _patch_pages(monkeypatch, [DDOS_GUARD_PAGE])
_dd, _calls = _patch_pages(monkeypatch, [DDOS_GUARD_PAGE])
selector = _Selector(["https://real.test", "https://other.test"])
with pytest.raises(dd.SearchUnavailableError, match="protection challenge"):
dd._fetch_search_table("https://real.test/search?q=dune", selector)
with pytest.raises(aa.SearchUnavailableError, match="protection challenge"):
aa._fetch_search_table("https://real.test/search?q=dune", selector)
assert selector.quarantined == []
def test_unreachable_mirror_raises_search_unavailable(monkeypatch):
dd, _calls = _patch_pages(monkeypatch, [""])
_dd, _calls = _patch_pages(monkeypatch, [""])
selector = _Selector(["https://real.test"])
try:
dd._fetch_search_table("https://real.test/search?q=dune", selector)
except dd.SearchUnavailableError:
aa._fetch_search_table("https://real.test/search?q=dune", selector)
except aa.SearchUnavailableError:
return
raise AssertionError("expected SearchUnavailableError")
def test_recorded_failure_reason_is_surfaced_to_the_caller(monkeypatch):
"""The concrete give-up reason html_get_page stashed replaces the generic line."""
import shelfmark.release_sources.direct_download as dd
reason = "Anna's Archive returned 403 (blocked) and no bypasser is enabled."
@@ -122,10 +123,10 @@ def test_recorded_failure_reason_is_surfaced_to_the_caller(monkeypatch):
selector.last_failure = reason
return ""
monkeypatch.setattr(dd.downloader, "html_get_page", fake_get)
monkeypatch.setattr(dd.network, "get_available_aa_urls", lambda: ["a"])
monkeypatch.setattr(aa.downloader, "html_get_page", fake_get)
monkeypatch.setattr(aa.network, "get_available_aa_urls", lambda: ["a"])
selector = _Selector(["https://real.test"])
selector.last_failure = None
with pytest.raises(dd.SearchUnavailableError, match="403 .blocked."):
dd._fetch_search_table("https://real.test/search?q=dune", selector)
with pytest.raises(aa.SearchUnavailableError, match="403 .blocked."):
aa._fetch_search_table("https://real.test/search?q=dune", selector)
tests/direct_download/test_search_queries.py
+39 -38
View File
@@ -1,8 +1,10 @@
from shelfmark.core.config import config
from shelfmark.core.models import SearchFilters
from shelfmark.core.search_plan import build_release_search_plan
from shelfmark.metadata_providers import BookMetadata
from shelfmark.release_sources import BrowseRecord
from shelfmark.release_sources.direct_download import DirectDownloadSource
from shelfmark.release_sources.direct_download import annas_archive as aa
def _browse_record(record_id: str, title: str) -> BrowseRecord:
@@ -10,9 +12,9 @@ def _browse_record(record_id: str, title: str) -> BrowseRecord:
def _enable_direct_download(monkeypatch):
import shelfmark.release_sources.direct_download as dd
from shelfmark.release_sources.direct_download import source as dd
original_get = dd.config.get
original_get = config.get
def _fake_get(key: str, default=None, user_id=None):
del user_id
@@ -20,7 +22,7 @@ def _enable_direct_download(monkeypatch):
return True
return original_get(key, default)
monkeypatch.setattr(dd.config, "get", _fake_get)
monkeypatch.setattr(config, "get", _fake_get)
monkeypatch.setattr("shelfmark.core.mirrors.has_aa_mirror_configuration", lambda: True)
return dd
@@ -33,9 +35,9 @@ class TestDirectDownloadSearchQueries:
captured.append(query)
return []
dd = _enable_direct_download(monkeypatch)
_enable_direct_download(monkeypatch)
monkeypatch.setattr(dd, "search_books", fake_search_books)
monkeypatch.setattr(aa, "search_books", fake_search_books)
source = DirectDownloadSource()
book = BookMetadata(
@@ -75,9 +77,9 @@ class TestDirectDownloadSearchQueries:
captured.append((query, filters.lang))
return records_by_query[query]
dd = _enable_direct_download(monkeypatch)
_enable_direct_download(monkeypatch)
monkeypatch.setattr(dd, "search_books", fake_search_books)
monkeypatch.setattr(aa, "search_books", fake_search_books)
source = DirectDownloadSource()
book = BookMetadata(
@@ -119,9 +121,9 @@ class TestDirectDownloadSearchQueries:
return []
return fallback_results[query]
dd = _enable_direct_download(monkeypatch)
_enable_direct_download(monkeypatch)
monkeypatch.setattr(dd, "search_books", fake_search_books)
monkeypatch.setattr(aa, "search_books", fake_search_books)
source = DirectDownloadSource()
book = BookMetadata(
@@ -157,9 +159,9 @@ class TestDirectDownloadSearchQueries:
return []
return [_browse_record("manual-1", "Manual result")]
dd = _enable_direct_download(monkeypatch)
_enable_direct_download(monkeypatch)
monkeypatch.setattr(dd, "search_books", fake_search_books)
monkeypatch.setattr(aa, "search_books", fake_search_books)
source = DirectDownloadSource()
book = BookMetadata(
@@ -188,9 +190,9 @@ class TestDirectDownloadSearchQueries:
def _patch_path_language(monkeypatch, enabled: bool = True):
import shelfmark.release_sources.direct_download as dd
from shelfmark.release_sources.direct_download import source as dd
original_get = dd.config.get
original_get = config.get
def _fake_get(key: str, default=None, user_id=None):
del user_id
@@ -198,7 +200,7 @@ def _patch_path_language(monkeypatch, enabled: bool = True):
return enabled
return original_get(key, default)
monkeypatch.setattr(dd.config, "get", _fake_get)
monkeypatch.setattr(config, "get", _fake_get)
return dd
@@ -228,59 +230,59 @@ def _make_row(distant_path: str, language: str = "", record_id: str = "rec-1") -
def test_detects_bracketed_language_from_distant_path(monkeypatch):
dd = _patch_path_language(monkeypatch)
_patch_path_language(monkeypatch)
row = _row_from_html(_make_row(r"lgli/N:\comics1\emule\2021.08.01\[BD FR] Scrameustache.cbz"))
record = dd._parse_search_result_row(row)
record = aa._parse_search_result_row(row)
assert record is not None
assert record.language == "fr"
assert record.download_path is not None
def test_detects_mixed_case_bracketed_language(monkeypatch):
dd = _patch_path_language(monkeypatch)
_patch_path_language(monkeypatch)
row = _row_from_html(_make_row(r"lgli/V:\comics\_0DAY3\[Fr]\BDs [Fr]\!Pdf\S\Book.pdf"))
record = dd._parse_search_result_row(row)
record = aa._parse_search_result_row(row)
assert record is not None
assert record.language == "fr"
def test_overrides_unknown_language_with_path_detection(monkeypatch):
dd = _patch_path_language(monkeypatch)
_patch_path_language(monkeypatch)
row = _row_from_html(_make_row(r"lgli/V:\comics\_0DAY3\[Fr]\Book.pdf", language="unknown"))
record = dd._parse_search_result_row(row)
record = aa._parse_search_result_row(row)
assert record is not None
assert record.language == "fr"
def test_sets_unknown_when_path_has_no_language(monkeypatch):
dd = _patch_path_language(monkeypatch)
_patch_path_language(monkeypatch)
row = _row_from_html(_make_row(r"lgli/N:\comics1\emule\NoLanguageHere.epub"))
record = dd._parse_search_result_row(row)
record = aa._parse_search_result_row(row)
assert record is not None
assert record.language == "unknown"
def test_avoids_en_false_positive_when_french_present(monkeypatch):
dd = _patch_path_language(monkeypatch)
_patch_path_language(monkeypatch)
row = _row_from_html(
_make_row(r"lgli/V:\comics\_0DAY2\Stripboeken Frans - BD en Français\[BD Fr] Book.cbr")
)
record = dd._parse_search_result_row(row)
record = aa._parse_search_result_row(row)
assert record is not None
assert record.language == "fr"
def test_keeps_row_with_missing_language_when_toggle_disabled(monkeypatch):
dd = _patch_path_language(monkeypatch, enabled=False)
_patch_path_language(monkeypatch, enabled=False)
row = _row_from_html(_make_row(r"lgli/N:\comics1\[BD FR] Scrameustache.cbz"))
record = dd._parse_search_result_row(row)
record = aa._parse_search_result_row(row)
assert record is not None
assert record.language is None
def test_keeps_sparse_lgli_row(monkeypatch):
"""lgli rows missing author/publisher/year must not be dropped."""
dd = _patch_path_language(monkeypatch)
_patch_path_language(monkeypatch)
html = r"""
<tr>
<td><a href="/md5/sparse-1"><img src="cover.jpg"></a></td>
@@ -292,7 +294,7 @@ def test_keeps_sparse_lgli_row(monkeypatch):
<td><span>lgli/N:\comics1\ftp\[BD.FR] French Comics\Book.cbz</span></td>
</tr>
"""
record = dd._parse_search_result_row(_row_from_html(html))
record = aa._parse_search_result_row(_row_from_html(html))
assert record is not None
assert record.id == "sparse-1"
assert record.language == "fr"
@@ -300,9 +302,9 @@ def test_keeps_sparse_lgli_row(monkeypatch):
def test_search_books_filters_locally_when_path_language_enabled(monkeypatch):
dd = _patch_path_language(monkeypatch)
monkeypatch.setattr(dd.network, "get_aa_base_url", lambda: "https://mirror.example")
monkeypatch.setattr(dd.network, "AAMirrorSelector", lambda: object())
_patch_path_language(monkeypatch)
monkeypatch.setattr(aa.network, "get_aa_base_url", lambda: "https://mirror.example")
monkeypatch.setattr(aa.network, "AAMirrorSelector", lambda: object())
captured_url: dict[str, str] = {}
@@ -332,9 +334,9 @@ def test_search_books_filters_locally_when_path_language_enabled(monkeypatch):
</table>
"""
monkeypatch.setattr(dd.downloader, "html_get_page", _fake_html_get_page)
monkeypatch.setattr(aa.downloader, "html_get_page", _fake_html_get_page)
records = dd.search_books("demo", SearchFilters(lang=["fr"], format=["pdf"]))
records = aa.search_books("demo", SearchFilters(lang=["fr"], format=["pdf"]))
assert "&lang=" not in captured_url["url"]
assert len(records) == 1
@@ -343,9 +345,8 @@ def test_search_books_filters_locally_when_path_language_enabled(monkeypatch):
def test_book_matches_requested_languages_logic():
import shelfmark.release_sources.direct_download as dd
assert dd._book_matches_requested_languages(None, {"fr"}) is True
assert dd._book_matches_requested_languages(None, set()) is True
assert dd._book_matches_requested_languages("en", {"fr"}) is False
assert dd._book_matches_requested_languages("fr", {"fr"}) is True
assert aa._book_matches_requested_languages(None, {"fr"}) is True
assert aa._book_matches_requested_languages(None, set()) is True
assert aa._book_matches_requested_languages("en", {"fr"}) is False
assert aa._book_matches_requested_languages("fr", {"fr"}) is True
tests/direct_download/test_source_availability.py
+9 -12
View File
@@ -2,7 +2,9 @@ from types import SimpleNamespace
import pytest
from shelfmark.core.config import config
from shelfmark.release_sources.direct_download import DirectDownloadSource, SearchUnavailableError
from shelfmark.release_sources.direct_download import annas_archive as aa
def _fake_config_get(values: dict[str, object]):
@@ -14,9 +16,8 @@ def _fake_config_get(values: dict[str, object]):
def test_direct_download_source_is_unavailable_when_disabled(monkeypatch):
import shelfmark.release_sources.direct_download as dd
monkeypatch.setattr(dd.config, "get", _fake_config_get({"DIRECT_DOWNLOAD_ENABLED": False}))
monkeypatch.setattr(config, "get", _fake_config_get({"DIRECT_DOWNLOAD_ENABLED": False}))
monkeypatch.setattr("shelfmark.core.mirrors.has_aa_mirror_configuration", lambda: True)
source = DirectDownloadSource()
@@ -28,9 +29,8 @@ def test_direct_download_source_is_unavailable_when_disabled(monkeypatch):
def test_direct_download_source_is_unavailable_without_aa_mirrors(monkeypatch):
import shelfmark.release_sources.direct_download as dd
monkeypatch.setattr(dd.config, "get", _fake_config_get({"DIRECT_DOWNLOAD_ENABLED": True}))
monkeypatch.setattr(config, "get", _fake_config_get({"DIRECT_DOWNLOAD_ENABLED": True}))
monkeypatch.setattr("shelfmark.core.mirrors.has_aa_mirror_configuration", lambda: False)
source = DirectDownloadSource()
@@ -42,9 +42,8 @@ def test_direct_download_source_is_unavailable_without_aa_mirrors(monkeypatch):
def test_direct_download_source_is_available_when_enabled_and_configured(monkeypatch):
import shelfmark.release_sources.direct_download as dd
monkeypatch.setattr(dd.config, "get", _fake_config_get({"DIRECT_DOWNLOAD_ENABLED": True}))
monkeypatch.setattr(config, "get", _fake_config_get({"DIRECT_DOWNLOAD_ENABLED": True}))
monkeypatch.setattr("shelfmark.core.mirrors.has_aa_mirror_configuration", lambda: True)
source = DirectDownloadSource()
@@ -53,10 +52,9 @@ def test_direct_download_source_is_available_when_enabled_and_configured(monkeyp
def test_get_source_priority_disables_entries_without_required_mirrors(monkeypatch):
import shelfmark.release_sources.direct_download as dd
monkeypatch.setattr(
dd.config,
config,
"get",
_fake_config_get(
{
@@ -77,7 +75,7 @@ def test_get_source_priority_disables_entries_without_required_mirrors(monkeypat
monkeypatch.setattr("shelfmark.core.mirrors.has_welib_mirror_configuration", lambda: False)
monkeypatch.setattr("shelfmark.core.mirrors.has_zlib_mirror_configuration", lambda: True)
priority = {item["id"]: item["enabled"] for item in dd._get_source_priority()}
priority = {item["id"]: item["enabled"] for item in aa._get_source_priority()}
assert priority["aa-fast"] is False
assert priority["libgen"] is True
@@ -86,12 +84,11 @@ def test_get_source_priority_disables_entries_without_required_mirrors(monkeypat
def test_is_configured_zlib_link_uses_configured_mirror_domains(monkeypatch):
import shelfmark.release_sources.direct_download as dd
monkeypatch.setattr(
"shelfmark.core.mirrors.get_zlib_cookie_domains",
lambda: {"custom-zlib.example"},
)
assert dd._is_configured_zlib_link("https://custom-zlib.example/books/example") is True
assert dd._is_configured_zlib_link("https://other-zlib.example/books/example") is False
assert aa._is_configured_zlib_link("https://custom-zlib.example/books/example") is True
assert aa._is_configured_zlib_link("https://other-zlib.example/books/example") is False
tests/direct_download/test_untabled_page_names_the_answering_mirror.py
+20 -22
View File
@@ -11,6 +11,8 @@ import logging
import pytest
from shelfmark.release_sources.direct_download import annas_archive as aa
# A protection challenge, so the fingerprint line fires without looking like AA.
CHALLENGE_PAGE = (
"<html><head><title>DDOS-GUARD</title>"
@@ -26,7 +28,6 @@ def search_logs():
setup_logger builds loggers outside the standard hierarchy, so their records never
reach the root handler caplog installs - see tests/bypass/test_ddg_cookie_reuse.py.
"""
import shelfmark.release_sources.direct_download as dd
messages: list[str] = []
@@ -35,15 +36,15 @@ def search_logs():
messages.append(record.getMessage())
handler = _Capture()
dd.logger.addHandler(handler)
previous = dd.logger.level
dd.logger.setLevel(logging.DEBUG)
dd.logger._cache.clear()
aa.logger.addHandler(handler)
previous = aa.logger.level
aa.logger.setLevel(logging.DEBUG)
aa.logger._cache.clear()
try:
yield messages
finally:
dd.logger.removeHandler(handler)
dd.logger.setLevel(previous)
aa.logger.removeHandler(handler)
aa.logger.setLevel(previous)
class _Selector:
@@ -62,7 +63,6 @@ ANSWERED = "https://annas-archive.pk/search?q=Ken+follett"
def test_the_fingerprint_names_the_mirror_that_answered(monkeypatch, search_logs):
import shelfmark.release_sources.direct_download as dd
def fake_get(url, **kwargs):
# The caller must ask for it, or there is nothing to report.
@@ -71,11 +71,11 @@ def test_the_fingerprint_names_the_mirror_that_answered(monkeypatch, search_logs
# What an internal rotation looks like from the outside: a different host.
return CHALLENGE_PAGE, ANSWERED
monkeypatch.setattr(dd.downloader, "html_get_page", fake_get)
monkeypatch.setattr(dd.network, "get_available_aa_urls", lambda: ["a"])
monkeypatch.setattr(aa.downloader, "html_get_page", fake_get)
monkeypatch.setattr(aa.network, "get_available_aa_urls", lambda: ["a"])
with pytest.raises(dd.SearchUnavailableError):
dd._fetch_search_table_uncached(REQUESTED, _Selector())
with pytest.raises(aa.SearchUnavailableError):
aa._fetch_search_table_uncached(REQUESTED, _Selector())
fingerprint = [m for m in search_logs if m.startswith("Search page has no results table")]
assert len(fingerprint) == 1
@@ -85,13 +85,12 @@ def test_the_fingerprint_names_the_mirror_that_answered(monkeypatch, search_logs
def test_a_downloader_that_reports_no_url_falls_back_to_the_request(monkeypatch, search_logs):
"""The plain-string shape stays supported; the line is still worth having."""
import shelfmark.release_sources.direct_download as dd
monkeypatch.setattr(dd.downloader, "html_get_page", lambda _url, **_k: CHALLENGE_PAGE)
monkeypatch.setattr(dd.network, "get_available_aa_urls", lambda: ["a"])
monkeypatch.setattr(aa.downloader, "html_get_page", lambda _url, **_k: CHALLENGE_PAGE)
monkeypatch.setattr(aa.network, "get_available_aa_urls", lambda: ["a"])
with pytest.raises(dd.SearchUnavailableError):
dd._fetch_search_table_uncached(REQUESTED, _Selector())
with pytest.raises(aa.SearchUnavailableError):
aa._fetch_search_table_uncached(REQUESTED, _Selector())
fingerprint = [m for m in search_logs if m.startswith("Search page has no results table")]
assert len(fingerprint) == 1
@@ -100,15 +99,14 @@ def test_a_downloader_that_reports_no_url_falls_back_to_the_request(monkeypatch,
def test_the_empty_body_give_up_survives_the_tuple_shape(monkeypatch):
"""`("", url)` is truthy, so the exhaustion check has to read the body."""
import shelfmark.release_sources.direct_download as dd
monkeypatch.setattr(dd.downloader, "html_get_page", lambda _url, **_k: ("", REQUESTED))
monkeypatch.setattr(dd.network, "get_available_aa_urls", lambda: ["a"])
monkeypatch.setattr(aa.downloader, "html_get_page", lambda _url, **_k: ("", REQUESTED))
monkeypatch.setattr(aa.network, "get_available_aa_urls", lambda: ["a"])
selector = _Selector()
selector.last_failure = "Every mirror refused the connection."
with pytest.raises(dd.SearchUnavailableError) as excinfo:
dd._fetch_search_table_uncached(REQUESTED, selector)
with pytest.raises(aa.SearchUnavailableError) as excinfo:
aa._fetch_search_table_uncached(REQUESTED, selector)
assert "Every mirror refused the connection." in str(excinfo.value)
tests/direct_download/test_web_providers.py
+185
View File
@@ -0,0 +1,185 @@
from types import SimpleNamespace
import pytest
from shelfmark.core.models import SearchFilters
from shelfmark.release_sources.direct_download import handler, registry
from shelfmark.release_sources.direct_download.common import ParsedSearchResult, parse_search_page
def test_shared_parser_accepts_provider_specific_extraction():
from bs4 import BeautifulSoup, Tag
soup = BeautifulSoup('<li data-url="https://books.example/book">Example</li>', "html.parser")
def extract(item: Tag) -> ParsedSearchResult:
return ParsedSearchResult(
key=item["data-url"],
title=item.get_text(strip=True),
formats=("epub", "pdf"),
language="English",
source_url=item["data-url"],
)
records = parse_search_page(
soup,
SearchFilters(format=["epub"], lang=["en"]),
provider_id="example",
item_selector="li",
extract_item=extract,
)
assert len(records) == 1
assert records[0].id.startswith("example:")
assert records[0].format == "epub"
assert records[0].language == "en"
def test_web_provider_registry_dispatches_without_source_changes(monkeypatch, tmp_path):
from shelfmark.release_sources.direct_download import source as dd
source_url = "https://books.example/example.epub"
class ExampleProvider:
id = "example"
display_name = "Example"
def is_enabled(self):
return True
def handles(self, url):
return url.startswith("https://books.example/")
def search(self, book, plan, **kwargs):
del book, plan, kwargs
return [
dd.BrowseRecord(
id="example:1",
title="Example",
source="direct_download",
format="epub",
source_url=source_url,
)
]
def download(self, book_info, book_path, *callbacks):
del book_info, callbacks
book_path.write_bytes(b"x" * 12_000)
return source_url
monkeypatch.setattr(registry, "PROVIDER_TYPES", (ExampleProvider,))
monkeypatch.setattr(
registry.config,
"get",
lambda key, default=None: True if key == "DIRECT_DOWNLOAD_ENABLED" else default,
)
source = dd.DirectDownloadSource()
plan = SimpleNamespace(
manual_query="Example",
primary_query=None,
source_filters=None,
languages=None,
)
releases = source.search(SimpleNamespace(title="Example"), plan)
assert len(releases) == 1
release = releases[0]
record = ExampleProvider().search(SimpleNamespace(), plan)[0]
destination = tmp_path / "example.epub"
assert release.extra["web_provider"] == "example"
assert handler._download_book(record, destination) == source_url
assert destination.stat().st_size == 12_000
def test_md5_record_routes_to_annas_archive_cascade(monkeypatch, tmp_path):
from threading import Event
from shelfmark.release_sources.direct_download import annas_archive
from shelfmark.release_sources.direct_download import source as dd
record = dd.BrowseRecord(
id="0123456789abcdef0123456789abcdef",
title="Example",
source="direct_download",
format="epub",
)
destination = tmp_path / "example.epub"
cancel_flag = Event()
calls = []
def download(*args):
calls.append(args)
return "https://mirror.example/file.epub"
monkeypatch.setattr(annas_archive, "download_book", download)
assert handler._download_book(record, destination, cancel_flag=cancel_flag) == (
"https://mirror.example/file.epub"
)
assert calls == [(record, destination, None, cancel_flag, None)]
def test_unknown_url_is_not_silently_routed_to_annas_archive(monkeypatch, tmp_path):
from shelfmark.release_sources.direct_download import annas_archive
monkeypatch.setattr(
annas_archive,
"download_book",
lambda *_args, **_kwargs: pytest.fail("unknown provider reached Anna's Archive"),
)
record = handler.BrowseRecord(
id="unknown-record",
title="Example",
source="direct_download",
source_url="https://unknown.example/book",
)
with pytest.raises(RuntimeError, match="No Direct Download provider owns"):
handler._download_book(record, tmp_path / "example.epub")
def test_provider_failure_is_suppressed_when_another_provider_succeeds(monkeypatch):
from shelfmark.release_sources import BrowseRecord
from shelfmark.release_sources.direct_download.common import (
DirectDownloadUnavailableError,
)
from shelfmark.release_sources.direct_download.source import DirectDownloadSource
class FailingProvider:
id = "failing"
display_name = "Failing"
def is_enabled(self):
return True
def search(self, *_args, **_kwargs):
raise DirectDownloadUnavailableError("provider unavailable")
class WorkingProvider:
id = "working"
display_name = "Working"
def is_enabled(self):
return True
def search(self, *_args, **_kwargs):
return [
BrowseRecord(
id="working:1",
title="Example",
source="direct_download",
source_url="https://working.example/book",
)
]
monkeypatch.setattr(
registry.config,
"get",
lambda key, default=None: True if key == "DIRECT_DOWNLOAD_ENABLED" else default,
)
source = DirectDownloadSource()
source._providers = (FailingProvider(), WorkingProvider())
releases = source.search(SimpleNamespace(title="Example"), SimpleNamespace())
assert [release.source_id for release in releases] == ["working:1"]