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 is skipped with a
    logged warning, so one broken package cannot break data access for every
    other model.
    """
    found, _ = _discover()
    return found

def fetch_for(model: str, data_root: str | Path | None = None) -> 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.
    """
    from fwl_io.fetch import create_fetcher

    model = model.lower()
    fetched: dict[str, list[Path]] = {}
    failures: dict[str, str] = {}
    providers, provider_errors = _discover()
    for datasets in providers.values():
        for ds in datasets:
            if model not in tuple(r.lower() for r in ds.required_by):
                continue
            try:
                fetcher = create_fetcher(
                    subdir=ds.subdir,
                    zenodo=ds.zenodo,
                    dataverse=ds.dataverse,
                    registry=ds.registry(),
                    data_root=data_root,
                    extract=ds.extract,
                )
                fetched[ds.key] = fetcher.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) could not be read:')
            report += [f'  {name}: {msg}' for name, msg in sorted(provider_errors.items())]
        raise RuntimeError('\n'.join(report))
    return fetched

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}'
            )
        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,
            )
        )
    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)