mirror of
https://github.com/calibrain/shelfmark.git
synced 2026-10-04 22:05:45 +01:00
feat(sources): add Libgen as a direct catalogue search source (#1326)
## What Adds **Libgen as a search source**. Today Libgen is only a download mirror (reached by an Anna's Archive md5), so anything in Libgen but not in AA's search index is invisible — and that's where most of the CBZ/CBR comics and manga live. A Libgen search for *One Piece*, for instance, turns up ~99 volumes that AA search never shows. It's a self-contained `release_sources/libgen/` package (source + handler + settings) plus one line to register it. **No changes to `direct_download.py`** — it reuses the existing `ads.php → get.php` resolution and the mirrors already configured in `LIBGEN_MIRROR_URLS`. Plain HTTP, no bypasser needed (libgen.li isn't behind DDoS-Guard). Opt-in via a settings toggle. ## Worth a look in review - **`source_id` is `libgen:<md5>`, not the bare md5.** The download queue keys on `task_id` (= `source_id`), and `direct_download` already uses the bare md5. Since AA indexes a lot of Libgen, the same md5 shows up from both sources — a bare id would collide in the queue. The handler strips the prefix before downloading. - **Reachable like the other non-default sources** (Prowlarr, AudiobookBay, …): it appears in the per-book release search, not the free-text box (that stays wired to `direct_download`). Tests in `tests/libgen/` cover parsing (both row layouts), the source, the handler, and `get_record`. Lint/format/typecheck clean.
This commit is contained in:
Alex Guerrieri
@@ -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",
|
||||
)
|
||||
|
||||
@@ -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
|
||||
@@ -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:<md5>" 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
|
||||
@@ -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=<hash> and an AA /md5/<hash> 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'<a\s+href=["\']([^"\']*get\.php\?md5=[^"\']+&key=[^"\']+)["\'][^>]*>\s*'
|
||||
r"<h2[^>]*>GET</h2>\s*</a>",
|
||||
re.IGNORECASE,
|
||||
),
|
||||
re.compile(
|
||||
r'<a[^>]+href=["\']([^"\']*get\.php\?md5=[^"\']+&(?:amp;)?key=[^"\']+)["\']',
|
||||
re.IGNORECASE,
|
||||
),
|
||||
re.compile(
|
||||
r'<a\s+href=["\']([^"\']*get\.php[^"\']*)["\'][^>]*>[\s\S]*?<h2[^>]*>GET</h2>',
|
||||
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=<md5>`` 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}",
|
||||
)
|
||||
@@ -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},
|
||||
),
|
||||
]
|
||||
@@ -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:<md5>`` 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"],
|
||||
)
|
||||
@@ -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"""
|
||||
<html><body>
|
||||
<table id="tablelibgen">
|
||||
<tr><th>Title</th><th>Author(s)</th><th>Publisher</th><th>Year</th><th>Language</th>
|
||||
<th>Pages</th><th>Size</th><th>Ext.</th><th>Mirrors</th></tr>
|
||||
<tr>
|
||||
<td><img src="/covers/thumb.jpg?md5={DECOY_MD5}"><a href="edition.php?id=1">One Piece, Vol. 1</a></td>
|
||||
<td>Eiichiro Oda</td><td>Viz Media</td><td>2003</td><td>English</td>
|
||||
<td>216</td><td>180 MB</td><td>epub</td>
|
||||
<td><a href="/get.php?md5={MD5_A}">Libgen</a>
|
||||
<a href="https://annas-archive.org/md5/{MD5_A}">Anna's</a></td>
|
||||
</tr>
|
||||
<tr>
|
||||
<td>Ninja High School #127 Naruto, One Piece & Kenshin</td>
|
||||
<td>Ben Dunn</td><td>Antarctic Press</td><td>2005</td><td>English</td>
|
||||
<td>24</td><td>8 MB</td><td>cbr</td>
|
||||
<td><a href="/get.php?md5={MD5_B}">Libgen</a></td>
|
||||
</tr>
|
||||
<tr>
|
||||
<td>One Piece 515</td><td>19</td><td>6 MB</td><td>cbr</td>
|
||||
<td><a href="/get.php?md5={MD5_C}">Libgen</a></td>
|
||||
</tr>
|
||||
<tr>
|
||||
<td>One Piece 516</td><td>20</td><td>7 MB</td><td>cbz</td>
|
||||
<td><a href="/get.php?md5={MD5_D}">Libgen</a></td>
|
||||
</tr>
|
||||
<tr><td colspan="9">-- section separator --</td></tr>
|
||||
<tr>
|
||||
<td>Advertisement</td><td></td><td></td><td></td>
|
||||
<td><a href="/promo">Sponsored</a></td>
|
||||
</tr>
|
||||
</table>
|
||||
</body></html>
|
||||
"""
|
||||
|
||||
# A challenge/error page: no results table -> _parse_results returns None (try next mirror).
|
||||
NO_TABLE_HTML = """
|
||||
<html><body><div id="challenge">Checking your browser...</div></body></html>
|
||||
"""
|
||||
|
||||
# A well-formed but empty results table -> _parse_results returns [] (accepted as final).
|
||||
EMPTY_TABLE_HTML = """
|
||||
<html><body>
|
||||
<table id="tablelibgen">
|
||||
<tr><th>Title</th><th>Ext.</th><th>Mirrors</th></tr>
|
||||
</table>
|
||||
</body></html>
|
||||
"""
|
||||
|
||||
# An ads.php page: keyed GET link + the labelled metadata block (as visible text).
|
||||
ADS_HTML = f"""
|
||||
<html><head><title>Library Genesis</title></head><body>
|
||||
<table>
|
||||
<tr><td><a href="/get.php?md5={MD5_A}&key={GET_KEY}"><h2>GET</h2></a></td></tr>
|
||||
<tr><td>Title: One Piece, Vol. 1 Series: One Piece Author(s): Eiichiro Oda
|
||||
Publisher: Viz Media Year: 2003 ISBN: 9781234567890 Language: English Pages: 216</td></tr>
|
||||
</table>
|
||||
</body></html>
|
||||
"""
|
||||
|
||||
# An ads.php page with no GET link (resolution should fail).
|
||||
ADS_HTML_NO_GET = """
|
||||
<html><body><p>File not found.</p></body></html>
|
||||
"""
|
||||
@@ -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
|
||||
@@ -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")
|
||||
@@ -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) == []
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user