Skip to content
Original file line number Diff line number Diff line change
Expand Up @@ -282,6 +282,8 @@ def resolve_dependencies(
*,
sink_id: str,
force_download: bool = False,
centers: list[str] | None = None,
bundle_centers: list[str] | None = None,
) -> tuple[DependencyResolution, AnyPath | None]:
"""Resolve all dependencies in a spec for the given date.

Expand All @@ -298,6 +300,8 @@ def resolve_dependencies(
date: Target date (timezone-aware datetime, midnight UTC).
sink_id: Local resource alias for storing resolved files
(e.g. ``"local"``).
centers: Optional remote resource IDs to search. When omitted,
all configured centers are eligible.

Returns:
A ``(DependencyResolution, lockfile_path)`` tuple.
Expand All @@ -318,4 +322,6 @@ def resolve_dependencies(
date,
sink_id=sink_id,
force_download=force_download,
centers=centers,
bundle_centers=bundle_centers,
)
Original file line number Diff line number Diff line change
Expand Up @@ -348,6 +348,13 @@ def search(self) -> list[FoundResource]:
continue
seen[key] = True
params = {p.name: p.value for p in rq.product.parameters if p.value is not None}
# Search targets may retain wildcard regexes for unconstrained
# filename axes (for example TTT/PPP on Wuhan ORBIT). Replace
# those patterns with metadata parsed from the actual match so
# callers can group candidates by their real product family.
classified = self._search_planner._product_registry.classify(filename)
if classified and classified.get("product") == rq.product.name:
params.update(classified.get("parameters", {}))
protocol = (rq.server.protocol or "").upper()
is_local = protocol in ("FILE", "LOCAL")
if is_local:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -2,22 +2,24 @@

import logging
import os
import random
import re
import shutil
import threading
import time
from contextlib import contextmanager
from pathlib import Path
from typing import Optional
from urllib.parse import quote, urlparse
from urllib.request import Request, urlopen

import fsspec
import fsspec.utils

logger = logging.getLogger(__name__)

_AIUB_DOWNLOAD_ROOT = "https://www.aiub.unibe.ch/download"

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

These should live in yaml files and go through the propery analysis center registry

_AIUB_LISTING_ENDPOINT = (
"https://code.aiub.unibe.ch/s3_script/aiub_s3_bucket_listing.php?path="
)
_AIUB_LISTING_ENDPOINT = "https://code.aiub.unibe.ch/s3_script/aiub_s3_bucket_listing.php?path="


class ConnectionPool:
Expand Down Expand Up @@ -260,12 +262,8 @@ def _ls(conn: "fsspec.AbstractFileSystem") -> list[str]:
html = conn.cat_file(listing_url)
if isinstance(html, bytes):
html = html.decode("utf-8", errors="replace")
download_prefix = re.escape(
f'{_AIUB_DOWNLOAD_ROOT}/{directory.strip("/")}/'
)
return sorted(
set(re.findall(rf'href="{download_prefix}([^"/]+)"', html))
)
download_prefix = re.escape(f"{_AIUB_DOWNLOAD_ROOT}/{directory.strip('/')}/")
return sorted(set(re.findall(rf'href="{download_prefix}([^"/]+)"', html)))
raw = conn.ls(full_path, detail=False)
return [Path(p).name for p in raw]

Expand Down Expand Up @@ -352,6 +350,30 @@ def download_file(self, hostname: str, remote_path: str, target_dir: str) -> Pat
filename = Path(remote_path).name
local_path = Path(target_dir) / filename

# GithubFileSystem's Contents API path can return only the first 4 MiB
# of larger blobs. Use GitHub's immutable raw-content endpoint for
# the actual transfer while retaining GithubFileSystem for listings.
if pool.protocol == "github":
match = re.fullmatch(r"github://([^:]+):([^@]+)@(.+)", hostname)
if match:
owner, repository, revision = match.groups()
raw_url = (
f"https://raw.githubusercontent.com/{owner}/{repository}/"
f"{revision}/{remote_path.lstrip('/')}"
)
try:
request = Request(raw_url, headers={"Accept-Encoding": "identity"})
with (
urlopen(request, timeout=60) as source,
local_path.open("wb") as destination,
):
shutil.copyfileobj(source, destination, length=1024 * 1024)
return local_path if local_path.stat().st_size > 0 else None
except Exception as exc:
logger.warning("Raw GitHub download failed for %s: %s", raw_url, exc)
local_path.unlink(missing_ok=True)
return None

def _get(conn: "fsspec.AbstractFileSystem") -> Path | None:
conn.get(full_path, str(local_path))
if local_path.exists() and local_path.stat().st_size > 0:
Expand All @@ -365,6 +387,8 @@ def _get(conn: "fsspec.AbstractFileSystem") -> Path | None:
return _get(conn)
except (BrokenPipeError, ConnectionError, EOFError, OSError) as e:
logger.debug("Stale connection for %s, reconnecting: %s", hostname, e)
if pool.protocol != "file":
time.sleep(random.uniform(1.0, 3.0))
fresh = pool.replace_connection(conn)
if fresh is None:
return None
Expand Down
Loading
Loading