diff --git a/shelfmark/release_sources/__init__.py b/shelfmark/release_sources/__init__.py index 8901ee7e..6495c50e 100644 --- a/shelfmark/release_sources/__init__.py +++ b/shelfmark/release_sources/__init__.py @@ -431,6 +431,7 @@ _BUILTIN_SOURCE_MODULES = ( "shelfmark.release_sources.audiobookbay", "shelfmark.release_sources.direct_download", "shelfmark.release_sources.irc", + "shelfmark.release_sources.libgen", "shelfmark.release_sources.newznab", "shelfmark.release_sources.prowlarr", ) diff --git a/shelfmark/release_sources/libgen/__init__.py b/shelfmark/release_sources/libgen/__init__.py new file mode 100644 index 00000000..b66fbc28 --- /dev/null +++ b/shelfmark/release_sources/libgen/__init__.py @@ -0,0 +1,6 @@ +"""Libgen release source - direct catalogue search over the libgen.li family.""" + +# Import to trigger registration +from shelfmark.release_sources.libgen import handler as handler +from shelfmark.release_sources.libgen import settings as settings +from shelfmark.release_sources.libgen import source as source diff --git a/shelfmark/release_sources/libgen/handler.py b/shelfmark/release_sources/libgen/handler.py new file mode 100644 index 00000000..4ca79825 --- /dev/null +++ b/shelfmark/release_sources/libgen/handler.py @@ -0,0 +1,114 @@ +"""Libgen download handler - resolves an md5 to a file via the ads.php cascade. + +Selected by ``get_handler(task.source)`` for ``source == "libgen"``. It mirrors +DirectDownloadHandler's shape (stage into TMP_DIR, let the orchestrator post-process) but +only knows the libgen ``ads.php?md5= -> get.php`` path, keyed on the md5 the search source +put in ``source_id``. +""" + +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 build_filename +from shelfmark.download import http as downloader +from shelfmark.release_sources import DownloadHandler, register_handler +from shelfmark.release_sources.libgen import scraper + +if TYPE_CHECKING: + from collections.abc import Callable + from threading import Event + + from shelfmark.core.models import DownloadTask + +logger = setup_logger(__name__) + +# Files under this size are almost certainly an error/challenge page, not a book. Same +# threshold direct_download uses; duplicated to keep the package self-contained. +_MIN_VALID_FILE_SIZE = 10 * 1024 + + +@register_handler("libgen") +class LibgenHandler(DownloadHandler): + """Download handler for Libgen search releases.""" + + def download( + self, + task: DownloadTask, + cancel_flag: Event, + progress_callback: Callable[[float], None], + status_callback: Callable[[str, str | None], None], + ) -> str | None: + """Resolve the md5 through each configured mirror and download the file. + + Returns the staged temp path on success (orchestrator handles post-processing) or + None if every mirror fails. + """ + from shelfmark.core import mirrors + + try: + if cancel_flag.is_set(): + status_callback("cancelled", "Cancelled") + return None + + # source_id was namespaced "libgen:" to avoid a queue-key collision with + # direct_download; strip it back to the bare (lowercase) md5 the download page expects. + md5 = task.task_id.split(":", 1)[-1].lower() + + if config.get("FILE_ORGANIZATION", "rename") == "none": + book_name = f"{md5}.{task.format or 'bin'}" + else: + book_name = build_filename(task.title, task.author, task.year, task.format) + book_path = TMP_DIR / book_name + + for base in mirrors.get_libgen_mirrors(): + if cancel_flag.is_set(): + status_callback("cancelled", "Cancelled") + return None + + ads_url = f"{base.rstrip('/')}/ads.php?md5={md5}" + status_callback("resolving", "Resolving Libgen") + ads_html = scraper.fetch_page(ads_url, (5, 10)) + if not ads_html: + continue + get_url = scraper.resolve_download_url(ads_html, base) + if not get_url: + continue + + # _selector=None: download_url builds its own AAMirrorSelector (a no-op for + # non-AA URLs), so we avoid initialising dead AA-mirror state here. + data = downloader.download_url( + get_url, + task.size or "", + progress_callback, + cancel_flag, + None, + status_callback, + referer=ads_url, + ) + if not data: + continue + if data.tell() < _MIN_VALID_FILE_SIZE: + logger.warning("Libgen file too small from %s, treating as failure", base) + continue + + data.seek(0) + with book_path.open("wb") as file: + file.write(data.getbuffer()) + return str(book_path) + except Exception as exc: + if cancel_flag.is_set(): + status_callback("cancelled", "Cancelled") + else: + logger.exception("Error downloading from Libgen") + status_callback("error", str(exc)) + return None + else: + # Loop exhausted without returning: every mirror failed to resolve/download. + status_callback("error", "All Libgen mirrors failed") + return None + + def cancel(self, task_id: str) -> bool: + """Cancellation is handled by the orchestrator via the cancel_flag.""" + return False diff --git a/shelfmark/release_sources/libgen/scraper.py b/shelfmark/release_sources/libgen/scraper.py new file mode 100644 index 00000000..c3a6f9a9 --- /dev/null +++ b/shelfmark/release_sources/libgen/scraper.py @@ -0,0 +1,252 @@ +"""Libgen catalogue scraping: search results and download-link resolution. + +This is the pure fetch+parse core of the Libgen source. It talks to the libgen.li +family of mirrors (``index.php?req=`` search, ``ads.php?md5=`` download pages) using +plain HTTP -- these mirrors are not behind DDoS-Guard, so no browser/bypasser is +needed. All shelfmark-stateful behaviour lives in source.py/handler.py. +""" + +import re +from http import HTTPStatus +from urllib.parse import quote + +import requests +from bs4 import BeautifulSoup, Tag + +from shelfmark.core.languages import normalize_language +from shelfmark.core.logger import setup_logger +from shelfmark.download import http as downloader +from shelfmark.download import network +from shelfmark.release_sources import BrowseRecord + +logger = setup_logger(__name__) + +# The libgen.li results table. Both full (9-cell) and compact (5-cell) rows live in it. +_RESULTS_TABLE_ID = "tablelibgen" + +# md5 appears in the row's Mirrors cell as get.php?md5= and an AA /md5/ link. +_MD5_RE = re.compile(r"md5=([0-9a-f]{32})", re.IGNORECASE) + +# Patterns for the keyed GET link on an ads.php page. Kept in sync with the resolution +# libgen download has always used (direct_download._LIBGEN_GET_PATTERNS); duplicated here +# on purpose so the Libgen source stays self-contained and does not import that module's +# internals (which an in-flight upstream refactor is relocating). +_GET_KEY_PATTERNS = [ + re.compile( + r']*>\s*' + r"]*>GET\s*", + re.IGNORECASE, + ), + re.compile( + r']+href=["\']([^"\']*get\.php\?md5=[^"\']+&(?:amp;)?key=[^"\']+)["\']', + re.IGNORECASE, + ), + re.compile( + r']*>[\s\S]*?]*>GET', + re.IGNORECASE, + ), + re.compile( + r'href=["\']([^"\']*get\.php\?[^"\']*md5=[^"\']*&[^"\']*key=[^"\']+)["\']', + re.IGNORECASE, + ), +] + +# Labels that terminate a metadata value on an ads.php page, so e.g. "Year: 2003 ISBN: ..." +# stops Year at "ISBN:" rather than swallowing it. We only emit a subset (see _parse_ads_metadata). +_METADATA_STOP_LABELS = [ + "Title", + "Series", + "Author(s)", + "Publisher", + "Year", + "Language", + "Pages", + "ISBN", + "Edition", + "Extension", + "Size", + "Time added", + "ID", + "Filename", + "Description", +] + + +def fetch_page(url: str, timeout: tuple[int, int] = (5, 15)) -> str | None: + """GET a libgen page, returning its text on HTTP 200 or None on any failure. + + Public (not underscore-prefixed) because handler.py fetches ads.php pages through it + and tests patch it. Uses the app's proxy/SSL/DNS configuration so egress stays on + whatever network the container is bound to (the VPN namespace, in the deployed stack). + """ + try: + response = requests.get( + url, + headers=downloader.DOWNLOAD_HEADERS, + timeout=timeout, + allow_redirects=True, + proxies=network.get_proxies(url), + verify=network.get_ssl_verify(url), + ) + except requests.exceptions.RequestException as exc: + logger.debug("Libgen fetch failed for %s: %s", url, exc) + return None + if response.status_code != HTTPStatus.OK: + logger.debug("Libgen fetch %s returned %s", url, response.status_code) + return None + return response.text + + +def search_libgen( + query: str, + mirrors: list[str], + *, + max_results: int, + timeout: tuple[int, int] = (5, 15), +) -> list[BrowseRecord]: + """Search each mirror's catalogue until one answers with a results table. + + The first mirror that returns a parseable ``#tablelibgen`` wins -- including when that + table is empty ([] is returned as final). Mirrors can lag independently, but falling + through on every empty result would multiply latency under the shared search deadline, + so an empty-but-well-formed answer is trusted rather than re-queried elsewhere. + """ + for base in mirrors: + url = f"{base.rstrip('/')}/index.php?req={quote(query)}&res={max_results}" + html = fetch_page(url, timeout) + if html is None: + continue + records = _parse_results(html, base) + if records is not None: + return records + return [] + + +def fetch_record_by_md5( + md5: str, + mirrors: list[str], + *, + timeout: tuple[int, int] = (5, 10), +) -> BrowseRecord | None: + """Resolve a single record from its md5 by parsing an ads.php page's metadata. + + libgen's ``index.php?req=`` does not match on md5 (req= indexes title/author/ + description), so md5 -> record must go through the ads.php page instead. + """ + for base in mirrors: + html = fetch_page(f"{base.rstrip('/')}/ads.php?md5={md5}", timeout) + if html is None: + continue + record = _parse_ads_metadata(html, md5, base) + if record is not None: + return record + return None + + +def resolve_download_url(ads_html: str, base_url: str) -> str | None: + """Extract the keyed get.php download URL from an ads.php page, or None.""" + if "get.php" not in ads_html: + return None + for pattern in _GET_KEY_PATTERNS: + match = pattern.search(ads_html) + if not match: + continue + url = match.group(1).replace("&", "&").replace(">", ">").replace("<", "<") + if not url.startswith("http"): + url = f"{base_url.rstrip('/')}/{url.lstrip('/')}" + return url + return None + + +def _cell_text(cell: Tag) -> str: + """Cell text with runs of whitespace (incl.   / \\xa0) collapsed to single spaces.""" + return re.sub(r"\s+", " ", cell.get_text(" ", strip=True)).strip() + + +def _parse_results(html: str, base_url: str) -> list[BrowseRecord] | None: + """Parse a libgen search page. + + Returns None when the page has no results table (a challenge/error page -> the caller + tries the next mirror), or a list (possibly empty) when the table is present. + + Row shapes vary and carry NO rowspans: full rows have 9 cells + ``[Title, Author, Publisher, Year, Language, Pages, Size, Ext, Mirrors]`` and compact + rows (extra files under one edition) have 5 ``[Title, Pages, Size, Ext, Mirrors]``. The + file-level columns are stable from the right, so index from the end: Mirrors[-1] (md5), + Ext[-2], Size[-3]. Title is always [0]. Author/Language exist only on full rows. + + The two shapes are the only ones libgen.li is known to emit; an unexpected width just + fails safe (author/language read as None) rather than mis-columning. + """ + soup = BeautifulSoup(html, "html.parser") + table = soup.find("table", id=_RESULTS_TABLE_ID) + if not isinstance(table, Tag): + return None + + records: list[BrowseRecord] = [] + for row in table.find_all("tr")[1:]: # skip the header row + cells = row.find_all("td") + if len(cells) < 5: + continue + # Scope the md5 to the Mirrors cell (last column): scanning the whole row could match + # an md5-shaped string elsewhere (e.g. a cover-image URL) and misattribute it. + md5_match = _MD5_RE.search(str(cells[-1])) + if not md5_match: + continue # spacer/section rows carry no md5 + md5 = md5_match.group(1).lower() + + title = _cell_text(cells[0]) + fmt = _cell_text(cells[-2]).lower() or None + size = _cell_text(cells[-3]) or None + + author = None + language = None + if len(cells) >= 9: # full row: middle metadata columns are present + author = _cell_text(cells[1]) or None + language = normalize_language(_cell_text(cells[4])) + + records.append( + BrowseRecord( + id=md5, + title=title, + source="libgen", + author=author, + language=language, + size=size, + format=fmt, + source_url=f"{base_url.rstrip('/')}/ads.php?md5={md5}", + ) + ) + return records + + +def _parse_ads_metadata(html: str, md5: str, base_url: str) -> BrowseRecord | None: + """Build a BrowseRecord from an ads.php page's labelled metadata. + + The page's metadata lives in a deeply nested table, so read it from the visible text by + label rather than by cell position -- the labels (Title:, Series:, Author(s): ...) are + stable even though the surrounding markup is not. Returns None if the page has no title. + """ + text = re.sub(r"\s+", " ", BeautifulSoup(html, "html.parser").get_text(" ", strip=True)) + + def field(name: str) -> str | None: + others = "|".join( + re.escape(other) + r":" for other in _METADATA_STOP_LABELS if other != name + ) + match = re.search(re.escape(name) + r":\s*(.*?)\s*(?:" + others + r"|$)", text) + value = match.group(1).strip() if match else "" + return value or None + + title = field("Title") + if not title: + return None + return BrowseRecord( + id=md5, + title=title, + source="libgen", + author=field("Author(s)"), + publisher=field("Publisher"), + year=field("Year"), + language=normalize_language(field("Language") or ""), + source_url=f"{base_url.rstrip('/')}/ads.php?md5={md5}", + ) diff --git a/shelfmark/release_sources/libgen/settings.py b/shelfmark/release_sources/libgen/settings.py new file mode 100644 index 00000000..a97273a0 --- /dev/null +++ b/shelfmark/release_sources/libgen/settings.py @@ -0,0 +1,34 @@ +"""Libgen search settings registration.""" + +from shelfmark.core.settings_registry import ( + CheckboxField, + NumberField, + SettingsField, + register_settings, +) + + +@register_settings("libgen_config", "Libgen Search", icon="download", order=46) +def libgen_config_settings() -> list[SettingsField]: + """Libgen search configuration settings.""" + return [ + CheckboxField( + key="LIBGEN_SEARCH_ENABLED", + label="Enable Libgen Search", + description=( + "Search the Libgen catalogue directly, including CBZ/CBR comics and manga " + "that Anna's Archive does not index. Uses the Libgen mirrors configured " + "under Mirrors for both search and download." + ), + default=False, + ), + NumberField( + key="LIBGEN_SEARCH_MAX_RESULTS", + label="Max Results", + description="Maximum number of results to request per search (1-100).", + default=25, + min_value=1, + max_value=100, + show_when={"field": "LIBGEN_SEARCH_ENABLED", "value": True}, + ), + ] diff --git a/shelfmark/release_sources/libgen/source.py b/shelfmark/release_sources/libgen/source.py new file mode 100644 index 00000000..c4988fa5 --- /dev/null +++ b/shelfmark/release_sources/libgen/source.py @@ -0,0 +1,198 @@ +"""Libgen release source - searches the libgen catalogue directly. + +Anna's Archive is shelfmark's only other web search source, and libgen appears there +purely as a download mirror keyed by an AA md5. This source searches libgen's own +catalogue, which surfaces content AA does not index -- most visibly CBZ/CBR comics and +manga volumes. Downloads reuse the existing ``ads.php?md5=`` resolution (see handler.py). +""" + +from typing import TYPE_CHECKING, ClassVar + +from shelfmark.core import mirrors +from shelfmark.core.config import config +from shelfmark.core.logger import setup_logger +from shelfmark.release_sources import ( + BrowseRecord, + ColumnAlign, + ColumnColorHint, + ColumnRenderType, + ColumnSchema, + Release, + ReleaseColumnConfig, + ReleaseProtocol, + ReleaseSource, + register_source, +) +from shelfmark.release_sources.libgen import scraper + +if TYPE_CHECKING: + from shelfmark.core.models import DownloadTask # noqa: F401 + from shelfmark.core.search_plan import ReleaseSearchPlan + from shelfmark.metadata_providers import BookMetadata + +logger = setup_logger(__name__) + +_DEFAULT_MAX_RESULTS = 25 + + +def _coerce_positive_int(value: object, default: int) -> int: + """Return a positive integer config value or the provided default.""" + if isinstance(value, bool): + return default + if isinstance(value, int) and value > 0: + return value + return default + + +def _build_query_candidates(plan: ReleaseSearchPlan, book: BookMetadata) -> list[str]: + """Build ordered, de-duplicated search queries from the plan (mirrors AudiobookBay).""" + candidates: list[str] = [] + if plan.manual_query: + candidates.append(plan.manual_query.strip()) + elif plan.title_variants: + variant = plan.title_variants[0] + combined = f"{variant.title} {variant.author}".strip() + title_only = (variant.title or "").strip() + if combined: + candidates.append(combined) + if title_only and title_only.lower() != combined.lower(): + candidates.append(title_only) + elif book.title: + candidates.append(book.title.strip()) + + deduped: list[str] = [] + seen: set[str] = set() + for candidate in candidates: + normalized = candidate.strip() + if not normalized or normalized.lower() in seen: + continue + seen.add(normalized.lower()) + deduped.append(normalized) + return deduped + + +@register_source("libgen") +class LibgenSource(ReleaseSource): + """Release source that searches the libgen catalogue for downloadable files.""" + + name = "libgen" + display_name = "Libgen" + supported_content_types: ClassVar[list[str]] = ["ebook"] # incl. comics/manga (cbz/cbr) + + def is_available(self) -> bool: + """Available only when explicitly enabled and libgen mirrors are configured. + + ``is True`` rather than ``bool(...)`` matches the AudiobookBay idiom and avoids a + truthy string ever enabling network egress to an unmoderated site. + """ + return ( + config.get("LIBGEN_SEARCH_ENABLED", False) is True + and mirrors.has_libgen_mirror_configuration() + ) + + def search( + self, + book: BookMetadata, + plan: ReleaseSearchPlan, + *, + expand_search: bool = False, + content_type: str = "ebook", + ) -> list[Release]: + """Search libgen for releases of a book.""" + if content_type != "ebook": + return [] + if not self.is_available(): + return [] + + queries = _build_query_candidates(plan, book) + if not queries: + return [] + max_results = _coerce_positive_int( + config.get("LIBGEN_SEARCH_MAX_RESULTS", _DEFAULT_MAX_RESULTS), _DEFAULT_MAX_RESULTS + ) + mirror_list = mirrors.get_libgen_mirrors() + + # One search_libgen call per candidate; it already retries every mirror internally. + # Worst case (all mirrors dead) stays within the shared search deadline. + for query in queries: + logger.info("Searching Libgen for: %s", query) + records = scraper.search_libgen(query, mirror_list, max_results=max_results) + if records: + return [self._record_to_release(record) for record in records] + return [] + + def _record_to_release(self, record: BrowseRecord) -> Release: + """Normalize a libgen catalogue record into a Release. + + ``source_id`` is namespaced ``libgen:`` so the download queue key never + collides with a direct_download release for the same md5 (Anna's Archive heavily + indexes libgen, so the same md5 routinely appears from both sources). The handler + strips the prefix back to the bare md5. + """ + return Release( + source="libgen", + source_id=f"libgen:{record.id}", + title=record.title, + format=record.format, + language=record.language, + size=record.size, + download_url=None, # handler builds ads.php?md5= from the md5 + info_url=record.source_url, + protocol=ReleaseProtocol.HTTP, + indexer="Libgen", + content_type="ebook", + extra={ + "author": record.author, + "year": record.year, + "md5": record.id, + "language": record.language, + }, + ) + + def search_results_are_releases(self) -> bool: + """Libgen search rows are concrete, directly downloadable releases.""" + return True + + def get_record( + self, + record_id: str, + *, + fetch_download_count: bool = True, + ) -> BrowseRecord | None: + """Resolve a libgen record by (possibly prefixed) md5, or None if not found.""" + md5 = record_id.split(":", 1)[-1].lower() + return scraper.fetch_record_by_md5(md5, mirrors.get_libgen_mirrors()) + + def get_column_config(self) -> ReleaseColumnConfig: + """Language, format and size badges -- same layout as Direct Download.""" + return ReleaseColumnConfig( + columns=[ + ColumnSchema( + key="extra.language", + label="Language", + render_type=ColumnRenderType.BADGE, + align=ColumnAlign.CENTER, + width="60px", + color_hint=ColumnColorHint(type="map", value="language"), + uppercase=True, + ), + ColumnSchema( + key="format", + label="Format", + render_type=ColumnRenderType.BADGE, + align=ColumnAlign.CENTER, + width="80px", + color_hint=ColumnColorHint(type="map", value="format"), + uppercase=True, + ), + ColumnSchema( + key="size", + label="Size", + render_type=ColumnRenderType.SIZE, + align=ColumnAlign.CENTER, + width="80px", + ), + ], + grid_template="minmax(0,2fr) 60px 80px 80px", + supported_filters=["format", "language"], + ) diff --git a/tests/libgen/__init__.py b/tests/libgen/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/libgen/sample_html.py b/tests/libgen/sample_html.py new file mode 100644 index 00000000..bd5c52bf --- /dev/null +++ b/tests/libgen/sample_html.py @@ -0,0 +1,82 @@ +"""Inline HTML samples for Libgen tests, trimmed to the real libgen.li structure. + +The results table mixes two row shapes with NO rowspans: full 9-cell rows +``[Title, Author, Publisher, Year, Language, Pages, Size, Ext, Mirrors]`` and compact +5-cell rows ``[Title, Pages, Size, Ext, Mirrors]``. The md5 lives in the Mirrors cell. +""" + +MD5_A = "a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1" # 9-cell epub row (author + language present) +# A decoy md5 planted in row A's Title cell (as a cover-image URL). The parser must NOT pick +# it: md5 extraction is scoped to the Mirrors cell, so row A must resolve to MD5_A, not this. +DECOY_MD5 = "0000000000000000000000000000dead" +MD5_B = "b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2" # 9-cell off-topic name-drop (cbr) - must survive +MD5_C = "c3c3c3c3c3c3c3c3c3c3c3c3c3c3c3c3" # 5-cell compact manga volume (cbr) +MD5_D = "d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4" # 5-cell compact manga volume (cbz) +GET_KEY = "TESTKEY0001" + +# A #tablelibgen with: header, 2 full rows (1 on-topic, 1 off-topic name-drop), 2 compact +# manga rows, a 1-cell spacer (len < 5 -> skipped), and a 5-cell row with no md5 (-> skipped). +SEARCH_HTML = f""" + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
TitleAuthor(s)PublisherYearLanguagePagesSizeExt.Mirrors
One Piece, Vol. 1Eiichiro OdaViz Media2003English216180 MBepubLibgen + Anna's
Ninja High School #127 Naruto, One Piece & KenshinBen DunnAntarctic Press2005English248 MBcbrLibgen
One Piece 515196 MBcbrLibgen
One Piece 516207 MBcbzLibgen
-- section separator --
AdvertisementSponsored
+ +""" + +# A challenge/error page: no results table -> _parse_results returns None (try next mirror). +NO_TABLE_HTML = """ +
Checking your browser...
+""" + +# A well-formed but empty results table -> _parse_results returns [] (accepted as final). +EMPTY_TABLE_HTML = """ + + + +
TitleExt.Mirrors
+ +""" + +# An ads.php page: keyed GET link + the labelled metadata block (as visible text). +ADS_HTML = f""" +Library Genesis + + + +

