Lint + dependabot

This commit is contained in:
√(noham)²
2026-05-17 14:24:45 +02:00
parent 4d148fb3aa
commit 751cf37b8a
2 changed files with 23 additions and 6 deletions

6
.github/dependabot.yml vendored Normal file
View File

@@ -0,0 +1,6 @@
version: 2
updates:
- package-ecosystem: "uv"
directory: "/"
schedule:
interval: "weekly"

23
main.py
View File

@@ -1,3 +1,5 @@
"""Download a Cafeyn issue from a reader URL and export it as a PDF."""
import argparse import argparse
import os import os
import re import re
@@ -7,19 +9,21 @@ from concurrent.futures import ThreadPoolExecutor, as_completed
from io import BytesIO from io import BytesIO
from pathlib import Path from pathlib import Path
import curl_cffi
from dotenv import load_dotenv
from PIL import Image
# AWS ELB in front of api.cafeyn.co does TLS-fingerprint filtering, so we # AWS ELB in front of api.cafeyn.co does TLS-fingerprint filtering, so we
# impersonate a real Chrome to avoid 404s from the ELB layer that never # impersonate a real Chrome to avoid 404s from the ELB layer that never
# reach the Kestrel backend. # reach the Kestrel backend.
TLS_IMPERSONATE = "chrome131" TLS_IMPERSONATE = "chrome131"
import curl_cffi
from dotenv import load_dotenv
from PIL import Image
try: try:
from tqdm import tqdm from tqdm import tqdm
except ImportError: except ImportError:
def tqdm(iterable, **kwargs): def tqdm(iterable, **_kwargs):
"""Fallback tqdm implementation when tqdm is unavailable."""
return iterable return iterable
@@ -30,12 +34,14 @@ USER_AGENT = (
def sanitize_filename(value): def sanitize_filename(value):
"""Convert a string to a safe filename."""
value = re.sub(r'[\\/:*?"<>|]+', "_", value) value = re.sub(r'[\\/:*?"<>|]+', "_", value)
value = re.sub(r"\s+", " ", value).strip() value = re.sub(r"\s+", " ", value).strip()
return value return value
def parse_reader_url(reader_url): def parse_reader_url(reader_url):
"""Validate and parse a Cafeyn reader URL."""
match = re.search(r"/ticket/reader/(\d+)/(\d+)", reader_url) match = re.search(r"/ticket/reader/(\d+)/(\d+)", reader_url)
if not match: if not match:
raise ValueError( raise ValueError(
@@ -46,6 +52,7 @@ def parse_reader_url(reader_url):
def get_env(name): def get_env(name):
"""Read a required environment variable."""
value = os.getenv(name) value = os.getenv(name)
if not value: if not value:
raise ValueError(f"Missing required env variable: {name}") raise ValueError(f"Missing required env variable: {name}")
@@ -53,6 +60,7 @@ def get_env(name):
def fetch_ticket(reader_url, jwt_token, websessionid): def fetch_ticket(reader_url, jwt_token, websessionid):
"""Retrieve a temporary ticket from the reader endpoint."""
headers = { headers = {
"accept": "application/json", "accept": "application/json",
"accesskeyid": "123456", "accesskeyid": "123456",
@@ -67,6 +75,7 @@ def fetch_ticket(reader_url, jwt_token, websessionid):
def fetch_material(ticket): def fetch_material(ticket):
"""Fetch issue metadata and page/tile information."""
headers = { headers = {
"accept": "application/json, text/plain, */*", "accept": "application/json, text/plain, */*",
"user-agent": USER_AGENT, "user-agent": USER_AGENT,
@@ -81,6 +90,7 @@ def fetch_material(ticket):
def apply_tile_transformation(image, transformation): def apply_tile_transformation(image, transformation):
"""Apply Cafeyn tile transform (vertical flip then rotation)."""
rotation, flip = transformation rotation, flip = transformation
if flip: if flip:
@@ -93,6 +103,7 @@ def apply_tile_transformation(image, transformation):
def download_and_construct_image(page_data, hosts_base, ticket_value, max_workers): def download_and_construct_image(page_data, hosts_base, ticket_value, max_workers):
"""Download all tiles for one page and merge them into a single image."""
hd = page_data["hd"] hd = page_data["hd"]
path = hd["path"] path = hd["path"]
tile_col_count = hd["tile_col_count"] tile_col_count = hd["tile_col_count"]
@@ -112,7 +123,6 @@ def download_and_construct_image(page_data, hosts_base, ticket_value, max_worker
dl_url = hosts_base + ticket_value + f"/{tile_path}" dl_url = hosts_base + ticket_value + f"/{tile_path}"
transformation = transformations.get(tile_name) transformation = transformations.get(tile_name)
# print(f"Downloading tile {row + 1},{col + 1}/{tile_row_count},{tile_col_count}: {tile_name}")
response = curl_cffi.get(dl_url, impersonate=TLS_IMPERSONATE) response = curl_cffi.get(dl_url, impersonate=TLS_IMPERSONATE)
response.raise_for_status() response.raise_for_status()
tile_image = Image.open(BytesIO(response.content)).convert("RGBA") tile_image = Image.open(BytesIO(response.content)).convert("RGBA")
@@ -152,6 +162,7 @@ def download_and_construct_image(page_data, hosts_base, ticket_value, max_worker
def build_pdf(material_data, output_dir, ticket, max_workers): def build_pdf(material_data, output_dir, ticket, max_workers):
"""Render all pages and write the final PDF to disk."""
publication_date = material_data["metadata"]["publication_date"] publication_date = material_data["metadata"]["publication_date"]
publication_title = material_data["metadata"]["title"] publication_title = material_data["metadata"]["title"]
hosts = material_data["hosts"]["pages"] hosts = material_data["hosts"]["pages"]
@@ -190,6 +201,7 @@ def build_pdf(material_data, output_dir, ticket, max_workers):
def main(): def main():
"""Parse CLI args, load credentials, and run the download pipeline."""
load_dotenv() load_dotenv()
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(
@@ -222,7 +234,6 @@ def main():
if not reader_url: if not reader_url:
raise ValueError("Reader URL is required.") raise ValueError("Reader URL is required.")
# Validate format early so user errors fail fast.
parse_reader_url(reader_url) parse_reader_url(reader_url)
jwt_token = get_env("JWT_TOKEN") jwt_token = get_env("JWT_TOKEN")