Initial commit
This commit is contained in:
+14
@@ -0,0 +1,14 @@
|
||||
.gitrepo/
|
||||
cookie*.txt
|
||||
auth*.txt
|
||||
headers*.txt
|
||||
token*.txt
|
||||
*.pdf
|
||||
downloaded_pages/
|
||||
pages/
|
||||
test_pages/
|
||||
cookie_test_pages/
|
||||
mzk_cookie_20_pages/
|
||||
generalized_test_pages/
|
||||
kniha/
|
||||
__pycache__/
|
||||
@@ -0,0 +1,139 @@
|
||||
# Kramerius PDF Downloader
|
||||
|
||||
Download page images from Digitalniknihovna / Kramerius and assemble them into a PDF.
|
||||
|
||||
The script:
|
||||
|
||||
- discovers all page UUIDs for a document
|
||||
- downloads each page image
|
||||
- embeds the downloaded JPEGs into a PDF
|
||||
- can resume interrupted downloads by skipping existing images
|
||||
- supports authenticated downloads with a Kramerius bearer token
|
||||
- can resolve other `digitalniknihovna.cz/<library>/...` links automatically
|
||||
|
||||
## Requirements
|
||||
|
||||
Python 3.10 or newer is recommended. The script uses only the Python standard library, so no `pip install` step is needed.
|
||||
|
||||
## Basic Usage
|
||||
|
||||
```bash
|
||||
python3 download_kramerius_pdf.py "https://www.digitalniknihovna.cz/mzk/view/uuid:0abf59c0-794e-11e4-8ce5-005056827e52?page=uuid:e57e2820-8547-11e4-a321-5ef3fc9bb22f" \
|
||||
--output-dir pages \
|
||||
--pdf book.pdf
|
||||
```
|
||||
|
||||
To test only the first few pages:
|
||||
|
||||
```bash
|
||||
python3 download_kramerius_pdf.py "URL_HERE" \
|
||||
--limit 20 \
|
||||
--output-dir test_pages \
|
||||
--pdf test.pdf
|
||||
```
|
||||
|
||||
If a run is interrupted, run the same command again. Existing images are skipped unless you pass `--overwrite`.
|
||||
|
||||
## Authenticated Downloads
|
||||
|
||||
Some pages require a logged-in account. The Digitalniknihovna web app sends protected image requests with:
|
||||
|
||||
```text
|
||||
Authorization: Bearer <token>
|
||||
Client: www.digitalniknihovna.cz
|
||||
```
|
||||
|
||||
Use either `--auth-token`:
|
||||
|
||||
```bash
|
||||
python3 download_kramerius_pdf.py "URL_HERE" \
|
||||
--auth-token "PASTE_TOKEN_HERE" \
|
||||
--output-dir pages \
|
||||
--pdf book.pdf
|
||||
```
|
||||
|
||||
Or put the token in `auth-token.txt` and use `--auth-file`:
|
||||
|
||||
```bash
|
||||
python3 download_kramerius_pdf.py "URL_HERE" \
|
||||
--auth-file auth-token.txt \
|
||||
--output-dir pages \
|
||||
--pdf book.pdf
|
||||
```
|
||||
|
||||
`auth-token.txt` may contain either the raw token or a copied header like:
|
||||
|
||||
```text
|
||||
Authorization: Bearer eyJ...
|
||||
```
|
||||
|
||||
## How To Get The Auth Token
|
||||
|
||||
1. Log in on `https://www.digitalniknihovna.cz`.
|
||||
2. Open the protected book page in the browser.
|
||||
3. Open browser developer tools:
|
||||
- Chrome / Edge: `F12` or `Ctrl+Shift+I`
|
||||
- Firefox: `F12`
|
||||
4. Go to the **Application** tab.
|
||||
5. Open **Local storage** in the left sidebar.
|
||||
6. Select:
|
||||
|
||||
```text
|
||||
https://www.digitalniknihovna.cz
|
||||
```
|
||||
|
||||
7. Find the key for your library. For MZK it is:
|
||||
|
||||
```text
|
||||
auth.token.mzk
|
||||
```
|
||||
|
||||
8. Copy the value and save it in `auth-token.txt`, or pass it directly with `--auth-token`.
|
||||
|
||||
For another library, the key usually follows the same pattern:
|
||||
|
||||
```text
|
||||
auth.token.<library-code>
|
||||
```
|
||||
|
||||
For example, a URL containing `/mzk/view/...` uses `auth.token.mzk`.
|
||||
|
||||
## Using Other Digitalniknihovna URLs
|
||||
|
||||
Paste the document URL directly:
|
||||
|
||||
```bash
|
||||
python3 download_kramerius_pdf.py "https://www.digitalniknihovna.cz/LIBRARY_CODE/view/uuid:..." \
|
||||
--auth-file auth-token.txt \
|
||||
--output-dir pages \
|
||||
--pdf book.pdf
|
||||
```
|
||||
|
||||
The script reads the library code from the URL and tries to resolve the correct Kramerius API backend automatically. If that fails, pass the API backend manually:
|
||||
|
||||
```bash
|
||||
python3 download_kramerius_pdf.py "URL_HERE" \
|
||||
--library mzk \
|
||||
--base-url "https://api.kramerius.mzk.cz" \
|
||||
--auth-file auth-token.txt \
|
||||
--output-dir pages \
|
||||
--pdf book.pdf
|
||||
```
|
||||
|
||||
## Useful Options
|
||||
|
||||
```text
|
||||
--limit N Download only the first N pages
|
||||
--overwrite Redownload images that already exist
|
||||
--delay SECONDS Wait between image downloads
|
||||
--cookies FILE Send cookies from a cookies.txt export
|
||||
--auth-token TOKEN Send Authorization: Bearer TOKEN
|
||||
--auth-file FILE Read token or headers from a file
|
||||
--client VALUE Override the Client header
|
||||
```
|
||||
|
||||
## Security Notes
|
||||
|
||||
Treat `auth-token.txt`, `cookie-dl.txt`, and copied request headers like passwords. Do not share them and do not commit them to version control.
|
||||
|
||||
This repository's `.gitignore` already ignores common token, cookie, header, PDF, and downloaded-page files.
|
||||
@@ -0,0 +1,745 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Download Kramerius page images and assemble them into a PDF.
|
||||
|
||||
Default target:
|
||||
https://www.digitalniknihovna.cz/mzk/view/uuid:0abf59c0-794e-11e4-8ce5-005056827e52
|
||||
|
||||
The script uses the same public Kramerius 7 APIs as the web viewer:
|
||||
1. discover child page UUIDs through the search endpoint
|
||||
2. download each page from the item image endpoint
|
||||
3. embed the JPEGs directly into a PDF
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import http.cookiejar
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
DEFAULT_URL = (
|
||||
"https://www.digitalniknihovna.cz/mzk/view/"
|
||||
"uuid:0abf59c0-794e-11e4-8ce5-005056827e52"
|
||||
"?page=uuid:e57e2820-8547-11e4-a321-5ef3fc9bb22f"
|
||||
)
|
||||
|
||||
KRAMERIUS_BASES = {
|
||||
"mzk": "https://api.kramerius.mzk.cz",
|
||||
}
|
||||
|
||||
DEFAULT_CLIENT_IDS = {
|
||||
"mzk": "www.digitalniknihovna.cz",
|
||||
}
|
||||
|
||||
USER_AGENT = "Mozilla/5.0 (compatible; kramerius-pdf-downloader/1.0)"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Page:
|
||||
index: int
|
||||
pid: str
|
||||
label: str
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class JpegInfo:
|
||||
width: int
|
||||
height: int
|
||||
components: int
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class LibraryConfig:
|
||||
code: str
|
||||
base_url: str
|
||||
version: int
|
||||
client_id: str | None
|
||||
|
||||
|
||||
def parse_args() -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Download all page images from a Kramerius document and create a PDF."
|
||||
)
|
||||
parser.add_argument(
|
||||
"url",
|
||||
nargs="?",
|
||||
default=DEFAULT_URL,
|
||||
help="Viewer URL, document UUID, or page/document URL from digitalniknihovna.cz.",
|
||||
)
|
||||
parser.add_argument("--library", default=None, help="Library code, e.g. mzk.")
|
||||
parser.add_argument(
|
||||
"--base-url",
|
||||
default=None,
|
||||
help="Kramerius API base URL. Defaults from --library when known.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--output-dir",
|
||||
default="downloaded_pages",
|
||||
help="Directory where images will be stored.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--pdf",
|
||||
default="kramerius_download.pdf",
|
||||
help="Output PDF path.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--limit",
|
||||
type=int,
|
||||
default=None,
|
||||
help="Download only the first N pages, useful for testing.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--delay",
|
||||
type=float,
|
||||
default=0.2,
|
||||
help="Seconds to wait between page downloads.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--retries",
|
||||
type=int,
|
||||
default=3,
|
||||
help="Download retries per request.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--overwrite",
|
||||
action="store_true",
|
||||
help="Redownload images that already exist.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--cookies",
|
||||
default=None,
|
||||
help=(
|
||||
"Cookie file to use for authenticated downloads. Supports Netscape "
|
||||
"cookies.txt exports and plain Cookie header text."
|
||||
),
|
||||
)
|
||||
parser.add_argument(
|
||||
"--auth-token",
|
||||
default=None,
|
||||
help=(
|
||||
"Kramerius auth token. The script sends it as "
|
||||
"Authorization: Bearer <token>."
|
||||
),
|
||||
)
|
||||
parser.add_argument(
|
||||
"--auth-file",
|
||||
default=None,
|
||||
help=(
|
||||
"File containing either a raw token, an Authorization header, or "
|
||||
"copied request headers from browser devtools."
|
||||
),
|
||||
)
|
||||
parser.add_argument(
|
||||
"--client",
|
||||
default=None,
|
||||
help="Value for the Kramerius Client header.",
|
||||
)
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def parse_viewer_input(value: str) -> tuple[str | None, str]:
|
||||
"""Return (library_code, document_pid)."""
|
||||
parsed = urllib.parse.urlparse(value)
|
||||
|
||||
if parsed.scheme and parsed.netloc:
|
||||
path_parts = [part for part in parsed.path.split("/") if part]
|
||||
library = path_parts[0] if path_parts else None
|
||||
|
||||
view_uuid = None
|
||||
if "view" in path_parts:
|
||||
view_index = path_parts.index("view")
|
||||
if len(path_parts) > view_index + 1:
|
||||
view_uuid = urllib.parse.unquote(path_parts[view_index + 1])
|
||||
|
||||
if view_uuid:
|
||||
return library, normalize_pid(view_uuid)
|
||||
|
||||
match = re.search(r"uuid:[0-9a-fA-F-]{36}", value)
|
||||
if match:
|
||||
return library, normalize_pid(match.group(0))
|
||||
|
||||
return None, normalize_pid(value)
|
||||
|
||||
|
||||
def normalize_pid(value: str) -> str:
|
||||
value = urllib.parse.unquote(value.strip())
|
||||
match = re.search(r"uuid:[0-9a-fA-F-]{36}", value)
|
||||
if match:
|
||||
return match.group(0)
|
||||
match = re.search(r"[0-9a-fA-F]{8}(?:-[0-9a-fA-F]{4}){3}-[0-9a-fA-F]{12}", value)
|
||||
if match:
|
||||
return "uuid:" + match.group(0)
|
||||
raise ValueError(f"Could not find a UUID in: {value}")
|
||||
|
||||
|
||||
def resolve_library_config(
|
||||
opener: urllib.request.OpenerDirector,
|
||||
library: str,
|
||||
explicit_base_url: str | None,
|
||||
retries: int,
|
||||
) -> LibraryConfig:
|
||||
if explicit_base_url:
|
||||
return LibraryConfig(
|
||||
code=library,
|
||||
base_url=explicit_base_url.rstrip("/"),
|
||||
version=7,
|
||||
client_id=DEFAULT_CLIENT_IDS.get(library),
|
||||
)
|
||||
|
||||
if library in KRAMERIUS_BASES:
|
||||
return LibraryConfig(
|
||||
code=library,
|
||||
base_url=KRAMERIUS_BASES[library],
|
||||
version=7,
|
||||
client_id=DEFAULT_CLIENT_IDS.get(library),
|
||||
)
|
||||
|
||||
try:
|
||||
config = fetch_digitalniknihovna_config(opener, library, retries)
|
||||
except RuntimeError as exc:
|
||||
raise SystemExit(
|
||||
f"Could not resolve library {library!r} from digitalniknihovna.cz. "
|
||||
"Pass --base-url explicitly.\n"
|
||||
f"Reason: {exc}"
|
||||
) from exc
|
||||
|
||||
return config
|
||||
|
||||
|
||||
def fetch_digitalniknihovna_config(
|
||||
opener: urllib.request.OpenerDirector, library: str, retries: int
|
||||
) -> LibraryConfig:
|
||||
url = "https://www.digitalniknihovna.cz/assets/shared/globals.js"
|
||||
data, content_type = request_bytes(opener, url, retries)
|
||||
if "javascript" not in content_type.lower() and "text" not in content_type.lower():
|
||||
raise RuntimeError(f"Expected JavaScript config, got {content_type or 'unknown'}")
|
||||
|
||||
text = data.decode("utf-8", errors="replace")
|
||||
block = find_library_config_block(text, library)
|
||||
if block is None:
|
||||
raise RuntimeError(f"Library code {library!r} was not found in globals.js")
|
||||
|
||||
base_url = extract_js_string_property(block, "url")
|
||||
if not base_url:
|
||||
raise RuntimeError(f"Library code {library!r} has no API URL in globals.js")
|
||||
|
||||
version = extract_js_int_property(block, "version") or 7
|
||||
client_id = extract_js_string_property(block, "ga4clientId") or DEFAULT_CLIENT_IDS.get(library)
|
||||
return LibraryConfig(
|
||||
code=library,
|
||||
base_url=base_url.rstrip("/"),
|
||||
version=version,
|
||||
client_id=client_id,
|
||||
)
|
||||
|
||||
|
||||
def find_library_config_block(text: str, library: str) -> str | None:
|
||||
code_pattern = re.compile(rf"\bcode\s*:\s*['\"]{re.escape(library)}['\"]")
|
||||
match = code_pattern.search(text)
|
||||
if not match:
|
||||
return None
|
||||
|
||||
start = text.rfind("{", 0, match.start())
|
||||
if start == -1:
|
||||
return None
|
||||
|
||||
depth = 0
|
||||
quote: str | None = None
|
||||
escaped = False
|
||||
for index in range(start, len(text)):
|
||||
char = text[index]
|
||||
if quote:
|
||||
if escaped:
|
||||
escaped = False
|
||||
elif char == "\\":
|
||||
escaped = True
|
||||
elif char == quote:
|
||||
quote = None
|
||||
continue
|
||||
|
||||
if char in {"'", '"'}:
|
||||
quote = char
|
||||
elif char == "{":
|
||||
depth += 1
|
||||
elif char == "}":
|
||||
depth -= 1
|
||||
if depth == 0:
|
||||
return text[start : index + 1]
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def extract_js_string_property(block: str, name: str) -> str | None:
|
||||
pattern = re.compile(rf"\b{re.escape(name)}\s*:\s*(['\"])(.*?)\1", re.S)
|
||||
match = pattern.search(block)
|
||||
return match.group(2) if match else None
|
||||
|
||||
|
||||
def extract_js_int_property(block: str, name: str) -> int | None:
|
||||
pattern = re.compile(rf"\b{re.escape(name)}\s*:\s*(\d+)")
|
||||
match = pattern.search(block)
|
||||
return int(match.group(1)) if match else None
|
||||
|
||||
|
||||
def build_opener(
|
||||
cookie_path: str | None,
|
||||
) -> tuple[urllib.request.OpenerDirector, int, list[str]]:
|
||||
if not cookie_path:
|
||||
return urllib.request.build_opener(), 0, []
|
||||
|
||||
path = Path(cookie_path)
|
||||
if not path.exists():
|
||||
raise FileNotFoundError(f"Cookie file not found: {path}")
|
||||
|
||||
cookie_jar = http.cookiejar.MozillaCookieJar()
|
||||
cookie_count = 0
|
||||
|
||||
try:
|
||||
cookie_jar.load(path, ignore_discard=True, ignore_expires=True)
|
||||
cookie_count = len(cookie_jar)
|
||||
except (http.cookiejar.LoadError, OSError):
|
||||
raw_cookie = parse_raw_cookie_file(path)
|
||||
if raw_cookie:
|
||||
opener = urllib.request.build_opener(RawCookieProcessor(raw_cookie))
|
||||
return opener, len([part for part in raw_cookie.split(";") if part.strip()]), ["*"]
|
||||
raise
|
||||
|
||||
domains = sorted({cookie.domain for cookie in cookie_jar})
|
||||
return urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cookie_jar)), cookie_count, domains
|
||||
|
||||
|
||||
def cookie_domains_match_host(domains: list[str], host: str) -> bool:
|
||||
if "*" in domains:
|
||||
return True
|
||||
|
||||
host = host.lower()
|
||||
for domain in domains:
|
||||
domain = domain.lower().lstrip(".")
|
||||
if host == domain or host.endswith("." + domain):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def parse_raw_cookie_file(path: Path) -> str:
|
||||
lines = []
|
||||
for line in path.read_text(encoding="utf-8").splitlines():
|
||||
line = line.strip()
|
||||
if not line or line.startswith("#"):
|
||||
continue
|
||||
if line.lower().startswith("cookie:"):
|
||||
line = line.split(":", 1)[1].strip()
|
||||
lines.append(line)
|
||||
return "; ".join(lines)
|
||||
|
||||
|
||||
class RawCookieProcessor(urllib.request.BaseHandler):
|
||||
def __init__(self, cookie_header: str) -> None:
|
||||
self.cookie_header = cookie_header
|
||||
|
||||
def http_request(self, request: urllib.request.Request) -> urllib.request.Request:
|
||||
request.add_unredirected_header("Cookie", self.cookie_header)
|
||||
return request
|
||||
|
||||
https_request = http_request
|
||||
|
||||
|
||||
class StaticHeaderProcessor(urllib.request.BaseHandler):
|
||||
def __init__(self, headers: dict[str, str]) -> None:
|
||||
self.headers = headers
|
||||
|
||||
def http_request(self, request: urllib.request.Request) -> urllib.request.Request:
|
||||
for key, value in self.headers.items():
|
||||
request.add_unredirected_header(key, value)
|
||||
return request
|
||||
|
||||
https_request = http_request
|
||||
|
||||
|
||||
def parse_auth_file(path: str) -> dict[str, str]:
|
||||
text = Path(path).read_text(encoding="utf-8").strip()
|
||||
if not text:
|
||||
return {}
|
||||
|
||||
headers: dict[str, str] = {}
|
||||
for line in text.splitlines():
|
||||
line = line.strip()
|
||||
if not line or line.startswith("#"):
|
||||
continue
|
||||
if ":" not in line:
|
||||
continue
|
||||
key, value = line.split(":", 1)
|
||||
key = key.strip()
|
||||
value = value.strip()
|
||||
if key.lower() in {"authorization", "client"} and value:
|
||||
headers[header_display_name(key)] = value
|
||||
|
||||
if headers:
|
||||
return headers
|
||||
|
||||
token = text
|
||||
if token.lower().startswith("bearer "):
|
||||
token = token.split(None, 1)[1].strip()
|
||||
return {"Authorization": f"Bearer {token}"}
|
||||
|
||||
|
||||
def header_display_name(key: str) -> str:
|
||||
lower = key.lower()
|
||||
if lower == "authorization":
|
||||
return "Authorization"
|
||||
if lower == "client":
|
||||
return "Client"
|
||||
return key
|
||||
|
||||
|
||||
def auth_headers(auth_token: str | None, auth_file: str | None, client: str | None) -> dict[str, str]:
|
||||
headers = parse_auth_file(auth_file) if auth_file else {}
|
||||
|
||||
if auth_token:
|
||||
token = auth_token.strip()
|
||||
if token.lower().startswith("bearer "):
|
||||
headers["Authorization"] = token
|
||||
else:
|
||||
headers["Authorization"] = f"Bearer {token}"
|
||||
|
||||
if client:
|
||||
headers["Client"] = client
|
||||
|
||||
return headers
|
||||
|
||||
|
||||
def install_headers(
|
||||
opener: urllib.request.OpenerDirector, headers: dict[str, str]
|
||||
) -> urllib.request.OpenerDirector:
|
||||
if not headers:
|
||||
return opener
|
||||
opener.add_handler(StaticHeaderProcessor(headers))
|
||||
return opener
|
||||
|
||||
|
||||
def request_bytes(
|
||||
opener: urllib.request.OpenerDirector, url: str, retries: int
|
||||
) -> tuple[bytes, str]:
|
||||
headers = {"User-Agent": USER_AGENT, "Accept": "*/*"}
|
||||
last_error: Exception | None = None
|
||||
|
||||
for attempt in range(1, retries + 1):
|
||||
request = urllib.request.Request(url, headers=headers)
|
||||
try:
|
||||
with opener.open(request, timeout=60) as response:
|
||||
content_type = response.headers.get("Content-Type", "")
|
||||
return response.read(), content_type
|
||||
except (urllib.error.URLError, TimeoutError) as exc:
|
||||
last_error = exc
|
||||
if attempt < retries:
|
||||
time.sleep(min(2**attempt, 10))
|
||||
|
||||
raise RuntimeError(f"Failed to fetch {url}: {last_error}")
|
||||
|
||||
|
||||
def request_json(opener: urllib.request.OpenerDirector, url: str, retries: int) -> dict:
|
||||
data, content_type = request_bytes(opener, url, retries)
|
||||
if "json" not in content_type.lower():
|
||||
raise RuntimeError(f"Expected JSON from {url}, got {content_type or 'unknown'}")
|
||||
return json.loads(data.decode("utf-8"))
|
||||
|
||||
|
||||
def discover_pages(
|
||||
opener: urllib.request.OpenerDirector, config: LibraryConfig, document_pid: str, retries: int
|
||||
) -> list[Page]:
|
||||
if config.version >= 7:
|
||||
query = {
|
||||
"fl": "pid,accessibility,model,title.search,page.type,page.number,page.placement",
|
||||
"q": f'own_parent.pid:"{document_pid}"',
|
||||
"sort": "rels_ext_index.sort asc",
|
||||
"rows": "10000",
|
||||
"wt": "json",
|
||||
}
|
||||
pid_field = "pid"
|
||||
label_fields = ("page.number", "title.search")
|
||||
else:
|
||||
query = {
|
||||
"fl": "PID,dostupnost,fedora.model,dc.title",
|
||||
"q": f'parent_pid:"{document_pid}"',
|
||||
"sort": "rels_ext_index asc",
|
||||
"rows": "10000",
|
||||
"wt": "json",
|
||||
}
|
||||
pid_field = "PID"
|
||||
label_fields = ("dc.title",)
|
||||
|
||||
url = f"{api_url(config)}/search?{urllib.parse.urlencode(query)}"
|
||||
payload = request_json(opener, url, retries)
|
||||
docs = payload.get("response", {}).get("docs", [])
|
||||
|
||||
pages: list[Page] = []
|
||||
for index, doc in enumerate(docs, start=1):
|
||||
pid = doc.get(pid_field)
|
||||
if not pid:
|
||||
continue
|
||||
label = next((doc.get(field) for field in label_fields if doc.get(field)), str(index))
|
||||
pages.append(Page(index=index, pid=pid, label=str(label)))
|
||||
|
||||
if not pages:
|
||||
raise RuntimeError(f"No pages found for {document_pid}. Search URL was: {url}")
|
||||
|
||||
return pages
|
||||
|
||||
|
||||
def api_url(config: LibraryConfig) -> str:
|
||||
if config.version >= 7:
|
||||
return f"{config.base_url}/search/api/client/v7.0"
|
||||
return f"{config.base_url}/search/api/v5.0"
|
||||
|
||||
|
||||
def page_image_url(config: LibraryConfig, pid: str) -> str:
|
||||
encoded_pid = urllib.parse.quote(pid, safe=":")
|
||||
if config.version >= 7:
|
||||
return f"{api_url(config)}/items/{encoded_pid}/image"
|
||||
return f"{api_url(config)}/item/{encoded_pid}/streams/IMG_FULL"
|
||||
|
||||
|
||||
def download_pages(
|
||||
opener: urllib.request.OpenerDirector,
|
||||
config: LibraryConfig,
|
||||
pages: list[Page],
|
||||
output_dir: Path,
|
||||
retries: int,
|
||||
delay: float,
|
||||
overwrite: bool,
|
||||
) -> list[Path]:
|
||||
output_dir.mkdir(parents=True, exist_ok=True)
|
||||
paths: list[Path] = []
|
||||
|
||||
for page in pages:
|
||||
path = output_dir / f"{page.index:04d}.jpg"
|
||||
paths.append(path)
|
||||
|
||||
if path.exists() and not overwrite:
|
||||
print(f"[{page.index:03d}/{len(pages):03d}] exists {path}")
|
||||
continue
|
||||
|
||||
url = page_image_url(config, page.pid)
|
||||
data, content_type = request_bytes(opener, url, retries)
|
||||
if "image" not in content_type.lower():
|
||||
preview = data[:200].decode("utf-8", errors="replace")
|
||||
raise RuntimeError(
|
||||
f"Expected image for {page.pid}, got {content_type or 'unknown'}.\n"
|
||||
f"Response starts with: {preview}"
|
||||
)
|
||||
|
||||
path.write_bytes(data)
|
||||
print(f"[{page.index:03d}/{len(pages):03d}] saved {path} ({len(data) // 1024} KiB)")
|
||||
if delay > 0:
|
||||
time.sleep(delay)
|
||||
|
||||
return paths
|
||||
|
||||
|
||||
def read_jpeg_info(data: bytes) -> JpegInfo:
|
||||
if not data.startswith(b"\xff\xd8"):
|
||||
raise ValueError("Not a JPEG file")
|
||||
|
||||
offset = 2
|
||||
while offset < len(data):
|
||||
while offset < len(data) and data[offset] == 0xFF:
|
||||
offset += 1
|
||||
if offset >= len(data):
|
||||
break
|
||||
|
||||
marker = data[offset]
|
||||
offset += 1
|
||||
|
||||
if marker in {0xD8, 0xD9}:
|
||||
continue
|
||||
if offset + 2 > len(data):
|
||||
break
|
||||
|
||||
segment_length = int.from_bytes(data[offset : offset + 2], "big")
|
||||
if segment_length < 2 or offset + segment_length > len(data):
|
||||
break
|
||||
|
||||
if marker in {
|
||||
0xC0,
|
||||
0xC1,
|
||||
0xC2,
|
||||
0xC3,
|
||||
0xC5,
|
||||
0xC6,
|
||||
0xC7,
|
||||
0xC9,
|
||||
0xCA,
|
||||
0xCB,
|
||||
0xCD,
|
||||
0xCE,
|
||||
0xCF,
|
||||
}:
|
||||
if segment_length < 8:
|
||||
break
|
||||
height = int.from_bytes(data[offset + 3 : offset + 5], "big")
|
||||
width = int.from_bytes(data[offset + 5 : offset + 7], "big")
|
||||
components = data[offset + 7]
|
||||
return JpegInfo(width=width, height=height, components=components)
|
||||
|
||||
offset += segment_length
|
||||
|
||||
raise ValueError("Could not find JPEG dimensions")
|
||||
|
||||
|
||||
def pdf_escape(value: str) -> str:
|
||||
return value.replace("\\", "\\\\").replace("(", "\\(").replace(")", "\\)")
|
||||
|
||||
|
||||
def build_pdf(image_paths: list[Path], pdf_path: Path) -> None:
|
||||
if not image_paths:
|
||||
raise RuntimeError("No images to write to PDF")
|
||||
|
||||
objects: list[bytes] = []
|
||||
|
||||
def add_object(body: bytes) -> int:
|
||||
objects.append(body)
|
||||
return len(objects)
|
||||
|
||||
catalog_id = add_object(b"<< /Type /Catalog /Pages 2 0 R >>")
|
||||
pages_id = add_object(b"")
|
||||
page_ids: list[int] = []
|
||||
|
||||
for index, image_path in enumerate(image_paths, start=1):
|
||||
image_data = image_path.read_bytes()
|
||||
info = read_jpeg_info(image_data)
|
||||
|
||||
color_space = b"/DeviceRGB"
|
||||
decode = b""
|
||||
if info.components == 1:
|
||||
color_space = b"/DeviceGray"
|
||||
elif info.components == 4:
|
||||
color_space = b"/DeviceCMYK"
|
||||
decode = b" /Decode [1 0 1 0 1 0 1 0]"
|
||||
|
||||
image_id = len(objects) + 2
|
||||
content_id = len(objects) + 3
|
||||
page_id = add_object(
|
||||
(
|
||||
f"<< /Type /Page /Parent {pages_id} 0 R "
|
||||
f"/MediaBox [0 0 {info.width} {info.height}] "
|
||||
f"/Resources << /XObject << /Im{index} {image_id} 0 R >> >> "
|
||||
f"/Contents {content_id} 0 R >>"
|
||||
).encode("ascii")
|
||||
)
|
||||
page_ids.append(page_id)
|
||||
|
||||
image_header = (
|
||||
f"<< /Type /XObject /Subtype /Image /Width {info.width} "
|
||||
f"/Height {info.height} /ColorSpace {color_space.decode('ascii')} "
|
||||
f"/BitsPerComponent 8 /Filter /DCTDecode{decode.decode('ascii')} "
|
||||
f"/Length {len(image_data)} >>\nstream\n"
|
||||
).encode("ascii")
|
||||
add_object(image_header + image_data + b"\nendstream")
|
||||
|
||||
content = f"q\n{info.width} 0 0 {info.height} 0 0 cm\n/Im{index} Do\nQ\n".encode(
|
||||
"ascii"
|
||||
)
|
||||
add_object(
|
||||
b"<< /Length "
|
||||
+ str(len(content)).encode("ascii")
|
||||
+ b" >>\nstream\n"
|
||||
+ content
|
||||
+ b"endstream"
|
||||
)
|
||||
|
||||
kids = " ".join(f"{page_id} 0 R" for page_id in page_ids)
|
||||
objects[pages_id - 1] = (
|
||||
f"<< /Type /Pages /Count {len(page_ids)} /Kids [{kids}] >>"
|
||||
).encode("ascii")
|
||||
|
||||
metadata = f"<< /Producer ({pdf_escape(Path(sys.argv[0]).name)}) >>".encode("ascii")
|
||||
info_id = add_object(metadata)
|
||||
|
||||
pdf_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with pdf_path.open("wb") as handle:
|
||||
handle.write(b"%PDF-1.4\n%\xe2\xe3\xcf\xd3\n")
|
||||
offsets = [0]
|
||||
for object_id, body in enumerate(objects, start=1):
|
||||
offsets.append(handle.tell())
|
||||
handle.write(f"{object_id} 0 obj\n".encode("ascii"))
|
||||
handle.write(body)
|
||||
handle.write(b"\nendobj\n")
|
||||
|
||||
xref_offset = handle.tell()
|
||||
handle.write(f"xref\n0 {len(objects) + 1}\n".encode("ascii"))
|
||||
handle.write(b"0000000000 65535 f \n")
|
||||
for offset in offsets[1:]:
|
||||
handle.write(f"{offset:010d} 00000 n \n".encode("ascii"))
|
||||
handle.write(
|
||||
(
|
||||
f"trailer\n<< /Size {len(objects) + 1} "
|
||||
f"/Root {catalog_id} 0 R /Info {info_id} 0 R >>\n"
|
||||
f"startxref\n{xref_offset}\n%%EOF\n"
|
||||
).encode("ascii")
|
||||
)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
args = parse_args()
|
||||
library_from_url, document_pid = parse_viewer_input(args.url)
|
||||
library = args.library or library_from_url or "mzk"
|
||||
opener, cookie_count, cookie_domains = build_opener(args.cookies)
|
||||
config = resolve_library_config(opener, library, args.base_url, args.retries)
|
||||
|
||||
print(f"Library: {config.code}")
|
||||
print(f"API: {config.base_url} (Kramerius {config.version})")
|
||||
print(f"Document: {document_pid}")
|
||||
client = args.client or config.client_id
|
||||
headers = auth_headers(
|
||||
args.auth_token,
|
||||
args.auth_file,
|
||||
client if (args.auth_token or args.auth_file) else None,
|
||||
)
|
||||
opener = install_headers(opener, headers)
|
||||
if args.cookies:
|
||||
if cookie_count:
|
||||
print(f"Loaded {cookie_count} cookies from {args.cookies}")
|
||||
api_host = urllib.parse.urlparse(config.base_url).hostname or ""
|
||||
if not cookie_domains_match_host(cookie_domains, api_host):
|
||||
print(
|
||||
"Warning: cookie domains do not match the API host "
|
||||
f"{api_host}; these cookies probably will not authenticate downloads."
|
||||
)
|
||||
else:
|
||||
print(f"Warning: loaded 0 cookies from {args.cookies}")
|
||||
if "Authorization" in headers:
|
||||
print("Using Authorization bearer token")
|
||||
if "Client" in headers:
|
||||
print(f"Using Client header: {headers['Client']}")
|
||||
print("Discovering pages...")
|
||||
|
||||
pages = discover_pages(opener, config, document_pid, args.retries)
|
||||
if args.limit:
|
||||
pages = pages[: args.limit]
|
||||
print(f"Found {len(pages)} pages")
|
||||
|
||||
image_paths = download_pages(
|
||||
opener=opener,
|
||||
config=config,
|
||||
pages=pages,
|
||||
output_dir=Path(args.output_dir),
|
||||
retries=args.retries,
|
||||
delay=args.delay,
|
||||
overwrite=args.overwrite,
|
||||
)
|
||||
|
||||
print(f"Writing PDF: {args.pdf}")
|
||||
build_pdf(image_paths, Path(args.pdf))
|
||||
print("Done")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user