GET

Title: One Piece, Vol. 1 Series: One Piece Author(s): Eiichiro Oda + Publisher: Viz Media Year: 2003 ISBN: 9781234567890 Language: English Pages: 216
+ +""" + +# An ads.php page with no GET link (resolution should fail). +ADS_HTML_NO_GET = """ +

File not found.

+""" diff --git a/tests/libgen/test_get_record.py b/tests/libgen/test_get_record.py new file mode 100644 index 00000000..4960eedc --- /dev/null +++ b/tests/libgen/test_get_record.py @@ -0,0 +1,43 @@ +"""Tests for md5 -> record resolution via the ads.php metadata block.""" + +from unittest.mock import patch + +from shelfmark.release_sources.libgen import scraper +from shelfmark.release_sources.libgen import source as libgen_source +from shelfmark.release_sources.libgen.source import LibgenSource +from tests.libgen import sample_html as html + + +def test_fetch_record_by_md5_parses_ads_metadata(): + with patch.object(scraper, "fetch_page", return_value=html.ADS_HTML): + record = scraper.fetch_record_by_md5(html.MD5_A, ["https://libgen.li"]) + assert record is not None + assert record.title == "One Piece, Vol. 1" + assert record.author == "Eiichiro Oda" + assert record.publisher == "Viz Media" + assert record.year == "2003" # stops at "ISBN:", not swallowed + assert record.language == "en" + assert record.source == "libgen" + assert record.id == html.MD5_A + + +def test_fetch_record_by_md5_titleless_page_returns_none(): + with patch.object(scraper, "fetch_page", return_value=html.ADS_HTML_NO_GET): + assert scraper.fetch_record_by_md5(html.MD5_A, ["https://libgen.li"]) is None + + +def test_fetch_record_by_md5_all_mirrors_miss_returns_none(): + with patch.object(scraper, "fetch_page", return_value=None): + assert scraper.fetch_record_by_md5(html.MD5_A, ["https://a", "https://b"]) is None + + +def test_get_record_strips_prefix_before_lookup(): + with ( + patch.object( + libgen_source.mirrors, "get_libgen_mirrors", return_value=["https://libgen.li"] + ), + patch.object(libgen_source.scraper, "fetch_record_by_md5", return_value=None) as mock_fetch, + ): + LibgenSource().get_record(f"libgen:{html.MD5_A}") + mock_fetch.assert_called_once() + assert mock_fetch.call_args.args[0] == html.MD5_A diff --git a/tests/libgen/test_handler.py b/tests/libgen/test_handler.py new file mode 100644 index 00000000..31aff762 --- /dev/null +++ b/tests/libgen/test_handler.py @@ -0,0 +1,131 @@ +"""Tests for LibgenHandler: prefix stripping, ads.php resolution, mirror fallthrough.""" + +import io +import threading +from unittest.mock import MagicMock, patch + +from shelfmark.core.models import DownloadTask +from shelfmark.release_sources.libgen import handler as libgen_handler +from shelfmark.release_sources.libgen.handler import LibgenHandler +from tests.libgen import sample_html as html + + +def _task(task_id, fmt="cbr"): + return DownloadTask( + task_id=task_id, source="libgen", title="One Piece 515", format=fmt, size="6 MB" + ) + + +def _buf(nbytes=20000): + buf = io.BytesIO(b"x" * nbytes) + buf.seek(0, io.SEEK_END) # download_url returns the buffer positioned at its end + return buf + + +def _run(task, tmp_path, *, mirrors_list, fetch_page, download_url): + status = MagicMock() + cancel = threading.Event() + with ( + patch.object(libgen_handler, "TMP_DIR", tmp_path), + patch.object( + libgen_handler.config, + "get", + side_effect=lambda k, d=None: "none" if k == "FILE_ORGANIZATION" else d, + ), + patch("shelfmark.core.mirrors.get_libgen_mirrors", return_value=mirrors_list), + patch.object(libgen_handler.scraper, "fetch_page", side_effect=fetch_page), + patch.object(libgen_handler.downloader, "download_url", side_effect=download_url), + ): + result = LibgenHandler().download(task, cancel, MagicMock(), status) + return result, status + + +def test_download_strips_prefix_resolves_and_writes(tmp_path): + captured = {} + + def fetch_page(url, timeout=(5, 10)): + return html.ADS_HTML + + def download_url(link, size, prog, cancel, sel, status, referer=None): + captured["link"] = link + captured["referer"] = referer + captured["selector"] = sel + return _buf() + + result, _ = _run( + _task(f"libgen:{html.MD5_A}"), + tmp_path, + mirrors_list=["https://libgen.li"], + fetch_page=fetch_page, + download_url=download_url, + ) + expected = tmp_path / f"{html.MD5_A}.cbr" + assert result == str(expected) + assert expected.exists() + assert captured["link"] == f"https://libgen.li/get.php?md5={html.MD5_A}&key={html.GET_KEY}" + assert captured["referer"] == f"https://libgen.li/ads.php?md5={html.MD5_A}" + assert captured["selector"] is None # no AAMirrorSelector constructed + + +def test_download_accepts_bare_md5_task_id(tmp_path): + result, _ = _run( + _task(html.MD5_A), + tmp_path, + mirrors_list=["https://libgen.li"], + fetch_page=lambda url, timeout=(5, 10): html.ADS_HTML, + download_url=lambda *a, **k: _buf(), + ) + assert result == str(tmp_path / f"{html.MD5_A}.cbr") + + +def test_download_falls_through_to_second_mirror(tmp_path): + def fetch_page(url, timeout=(5, 10)): + return None if "dead" in url else html.ADS_HTML + + result, _ = _run( + _task(f"libgen:{html.MD5_A}"), + tmp_path, + mirrors_list=["https://dead.example", "https://libgen.li"], + fetch_page=fetch_page, + download_url=lambda *a, **k: _buf(), + ) + assert result == str(tmp_path / f"{html.MD5_A}.cbr") + + +def test_download_all_mirrors_fail_returns_none(tmp_path): + result, status = _run( + _task(f"libgen:{html.MD5_A}"), + tmp_path, + mirrors_list=["https://a", "https://b"], + fetch_page=lambda url, timeout=(5, 10): None, + download_url=lambda *a, **k: _buf(), + ) + assert result is None + status.assert_any_call("error", "All Libgen mirrors failed") + + +def test_download_too_small_file_is_rejected(tmp_path): + result, status = _run( + _task(f"libgen:{html.MD5_A}"), + tmp_path, + mirrors_list=["https://libgen.li"], + fetch_page=lambda url, timeout=(5, 10): html.ADS_HTML, + download_url=lambda *a, **k: _buf(100), # below _MIN_VALID_FILE_SIZE + ) + assert result is None + status.assert_any_call("error", "All Libgen mirrors failed") + + +def test_download_cancelled_before_start(tmp_path): + status = MagicMock() + cancel = threading.Event() + cancel.set() + with ( + patch.object(libgen_handler, "TMP_DIR", tmp_path), + patch("shelfmark.core.mirrors.get_libgen_mirrors", return_value=["https://libgen.li"]), + ): + result = LibgenHandler().download( + _task(f"libgen:{html.MD5_A}"), cancel, MagicMock(), status + ) + assert result is None + status.assert_any_call("cancelled", "Cancelled") diff --git a/tests/libgen/test_scraper.py b/tests/libgen/test_scraper.py new file mode 100644 index 00000000..1f70a8fe --- /dev/null +++ b/tests/libgen/test_scraper.py @@ -0,0 +1,92 @@ +"""Tests for the Libgen scraper: results parsing and download-link resolution.""" + +from unittest.mock import patch + +from shelfmark.release_sources.libgen import scraper +from tests.libgen import sample_html as html + + +def test_parse_results_extracts_all_md5_rows(): + records = scraper._parse_results(html.SEARCH_HTML, "https://libgen.li") + assert records is not None + assert [r.id for r in records] == [html.MD5_A, html.MD5_B, html.MD5_C, html.MD5_D] + + +def test_parse_results_full_row_fields(): + records = scraper._parse_results(html.SEARCH_HTML, "https://libgen.li") + a = records[0] + assert a.title == "One Piece, Vol. 1" + assert a.author == "Eiichiro Oda" + assert a.format == "epub" + assert a.size == "180 MB" #   normalized to a plain space + assert a.language == "en" + assert a.source == "libgen" + assert a.source_url == f"https://libgen.li/ads.php?md5={html.MD5_A}" + + +def test_parse_results_md5_scoped_to_mirrors_cell(): + # Row A's Title cell carries a decoy md5-shaped cover URL that appears BEFORE the real + # md5 in document order; the parser must resolve to the Mirrors-cell md5, not the decoy. + records = scraper._parse_results(html.SEARCH_HTML, "https://libgen.li") + assert records[0].id == html.MD5_A + assert html.DECOY_MD5 not in {r.id for r in records} + + +def test_parse_results_offtopic_row_survives(): + # No relevance filter: a row that merely name-drops the query is kept, same as AA. + records = scraper._parse_results(html.SEARCH_HTML, "https://libgen.li") + assert any(r.id == html.MD5_B and r.format == "cbr" for r in records) + + +def test_parse_results_compact_rows_have_no_author_or_language(): + records = scraper._parse_results(html.SEARCH_HTML, "https://libgen.li") + c = next(r for r in records if r.id == html.MD5_C) + assert c.title == "One Piece 515" + assert c.format == "cbr" + assert c.size == "6 MB" + assert c.author is None + assert c.language is None + + +def test_parse_results_comics_not_dropped_by_format(): + records = scraper._parse_results(html.SEARCH_HTML, "https://libgen.li") + assert {"cbr", "cbz"} <= {r.format for r in records} + + +def test_parse_results_no_table_returns_none(): + assert scraper._parse_results(html.NO_TABLE_HTML, "https://libgen.li") is None + + +def test_parse_results_empty_table_returns_empty_list(): + result = scraper._parse_results(html.EMPTY_TABLE_HTML, "https://libgen.li") + assert result == [] + assert result is not None # distinct from the no-table case + + +def test_resolve_download_url_extracts_keyed_get(): + url = scraper.resolve_download_url(html.ADS_HTML, "https://libgen.li") + assert url == f"https://libgen.li/get.php?md5={html.MD5_A}&key={html.GET_KEY}" + + +def test_resolve_download_url_missing_get_returns_none(): + assert scraper.resolve_download_url(html.ADS_HTML_NO_GET, "https://libgen.li") is None + + +def test_search_libgen_falls_through_dead_mirror(): + calls = [] + + def fake_fetch(url, timeout=(5, 15)): + calls.append(url) + return None if "dead" in url else html.SEARCH_HTML + + with patch.object(scraper, "fetch_page", side_effect=fake_fetch): + records = scraper.search_libgen( + "one piece", ["https://dead.example", "https://libgen.li"], max_results=25 + ) + assert len(records) == 4 + assert len(calls) == 2 # dead mirror tried first, then the live one + + +def test_search_libgen_all_mirrors_dead_returns_empty(): + with patch.object(scraper, "fetch_page", return_value=None): + assert scraper.search_libgen("q", ["https://a", "https://b"], max_results=25) == [] diff --git a/tests/libgen/test_source.py b/tests/libgen/test_source.py new file mode 100644 index 00000000..ab070356 --- /dev/null +++ b/tests/libgen/test_source.py @@ -0,0 +1,138 @@ +"""Tests for LibgenSource: availability gating, query building, record mapping.""" + +import types +from unittest.mock import patch + +from shelfmark.release_sources import BrowseRecord, ReleaseProtocol +from shelfmark.release_sources.libgen import source as libgen_source +from shelfmark.release_sources.libgen.source import LibgenSource, _build_query_candidates +from tests.libgen import sample_html as html + + +def _plan(manual_query=None, variants=None): + return types.SimpleNamespace( + manual_query=manual_query, title_variants=variants or [], author="" + ) + + +def _variant(title, author): + return types.SimpleNamespace(title=title, author=author) + + +def _book(title): + return types.SimpleNamespace(title=title) + + +class TestIsAvailable: + def _patches(self, *, enabled, has_mirrors): + return ( + patch.object( + libgen_source.config, + "get", + side_effect=lambda k, d=None: enabled if k == "LIBGEN_SEARCH_ENABLED" else d, + ), + patch.object( + libgen_source.mirrors, "has_libgen_mirror_configuration", return_value=has_mirrors + ), + ) + + def test_enabled_with_mirrors(self): + cfg, mir = self._patches(enabled=True, has_mirrors=True) + with cfg, mir: + assert LibgenSource().is_available() is True + + def test_disabled(self): + cfg, mir = self._patches(enabled=False, has_mirrors=True) + with cfg, mir: + assert LibgenSource().is_available() is False + + def test_enabled_without_mirrors(self): + cfg, mir = self._patches(enabled=True, has_mirrors=False) + with cfg, mir: + assert LibgenSource().is_available() is False + + def test_truthy_string_does_not_enable(self): + cfg = patch.object( + libgen_source.config, + "get", + side_effect=lambda k, d=None: "true" if k == "LIBGEN_SEARCH_ENABLED" else d, + ) + mir = patch.object( + libgen_source.mirrors, "has_libgen_mirror_configuration", return_value=True + ) + with cfg, mir: + assert LibgenSource().is_available() is False + + +def test_build_query_candidates_manual_query_wins(): + plan = _plan(manual_query=" attack on titan ") + assert _build_query_candidates(plan, _book("ignored")) == ["attack on titan"] + + +def test_build_query_candidates_combined_then_title_only(): + plan = _plan(variants=[_variant("One Piece", "Oda")]) + assert _build_query_candidates(plan, _book("x")) == ["One Piece Oda", "One Piece"] + + +def test_build_query_candidates_dedups_when_no_author(): + plan = _plan(variants=[_variant("Dune", "")]) + assert _build_query_candidates(plan, _book("x")) == ["Dune"] + + +def test_build_query_candidates_book_title_fallback(): + assert _build_query_candidates(_plan(), _book("Fallback Title")) == ["Fallback Title"] + + +def test_search_non_ebook_returns_empty(): + with patch.object(LibgenSource, "is_available", return_value=True): + result = LibgenSource().search( + _book("x"), _plan(manual_query="x"), content_type="audiobook" + ) + assert result == [] + + +def test_search_unavailable_returns_empty(): + with patch.object(LibgenSource, "is_available", return_value=False): + assert LibgenSource().search(_book("x"), _plan(manual_query="x")) == [] + + +def test_search_maps_records_to_releases(): + record = BrowseRecord( + id=html.MD5_A, + title="One Piece, Vol. 1", + source="libgen", + format="epub", + size="180 MB", + language="en", + author="Oda", + ) + with ( + patch.object(LibgenSource, "is_available", return_value=True), + patch.object( + libgen_source.mirrors, "get_libgen_mirrors", return_value=["https://libgen.li"] + ), + patch.object(libgen_source.config, "get", side_effect=lambda k, d=None: d), + patch.object(libgen_source.scraper, "search_libgen", return_value=[record]) as mock_search, + ): + releases = LibgenSource().search(_book("One Piece"), _plan(manual_query="One Piece")) + assert len(releases) == 1 + assert releases[0].source_id == f"libgen:{html.MD5_A}" + mock_search.assert_called_once() + + +def test_record_to_release_namespaces_source_id_and_fields(): + record = BrowseRecord( + id=html.MD5_C, title="One Piece 515", source="libgen", format="cbr", size="6 MB" + ) + release = LibgenSource()._record_to_release(record) + assert release.source == "libgen" + assert release.source_id == f"libgen:{html.MD5_C}" + assert release.protocol == ReleaseProtocol.HTTP + assert release.indexer == "Libgen" + assert release.content_type == "ebook" + assert release.extra["md5"] == html.MD5_C + assert release.download_url is None + + +def test_search_results_are_releases(): + assert LibgenSource().search_results_are_releases() is True