Skip to content
GitLab
Commit 800a5674 authored by Eric Duminil's avatar Eric Duminil
Browse files

Merge branch 'refactor'

parents 45363a95 70340023
Hide whitespace changes
Inline Side-by-side
02_bayern.py
View file @ 800a5674
#!/usr/bin/env python3
# https://geodaten.bayern.de/opengeodata/OpenDataDetail.html?pn=lod2
# Lizenz CC BY 4.0
# Koordinatensystem(e) UTM32 (EPSG:25832)
# Datenformat(e) CITYGML
# Abgabeeinteilung Kachelung 2km x 2km
# Aktualisierung wöchentlich
# Datenmenge ca. 150 GB
# Weitere Informationen Hinweise zur Datenabgabe
# More info : https://geodaten.bayern.de/odd/m/3/pdf/hinweise_daten_lod2_download.pdf
# Lizenz: CC BY 4.0
# Koordinatensystem(e): UTM32 (EPSG:25832)
# Datenformat(e): CITYGML
# Abgabeeinteilung: Kachelung 2km x 2km
# Aktualisierung: wöchentlich
# Datenmenge: ca. 150 GB
# Weitere Informationen: https://geodaten.bayern.de/odd/m/3/pdf/hinweise_daten_lod2_download.pdf
import random
import xml.etree.ElementTree as ET
from pathlib import Path
import download_metalink
from citygml_download import TMP_DIR, Bundesland, CityGMLWithHash, download_all_files, download_file
BAYERN = Bundesland(
"Bayern",
source="https://geodaten.bayern.de/odd/a/lod2/citygml/meta/metalink/09.meta4",
info="https://www.ldbv.bayern.de/",
)
def download_metalink(metalink_url: str, tmp_path: Path = TMP_DIR) -> Path:
"""Download metalink file to temporary directory."""
print(f"Downloading {metalink_url}")
basename = metalink_url.split("/")[-1] or "metalink.xml"
metalink_path = tmp_path / basename
download_file(metalink_url, metalink_path)
return metalink_path
def parse_metalink(metalink_path: Path, bundesland: Bundesland) -> list[CityGMLWithHash]:
"""Parse metalink XML file and return CityGMLWithHash objects."""
print(f"Parsing metalink file: {metalink_path}")
tree = ET.parse(metalink_path)
root = tree.getroot()
SCRIPT_DIR = Path(__file__).resolve().parent
# Handle namespace
ns = {"ml": "urn:ietf:params:xml:ns:metalink"}
files = []
for file_elem in root.findall("ml:file", ns):
filename = file_elem.get("name")
if filename is None:
raise ValueError(f"Not enough information from {file_elem}")
hash_node = file_elem.find('ml:hash[@type="sha-256"]', ns)
if hash_node is None or hash_node.text is None:
raise ValueError(f"{filename} has no sha256")
urls = [url.text for url in file_elem.findall("ml:url", ns) if url.text]
if len(urls) == 0:
raise ValueError(f"No URL for {file_elem}")
else:
# Randomize URL order to distribute load across servers
random.shuffle(urls)
url = urls[0]
files.append(
CityGMLWithHash(
url=url,
coordinate_reference_system="EPSG:25832",
bundesland=bundesland,
expected_sha256=hash_node.text,
)
)
return files
BAYERN_METALINK = "https://geodaten.bayern.de/odd/a/lod2/citygml/meta/metalink/09.meta4"
if __name__ == "__main__":
download_metalink.download_all_files(BAYERN_METALINK, SCRIPT_DIR / "citygml" / "bayern")
metalink_path = download_metalink(BAYERN.source)
citygmls = parse_metalink(metalink_path, BAYERN)
# Download all files
download_all_files(citygmls)
09_niedersachsen.py
View file @ 800a5674
# https://www.lgln.niedersachsen.de/startseite/geodaten_karten/3d_geobasisdaten/3d_gebaudemodelle/3d-gebaudemodelle-lod1-und-lod2-142891.html
# Für alle offenen Geodaten des LGLN gilt die Lizenz CC BY 4.0.
import json
import os
import sys
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime
from pathlib import Path
from download_metalink import SCRIPT_DIR, download_file
from citygml_download import TMP_DIR, Bundesland, CityGMLWithDate, download_all_files, download_file
NIEDERSACHSEN_GEOJSON_URL = (
"https://arcgis-geojson.s3.eu-de.cloud-object-storage.appdomain.cloud/lod2/lgln-opengeodata-lod2.geojson"
NIEDERSACHSEN = Bundesland(
"Niedersachsen",
source="https://arcgis-geojson.s3.eu-de.cloud-object-storage.appdomain.cloud/lod2/lgln-opengeodata-lod2.geojson",
info="https://www.lgln.niedersachsen.de/startseite/geodaten_karten/3d_geobasisdaten/3d_gebaudemodelle/3d-gebaudemodelle-lod1-und-lod2-142891.html",
)
......@@ -40,125 +36,22 @@ NIEDERSACHSEN_GEOJSON_URL = (
# },
# TODO: Dry with 02_bayern.py and 10_nordrhein_westfalen.py, and move some functions to utils.py?
# NOTE: Could define a CityGML class: URL, Path, Size, SHA, Date, UTM, Source, Bundesland. And "Up to date?", "Check" method, as well as "Download". "Select", "Compress" too?
def download_and_verify(file_info: dict, tmp_dir: Path, download_dir: Path, max_retries=3) -> bool:
"""Download a file, verify its age, and move to final location."""
url = file_info["xml"]
filename = url.split("/")[-1]
last_updated_server = datetime.strptime(file_info["Aktualitaet"], "%Y-%m-%d %H:%M:%S")
urls = [url]
final_path = download_dir / filename
# Check if file already exists and is valid
if final_path.exists():
print(f"Checking existing file: {filename}")
if (
os.path.getsize(final_path) > 0
and datetime.fromtimestamp(os.path.getmtime(final_path)) > last_updated_server
):
print(f"✓ {filename} already downloaded and verified")
return True
else:
print(f" Existing file is too old, 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
print(f" Verifying size for {filename}")
actual_size = os.path.getsize(tmp_path)
if actual_size > 0:
# 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("Empty GML")
tmp_path.unlink()
print(f"✗ Failed to download {filename} after {max_retries} attempts")
return False
def download_all_files(
json_url: str,
local_json: 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)
local_json_path = tmp_path / local_json
if __name__ == "__main__":
local_json_path = TMP_DIR / "niedersachsen_lod2.geojson"
if not local_json_path.exists():
download_file(json_url, local_json_path)
download_file(NIEDERSACHSEN.source, local_json_path)
with open(local_json_path) as json_file:
data = json.load(json_file)
features = data["features"]
files = [feature["properties"] for feature in features]
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["xml"]
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__":
download_all_files(
NIEDERSACHSEN_GEOJSON_URL,
"niedersachsen_lod2.geojson",
SCRIPT_DIR / "citygml" / "niedersachsen",
retries=2,
)
citygmls = []
for feature in data["features"]:
citygmls.append(
CityGMLWithDate(
url=feature["properties"]["xml"],
bundesland=NIEDERSACHSEN,
source_date=datetime.strptime(feature["properties"]["Aktualitaet"], "%Y-%m-%d %H:%M:%S"),
# coordinate_reference_system ?
)
)
download_all_files(citygmls)
10_nordrhein_westfalen.py
View file @ 800a5674
# https://www.opengeodata.nrw.de/produkte/geobasis/3dg/lod2_gml/
import json
import os
import sys
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
from download_metalink import SCRIPT_DIR, download_file
NRW_SERVER = "https://www.opengeodata.nrw.de/produkte/geobasis/3dg/lod2_gml/lod2_gml/"
NRW_JSON = SCRIPT_DIR / "tmp" / "nrw_lod2.json"
# TODO: Dry with 02_bayern.py, and move some functions to utils.py?
def download_and_verify(file_info: dict, tmp_dir: Path, download_dir: Path, max_retries=3) -> bool:
"""Download a file, verify its size, and move to final location."""
filename = file_info["name"]
expected_size = int(file_info["size"])
urls = [NRW_SERVER + filename]
final_path = download_dir / filename
# Check if file already exists and is valid
if final_path.exists():
print(f"Checking existing file: {filename}")
if os.path.getsize(final_path) == expected_size:
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
print(f" Verifying size for {filename}")
actual_size = os.path.getsize(tmp_path)
if actual_size == expected_size:
# 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" Size mismatch for {filename}")
print(f" Expected: {expected_size}")
print(f" Got: {actual_size}")
tmp_path.unlink()
from citygml_download import TMP_DIR, Bundesland, CityGMLWithSize, download_all_files, download_file
print(f"✗ Failed to download {filename} after {max_retries} attempts")
return False
NORDRHEIN_WESTFALEN = Bundesland(
"Nordrhein-Westfalen",
source="https://www.opengeodata.nrw.de/produkte/geobasis/3dg/lod2_gml/lod2_gml/",
info="https://www.opengeodata.nrw.de/produkte/geobasis/3dg/lod2_gml",
)
def download_all_files(
json_url: 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 __name__ == "__main__":
nrw_json = TMP_DIR / "nrw_lod2.json"
download_file(
json_url,
NRW_JSON,
NORDRHEIN_WESTFALEN.source + "index.json",
nrw_json,
)
with open(NRW_JSON) as json_file:
with open(nrw_json) as json_file:
data = json.load(json_file)
# Parse metalink file
files = [file for sets in data["datasets"] for file in sets["files"]]
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__":
download_all_files(NRW_SERVER + "index.json", SCRIPT_DIR / "citygml" / "nordrhein_westfalen", retries=2)
citygmls = []
for sets in data["datasets"]:
for file in sets["files"]:
name = file["name"]
citygmls.append(
CityGMLWithSize(
url=NORDRHEIN_WESTFALEN.source + name,
bundesland=NORDRHEIN_WESTFALEN,
expected_size=int(file["size"]),
# coordinate_reference_system ?
)
)
download_all_files(citygmls)
citygml_download.py 0 → 100644
View file @ 800a5674
#!/usr/bin/env python3
"""
CityGML dataclass for German federal states building data.
"""
import hashlib
import os
import sys
import time
from abc import ABC, abstractmethod
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Callable, Sequence
from urllib.error import URLError
from urllib.request import urlopen
SCRIPT_DIR = Path(__file__).resolve().parent
TMP_DIR = SCRIPT_DIR / "tmp"
DOWNLOAD_DIR = SCRIPT_DIR / "citygml"
TMP_DIR.mkdir(parents=True, exist_ok=True)
def download_all_files(
files: Sequence["CityGML"], tmp_dir: Path = TMP_DIR, jobs: int = 3, retries: int = 2, sleep: int = 5
):
"""Download all CityGML files with parallel downloads."""
print(f"Found {len(files)} files to download\n")
# Download files in parallel
successful = 0
failed = 0
with ThreadPoolExecutor(max_workers=jobs) as executor:
futures = {executor.submit(file.download, tmp_dir, retries, sleep): file for file in files}
for future in as_completed(futures):
file = futures[future]
try:
if future.result():
successful += 1
else:
failed += 1
except Exception as e:
print(f"✗ Exception while processing {file.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)
def download_file(url: str, dest_path: Path, sleep: int = 0, chunk_size: int = 8192) -> bool:
"""Download file from URL to destination path."""
try:
if sleep:
time.sleep(sleep)
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
@dataclass
class Bundesland:
name: str
source: str ="Unknown"
info: str ="Unknown"
license: str = "CC BY 4.0."
# TODO: Add Sequence[CityGML]
# TODO: Get size
# TODO: Get count
# TODO: Define Germany
# TODO: Tmp file, TMP_DIR / source.ext?
def __post_init__(self):
self.download_folder.mkdir(parents=True, exist_ok=True)
@property
def download_folder(self) -> Path:
return DOWNLOAD_DIR / self.name.lower().replace("-", "_").replace("ü", "ue")
@dataclass
class CityGML(ABC):
"""
Represents a CityGML file with metadata and download capabilities.
All properties must be defined - no optionals.
Subclasses or instances must implement the verify method.
"""
url: str
bundesland: Bundesland
coordinate_reference_system: str = "Unknown"
@property
def download_folder(self) -> Path:
return self.bundesland.download_folder
@property
def path(self) -> Path:
return self.download_folder / self.filename
@property
def filename(self) -> str:
"""Extract filename from URL."""
return self.url.split("/")[-1]
def size(self) -> int:
"""Calculate file size if path exists, otherwise return 0."""
if self.path.exists():
return os.path.getsize(self.path)
return 0
def sha256(self) -> str:
"""Calculate SHA-256 hash if path exists, otherwise return empty string."""
if not self.path.exists():
return ""
sha256_hash = hashlib.sha256()
with open(self.path, "rb") as f:
for chunk in iter(lambda: f.read(8192), b""):
sha256_hash.update(chunk)
return sha256_hash.hexdigest()
def date(self) -> datetime:
"""Get file modification datetime if path exists, otherwise return epoch."""
if self.path.exists():
return datetime.fromtimestamp(os.path.getmtime(self.path))
return datetime.fromtimestamp(0)
@abstractmethod
def verify(self, file_path: Path) -> bool:
"""
Verify that the downloaded file is valid.
Args:
file_path: Path to the file to verify
Returns:
True if file is valid, False otherwise
"""
pass
def is_up_to_date(self) -> bool:
"""Check if file exists and is up to date."""
if not self.path.exists():
return False
if self.size() == 0:
return False
return self.verify(self.path)
def download(self, tmp_dir: Path = TMP_DIR, max_retries: int = 3, sleep: int = 0) -> bool:
"""
Download and verify the file.
Args:
tmp_dir: Temporary directory for downloads
max_retries: Maximum number of retry attempts
Returns:
True if download and verification successful, False otherwise
"""
# Check if already up to date
if self.is_up_to_date():
print(f"✓ {self.filename} already downloaded and verified")
return True
# Remove old file if it exists but is invalid
if self.path.exists():
print(f" Existing file is invalid, will re-download")
self.path.unlink()
tmp_path = tmp_dir / self.filename
for attempt in range(max_retries):
print(f"Downloading {self.filename} (attempt {attempt + 1}/{max_retries})")
print(f" Trying {self.url}")
# Try to download
if not download_file(self.url, tmp_path, sleep):
print(f" Failed to download")
if tmp_path.exists():
tmp_path.unlink()
continue
# Verify the downloaded file
print(f" Verifying {self.filename}")
if self.verify(tmp_path):
# Move to final location
tmp_path.rename(self.path)
print(f"✓ {self.filename} downloaded and verified successfully")
return True
else:
print(f" Verification failed")
if tmp_path.exists():
tmp_path.unlink()
print(f"✗ Failed to download {self.filename} after {max_retries} attempts")
return False
@dataclass
class CityGMLWithHash(CityGML):
"""CityGML file verified by SHA-256 hash."""
expected_sha256: str = ""
def verify(self, file_path: Path) -> bool:
"""Verify file using SHA-256 hash."""
# Calculate hash for the file to verify
sha256_hash = hashlib.sha256()
with open(file_path, "rb") as f:
for chunk in iter(lambda: f.read(8192), b""):
sha256_hash.update(chunk)
actual_hash = sha256_hash.hexdigest()
if actual_hash != self.expected_sha256:
print(f" Hash mismatch: expected {self.expected_sha256}, got {actual_hash}")
return False
return True
@dataclass
class CityGMLWithSize(CityGML):
"""CityGML file verified by file size."""
expected_size: int = 0
def verify(self, file_path: Path) -> bool:
"""Verify file using size."""
actual_size = os.path.getsize(file_path)
if actual_size != self.expected_size:
print(f" Size mismatch: expected {self.expected_size}, got {actual_size}")
return False
return True
@dataclass
class CityGMLWithDate(CityGML):
"""CityGML file verified by modification date (file must be newer than source)."""
source_date: datetime = datetime.fromtimestamp(0)
def verify(self, file_path: Path) -> bool:
"""Verify file is newer than the source date and non-empty."""
file_mtime = datetime.fromtimestamp(os.path.getmtime(file_path))
if file_mtime <= self.source_date:
print(f" File too old: {file_mtime} <= {self.source_date}")
return False
return True
@dataclass
class CityGMLWithCustomVerify(CityGML):
"""
CityGML file with custom verification function.
The verify_func should be set after instantiation:
obj = CityGMLWithCustomVerify(...)
obj.verify_func = lambda path: os.path.getsize(path) > 1000
"""
verify_func: Callable[[Path], bool] = lambda _path: False
def verify(self, file_path: Path) -> bool:
"""Verify file using custom function."""
return self.verify_func(file_path)
download_metalink.py deleted 100644 → 0
View file @ 45363a95
#!/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)
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment

Dies ist die Gitlab-Instanz des Transferportals der Hochschule für Technik Stuttgart. Hier geht es zurück zum Portal