manifest

def discover_manifests() -> dict[str, list[Dataset]]:
    """Collect datasets from every installed ``fwl_io.manifests`` entry point.

    Entry points must resolve to a zero-argument callable returning the
    manifest path. A provider whose manifest fails to load, or that conflicts
    with another provider, is skipped and reported by ``fwl-io list`` and
    ``check_for``, so one broken package cannot break data access for every
    other model.
    """
    return _discover_all().found

def fetch_for(
    model: str, data_root: str | Path | None = None, progress: bool = False
) -> dict[str, list[Path]]:
    """Fetch every dataset a given model requires; return paths per dataset.

    All matching datasets are attempted; failures are collected and raised
    together at the end so one broken dataset does not block the others
    (files fetched before the error remain in place).

    An unreadable manifest counts as a failure here, even when other datasets
    arrived: a manifest that cannot be parsed may be the one declaring this
    model, and a model is routinely served by both its own manifest and the
    shared one. Listing is more forgiving, since it reports per provider and
    a reader can see which one is missing.

    Parameters
    ----------
    model : str
        Model name matched (case-insensitively) against ``required_by``.
    data_root : str | Path | None
        Override for the data root; defaults to the resolved FWL_DATA tree.
    progress : bool
        Show a per-file download progress bar. When tqdm is not installed, or
        there is no ``sys.stderr`` to draw on, the bar is skipped and the fetch
        continues.
    """
    progress = _usable_progress(progress)
    model = model.lower()
    fetched: dict[str, list[Path]] = {}
    failures: dict[str, str] = {}
    discovery = _discover_all()
    providers = discovery.found
    provider_errors = discovery.errors_for(model)
    for datasets in providers.values():
        for ds in datasets:
            if model not in tuple(r.lower() for r in ds.required_by):
                continue
            try:
                fetched[ds.key] = _fetcher_of(ds, data_root, progress).fetch_all()
            except Exception as exc:  # noqa: BLE001 -- aggregate and re-raise below
                failures[ds.key] = str(exc)
    if failures or provider_errors:
        report = [f'fetching data for model {model!r} failed ({len(fetched)} dataset(s) arrived)']
        if failures:
            report.append(f'{len(failures)} dataset(s) failed:')
            report += [f'  {key}: {msg}' for key, msg in sorted(failures.items())]
        if provider_errors:
            report.append(f'{len(provider_errors)} manifest(s) not used:')
            by_message: dict[str, list[str]] = {}
            for name, error in sorted(provider_errors.items()):
                by_message.setdefault(error.message, []).append(name)
            report += [f'  {", ".join(names)}: {msg}' for msg, names in by_message.items()]
        raise RuntimeError('\n'.join(report))
    return fetched

def fetcher_for_key(key: str, data_root: str | Path | None = None, progress: bool = False):
    """Return the fetcher of the dataset ``key`` declared in an installed manifest.

    It is the fetcher :func:`fetch_for` uses, with the same mirrors and checks, so a code
    that cannot import fwl-io can fetch or locate one dataset through the CLI.

    Parameters
    ----------
    key : str
        Dotted manifest key of the dataset, e.g. ``atmos_clim.refractive.agni_aerosols``.
    data_root : str | Path | None
        Override for the data root; defaults to the resolved FWL_DATA tree.
    progress : bool
        Show a per-file download progress bar when one can be drawn.

    Returns
    -------
    fwl_io.Fetcher
        Fetcher bound to the dataset's version directory.

    Raises
    ------
    LookupError
        No usable installed manifest declares ``key``.
    """
    discovery = _discover_all()
    for datasets in discovery.found.values():
        for ds in datasets:
            if ds.key == key:
                return _fetcher_of(ds, data_root, _usable_progress(progress))
    n = len(discovery.errors)
    unused = f'; {n} manifest(s) not used, see fwl-io list' if n else ''
    raise LookupError(f'no installed manifest declares the dataset {key!r}{unused}')

def load_manifest(path: str | Path) -> list[Dataset]:
    """Load and validate all datasets declared in one manifest file."""
    path = Path(path)
    with path.open('rb') as fh:
        tree = tomllib.load(fh)

    # Read the declaration first: a manifest above this code's schema cannot be
    # judged by this code's rules, so it must be refused before they are applied.
    declared_schema = _read_declared_schema(tree)
    _reject_declared_subdir('the manifest root', tree)
    datasets: list[Dataset] = []
    folded: dict[str, str] = {}
    for key, table in _walk_tables(tree, declared_schema=declared_schema):
        clash = folded.setdefault(key.lower(), key)
        if clash != key:
            raise ValueError(
                f'datasets {clash!r} and {key!r} differ only in case; they would share one '
                f'directory and one registry file on a case-insensitive filesystem'
            )
        zenodo = table.get('zenodo')
        dataverse = table.get('dataverse')
        if zenodo == '':
            raise ValueError(
                f'dataset {key!r}: the "zenodo" value is empty; a Zenodo version DOI is '
                f'required ("dataverse" is a download mirror, not a primary source)'
            )
        if not isinstance(zenodo, str) or not ZENODO_DOI_PATTERN.fullmatch(zenodo):
            raise ValueError(
                f'dataset {key!r}: zenodo value {zenodo!r} is not a Zenodo DOI '
                f'of the form 10.5281/zenodo.<record-id>'
            )
        if dataverse is not None and (
            not isinstance(dataverse, str) or not _GENERIC_DOI_PATTERN.fullmatch(dataverse)
        ):
            raise ValueError(f'dataset {key!r}: dataverse value {dataverse!r} is not a DOI')
        name = table.get('name', key)
        if not isinstance(name, str) or not name.strip():
            raise ValueError(f'dataset {key!r}: "name" must be non-empty text, got {name!r}')
        required_by = table.get('required_by', ())
        if not isinstance(required_by, list | tuple) or not all(
            isinstance(model, str) for model in required_by
        ):
            raise ValueError(
                f'dataset {key!r}: "required_by" must be a list of model names, got {required_by!r}'
            )
        extract = table.get('extract')
        if extract is not None and extract not in ARCHIVE_KINDS:
            raise ValueError(
                f'dataset {key!r}: extract value {extract!r} must be one of {ARCHIVE_KINDS}'
            )
        files = table.get('files')
        if files is not None:
            files = _validate_files(key, files, extract)
        datasets.append(
            Dataset(
                key=key,
                name=name,
                zenodo=zenodo,
                dataverse=dataverse,
                required_by=tuple(required_by),
                registry_path=path.parent / f'{key}.registry.txt',
                extract=extract,
                files=files,
            )
        )
    return datasets

def shared_manifest_path() -> Path:
    """Entry-point target: the manifest of datasets shared across models."""
    return Path(__file__).parent / 'data' / 'shared_manifest.toml'

def zenodo_record_id(doi: str) -> str:
    """Return the numeric record id of a Zenodo version DOI.

    Parameters
    ----------
    doi : str
        A Zenodo DOI of the form ``10.5281/zenodo.<record-id>`` (an optional
        ``doi:`` prefix is accepted).

    Returns
    -------
    str
        The trailing record-id digits.

    Raises
    ------
    ValueError
        When the string is not a Zenodo DOI.
    """
    match = ZENODO_DOI_PATTERN.fullmatch(doi.strip())
    if not match:
        raise ValueError(f'{doi!r} is not a Zenodo DOI of the form 10.5281/zenodo.<id>')
    return match.group(2)