Script to download from metalink files

e.g. for Bayern

.gitignore

+citygml/
+tmp/

download_metalink.py

+#!/usr/bin/env python3
+"""
+Metalink file downloader with parallel downloads and SHA-256 verification.
+
+❯ python download_metalink.py --help
+usage: download_metalink.py [-h] [-o OUTPUT] [-t TMP] [-j JOBS] [-r RETRIES] metalink
+
+Download files from a metalink file with verification
+
+positional arguments:
+  metalink              Path to the metalink XML file
+
+options:
+  -h, --help            show this help message and exit
+  -o OUTPUT, --output OUTPUT
+                        Output directory for downloaded files (default: download/)
+  -t TMP, --tmp TMP     Temporary directory for downloads (default: tmp/)
+  -j JOBS, --jobs JOBS  Number of parallel downloads (default: 4)
+  -r RETRIES, --retries RETRIES
+                        Maximum number of retry attempts per file (default: 3)
+
+Example for Bayern:
+
+# From https://geodaten.bayern.de/opengeodata/OpenDataDetail.html?pn=lod2
+python download_metalink.py https://geodaten.bayern.de/odd/a/lod2/citygml/meta/metalink/09.meta4
+"""
+
+import argparse
+import hashlib
+import random
+import sys
+import xml.etree.ElementTree as ET
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from pathlib import Path
+from urllib.error import URLError
+from urllib.request import urlopen
+
+SCRIPT_DIR = Path(__file__).resolve().parent
+
+
+def download_metalink(metalink: str, tmp_path: Path) -> Path:
+    print(f"Downloading {metalink}")
+    basename = metalink.split("/")[-1] or "metalink.xml"
+    metalink_path = tmp_path / basename
+    download_file(metalink, metalink_path)
+    return metalink_path
+
+
+def parse_metalink(metalink_path: Path) -> list[dict]:
+    """Parse metalink XML file and extract file information."""
+    print(f"Parsing metalink file: {metalink_path}")
+    tree = ET.parse(metalink_path)
+    root = tree.getroot()
+
+    # Handle namespace
+    ns = {"ml": "urn:ietf:params:xml:ns:metalink"}
+
+    files = []
+    for file_elem in root.findall("ml:file", ns):
+        hash_node = file_elem.find('ml:hash[@type="sha-256"]', ns)
+        if hash_node is None:
+            raise ValueError(f"{file_elem} has no sha256")
+
+        file_info = {
+            "name": file_elem.get("name"),
+            "hash": hash_node.text,
+            "urls": [url.text for url in file_elem.findall("ml:url", ns)],
+        }
+        files.append(file_info)
+
+    return files
+
+
+def calculate_sha256(file_path: Path) -> str:
+    """Calculate SHA-256 hash of a file."""
+    sha256_hash = hashlib.sha256()
+    with open(file_path, "rb") as f:
+        for chunk in iter(lambda: f.read(8192), b""):
+            sha256_hash.update(chunk)
+    return sha256_hash.hexdigest()
+
+
+def download_file(url: str, dest_path: Path, chunk_size=8192) -> bool:
+    """Download file from URL to destination path."""
+    try:
+        with urlopen(url, timeout=30) as response:
+            dest_path.parent.mkdir(parents=True, exist_ok=True)
+            with open(dest_path, "wb") as f:
+                while True:
+                    chunk = response.read(chunk_size)
+                    if not chunk:
+                        break
+                    f.write(chunk)
+        return True
+    except (URLError, OSError) as e:
+        print(f"  Error downloading from {url}: {e}")
+        return False
+
+
+def download_and_verify(file_info: dict, tmp_dir: Path, download_dir: Path, max_retries=3) -> bool:
+    """Download a file, verify its hash, and move to final location."""
+    filename = file_info["name"]
+    expected_hash = file_info["hash"]
+    urls = file_info["urls"]
+    # In order to not always download from the same server
+    random.shuffle(urls)
+
+    final_path = download_dir / filename
+
+    # Check if file already exists and is valid
+    if final_path.exists():
+        print(f"Checking existing file: {filename}")
+        if calculate_sha256(final_path) == expected_hash:
+            print(f"✓ {filename} already downloaded and verified")
+            return True
+        else:
+            print(f"  Existing file has incorrect hash, will re-download")
+            final_path.unlink()
+
+    tmp_path = tmp_dir / filename
+
+    for attempt in range(max_retries):
+        print(f"Downloading {filename} (attempt {attempt + 1}/{max_retries})")
+
+        # Try each URL until one succeeds
+        download_success = False
+        for url in urls:
+            print(f"  Trying {url}")
+            if download_file(url, tmp_path):
+                download_success = True
+                break
+
+        if not download_success:
+            print(f"  Failed to download from all URLs")
+            if tmp_path.exists():
+                tmp_path.unlink()
+            continue
+
+        # Verify hash
+        print(f"  Verifying hash for {filename}")
+        actual_hash = calculate_sha256(tmp_path)
+
+        if actual_hash == expected_hash:
+            # Move to final location
+            final_path.parent.mkdir(parents=True, exist_ok=True)
+            tmp_path.rename(final_path)
+            print(f"✓ {filename} downloaded and verified successfully")
+            return True
+        else:
+            print(f"  Hash mismatch for {filename}")
+            print(f"    Expected: {expected_hash}")
+            print(f"    Got:      {actual_hash}")
+            tmp_path.unlink()
+
+    print(f"✗ Failed to download {filename} after {max_retries} attempts")
+    return False
+
+
+def download_all_files(
+    metalink: str, download_path: Path, tmp_path: Path, jobs: int = 4, retries: int = 3
+):
+    # Create directories
+    tmp_path.mkdir(parents=True, exist_ok=True)
+
+    if metalink.startswith("https://"):
+        metalink_path = download_metalink(metalink, tmp_path)
+    else:
+        metalink_path = Path(metalink)
+
+    if not metalink_path.exists():
+        print(f"Error: Metalink file not found: {metalink_path}")
+        sys.exit(1)
+
+    # Parse metalink file
+    files = parse_metalink(metalink_path)
+    print(f"Found {len(files)} files to download\n")
+
+    download_path.mkdir(parents=True, exist_ok=True)
+
+    # Download files in parallel
+    successful = 0
+    failed = 0
+
+    with ThreadPoolExecutor(max_workers=jobs) as executor:
+        futures = {
+            executor.submit(download_and_verify, file_info, tmp_path, download_path, retries): file_info["name"]
+            for file_info in files
+        }
+
+        for future in as_completed(futures):
+            filename = futures[future]
+            try:
+                if future.result():
+                    successful += 1
+                else:
+                    failed += 1
+            except Exception as e:
+                print(f"✗ Exception while processing {filename}: {e}")
+                failed += 1
+
+    # Summary
+    print(f"\n{'='*60}")
+    print(f"Download complete: {successful} successful, {failed} failed")
+    print(f"{'='*60}")
+
+    if failed > 0:
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Download files from a metalink file with verification")
+    parser.add_argument("metalink", type=str, help="Path to the metalink XML file")
+    parser.add_argument(
+        "-o",
+        "--output",
+        type=Path,
+        default=SCRIPT_DIR / "download",
+        help="Output directory for downloaded files (default: download/)",
+    )
+    parser.add_argument(
+        "-t", "--tmp", type=Path, default=SCRIPT_DIR / "tmp", help="Temporary directory for downloads (default: tmp/)"
+    )
+    parser.add_argument("-j", "--jobs", type=int, default=4, help="Number of parallel downloads (default: 4)")
+    parser.add_argument(
+        "-r", "--retries", type=int, default=3, help="Maximum number of retry attempts per file (default: 3)"
+    )
+
+    args = parser.parse_args()
+    download_all_files(args.metalink, args.output, args.tmp, args.jobs, args.retries)