Not needed anymore

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 = SCRIPT_DIR / "tmp", 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)"
-    )
-    # TODO: Add sleep too? ~ 3s by default?
-
-    args = parser.parse_args()
-    download_all_files(args.metalink, args.output, args.tmp, args.jobs, args.retries)