376 lines
18 KiB
Python
376 lines
18 KiB
Python
"""Legitimate places to find a magazine issue to import: OPDS catalogues and the Internet Archive.
|
|
|
|
Each source is a provider behind the same small interface (search + resolve a PDF), so one going away
|
|
or changing its API does not break the others. Nothing here scrapes a website: OPDS is the open
|
|
catalogue standard (Calibre-Web, Kavita and Komga serve it) and the Internet Archive has a documented
|
|
search and metadata API. See TODO.md ("Finding magazines to read") for why commercial redistribution
|
|
sites are deliberately not supported.
|
|
|
|
Security boundaries:
|
|
* a download is only ever fetched from the provider's own host (the OPDS catalogue's host, or
|
|
archive.org), never from a URL an arbitrary feed entry names -- a feed cannot aim the server at
|
|
another machine (SSRF);
|
|
* redirects are followed by hand and re-checked against the same host rule;
|
|
* size and time are bounded, and the result must really be a PDF.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from dataclasses import dataclass, asdict
|
|
from typing import Iterable, Protocol
|
|
from urllib.parse import urljoin, urlsplit, quote
|
|
|
|
import requests
|
|
import xml.etree.ElementTree as ET
|
|
|
|
USER_AGENT = "tts-voice-creator-magazine-import/1.0 (self-hosted; contact via repository)"
|
|
SEARCH_TIMEOUT = (5, 20) # connect, read (seconds)
|
|
DOWNLOAD_TIMEOUT = (5, 60)
|
|
MAX_DOWNLOAD_BYTES = 400 * 1024 * 1024
|
|
MAX_FEED_BYTES = 4 * 1024 * 1024
|
|
MAX_REDIRECTS = 4
|
|
PAGE_SIZE = 20
|
|
|
|
_ATOM = "{http://www.w3.org/2005/Atom}"
|
|
_OPDS_ACQ = "http://opds-spec.org/acquisition"
|
|
_OPDS_IMAGE = ("http://opds-spec.org/image", "http://opds-spec.org/image/thumbnail")
|
|
_IA_ID = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}")
|
|
|
|
|
|
class SourceError(Exception):
|
|
"""A source could not be searched or downloaded; the message is safe to show the user."""
|
|
|
|
|
|
@dataclass
|
|
class SourceItem:
|
|
id: str # provider-specific handle, e.g. the Internet Archive identifier or an OPDS entry URL
|
|
title: str
|
|
summary: str = ""
|
|
author: str = ""
|
|
date: str = ""
|
|
license: str = "" # a licence URL or name when the source states one, else empty (unknown)
|
|
page_url: str = "" # human-readable page for the item
|
|
thumb_url: str = ""
|
|
downloadable: bool = True
|
|
|
|
def as_dict(self) -> dict:
|
|
return asdict(self)
|
|
|
|
|
|
class Provider(Protocol):
|
|
key: str
|
|
label: str
|
|
|
|
def search(self, query: str, page: int = 1) -> list[SourceItem]: ...
|
|
def resolve_pdf(self, item_id: str) -> str: ...
|
|
def allowed_host(self, url: str) -> bool: ...
|
|
|
|
|
|
# ── helpers ──────────────────────────────────────────────────────────────────
|
|
|
|
def _clip(value, limit: int) -> str:
|
|
return re.sub(r"\s+", " ", str(value or "")).strip()[:limit]
|
|
|
|
|
|
def _get(session: requests.Session, url: str, *, timeout, stream: bool = False, host_ok=None,
|
|
accept: str = "*/*", auth=None) -> requests.Response:
|
|
"""GET with manual, host-checked redirects."""
|
|
for _ in range(MAX_REDIRECTS + 1):
|
|
if host_ok is not None and not host_ok(url):
|
|
raise SourceError("The source pointed to a different host; refusing to follow it")
|
|
try:
|
|
resp = session.get(url, timeout=timeout, stream=stream, allow_redirects=False,
|
|
headers={"User-Agent": USER_AGENT, "Accept": accept}, **({"auth": auth} if auth else {}))
|
|
except requests.RequestException as e:
|
|
raise SourceError(f"Could not reach the source ({type(e).__name__})") from e
|
|
if resp.is_redirect or resp.status_code in (301, 302, 303, 307, 308):
|
|
target = resp.headers.get("location") or ""
|
|
resp.close()
|
|
if not target:
|
|
raise SourceError("The source sent an empty redirect")
|
|
url = urljoin(url, target)
|
|
continue
|
|
if resp.status_code == 429:
|
|
resp.close()
|
|
raise SourceError("The source is rate-limiting requests; try again in a minute")
|
|
if resp.status_code >= 400:
|
|
resp.close()
|
|
raise SourceError(f"The source answered HTTP {resp.status_code}")
|
|
return resp
|
|
raise SourceError("Too many redirects")
|
|
|
|
|
|
def _read_limited(resp: requests.Response, limit: int) -> bytes:
|
|
out = bytearray()
|
|
for chunk in resp.iter_content(64 * 1024):
|
|
out.extend(chunk)
|
|
if len(out) > limit:
|
|
resp.close()
|
|
raise SourceError("The response is larger than the allowed size")
|
|
return bytes(out)
|
|
|
|
|
|
def parse_xml_safely(payload: bytes) -> ET.Element:
|
|
"""Parse a feed while refusing DTDs and entity declarations (billion-laughs / XXE class)."""
|
|
head = payload[:4096].lower()
|
|
if b"<!doctype" in head or b"<!entity" in payload[:65536].lower():
|
|
raise SourceError("The feed uses an XML feature that is not allowed")
|
|
try:
|
|
return ET.fromstring(payload)
|
|
except ET.ParseError as e:
|
|
raise SourceError("The feed is not valid XML") from e
|
|
|
|
|
|
def download_pdf(provider: Provider, item_id: str, dest, *, session: requests.Session | None = None,
|
|
progress=None) -> int:
|
|
"""Download the item's PDF to `dest` (a pathlib.Path); returns the byte count.
|
|
|
|
Written to a temporary sibling first and moved into place only once it is verified to be a PDF.
|
|
"""
|
|
from pathlib import Path
|
|
dest = Path(dest)
|
|
session = session or requests.Session()
|
|
url = provider.resolve_pdf(item_id)
|
|
resp = _get(session, url, timeout=DOWNLOAD_TIMEOUT, stream=True, host_ok=provider.allowed_host,
|
|
accept="application/pdf,*/*", auth=getattr(provider, "auth", lambda: None)())
|
|
tmp = dest.with_suffix(".part")
|
|
total = 0
|
|
try:
|
|
declared = int(resp.headers.get("content-length") or 0)
|
|
if declared > MAX_DOWNLOAD_BYTES:
|
|
raise SourceError("This PDF is larger than the allowed size")
|
|
with open(tmp, "wb") as fh:
|
|
first = True
|
|
for chunk in resp.iter_content(256 * 1024):
|
|
if not chunk:
|
|
continue
|
|
if first:
|
|
if not chunk.lstrip()[:5].startswith(b"%PDF"):
|
|
raise SourceError("The source did not return a PDF")
|
|
first = False
|
|
total += len(chunk)
|
|
if total > MAX_DOWNLOAD_BYTES:
|
|
raise SourceError("This PDF is larger than the allowed size")
|
|
fh.write(chunk)
|
|
if progress:
|
|
progress(total, declared)
|
|
if total == 0:
|
|
raise SourceError("The source returned an empty file")
|
|
tmp.replace(dest)
|
|
return total
|
|
except BaseException:
|
|
tmp.unlink(missing_ok=True)
|
|
raise
|
|
finally:
|
|
resp.close()
|
|
|
|
|
|
# ── Internet Archive ─────────────────────────────────────────────────────────
|
|
|
|
class InternetArchiveProvider:
|
|
"""archive.org: documented advanced-search and metadata APIs; only freely downloadable items are listed."""
|
|
key = "internet-archive"
|
|
label = "Internet Archive"
|
|
_HOST = "archive.org"
|
|
|
|
def __init__(self, session: requests.Session | None = None):
|
|
self._s = session or requests.Session()
|
|
|
|
def allowed_host(self, url: str) -> bool:
|
|
parts = urlsplit(url)
|
|
host = (parts.hostname or "").lower()
|
|
return parts.scheme == "https" and (host == self._HOST or host.endswith("." + self._HOST))
|
|
|
|
def search(self, query: str, page: int = 1) -> list[SourceItem]:
|
|
q = _clip(query, 200)
|
|
if not q:
|
|
raise SourceError("Enter something to search for")
|
|
# Lucene-escape quotes only; the rest is passed as the user's free text.
|
|
text = q.replace('"', " ")
|
|
# Loan-only scans are not downloadable, and a PDF is what this app ingests.
|
|
lucene = f'({text}) AND mediatype:texts AND format:pdf AND -access-restricted-item:true AND (subject:magazine OR collection:magazine_rack OR collection:periodicals)'
|
|
params = {
|
|
"q": lucene, "fl[]": ["identifier", "title", "description", "creator", "date", "licenseurl"],
|
|
"rows": PAGE_SIZE, "page": max(1, int(page)), "output": "json",
|
|
}
|
|
try:
|
|
resp = self._s.get(f"https://{self._HOST}/advancedsearch.php", params=params, timeout=SEARCH_TIMEOUT,
|
|
headers={"User-Agent": USER_AGENT})
|
|
except requests.RequestException as e:
|
|
raise SourceError(f"Could not reach the Internet Archive ({type(e).__name__})") from e
|
|
if resp.status_code == 429:
|
|
raise SourceError("The Internet Archive is rate-limiting requests; try again in a minute")
|
|
if resp.status_code != 200:
|
|
raise SourceError(f"The Internet Archive answered HTTP {resp.status_code}")
|
|
try:
|
|
docs = resp.json()["response"]["docs"]
|
|
except (ValueError, KeyError, TypeError) as e:
|
|
raise SourceError("The Internet Archive returned an unexpected answer") from e
|
|
return list(self._items(docs))
|
|
|
|
@staticmethod
|
|
def _first(v) -> str:
|
|
return _clip(v[0] if isinstance(v, list) and v else v, 500)
|
|
|
|
def _items(self, docs: Iterable[dict]) -> Iterable[SourceItem]:
|
|
for d in docs or []:
|
|
ident = self._first(d.get("identifier"))
|
|
if not _IA_ID.fullmatch(ident):
|
|
continue
|
|
yield SourceItem(
|
|
id=ident, title=self._first(d.get("title")) or ident,
|
|
summary=_clip(self._first(d.get("description")), 400), author=self._first(d.get("creator")),
|
|
date=self._first(d.get("date"))[:10], license=self._first(d.get("licenseurl")),
|
|
page_url=f"https://{self._HOST}/details/{quote(ident)}",
|
|
thumb_url=f"https://{self._HOST}/services/img/{quote(ident)}",
|
|
)
|
|
|
|
def resolve_pdf(self, item_id: str) -> str:
|
|
if not _IA_ID.fullmatch(item_id or ""):
|
|
raise SourceError("Not a valid Internet Archive identifier")
|
|
resp = _get(self._s, f"https://{self._HOST}/metadata/{quote(item_id)}", timeout=SEARCH_TIMEOUT,
|
|
host_ok=self.allowed_host, accept="application/json")
|
|
try:
|
|
meta = resp.json()
|
|
except ValueError as e:
|
|
raise SourceError("The Internet Archive returned an unexpected answer") from e
|
|
finally:
|
|
resp.close()
|
|
if (meta.get("metadata") or {}).get("access-restricted-item") in ("true", True):
|
|
raise SourceError("This item is lending-only and cannot be downloaded")
|
|
files = [f for f in meta.get("files") or [] if str(f.get("name", "")).lower().endswith(".pdf")]
|
|
# The original upload beats a derived "Text PDF"; among equals the largest is the full issue.
|
|
def rank(f):
|
|
return (0 if f.get("source") == "original" else 1, -int(f.get("size") or 0))
|
|
files.sort(key=rank)
|
|
if not files:
|
|
raise SourceError("This item has no PDF file")
|
|
return f"https://{self._HOST}/download/{quote(item_id)}/{quote(str(files[0]['name']))}"
|
|
|
|
|
|
# ── OPDS ─────────────────────────────────────────────────────────────────────
|
|
|
|
class OpdsProvider:
|
|
"""A user-configured OPDS catalogue (their own Calibre-Web/Kavita/Komga, or any public one).
|
|
|
|
Login, when the catalogue needs one, is HTTP Basic with a username from the settings and a password read from
|
|
the *environment variable whose name the settings give* (`password_env`): the password itself is never stored
|
|
in the settings, the URL, a log line or an error message. It is only ever sent to the catalogue's own host.
|
|
"""
|
|
MAX_FEED_PAGES = 5 # how many of the catalogue's result pages one search may walk
|
|
|
|
def __init__(self, key: str, label: str, root_url: str, session: requests.Session | None = None,
|
|
username: str = "", password_env: str = ""):
|
|
self.key = key
|
|
self.label = label
|
|
self._root = root_url
|
|
self._host = (urlsplit(root_url).hostname or "").lower()
|
|
self._s = session or requests.Session()
|
|
self._username = username
|
|
self._password_env = password_env
|
|
|
|
def auth(self):
|
|
"""(user, password) for this catalogue, or None when it is open. Raises SourceError if configured but unset."""
|
|
if not self._username and not self._password_env:
|
|
return None
|
|
import os
|
|
password = os.environ.get(self._password_env or "", "")
|
|
if not self._username or not password:
|
|
raise SourceError(f"The catalogue \u201c{self.label}\u201d needs a login: set a username and the environment "
|
|
f"variable named in password_env ({self._password_env or 'not configured'})")
|
|
return (self._username, password)
|
|
|
|
def allowed_host(self, url: str) -> bool:
|
|
parts = urlsplit(url)
|
|
return parts.scheme in ("http", "https") and (parts.hostname or "").lower() == self._host
|
|
|
|
def _feed(self, url: str) -> ET.Element:
|
|
resp = _get(self._s, url, timeout=SEARCH_TIMEOUT, stream=True, host_ok=self.allowed_host,
|
|
accept="application/atom+xml,application/xml,text/xml,*/*", auth=self.auth())
|
|
try:
|
|
return parse_xml_safely(_read_limited(resp, MAX_FEED_BYTES))
|
|
finally:
|
|
resp.close()
|
|
|
|
def _search_url(self) -> str:
|
|
"""The catalogue's own OpenSearch template if it advertises one, else the conventional path."""
|
|
root = self._feed(self._root)
|
|
for link in root.findall(f"{_ATOM}link"):
|
|
if link.get("rel") == "search" and link.get("type", "").startswith("application/atom+xml"):
|
|
href = urljoin(self._root, link.get("href", ""))
|
|
if "{searchTerms}" in href:
|
|
return href
|
|
return urljoin(self._root, "search?query={searchTerms}")
|
|
|
|
def search(self, query: str, page: int = 1) -> list[SourceItem]:
|
|
q = _clip(query, 200)
|
|
if not q:
|
|
raise SourceError("Enter something to search for")
|
|
page = max(1, int(page))
|
|
url = self._search_url().replace("{searchTerms}", quote(q))
|
|
need = page * PAGE_SIZE
|
|
items: list[SourceItem] = []
|
|
seen: set[str] = set()
|
|
# A catalogue answers a search one result page at a time and points at the next with rel="next".
|
|
# Entries without a PDF are dropped, so walk on (bounded) until this page can be filled.
|
|
for _ in range(self.MAX_FEED_PAGES):
|
|
if not url or url in seen or len(items) >= need:
|
|
break
|
|
seen.add(url)
|
|
root = self._feed(url)
|
|
items.extend(self._entries(root, url))
|
|
nxt = next((urljoin(url, l.get("href", "")) for l in root.findall(f"{_ATOM}link")
|
|
if l.get("rel") == "next" and l.get("href")), "")
|
|
url = nxt if nxt and self.allowed_host(nxt) else ""
|
|
return items[(page - 1) * PAGE_SIZE: page * PAGE_SIZE]
|
|
|
|
def _entries(self, root: ET.Element, base: str) -> Iterable[SourceItem]:
|
|
for e in root.findall(f"{_ATOM}entry"):
|
|
pdf = ""
|
|
thumb = ""
|
|
page_url = ""
|
|
for link in e.findall(f"{_ATOM}link"):
|
|
href = urljoin(base, link.get("href", ""))
|
|
rel = link.get("rel", "")
|
|
if rel.startswith(_OPDS_ACQ) and "pdf" in (link.get("type") or "").lower() and not pdf:
|
|
pdf = href
|
|
elif rel in _OPDS_IMAGE and not thumb:
|
|
thumb = href
|
|
elif rel == "alternate" and not page_url:
|
|
page_url = href
|
|
# An entry with no PDF is a navigation entry or an EPUB-only book; not importable here.
|
|
if not pdf or not self.allowed_host(pdf):
|
|
continue
|
|
author = e.find(f"{_ATOM}author/{_ATOM}name")
|
|
yield SourceItem(
|
|
id=pdf, title=_clip(e.findtext(f"{_ATOM}title"), 300) or "Untitled",
|
|
summary=_clip(e.findtext(f"{_ATOM}summary") or e.findtext(f"{_ATOM}content"), 400),
|
|
author=_clip(author.text if author is not None else "", 200),
|
|
date=_clip(e.findtext(f"{_ATOM}updated"), 10),
|
|
page_url=page_url if self.allowed_host(page_url or "") else "",
|
|
thumb_url=thumb if self.allowed_host(thumb or "") else "",
|
|
)
|
|
|
|
def resolve_pdf(self, item_id: str) -> str:
|
|
# The id *is* the acquisition URL the search returned; accept only this catalogue's host.
|
|
if not self.allowed_host(item_id or ""):
|
|
raise SourceError("That item does not belong to this catalogue")
|
|
return item_id
|
|
|
|
|
|
def build_providers(catalogs: list[dict] | None, session: requests.Session | None = None) -> dict[str, Provider]:
|
|
"""The Internet Archive plus every configured OPDS catalogue, keyed for the API."""
|
|
providers: dict[str, Provider] = {InternetArchiveProvider.key: InternetArchiveProvider(session)}
|
|
for n, cat in enumerate(catalogs or [], 1):
|
|
url = str((cat or {}).get("url") or "").strip()
|
|
parts = urlsplit(url)
|
|
if parts.scheme not in ("http", "https") or not parts.hostname or parts.username or parts.password:
|
|
continue
|
|
key = f"opds-{n}"
|
|
env = str((cat or {}).get("password_env") or "").strip()
|
|
if env and not re.fullmatch(r"[A-Z_][A-Z0-9_]{0,63}", env):
|
|
continue # not a plausible variable name: refuse rather than read an arbitrary one
|
|
providers[key] = OpdsProvider(key, _clip((cat or {}).get("name"), 80) or parts.hostname, url, session,
|
|
username=_clip((cat or {}).get("username"), 120), password_env=env)
|
|
return providers
|