1089 lines
39 KiB
Python
1089 lines
39 KiB
Python
#!/usr/bin/env python3
|
|
"""Download the latest German amateur-radio exam question catalog from
|
|
the Bundesnetzagentur (BNetzA), validate it, and extract it into a
|
|
per-edition directory on disk.
|
|
|
|
The full design and contract live in DESIGN.md. As an overview the
|
|
script does:
|
|
|
|
1. Ask the BNetzA server for the file's HTTP `Last-Modified` header.
|
|
2. Skip everything if the previous run already extracted that exact
|
|
version (we recorded the header in manifest-latest.json).
|
|
3. Otherwise stream the ZIP to a scratch file, computing its sha256
|
|
as bytes arrive, and reject anything unsafe to unpack (zip-slip
|
|
paths, symlinks, oversized archives, etc.).
|
|
4. Validate the JSON schema and that every picture reference resolves
|
|
to a file under svgs/. Missing pictures are a SOFT check: they get
|
|
recorded in the manifest but do not fail the extraction.
|
|
5. Extract into a per-edition directory like
|
|
`data/2024-03-20-3-auflage/`, with a `manifest.json` next to the
|
|
data.
|
|
6. Atomically update `data/manifest-latest.json` so a future run can
|
|
do step 1 and skip immediately.
|
|
|
|
The script is intentionally a single file with stdlib only. Readability
|
|
beats cleverness here — most of the bytes below are docstrings and
|
|
comments. Performance is a non-goal (the catalog is ~3 MB and changes
|
|
maybe once a quarter).
|
|
"""
|
|
|
|
import argparse
|
|
import dataclasses
|
|
import datetime as dt
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import posixpath
|
|
import re
|
|
import shutil
|
|
import stat
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
|
|
# ============================================================================
|
|
# Constants
|
|
# ============================================================================
|
|
|
|
# Canonical download URL. BNetzA replaces the file in-place across
|
|
# editions, so this URL is stable; we rely on the HTTP Last-Modified
|
|
# header to notice updates.
|
|
DEFAULT_SOURCE_URL = (
|
|
"https://www.bundesnetzagentur.de/SharedDocs/Downloads/DE/Sachgebiete/"
|
|
"Telekommunikation/Unternehmen_Institutionen/Frequenzen/Amateurfunk/"
|
|
"Fragenkatalog/PruefungsfragenZIP.zip?__blob=publicationFile"
|
|
)
|
|
|
|
# Where extracted editions land by default. Created if it doesn't exist.
|
|
DEFAULT_OUT_DIR = Path("data")
|
|
|
|
# Polite identifier for the BNetzA web server's access log. Not required
|
|
# by the server, but it's a friendly thing to send.
|
|
USER_AGENT = (
|
|
"amateurfunk-fetch/0.1 "
|
|
"(+https://www.bnetza.de/amateurfunk-fragenkatalog)"
|
|
)
|
|
|
|
# Compressed-size guard. The real archive is ~3 MB; 50 MB gives a lot
|
|
# of headroom while still rejecting a catastrophic upstream swap (such
|
|
# as an accidentally-published 10 GB file).
|
|
MAX_COMPRESSED_BYTES = 50 * 1024 * 1024
|
|
|
|
# Uncompressed-size guard. Defends against zip-bomb-style upstream
|
|
# regressions where a small ZIP expands to gigabytes. The real archive
|
|
# expands to ~8 MB.
|
|
MAX_UNCOMPRESSED_BYTES = 200 * 1024 * 1024
|
|
|
|
# Sanity floor for the figures directory. The real archive contains
|
|
# ~700 SVGs; an archive with fewer than 100 is almost certainly broken.
|
|
MIN_SVG_ENTRIES = 100
|
|
|
|
# Connection timeout in seconds for all HTTP requests.
|
|
NETWORK_TIMEOUT = 60
|
|
|
|
# Exit codes. See DESIGN.md for the contract behind each one.
|
|
EXIT_OK = 0 # Extracted, or already up to date.
|
|
EXIT_ERROR = 1 # Network or validation problem.
|
|
EXIT_BAD_STATE = 2 # Local state conflict the operator must resolve
|
|
# (stale .bak/.tmp dir, or different content
|
|
# already present without --force).
|
|
|
|
|
|
# ============================================================================
|
|
# Exception types and data classes
|
|
# ============================================================================
|
|
|
|
|
|
class ValidationError(Exception):
|
|
"""Raised when the downloaded archive fails a structural check."""
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class ValidatedArchive:
|
|
"""The result of a successful validation pass.
|
|
|
|
Carries everything the caller needs to write a per-edition manifest
|
|
and to know what to expect from the extracted tree.
|
|
"""
|
|
|
|
# Filename of the catalog JSON inside the archive. The name encodes
|
|
# the edition (e.g. `fragenkatalog3b.json` for "3rd edition,
|
|
# revision b").
|
|
json_filename: str
|
|
|
|
# The parsed JSON catalog itself.
|
|
json_data: dict
|
|
|
|
# Picture references that point at files not present in the archive.
|
|
# Each entry has keys: question_number, field, file.
|
|
# Soft check: recorded for downstream consumers; doesn't fail the run.
|
|
missing_pictures: list
|
|
|
|
# Raw text of README.txt from the archive (license + schema docs).
|
|
# Preserved verbatim into the extracted tree.
|
|
readme_text: str
|
|
|
|
# The Quellenvermerk (attribution) sentence pulled out of the
|
|
# README. Empty string if the README is missing or has no quoted
|
|
# block.
|
|
attribution: str
|
|
|
|
|
|
# ============================================================================
|
|
# HTTP
|
|
# ============================================================================
|
|
|
|
|
|
def http_head_last_modified(url, timeout=NETWORK_TIMEOUT):
|
|
"""Ask the server for the file's `Last-Modified` header via HEAD.
|
|
|
|
Returns the header value as-is (a string like `"Wed, 20 May 2026
|
|
11:30:40 GMT"`) or `None` if the server didn't send one.
|
|
|
|
This is the foundation of the idempotency check: on subsequent runs
|
|
we compare this value against what we recorded last time and skip
|
|
the actual download if they match.
|
|
"""
|
|
request = urllib.request.Request(
|
|
url, method="HEAD", headers={"User-Agent": USER_AGENT},
|
|
)
|
|
with urllib.request.urlopen(request, timeout=timeout) as response:
|
|
return response.headers.get("Last-Modified")
|
|
|
|
|
|
def http_get_to_file(url, destination, max_bytes, timeout=NETWORK_TIMEOUT):
|
|
"""Stream `url` to a local file, computing sha256 as bytes arrive.
|
|
|
|
Returns a 3-tuple `(sha256_hex, last_modified, byte_count)`. The
|
|
sha256 is computed on the fly so we never need to re-read the file
|
|
to hash it.
|
|
|
|
Raises `ValidationError` if the response exceeds `max_bytes`, and
|
|
`urllib.error.URLError` on connection problems.
|
|
"""
|
|
request = urllib.request.Request(
|
|
url, headers={"User-Agent": USER_AGENT},
|
|
)
|
|
hasher = hashlib.sha256()
|
|
bytes_written = 0
|
|
|
|
with urllib.request.urlopen(request, timeout=timeout) as response, \
|
|
destination.open("wb") as out_file:
|
|
last_modified = response.headers.get("Last-Modified")
|
|
while True:
|
|
chunk = response.read(64 * 1024)
|
|
if not chunk:
|
|
break
|
|
bytes_written += len(chunk)
|
|
if bytes_written > max_bytes:
|
|
raise ValidationError(
|
|
f"upstream ZIP exceeds the "
|
|
f"{max_bytes}-byte compressed cap"
|
|
)
|
|
hasher.update(chunk)
|
|
out_file.write(chunk)
|
|
|
|
return hasher.hexdigest(), last_modified, bytes_written
|
|
|
|
|
|
# ============================================================================
|
|
# ZIP safety
|
|
# ============================================================================
|
|
|
|
|
|
def _normalize_zip_name(name):
|
|
"""Normalize a ZIP member's path for comparison.
|
|
|
|
Backslashes (from Windows-produced archives) become forward
|
|
slashes, and `posixpath.normpath` collapses redundant separators
|
|
and `.` segments.
|
|
"""
|
|
return posixpath.normpath(name.replace("\\", "/"))
|
|
|
|
|
|
def _is_symlink_entry(zi):
|
|
"""Return True if a ZIP member encodes a POSIX symlink.
|
|
|
|
ZIPs created on Unix store the file mode in the upper 16 bits of
|
|
`external_attr`. The S_IFLNK file-type bit identifies symlinks.
|
|
|
|
Path-based checks (rejecting absolute paths and `..` segments) are
|
|
*not* enough on their own: a symlink with an innocuous name could
|
|
still point outside the destination root.
|
|
"""
|
|
upper_mode_bits = (zi.external_attr >> 16) & 0o170000
|
|
return upper_mode_bits == stat.S_IFLNK
|
|
|
|
|
|
def _check_zip_member_safe(zi):
|
|
"""Raise `ValidationError` if extracting this member could escape
|
|
the destination directory, or if it's a symlink.
|
|
|
|
The whole archive is rejected on the first offence — we never
|
|
partially extract a suspicious one.
|
|
"""
|
|
name = zi.filename
|
|
|
|
# Some ZIP producers (mainly on Windows) use backslash separators.
|
|
# Reject these outright to avoid ambiguity about what the path means.
|
|
if "\\" in name:
|
|
raise ValidationError(f"backslash in path: {name!r}")
|
|
|
|
# An absolute path would land outside the chosen destination root.
|
|
if posixpath.isabs(name):
|
|
raise ValidationError(f"absolute path: {name!r}")
|
|
|
|
# Reject ANY `..` segment in the raw path, not only entries that
|
|
# would escape after normalization. `posixpath.normpath` would
|
|
# happily collapse "foo/../bar" to "bar" — which lands in-bounds —
|
|
# but the design contract says fail closed on suspicious paths
|
|
# regardless of whether the normalized form happens to escape. A
|
|
# well-formed archive has no reason to ship `..` anywhere.
|
|
raw_segments = name.split("/")
|
|
if ".." in raw_segments:
|
|
raise ValidationError(f"path escape (.. segment): {name!r}")
|
|
|
|
# Reject symlinks identified by the POSIX file-type bits.
|
|
if _is_symlink_entry(zi):
|
|
raise ValidationError(f"symlink entry: {name!r}")
|
|
|
|
|
|
# ============================================================================
|
|
# Validation
|
|
# ============================================================================
|
|
|
|
|
|
def validate_zip(
|
|
zip_path,
|
|
max_uncompressed=MAX_UNCOMPRESSED_BYTES,
|
|
min_svg_entries=MIN_SVG_ENTRIES,
|
|
):
|
|
"""Open the ZIP at `zip_path`, run every structural check, and
|
|
return a `ValidatedArchive` describing what we found.
|
|
|
|
Raises `ValidationError` on the first structural problem. Soft
|
|
issues (missing picture references) are collected and returned in
|
|
the result rather than raised.
|
|
|
|
See DESIGN.md §4 "Validate" for the full list of checks.
|
|
"""
|
|
if not zipfile.is_zipfile(zip_path):
|
|
raise ValidationError(f"not a valid ZIP: {zip_path}")
|
|
|
|
with zipfile.ZipFile(zip_path) as zf:
|
|
members = zf.infolist()
|
|
|
|
# First pass: per-member safety check + accumulate uncompressed size.
|
|
total_uncompressed = 0
|
|
for member in members:
|
|
_check_zip_member_safe(member)
|
|
total_uncompressed += member.file_size
|
|
if total_uncompressed > max_uncompressed:
|
|
raise ValidationError(
|
|
f"uncompressed total {total_uncompressed} exceeds cap "
|
|
f"{max_uncompressed} (zip-bomb guard)"
|
|
)
|
|
|
|
# Find the catalog JSON. There must be exactly one at the
|
|
# archive root — more than one would mean we don't know which
|
|
# one is canonical.
|
|
root_jsons = _find_root_catalog_jsons(members)
|
|
if len(root_jsons) != 1:
|
|
raise ValidationError(
|
|
f"expected exactly one root-level fragenkatalog*.json, "
|
|
f"found {len(root_jsons)}: {root_jsons}"
|
|
)
|
|
json_filename = root_jsons[0]
|
|
|
|
# Find the figure files. We require more than a configurable
|
|
# floor (default 100) so that we'd notice if BNetzA shipped
|
|
# only a handful of figures by accident.
|
|
svg_members = _find_svg_members(members)
|
|
if len(svg_members) <= min_svg_entries:
|
|
raise ValidationError(
|
|
f"expected more than {min_svg_entries} svgs/ file "
|
|
f"entries, found {len(svg_members)}"
|
|
)
|
|
svg_stems = _build_svg_stem_index(svg_members)
|
|
|
|
# Parse the catalog JSON. `utf-8-sig` tolerates an optional
|
|
# UTF-8 BOM at the very start of the file.
|
|
with zf.open(json_filename) as f:
|
|
try:
|
|
catalog = json.loads(f.read().decode("utf-8-sig"))
|
|
except json.JSONDecodeError as e:
|
|
raise ValidationError(
|
|
f"malformed JSON in {json_filename}: {e}"
|
|
) from e
|
|
|
|
# Pull the README into memory if it's there. It carries the
|
|
# attribution string mandated by the DL-DE license.
|
|
readme_text = ""
|
|
if "README.txt" in zf.namelist():
|
|
readme_bytes = zf.read("README.txt")
|
|
readme_text = readme_bytes.decode(
|
|
"utf-8-sig", errors="replace",
|
|
)
|
|
|
|
# The JSON-shape checks happen after we've closed the ZIP; they
|
|
# only need the parsed catalog and the precomputed SVG stem index.
|
|
_validate_catalog_shape(catalog)
|
|
missing_pictures = _find_missing_picture_references(catalog, svg_stems)
|
|
attribution = _extract_attribution_from_readme(readme_text)
|
|
|
|
return ValidatedArchive(
|
|
json_filename=json_filename,
|
|
json_data=catalog,
|
|
missing_pictures=missing_pictures,
|
|
readme_text=readme_text,
|
|
attribution=attribution,
|
|
)
|
|
|
|
|
|
def _find_root_catalog_jsons(members):
|
|
"""Return ZIP-member filenames that look like the catalog JSON file
|
|
at the archive root.
|
|
|
|
"Root level" means the path contains no `/`. The actual name
|
|
encodes the edition (e.g. `fragenkatalog3b.json` for the 3rd
|
|
edition, revision b), so we match against a pattern rather than
|
|
hard-coding the current filename.
|
|
"""
|
|
pattern = re.compile(r"fragenkatalog.+\.json")
|
|
matches = []
|
|
for member in members:
|
|
if member.is_dir():
|
|
continue
|
|
if "/" in member.filename:
|
|
continue
|
|
if pattern.fullmatch(member.filename):
|
|
matches.append(member.filename)
|
|
return matches
|
|
|
|
|
|
def _find_svg_members(members):
|
|
"""Return ZIP members whose normalized path starts with `svgs/`.
|
|
|
|
Directory entries are skipped, and we deliberately don't require a
|
|
standalone `svgs/` directory record — many ZIP producers omit
|
|
directory entries and emit only file entries like
|
|
`svgs/AB108_q.svg`.
|
|
"""
|
|
result = []
|
|
for member in members:
|
|
if member.is_dir():
|
|
continue
|
|
if _normalize_zip_name(member.filename).startswith("svgs/"):
|
|
result.append(member)
|
|
return result
|
|
|
|
|
|
def _build_svg_stem_index(svg_members):
|
|
"""Build a set of basenames-without-extension, keyed for picture lookup.
|
|
|
|
The catalog references pictures by stem (e.g. `"AB109_q"`), not by
|
|
full filename (`"AB109_q.svg"`). Indexing by stem keeps the soft
|
|
cross-reference check accurate.
|
|
"""
|
|
stems = set()
|
|
for member in svg_members:
|
|
basename = posixpath.basename(_normalize_zip_name(member.filename))
|
|
stem, _extension = posixpath.splitext(basename)
|
|
stems.add(stem)
|
|
return stems
|
|
|
|
|
|
# ============================================================================
|
|
# JSON schema
|
|
# ============================================================================
|
|
|
|
|
|
# Every question MUST carry these keys. Missing any is a hard failure:
|
|
# we can't usefully present a question without four answer slots plus
|
|
# the basic identifying fields.
|
|
REQUIRED_QUESTION_KEYS = (
|
|
"number",
|
|
"class",
|
|
"question",
|
|
"answer_a",
|
|
"answer_b",
|
|
"answer_c",
|
|
"answer_d",
|
|
)
|
|
|
|
# Optional fields that, when set, point at an image file under svgs/.
|
|
# Soft-validated: a broken reference goes into the manifest but does
|
|
# not stop the extraction.
|
|
PICTURE_FIELDS = (
|
|
"picture_question",
|
|
"picture_a",
|
|
"picture_b",
|
|
"picture_c",
|
|
"picture_d",
|
|
)
|
|
|
|
|
|
def _validate_catalog_shape(catalog):
|
|
"""Verify the top-level catalog object has the required structure.
|
|
|
|
Hard validation: if the schema doesn't match we won't write a
|
|
manifest that claims this is a BNetzA catalog.
|
|
"""
|
|
if not isinstance(catalog, dict):
|
|
raise ValidationError("top-level JSON is not an object")
|
|
|
|
for key in ("metadata", "sections"):
|
|
if key not in catalog:
|
|
raise ValidationError(
|
|
f"top-level JSON missing required key {key!r}"
|
|
)
|
|
|
|
metadata = catalog["metadata"]
|
|
if not isinstance(metadata, dict):
|
|
raise ValidationError("metadata is not an object")
|
|
for key in ("edition", "issued_on", "valid_from", "license"):
|
|
if key not in metadata:
|
|
raise ValidationError(
|
|
f"metadata missing required key {key!r}"
|
|
)
|
|
|
|
if not isinstance(catalog["sections"], list):
|
|
raise ValidationError("sections is not a list")
|
|
|
|
for section in catalog["sections"]:
|
|
_validate_section(section)
|
|
|
|
|
|
def _validate_section(node):
|
|
"""Recursive check for one node in the sections tree.
|
|
|
|
Each node is either an inner section (carries nested `sections`)
|
|
or a leaf (carries `questions`). Carrying both is ambiguous, and
|
|
carrying neither means the node holds no information. We treat the
|
|
*presence* of the key as the signal, not whether the list inside
|
|
it is empty.
|
|
"""
|
|
if not isinstance(node, dict):
|
|
raise ValidationError("section is not an object")
|
|
|
|
# Use `is not None` (rather than truthiness) so that an empty list
|
|
# still counts as "the key is present" — that's what the original
|
|
# contract calls for.
|
|
has_subsections = node.get("sections") is not None
|
|
has_questions = node.get("questions") is not None
|
|
|
|
if has_subsections and has_questions:
|
|
raise ValidationError(
|
|
f"section {node.get('title')!r} has both "
|
|
f"sections and questions"
|
|
)
|
|
if not has_subsections and not has_questions:
|
|
raise ValidationError(
|
|
f"section {node.get('title')!r} has neither "
|
|
f"sections nor questions"
|
|
)
|
|
|
|
if has_subsections:
|
|
for child in node["sections"]:
|
|
_validate_section(child)
|
|
else:
|
|
for question in node["questions"]:
|
|
_validate_question(question)
|
|
|
|
|
|
def _validate_question(question):
|
|
"""Check that a question object has every required field."""
|
|
if not isinstance(question, dict):
|
|
raise ValidationError("question is not an object")
|
|
for key in REQUIRED_QUESTION_KEYS:
|
|
if key not in question:
|
|
raise ValidationError(
|
|
f"question {question.get('number', '?')!r} "
|
|
f"missing required key {key!r}"
|
|
)
|
|
|
|
|
|
# ============================================================================
|
|
# Picture cross-references (soft check)
|
|
# ============================================================================
|
|
|
|
|
|
def _picture_reference_resolves(reference, svg_stems):
|
|
"""Return True if a picture reference matches a known SVG/PNG file.
|
|
|
|
Accepts both forms:
|
|
- the canonical stem (`"AB109_q"`) — matches directly against
|
|
the precomputed stem index
|
|
- a filename with extension (`"AB109_q.svg"`) — defensive
|
|
fallback in case a future BNetzA edition changes the
|
|
convention; we strip the extension and try again
|
|
|
|
Returning False here is what marks a reference as "missing" — but
|
|
that is a soft signal that goes into the manifest, not a hard
|
|
failure.
|
|
"""
|
|
if reference in svg_stems:
|
|
return True
|
|
stem, _extension = posixpath.splitext(reference)
|
|
return stem in svg_stems
|
|
|
|
|
|
def _find_missing_picture_references(catalog, svg_stems):
|
|
"""Walk the catalog and return picture references that don't resolve.
|
|
|
|
Soft check: a single upstream typo should not brick the downloader
|
|
for everyone until BNetzA ships a fix. Downstream consumers
|
|
decide whether to ignore, warn, or hide the affected questions.
|
|
|
|
Returns a list of dicts with keys: question_number, field, file.
|
|
"""
|
|
missing = []
|
|
|
|
def walk(node):
|
|
# Inspect every question in this node, if any.
|
|
for question in node.get("questions") or []:
|
|
for field in PICTURE_FIELDS:
|
|
value = question.get(field)
|
|
if value and not _picture_reference_resolves(value, svg_stems):
|
|
missing.append({
|
|
"question_number": question.get("number"),
|
|
"field": field,
|
|
"file": value,
|
|
})
|
|
# Recurse into sub-sections, if any.
|
|
for child in node.get("sections") or []:
|
|
walk(child)
|
|
|
|
walk({"sections": catalog["sections"]})
|
|
return missing
|
|
|
|
|
|
# ============================================================================
|
|
# Attribution
|
|
# ============================================================================
|
|
|
|
|
|
def _extract_attribution_from_readme(readme_text):
|
|
"""Pull the Quellenvermerk (attribution) string out of the README.
|
|
|
|
The BNetzA README contains exactly one multi-line quoted block —
|
|
the Quellenvermerk that consumers must reproduce verbatim when
|
|
redistributing the data under the DL-DE license. We return the
|
|
matched text exactly as it appears (multi-line, original spacing
|
|
preserved), so a downstream consumer can reproduce it byte-for-
|
|
byte from `manifest["attribution"]`. The extracted `README.txt`
|
|
file in the per-edition directory remains the lossless source of
|
|
truth on disk regardless.
|
|
|
|
Returns "" if no quoted block is found.
|
|
"""
|
|
match = re.search(r'"([^"]+)"', readme_text, flags=re.DOTALL)
|
|
if not match:
|
|
return ""
|
|
return match.group(1)
|
|
|
|
|
|
# ============================================================================
|
|
# Slug derivation
|
|
# ============================================================================
|
|
|
|
|
|
def derive_slug(metadata):
|
|
"""Build a filesystem-safe, sortable directory name for this edition.
|
|
|
|
The slug combines the issue date and the edition ordinal so that
|
|
directories sort chronologically and we never have to parse German
|
|
month names from the human-readable `edition` field.
|
|
|
|
Example:
|
|
`{"edition": "3. Auflage, März 2024", "issued_on": "2024-03-20"}`
|
|
-> `"2024-03-20-3-auflage"`
|
|
|
|
Both inputs are best-effort. Missing pieces become `"unknown"` so
|
|
the slug is still a valid directory name.
|
|
"""
|
|
issued_on = (metadata.get("issued_on") or "").strip() or "unknown"
|
|
edition_text = metadata.get("edition") or ""
|
|
|
|
# The German edition string starts with the ordinal
|
|
# ("3. Auflage..."), so grab the leading run of digits.
|
|
match = re.match(r"\s*(\d+)", edition_text)
|
|
ordinal = match.group(1) if match else "unknown"
|
|
|
|
return f"{issued_on}-{ordinal}-auflage"
|
|
|
|
|
|
# ============================================================================
|
|
# Extraction
|
|
# ============================================================================
|
|
|
|
|
|
def extract_zip(zip_path, destination):
|
|
"""Extract every file from `zip_path` into `destination`.
|
|
|
|
Callers must run `validate_zip()` first — this function assumes
|
|
the archive has already been checked for path traversal and
|
|
symlinks. Directory entries inside the archive are skipped; parent
|
|
directories are created on the fly.
|
|
"""
|
|
destination.mkdir(parents=True, exist_ok=True)
|
|
with zipfile.ZipFile(zip_path) as zf:
|
|
for member in zf.infolist():
|
|
if member.is_dir():
|
|
continue
|
|
relative_path = _normalize_zip_name(member.filename)
|
|
target_path = destination / relative_path
|
|
target_path.parent.mkdir(parents=True, exist_ok=True)
|
|
with zf.open(member) as source, target_path.open("wb") as out:
|
|
shutil.copyfileobj(source, out)
|
|
|
|
|
|
# ============================================================================
|
|
# Atomic file write
|
|
# ============================================================================
|
|
|
|
|
|
def atomic_write_json(path, payload):
|
|
"""Write `payload` as pretty JSON to `path`, atomically.
|
|
|
|
Sequence (see DESIGN.md §4 step 7):
|
|
1. Write the bytes to a sibling `<name>.tmp` so any failure
|
|
doesn't leave a half-written `path`.
|
|
2. fsync the tmp file so the bytes are on disk, not just buffered.
|
|
3. `os.replace` performs the atomic swap.
|
|
4. Best-effort fsync the containing directory so the rename
|
|
itself is durable on POSIX. Windows doesn't support directory
|
|
fsync — we catch and ignore that.
|
|
|
|
A failure during step 1-2 removes the tmp file and the old `path`
|
|
stays intact. A failure after step 3 leaves the new file in place.
|
|
Either way we never end up with a corrupted `path`.
|
|
"""
|
|
parent = path.parent
|
|
parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
tmp_path = parent / (path.name + ".tmp")
|
|
encoded = json.dumps(payload, ensure_ascii=False, indent=2).encode("utf-8")
|
|
|
|
# Open with low-level os.open so we can fsync the file descriptor
|
|
# before closing — important for crash safety.
|
|
fd = os.open(
|
|
tmp_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o644,
|
|
)
|
|
try:
|
|
with os.fdopen(fd, "wb") as f:
|
|
f.write(encoded)
|
|
f.flush()
|
|
os.fsync(f.fileno())
|
|
except Exception:
|
|
# If anything went wrong before we got to `os.replace`, drop
|
|
# the tmp file so we don't leak garbage on disk.
|
|
tmp_path.unlink(missing_ok=True)
|
|
raise
|
|
|
|
# The atomic swap.
|
|
os.replace(tmp_path, path)
|
|
|
|
# Best-effort: persist the rename itself.
|
|
_fsync_directory_best_effort(parent)
|
|
|
|
|
|
def _fsync_directory_best_effort(directory):
|
|
"""fsync a directory so the most recent rename inside it is durable.
|
|
|
|
Some platforms (notably Windows) don't support opening a directory
|
|
for fsync — that's fine, the `os.replace` itself is still atomic,
|
|
just slightly less crash-durable on those platforms.
|
|
"""
|
|
try:
|
|
dir_fd = os.open(directory, os.O_DIRECTORY)
|
|
except OSError:
|
|
return
|
|
try:
|
|
os.fsync(dir_fd)
|
|
except OSError:
|
|
pass
|
|
finally:
|
|
os.close(dir_fd)
|
|
|
|
|
|
# ============================================================================
|
|
# Idempotency check
|
|
# ============================================================================
|
|
|
|
|
|
def already_up_to_date(out_dir, upstream_last_modified):
|
|
"""Return True if the previous run already fetched this exact version.
|
|
|
|
Compares the upstream Last-Modified value against the one we
|
|
recorded in `manifest-latest.json` on the previous successful run.
|
|
Also verifies the per-edition `manifest.json` still exists and
|
|
parses, so a half-cleaned-up `data/` doesn't fool us.
|
|
|
|
Returns False if anything is missing, mismatched, or unparseable —
|
|
the safe choice is always to re-download.
|
|
"""
|
|
if upstream_last_modified is None:
|
|
# Server didn't give us a Last-Modified; we can't trust the
|
|
# cache for this run.
|
|
return False
|
|
|
|
latest_pointer = out_dir / "manifest-latest.json"
|
|
if not latest_pointer.exists():
|
|
return False
|
|
|
|
# Read the pointer file. Anything unparseable means we proceed
|
|
# with a fresh download.
|
|
try:
|
|
latest = json.loads(latest_pointer.read_text("utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
return False
|
|
if not isinstance(latest, dict):
|
|
return False
|
|
|
|
# The header on disk must exactly equal the one the server just
|
|
# told us about.
|
|
if latest.get("http_last_modified") != upstream_last_modified:
|
|
return False
|
|
|
|
# Cross-check that the edition the pointer references is still on
|
|
# disk and not just a stale pointer.
|
|
cached_slug = latest.get("slug") or ""
|
|
if not cached_slug:
|
|
return False
|
|
cached_manifest = out_dir / cached_slug / "manifest.json"
|
|
if not cached_manifest.exists():
|
|
return False
|
|
try:
|
|
json.loads(cached_manifest.read_text("utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
# ============================================================================
|
|
# Main orchestration
|
|
# ============================================================================
|
|
|
|
|
|
def _parse_args(argv):
|
|
"""Build the argparse object and parse `argv`."""
|
|
parser = argparse.ArgumentParser(
|
|
description=(
|
|
"Download the BNetzA German amateur-radio exam question "
|
|
"catalog, validate it, and extract it into a per-edition "
|
|
"directory."
|
|
),
|
|
)
|
|
parser.add_argument(
|
|
"--out",
|
|
type=Path,
|
|
default=DEFAULT_OUT_DIR,
|
|
help="output root directory (default: ./data)",
|
|
)
|
|
parser.add_argument(
|
|
"--force",
|
|
action="store_true",
|
|
help=(
|
|
"re-download and replace an existing edition directory "
|
|
"when its content has changed"
|
|
),
|
|
)
|
|
parser.add_argument(
|
|
"--keep-zip",
|
|
action="store_true",
|
|
help=(
|
|
"keep the raw downloaded ZIP next to the extracted tree, "
|
|
"as <slug>/source.zip"
|
|
),
|
|
)
|
|
# Hidden flag, useful for tests that want to point at a local
|
|
# mirror or fixture URL.
|
|
parser.add_argument(
|
|
"--source-url",
|
|
default=DEFAULT_SOURCE_URL,
|
|
help=argparse.SUPPRESS,
|
|
)
|
|
return parser.parse_args(argv)
|
|
|
|
|
|
def main(argv=None):
|
|
"""Top-level entry point. Returns an exit code; never raises."""
|
|
args = _parse_args(argv)
|
|
out_dir = args.out
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# ---- Step 1: HEAD + idempotency skip.
|
|
# If the upstream file hasn't been touched since our last
|
|
# successful run, there's no useful work left to do.
|
|
try:
|
|
upstream_last_modified = http_head_last_modified(args.source_url)
|
|
except (urllib.error.URLError, TimeoutError) as e:
|
|
print(f"error: HEAD request failed: {e}", file=sys.stderr)
|
|
return EXIT_ERROR
|
|
|
|
if not args.force and already_up_to_date(out_dir, upstream_last_modified):
|
|
cached = json.loads(
|
|
(out_dir / "manifest-latest.json").read_text("utf-8"),
|
|
)
|
|
print(
|
|
f"up to date: {cached['slug']} "
|
|
f"(Last-Modified={upstream_last_modified})"
|
|
)
|
|
return EXIT_OK
|
|
|
|
# ---- Step 2: stream the ZIP to a scratch location.
|
|
# We put the scratch file inside `out_dir/.tmp/` (rather than
|
|
# /tmp/) so that the later `os.replace` happens within the same
|
|
# filesystem and is therefore atomic.
|
|
scratch_dir = out_dir / ".tmp"
|
|
scratch_dir.mkdir(parents=True, exist_ok=True)
|
|
zip_scratch = scratch_dir / "download.zip"
|
|
|
|
try:
|
|
sha256, get_last_modified, zip_size = http_get_to_file(
|
|
args.source_url, zip_scratch, max_bytes=MAX_COMPRESSED_BYTES,
|
|
)
|
|
except (urllib.error.URLError, TimeoutError, ValidationError) as e:
|
|
print(f"error: download failed: {e}", file=sys.stderr)
|
|
zip_scratch.unlink(missing_ok=True)
|
|
return EXIT_ERROR
|
|
|
|
# Prefer the Last-Modified from the GET; fall back to the HEAD
|
|
# value if the GET response didn't include one.
|
|
effective_last_modified = get_last_modified or upstream_last_modified
|
|
|
|
# ---- Step 3: validate the archive. Stops here on any structural
|
|
# problem.
|
|
try:
|
|
archive = validate_zip(zip_scratch)
|
|
except ValidationError as e:
|
|
print(f"error: validation failed: {e}", file=sys.stderr)
|
|
zip_scratch.unlink(missing_ok=True)
|
|
return EXIT_ERROR
|
|
|
|
# ---- Step 4: derive the per-edition directory name.
|
|
slug = derive_slug(archive.json_data["metadata"])
|
|
edition_dir = out_dir / slug
|
|
edition_tmp_dir = out_dir / f"{slug}.tmp"
|
|
edition_bak_dir = out_dir / f"{slug}.bak"
|
|
|
|
# ---- Step 5: decide what to do given any existing edition_dir.
|
|
|
|
# When the operator did NOT pass --force, we honor any existing
|
|
# extraction:
|
|
# * same bytes already on disk: skip re-extraction, just
|
|
# refresh the latest pointer
|
|
# * different bytes already on disk: refuse to clobber
|
|
#
|
|
# When --force IS set, we always proceed to re-extract. The
|
|
# sha-match shortcut would mask a corrupted on-disk tree (the
|
|
# manifest could record the correct sha while files inside are
|
|
# missing or truncated), and that is exactly the situation
|
|
# --force exists to repair.
|
|
if not args.force and edition_dir.exists():
|
|
existing_sha = _read_existing_zip_sha(edition_dir)
|
|
if existing_sha == sha256:
|
|
print(f"already extracted (sha256 match): {slug}")
|
|
_handle_zip_after_success(
|
|
zip_scratch, edition_dir, args.keep_zip,
|
|
)
|
|
_write_latest_pointer(
|
|
out_dir, slug, effective_last_modified, sha256,
|
|
)
|
|
_try_remove_empty(scratch_dir)
|
|
return EXIT_OK
|
|
|
|
print(
|
|
f"error: {edition_dir} already exists with different "
|
|
f"content. Re-run with --force to replace.",
|
|
file=sys.stderr,
|
|
)
|
|
zip_scratch.unlink(missing_ok=True)
|
|
return EXIT_BAD_STATE
|
|
|
|
# ---- Step 6: refuse to clobber any stale .tmp/.bak siblings.
|
|
# Each indicates a previous crash and might hold data the operator
|
|
# wants to inspect; we never touch them without explicit human
|
|
# review.
|
|
for stale_path in (edition_tmp_dir, edition_bak_dir):
|
|
if stale_path.exists():
|
|
print(
|
|
f"error: stale path {stale_path} exists. Inspect "
|
|
f"and remove manually before retrying.",
|
|
file=sys.stderr,
|
|
)
|
|
zip_scratch.unlink(missing_ok=True)
|
|
return EXIT_BAD_STATE
|
|
|
|
# ---- Step 7: extract into edition_tmp_dir, write its manifest.
|
|
try:
|
|
extract_zip(zip_scratch, edition_tmp_dir)
|
|
except Exception as e: # noqa: BLE001 (surface unexpected fails as exit 1)
|
|
print(f"error: extraction failed: {e}", file=sys.stderr)
|
|
shutil.rmtree(edition_tmp_dir, ignore_errors=True)
|
|
zip_scratch.unlink(missing_ok=True)
|
|
return EXIT_ERROR
|
|
|
|
# Wrap the remaining steps (manifest write, atomic swap, post-
|
|
# swap housekeeping) so any failure becomes EXIT_ERROR instead of
|
|
# an exception escaping main(). The docstring promises we never
|
|
# raise; this `try` is what enforces it.
|
|
try:
|
|
# Belt-and-braces: extraction should already have written
|
|
# README.txt, but if a hypothetical future archive lacks one
|
|
# we fall back to the text we read into memory during
|
|
# validation.
|
|
readme_in_tmp = edition_tmp_dir / "README.txt"
|
|
if not readme_in_tmp.exists() and archive.readme_text:
|
|
readme_in_tmp.write_text(archive.readme_text, encoding="utf-8")
|
|
|
|
atomic_write_json(
|
|
edition_tmp_dir / "manifest.json",
|
|
_build_manifest(
|
|
source_url=args.source_url,
|
|
archive=archive,
|
|
slug=slug,
|
|
sha256=sha256,
|
|
zip_size=zip_size,
|
|
last_modified=effective_last_modified,
|
|
),
|
|
)
|
|
|
|
# ---- Step 8: atomic directory swap.
|
|
# If a previous edition_dir is present, rename it out of the
|
|
# way first. If the new rename fails we attempt to restore
|
|
# the old one so the operator doesn't end up empty-handed.
|
|
# If the restore ALSO fails we leave both `.bak/` and `.tmp/`
|
|
# on disk: the operator can recover by hand
|
|
# (`mv <slug>.bak/ <slug>/`).
|
|
if edition_dir.exists():
|
|
os.replace(edition_dir, edition_bak_dir)
|
|
try:
|
|
os.replace(edition_tmp_dir, edition_dir)
|
|
except Exception:
|
|
os.replace(edition_bak_dir, edition_dir)
|
|
raise
|
|
shutil.rmtree(edition_bak_dir, ignore_errors=True)
|
|
else:
|
|
os.replace(edition_tmp_dir, edition_dir)
|
|
|
|
# ---- Step 9: post-success housekeeping.
|
|
# These don't undo the extraction, but a failure here (disk
|
|
# full, permission flip) is something the operator should
|
|
# see — so we surface it as EXIT_ERROR rather than swallow.
|
|
_handle_zip_after_success(
|
|
zip_scratch, edition_dir, args.keep_zip,
|
|
)
|
|
_write_latest_pointer(
|
|
out_dir, slug, effective_last_modified, sha256,
|
|
)
|
|
except Exception as e: # noqa: BLE001 (main() promises not to raise)
|
|
print(
|
|
f"error: failure after extraction: {e}",
|
|
file=sys.stderr,
|
|
)
|
|
# If the tmp dir is still on disk it's the abandoned new
|
|
# extraction (either pre-swap, or rolled back after a swap
|
|
# failure) — safe to drop. We deliberately don't touch
|
|
# edition_bak_dir: if it still exists it holds the previous
|
|
# edition's content and the operator may want it for manual
|
|
# recovery.
|
|
if edition_tmp_dir.exists():
|
|
shutil.rmtree(edition_tmp_dir, ignore_errors=True)
|
|
zip_scratch.unlink(missing_ok=True)
|
|
return EXIT_ERROR
|
|
|
|
_try_remove_empty(scratch_dir)
|
|
|
|
print(f"extracted: {edition_dir}")
|
|
if archive.missing_pictures:
|
|
print(
|
|
f"note: {len(archive.missing_pictures)} missing picture "
|
|
f"reference(s) recorded in manifest (soft check)"
|
|
)
|
|
return EXIT_OK
|
|
|
|
|
|
# ============================================================================
|
|
# Small helpers used by main()
|
|
# ============================================================================
|
|
|
|
|
|
def _read_existing_zip_sha(edition_dir):
|
|
"""Return the `zip_sha256` from an existing edition's manifest,
|
|
or `None` if it's missing or unreadable."""
|
|
manifest_path = edition_dir / "manifest.json"
|
|
try:
|
|
existing = json.loads(manifest_path.read_text("utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
return None
|
|
if not isinstance(existing, dict):
|
|
return None
|
|
return existing.get("zip_sha256")
|
|
|
|
|
|
def _build_manifest(
|
|
source_url, archive, slug, sha256, zip_size, last_modified,
|
|
):
|
|
"""Construct the per-edition `manifest.json` payload."""
|
|
return {
|
|
"source_url": source_url,
|
|
"fetched_at": dt.datetime.now(dt.timezone.utc).isoformat(
|
|
timespec="seconds",
|
|
),
|
|
"http_last_modified": last_modified,
|
|
"zip_sha256": sha256,
|
|
"zip_size": zip_size,
|
|
"json_filename": archive.json_filename,
|
|
"metadata": archive.json_data["metadata"],
|
|
"attribution": archive.attribution,
|
|
"missing_pictures": archive.missing_pictures,
|
|
"slug": slug,
|
|
}
|
|
|
|
|
|
def _handle_zip_after_success(zip_path, edition_dir, keep_zip):
|
|
"""Either move the downloaded ZIP into `edition_dir` (when
|
|
`--keep-zip` is set) or remove it.
|
|
|
|
Either way, the scratch ZIP is gone by the time this returns.
|
|
"""
|
|
if keep_zip and zip_path.exists():
|
|
try:
|
|
shutil.move(str(zip_path), str(edition_dir / "source.zip"))
|
|
return
|
|
except OSError:
|
|
# Falling through to unlink: archival is best-effort and
|
|
# never worth failing the whole run over.
|
|
pass
|
|
zip_path.unlink(missing_ok=True)
|
|
|
|
|
|
def _write_latest_pointer(out_dir, slug, last_modified, sha256):
|
|
"""Atomically update the pointer that drives the idempotency check."""
|
|
atomic_write_json(
|
|
out_dir / "manifest-latest.json",
|
|
{
|
|
"slug": slug,
|
|
"http_last_modified": last_modified,
|
|
"zip_sha256": sha256,
|
|
},
|
|
)
|
|
|
|
|
|
def _try_remove_empty(directory):
|
|
"""Remove `directory` if it's empty. Silent if it isn't or doesn't exist."""
|
|
try:
|
|
directory.rmdir()
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|