"""Legacy-compatible HubApi wrapper. Provides the same interface as ``modelscope.hub.api.HubApi`` (old SDK) by wrapping the new ``modelscope_hub.HubApi``. """ from __future__ import annotations import os import time import warnings from collections import defaultdict from typing import TYPE_CHECKING, Any from urllib.parse import urlencode from ..api import HubApi from ..constants import RepoType, Visibility from ..errors import ( AlreadyExistsError, AuthenticationError, InvalidParameter, NetworkError, NotExistError, PermissionDeniedError, RequestTimeoutError, is_repo_exists_error, ) from ..utils.logger import get_logger if TYPE_CHECKING: from ..types import PagedResult, RepoInfo logger = get_logger("compat") DEFAULT_DATASET_REVISION = "master" META_FILES_FORMAT = {".json", ".csv", ".jsonl", ".tsv", ".py"} # Transport/credential arguments the legacy surface accepts on almost every # method. They steer *how* a call is made and are never part of a request body. _CONTROL_KWARGS: frozenset[str] = frozenset({"token", "endpoint", "cookies", "headers", "timeout", "max_retries"}) def _split_control_kwargs(kwargs: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]: """Split legacy ``**kwargs`` into ``(control, passthrough)``. The old ``modelscope.hub.api.HubApi`` signature let callers append ``token=`` / ``endpoint=`` to any method, so shims declared ``**kwargs`` and forwarded it wholesale. For the Studio endpoints that turned out to be actively harmful: ``update_studio_settings`` forwards its kwargs as the settings payload, so the caller's API token ended up inside the ``PATCH`` request body (and therefore in server-side request logs), while ``get_studio_logs`` forwards into a keyword-only signature and raised ``TypeError: got an unexpected keyword argument 'token'``. Splitting them apart keeps the permissive legacy signature while ensuring only genuine business fields reach the wire. """ control = {k: v for k, v in kwargs.items() if k in _CONTROL_KWARGS} passthrough = {k: v for k, v in kwargs.items() if k not in _CONTROL_KWARGS} return control, passthrough class _AigcUploadAdapter: """Expose model-only upload methods expected by legacy ``AigcModel``.""" def __init__(self, api: HubApi) -> None: self._api = api def upload_file(self, *, repo_id: str, path_or_fileobj: Any, path_in_repo: str, **kwargs: Any) -> dict: kwargs.pop("token", None) return self._api.upload_file( repo_id=repo_id, repo_type=RepoType.MODEL, path_or_fileobj=path_or_fileobj, path_in_repo=path_in_repo, **kwargs, ) def upload_folder(self, *, repo_id: str, folder_path: Any, **kwargs: Any) -> dict | list[dict] | None: kwargs.pop("token", None) return self._api.upload_folder( repo_id=repo_id, repo_type=RepoType.MODEL, folder_path=folder_path, **kwargs, ) def _resolve_aigc_readme_content(aigc_model: Any, override: str | None = None) -> str | None: content = override if override is not None else getattr(aigc_model, "readme_content", None) if content is not None and not isinstance(content, str): raise InvalidParameter("readme_content must be a string or None") return content def _upload_aigc_readme(api: HubApi, model_id: str, content: str | None, version: str) -> None: if content is None: return api.upload_file( repo_id=model_id, repo_type=RepoType.MODEL, path_or_fileobj=content.encode("utf-8"), path_in_repo="README.md", revision="master", commit_message=f"Update README.md for AIGC version {version}", ) class LegacyHubApi: """Drop-in replacement for the old ``modelscope.hub.api.HubApi``. Accepts the old constructor signature and maps method calls to the new HubApi implementation. """ _api: HubApi _endpoint: str | None def __init__( self, endpoint: str | None = None, token: str | None = None, ) -> None: if endpoint and not endpoint.startswith("http"): endpoint = f"https://{endpoint}" self._endpoint = endpoint self._api = HubApi(endpoint=endpoint, token=token) # ------------------------------------------------------------------ # Authentication # ------------------------------------------------------------------ def login(self, token: str) -> tuple[str | None, Any]: """Login with token (old style returns ``(git_token, cookies)``). Preserves the legacy return contract: a 2-tuple of ``(git_access_token, cookies)`` so that existing callers doing ``token, cookies = api.login(...)`` continue to work. """ if not token: return (None, None) self._api.login(token) git_token = self._api._config.load_git_token() or token cookies = self._api.get_cookies() return (git_token, cookies) def get_cookies(self, access_token: str | None = None, cookies_required: bool = False): """Get cookies for legacy API authentication. Delegates to :meth:`HubApi.get_cookies`. """ return self._api.get_cookies(access_token=access_token, cookies_required=cookies_required) # ------------------------------------------------------------------ # Repository operations # ------------------------------------------------------------------ def get_model(self, model_id: str, revision: str | None = None) -> dict: """Get model info as a raw dict. Parameters ---------- model_id : str Model identifier, e.g. ``"owner/model_name"``. revision : str, optional Branch or tag to query. Reserved for future use. """ info = self._api.get_repo(model_id, RepoType.MODEL, revision=revision) return _repo_info_to_dict(info) def get_model_files( self, model_id: str, revision: str | None = None, root: str | None = None, recursive: bool = True, **kwargs: Any, ) -> list[dict]: """List files in a model repo. Legacy-compatible signature: ``revision`` selects a branch/tag/commit, ``root`` restricts results to files under a sub-path, and ``recursive`` walks subdirectories. Other legacy transport kwargs (e.g. ``use_cookies``, ``headers``) are accepted and ignored — auth is handled by the configured token/session. """ files = self._api.list_repo_files( model_id, RepoType.MODEL, revision=revision, recursive=recursive, ) result = [{"Path": f.path, "Size": f.size} for f in files] if root: prefix = root.strip("/") result = [f for f in result if f["Path"] == prefix or str(f["Path"]).startswith(prefix + "/")] return result def create_repo( self, repo_id: str, *, token: str | None = None, visibility: int | str | None = None, repo_type: str = "model", chinese_name: str | None = None, license: str | None = None, exist_ok: bool = False, create_default_config: bool = False, endpoint: str | None = None, **kwargs: Any, ) -> RepoInfo | None: """Create a repository (legacy signature).""" api = self._api if token or endpoint: api = HubApi(token=token, endpoint=endpoint or self._endpoint) if create_default_config: kwargs["create_default_config"] = True try: return api.create_repo( repo_id, repo_type=repo_type, visibility=visibility, license=license, chinese_name=chinese_name, **kwargs, ) except AlreadyExistsError: if exist_ok: return None raise except Exception as exc: if exist_ok and is_repo_exists_error(exc): return None raise def create_model(self, model_id: str, **kwargs: Any) -> str: """Create a model repo (legacy signature). AIGC models retain their dedicated endpoint and payload mapping. Plain models continue to use the unified :meth:`create_repo` path. """ # Pre-normalize: convert numeric string to int for backward compatibility visibility = kwargs.get("visibility") if isinstance(visibility, str) and visibility.isdigit(): kwargs["visibility"] = int(visibility) try: aigc_model = kwargs.pop("aigc_model", None) if aigc_model is not None: return self._create_aigc_model(model_id, aigc_model, kwargs) self.create_repo(model_id, repo_type="model", **kwargs) except (AuthenticationError, InvalidParameter) as e: if _is_auth_related(e): raise ValueError("Token does not exist, please login first.") from e raise ep = self._endpoint or self._api._config.endpoint return f"{ep}/models/{model_id}" def _create_aigc_model(self, model_id: str, aigc_model: Any, kwargs: dict[str, Any]) -> str: """Create an AIGC model without changing the plain-model code path.""" token = kwargs.pop("token", None) endpoint = kwargs.pop("endpoint", None) visibility = kwargs.pop("visibility", None) license_name = kwargs.pop("license", None) chinese_name = kwargs.pop("chinese_name", None) original_model_id = kwargs.pop("original_model_id", "") gated_mode = kwargs.pop("gated_mode", None) readme_content = _resolve_aigc_readme_content(aigc_model, kwargs.pop("readme_content", None)) if kwargs: unexpected = ", ".join(sorted(kwargs)) raise TypeError(f"create_model() got unexpected keyword argument(s): {unexpected}") api = self._api if token or endpoint: api = HubApi(token=token, endpoint=endpoint or self._endpoint) owner, name = api._parse_repo_id(model_id) normalised_visibility = api._normalize_visibility(visibility) if normalised_visibility is None: normalised_visibility = int(Visibility.PUBLIC) body: dict[str, Any] = { "Path": owner, "Name": name, "ChineseName": chinese_name, "Visibility": normalised_visibility, "License": license_name or "Apache License 2.0", "OriginalModelId": original_model_id, "TrainId": os.environ.get("MODELSCOPE_TRAIN_ID", ""), "TagShowName": aigc_model.tag, "CoverImages": aigc_model.cover_images, "AigcType": aigc_model.aigc_type, "TagDescription": aigc_model.description, "VisionFoundation": aigc_model.base_model_type, "BaseModel": aigc_model.base_model_id or original_model_id, "WeightsName": aigc_model.weight_filename, "WeightsSha256": aigc_model.weight_sha256, "WeightsSize": aigc_model.weight_size, "ModelPath": aigc_model.model_path, "TriggerWords": aigc_model.trigger_words, "ModelSource": aigc_model.model_source, "SubVisionFoundation": aigc_model.base_model_sub_type, } if aigc_model.official_tags: body["OfficialTags"] = aigc_model.official_tags if gated_mode is not None: if normalised_visibility == int(Visibility.PRIVATE): body["ProtectedMode"] = 1 if gated_mode else 2 else: logger.warning("gated_mode is only effective when visibility is PRIVATE, ignored.") cookies = api.get_cookies(access_token=token, cookies_required=True) aigc_model.preupload_weights( cookies=cookies, headers={}, endpoint=api._config.endpoint, ) api.legacy.create_aigc_model(body) aigc_model.upload_to_repo(_AigcUploadAdapter(api), model_id, token) _upload_aigc_readme(api, model_id, readme_content, aigc_model.tag) return f"{api._config.endpoint}/models/{model_id}" def create_model_tag( self, model_id: str, tag_name: str, endpoint: str | None = None, token: str | None = None, aigc_model: Any = None, readme_content: str | None = None, ) -> str: """Create a model tag while preserving the AIGC-specific endpoint. For AIGC models, ``readme_content`` is committed to ``master`` before the immutable version snapshot is created. When omitted, the value on ``aigc_model.readme_content`` is used. """ if not model_id: raise InvalidParameter("model_id is required!") if not tag_name: raise InvalidParameter("tag_name is required!") if tag_name.lower() in {"main", "master"}: raise InvalidParameter( f'tag_name "{tag_name}" is not allowed. ' 'Please use a different tag name (e.g., "v1.0", "v1.1", "latest"). ' "Reserved names: main, master" ) api = self._api if token or endpoint: api = HubApi(token=token, endpoint=endpoint or self._endpoint) if aigc_model is None: if readme_content is not None: raise InvalidParameter("readme_content is only supported for AIGC model tags") api.create_repo_tag(model_id, RepoType.MODEL, tag_name, revision="master") else: owner, name = api._parse_repo_id(model_id) content = _resolve_aigc_readme_content(aigc_model, readme_content) cookies = api.get_cookies(access_token=token, cookies_required=True) aigc_model.preupload_weights( cookies=cookies, headers={}, endpoint=api._config.endpoint, ) _upload_aigc_readme(api, model_id, content, tag_name) try: api.legacy.create_aigc_model_tag( { "CoverImages": aigc_model.cover_images, "Name": name, "Path": owner, "TagShowName": tag_name, "WeightsName": aigc_model.weight_filename, "WeightsSha256": aigc_model.weight_sha256, "WeightsSize": aigc_model.weight_size, "TriggerWords": aigc_model.trigger_words, "AigcType": aigc_model.aigc_type, "VisionFoundation": aigc_model.base_model_type, } ) except (NetworkError, RequestTimeoutError): for attempt in range(5): try: _, tags = api.legacy.list_revisions_detail(model_id, "model") if any(item.get("ShowName") == tag_name or item.get("Revision") == tag_name for item in tags): break except (NetworkError, RequestTimeoutError): pass if attempt < 4: time.sleep(2) else: raise return f"{api._config.endpoint}/models/{model_id}/tags/{tag_name}" def push_model(self, model_id: str, model_dir: str, **kwargs: Any) -> None: """Upload a model directory (legacy signature).""" # Pre-validate model_dir if not os.path.isdir(model_dir): raise ValueError(f"model_dir '{model_dir}' does not exist or is not a directory.") config_files = ("configuration.json", "configuration.yaml", "configuration.yml") if not any(os.path.isfile(os.path.join(model_dir, f)) for f in config_files): logger.warning( "No model configuration file found in '%s'. " "Expected one of: %s. The upload will proceed, " "but the directory may not contain a valid model.", model_dir, ", ".join(config_files), ) # Pre-normalize: convert numeric string to int for backward compatibility visibility = kwargs.get("visibility") if isinstance(visibility, str) and visibility.isdigit(): kwargs["visibility"] = int(visibility) try: try: self._api.create_repo( model_id, repo_type=RepoType.MODEL, visibility=kwargs.get("visibility"), license=kwargs.get("license"), chinese_name=kwargs.get("chinese_name"), ) except AlreadyExistsError: logger.info("Repository '%s' already exists, proceeding with upload.", model_id) except Exception as exc: if not is_repo_exists_error(exc): raise logger.info("Repository '%s' already exists, proceeding with upload.", model_id) self._api.upload_folder( model_id, RepoType.MODEL, model_dir, path_in_repo=kwargs.get("path_in_repo", ""), commit_message=kwargs.get("commit_message"), commit_description=kwargs.get("commit_description"), revision=kwargs.get("revision"), allow_patterns=kwargs.get("allow_patterns"), ignore_patterns=kwargs.get("ignore_patterns"), max_workers=kwargs.get("max_workers"), use_cache=kwargs.get("use_cache"), ) except (AuthenticationError, InvalidParameter) as e: if _is_auth_related(e): raise ValueError("Token does not exist, please login first.") from e raise # ------------------------------------------------------------------ # Endpoint resolution # ------------------------------------------------------------------ def get_endpoint_for_read( self, repo_id: str, *, repo_type: str | None = None, token: str | None = None, ) -> str: """Resolve the best endpoint for read operations. Backward-compatible with the old SDK's ``get_endpoint_for_read()``. Honors ``MODELSCOPE_ENDPOINT`` (or deprecated ``MODELSCOPE_DOMAIN``) and ``MODELSCOPE_PREFER_AI_SITE`` env vars. """ return self._api.resolve_endpoint_for_read( repo_id, repo_type=repo_type or "model", token=token, ) def repo_exists( self, repo_id: str, *, repo_type: str | None = None, endpoint: str | None = None, re_raise: bool = False, token: str | None = None, ) -> bool: """Check if a repo exists (legacy signature with endpoint/token override).""" api = self._api if endpoint is not None or token is not None: api = HubApi( endpoint=endpoint or self._api._config.endpoint, token=token or self._api._config.token, ) try: return api.repo_exists(repo_id, repo_type or "model") except Exception: if re_raise: raise return False def list_repos( self, repo_type: str | RepoType, *, owner: str | None = None, search: str | None = None, sort: str | None = None, page_number: int = 1, page_size: int = 10, **filters: Any, ) -> PagedResult[RepoInfo]: """List repositories of the given type. Delegates to :meth:`HubApi.list_repos`. """ return self._api.list_repos( repo_type, owner=owner, search=search, sort=sort, page_number=page_number, page_size=page_size, **filters, ) def get_repo( self, repo_id: str, repo_type: str | RepoType, *, revision: str | None = None, ) -> RepoInfo: """Get repository information. Delegates to :meth:`HubApi.get_repo`. """ return self._api.get_repo(repo_id, repo_type, revision=revision) # ------------------------------------------------------------------ # Download operations # ------------------------------------------------------------------ def download_model( self, model_id: str, revision: str | None = None, cache_dir: str | None = None, local_dir: str | None = None, ) -> str: """Download a model snapshot.""" import requests as _requests try: result = self._api.download_repo( model_id, repo_type=RepoType.MODEL, revision=revision, cache_dir=cache_dir, local_dir=local_dir, ) except (NotExistError, AuthenticationError, PermissionDeniedError) as e: raise _requests.exceptions.HTTPError(str(e), response=getattr(e, "response", None)) from e return str(result) # ------------------------------------------------------------------ # Studio operations # ------------------------------------------------------------------ def _api_for(self, control: dict[str, Any]) -> HubApi: """Resolve the :class:`HubApi` a single legacy call should run against. The legacy surface treats ``token`` / ``endpoint`` as per-call overrides. Honouring them requires a separate client, because the ambient one is bound to the credential it was constructed with -- which is why passing ``token=`` used to be silently ignored here. """ token = control.get("token") endpoint = control.get("endpoint") config = self._api._config if (token and token != config.token) or (endpoint and endpoint != config.endpoint): return HubApi(token=token or config.token, endpoint=endpoint or config.endpoint) return self._api def deploy_studio(self, studio_id: str, **kwargs: Any) -> dict: control, passthrough = _split_control_kwargs(kwargs) return self._api_for(control).deploy_repo( studio_id, RepoType.STUDIO, payload=passthrough.get("payload"), ) def stop_studio(self, studio_id: str, **kwargs: Any) -> dict: control, _ = _split_control_kwargs(kwargs) return self._api_for(control).stop_repo(studio_id, RepoType.STUDIO) def get_studio_logs(self, studio_id: str, **kwargs: Any) -> dict: control, passthrough = _split_control_kwargs(kwargs) return self._api_for(control).get_repo_logs(studio_id, RepoType.STUDIO, **passthrough) def update_studio_settings(self, studio_id: str, **kwargs: Any) -> dict: control, settings = _split_control_kwargs(kwargs) return self._api_for(control).update_repo_settings(studio_id, RepoType.STUDIO, **settings) def list_studio_secrets(self, studio_id: str, **kwargs: Any) -> list: control, _ = _split_control_kwargs(kwargs) return self._api_for(control).list_secrets(studio_id, RepoType.STUDIO) def add_studio_secret(self, studio_id: str, key: str, value: str, **kwargs: Any) -> None: control, _ = _split_control_kwargs(kwargs) self._api_for(control).add_secret(studio_id, key, value, RepoType.STUDIO) def update_studio_secret(self, studio_id: str, key: str, value: str, **kwargs: Any) -> None: control, _ = _split_control_kwargs(kwargs) self._api_for(control).update_secret(studio_id, key, value, RepoType.STUDIO) def delete_studio_secret(self, studio_id: str, key: str, **kwargs: Any) -> None: control, _ = _split_control_kwargs(kwargs) self._api_for(control).delete_secret(studio_id, key, RepoType.STUDIO) def list_studios(self, **kwargs: Any) -> dict: """List Studio spaces as a raw paginated dict.""" control, passthrough = _split_control_kwargs(kwargs) result = self._api_for(control).list_repos(RepoType.STUDIO, **passthrough) return result.to_dict() def list_studio_variables(self, studio_id: str, **kwargs: Any) -> list: control, _ = _split_control_kwargs(kwargs) return self._api_for(control).list_variables(studio_id, RepoType.STUDIO) def add_studio_variable(self, studio_id: str, key: str, value: str, **kwargs: Any) -> None: control, _ = _split_control_kwargs(kwargs) self._api_for(control).add_variable(studio_id, key, value, RepoType.STUDIO) def update_studio_variable(self, studio_id: str, key: str, value: str, **kwargs: Any) -> None: control, _ = _split_control_kwargs(kwargs) self._api_for(control).update_variable(studio_id, key, value, RepoType.STUDIO) def delete_studio_variable(self, studio_id: str, key: str, **kwargs: Any) -> None: control, _ = _split_control_kwargs(kwargs) self._api_for(control).delete_variable(studio_id, key, RepoType.STUDIO) def list_studio_hardware(self, **kwargs: Any) -> list: control, passthrough = _split_control_kwargs(kwargs) return self._api_for(control).list_studio_hardware(**passthrough) def list_studio_base_images(self, **kwargs: Any) -> list: control, _ = _split_control_kwargs(kwargs) return self._api_for(control).list_studio_base_images() def list_studio_sdk_versions(self, **kwargs: Any) -> list: control, passthrough = _split_control_kwargs(kwargs) return self._api_for(control).list_studio_sdk_versions(**passthrough) # ------------------------------------------------------------------ # Revision resolution # ------------------------------------------------------------------ def get_model_branches_and_tags_details( self, model_id: str, **kwargs: Any, ) -> tuple[list[dict], list[dict]]: """Get model branches and tags as two separate detail lists. Returns ``(branches_detail, tags_detail)`` where each item is a dict with at least ``Revision`` and ``CreatedAt`` keys. """ return self._api.legacy.list_revisions_detail(model_id, "model") def get_model_branches_and_tags( self, model_id: str, **kwargs: Any, ) -> tuple[list[str], list[str]]: """Get model branch and tag names.""" branches_detail, tags_detail = self.get_model_branches_and_tags_details(model_id) branches = [x["Revision"] for x in branches_detail] if branches_detail else [] tags = [x["Revision"] for x in tags_detail] if tags_detail else [] return branches, tags def get_valid_revision_detail( self, model_id: str, revision: str | None = None, cookies: Any = None, endpoint: str | None = None, *, release_timestamp: int | None = None, ) -> dict: """Resolve a model revision to a concrete branch/tag detail dict. Replicates the old ``modelscope.hub.api.HubApi.get_valid_revision_detail`` behavior. When *release_timestamp* is supplied (the old SDK passes ``modelscope.version.__release_datetime__`` converted to epoch seconds), the full version-selection logic is used: * **Dev mode** (``release_timestamp > now + 1 year``): default to ``master``, validate existence. * **Release mode**: pick the newest tag whose ``CreatedAt <= release_timestamp``, fall back to ``master`` if none match. Without *release_timestamp* the simplified rule applies: explicit *revision* is validated, ``None`` defaults to ``master``. """ _ONE_YEAR = 365 * 24 * 60 * 60 branches_detail, tags_detail = self.get_model_branches_and_tags_details(model_id) all_branches = [x["Revision"] for x in branches_detail] if branches_detail else [] all_tags = [x["Revision"] for x in tags_detail] if tags_detail else [] def _find(details: list[dict], name: str) -> dict | None: for item in details: if item.get("Revision") == name: return item return None def _created_at(tag: dict) -> int: """Safely coerce CreatedAt to epoch seconds (handles str, float, ms, None).""" raw = tag.get("CreatedAt") if raw is None: return 0 try: ts = int(float(raw)) except (TypeError, ValueError): return 0 # Normalize millisecond timestamps to seconds if ts > 9_999_999_999: ts = ts // 1000 return ts # --- Dev mode or no release_timestamp --------------------------------- if release_timestamp is None or release_timestamp > int(time.time()) + _ONE_YEAR: if revision is None: revision = "master" if revision not in all_branches and revision not in all_tags: raise NotExistError(f"The model: {model_id} has no revision: {revision}") detail = _find(tags_detail, revision) or _find(branches_detail, revision) return detail or {"Revision": revision} # --- Release mode ----------------------------------------------------- # Explicit branch name → return immediately if revision is not None and revision in all_branches: return _find(branches_detail, revision) or {"Revision": revision} # No tags at all → master (or validate explicit revision) if not tags_detail: if revision is None or revision == "master": return _find(branches_detail, "master") or {"Revision": "master"} raise NotExistError(f"The model: {model_id} has no revision: {revision}") # Has tags if revision is None: candidates = [t for t in tags_detail if _created_at(t) <= release_timestamp] if candidates: return max(candidates, key=_created_at) return _find(branches_detail, "master") or {"Revision": "master"} # Explicit revision if revision in all_tags: return _find(tags_detail, revision) or {"Revision": revision} if revision == "master": return _find(branches_detail, "master") or {"Revision": "master"} valid = ", ".join(all_tags) raise NotExistError(f"The model: {model_id} has no revision: {revision} (valid tags: {valid})") def get_valid_revision( self, model_id: str, revision: str | None = None, cookies: Any = None, endpoint: str | None = None, ) -> str: """Resolve a model revision to a concrete revision string.""" return self.get_valid_revision_detail( model_id, revision=revision, cookies=cookies, endpoint=endpoint, )["Revision"] # ------------------------------------------------------------------ # Collection / Skills # ------------------------------------------------------------------ def get_collection(self, collection_id: str, **kwargs: Any) -> dict: """Fetch collection data — delegates to legacy API.""" return self._api.legacy.get_collection(collection_id) def download_skill(self, skill_id: str, local_dir: str | None = None, **kwargs: Any) -> str: """Download a skill to local directory.""" result = self._api.download_repo( skill_id, repo_type=RepoType.SKILL, local_dir=local_dir, ) return str(result) # ------------------------------------------------------------------ # Type-specific list/create/get/delete (old SDK compatibility) # ------------------------------------------------------------------ _OPENAPI_MAX_PAGE_SIZE = 50 def list_models( self, owner_or_group: str, page_number: int = 1, page_size: int = 10, endpoint: str | None = None, token: str | None = None, ) -> dict: """List models owned by a user/org.""" api = self._api if token: api = HubApi(endpoint=endpoint or self._endpoint, token=token) page = api.list_repos( RepoType.MODEL, owner=owner_or_group, page_number=page_number, page_size=min(page_size, self._OPENAPI_MAX_PAGE_SIZE), ) return { "Models": [_repo_info_to_dict(r) for r in page.items], "TotalCount": page.total_count, } def list_datasets( self, owner: str | None = None, page_size: int = 50, page_number: int = 1, *, owner_or_group: str | None = None, sort: str | None = None, search: str | None = None, endpoint: str | None = None, token: str | None = None, ) -> PagedResult: """List datasets owned by a user/org. .. deprecated:: Use ``list_repos(owner, repo_type='dataset')`` instead. Parameters ---------- owner : str, optional Filter by owner (aligned with ``list_repos``'s ``owner`` param). page_size : int, optional Items per page. Default is 50. page_number : int, optional 1-based page index. Default is 1. owner_or_group : str, optional **Deprecated** alias for ``owner``. Kept for backward compatibility. sort : str, optional Sort key (e.g. ``"downloads"``). search : str, optional Free-text search query. endpoint : str, optional Override API endpoint. token : str, optional Override API token. """ warnings.warn( "list_datasets() is deprecated, use list_repos(owner, repo_type='dataset') instead.", DeprecationWarning, stacklevel=2, ) # Resolve owner: explicit 'owner' takes precedence over deprecated alias actual_owner = owner or owner_or_group api = self._api if token: api = HubApi(endpoint=endpoint or self._endpoint, token=token) return api.list_repos( RepoType.DATASET, owner=actual_owner, search=search, sort=sort, page_number=page_number, page_size=min(page_size, self._OPENAPI_MAX_PAGE_SIZE), ) def create_dataset( self, dataset_name: str, namespace: str, chinese_name: str = "", license: str = "Apache License 2.0", visibility: int = 1, description: str = "", endpoint: str | None = None, token: str | None = None, ) -> str: """Create a dataset repository, return its URL.""" repo_id = f"{namespace}/{dataset_name}" api = self._api if token: api = HubApi(endpoint=endpoint or self._endpoint, token=token) api.create_repo( repo_id, repo_type=RepoType.DATASET, visibility=visibility, license=license, chinese_name=chinese_name, ) ep = endpoint or self._endpoint or api._config.endpoint return f"{ep}/datasets/{namespace}/{dataset_name}" def get_dataset( self, dataset_id: str, revision: str | None = None, *, endpoint: str | None = None, token: str | None = None, ) -> RepoInfo: """Get dataset information via OpenAPI. .. deprecated:: Use ``get_repo(repo_id, repo_type='dataset')`` instead. Parameters ---------- dataset_id : str Dataset identifier (corresponds to ``repo_id`` in ``get_repo``). revision : str, optional Branch or tag to query (aligned with ``get_repo``'s ``revision``). endpoint : str, optional Override API endpoint. token : str, optional Override API token. """ warnings.warn( "get_dataset() is deprecated, use get_repo(repo_id, repo_type='dataset') instead.", DeprecationWarning, stacklevel=2, ) api = self._api if token: api = HubApi(endpoint=endpoint or self._endpoint, token=token) return api.get_repo(dataset_id, RepoType.DATASET, revision=revision) def delete_model( self, model_id: str, endpoint: str | None = None, token: str | None = None, ) -> None: """Delete a model repository.""" api = self._api if token: api = HubApi(endpoint=endpoint or self._endpoint, token=token) api.delete_repo(model_id, RepoType.MODEL) def delete_dataset( self, dataset_id: str, endpoint: str | None = None, token: str | None = None, ) -> None: """Delete a dataset repository.""" warnings.warn( "This function is deprecated due to security reasons, " "and will be recovered in future versions with proper token authentication.", DeprecationWarning, stacklevel=2, ) api = self._api if token: api = HubApi(endpoint=endpoint or self._endpoint, token=token) api.delete_repo(dataset_id, RepoType.DATASET) def get_dataset_files( self, repo_id: str, *, revision: str = DEFAULT_DATASET_REVISION, root_path: str = "/", recursive: bool = True, page_number: int = 1, page_size: int = 100, endpoint: str | None = None, token: str | None = None, dataset_hub_id: str | None = None, ) -> list: """Get dataset file tree via the datahub API.""" api = self._api if token: api = HubApi(endpoint=endpoint or self._endpoint, token=token) if dataset_hub_id is None: if "/" in repo_id: _owner, _name = repo_id.split("/", 1) else: raise ValueError(f"Invalid repo_id: {repo_id}") dataset_hub_id, _ = self.get_dataset_id_and_type( dataset_name=_name, namespace=_owner, endpoint=endpoint, token=token ) params: dict[str, Any] = { "Revision": revision, "Root": root_path, "Recursive": "True" if recursive else "False", "PageNumber": page_number, "PageSize": page_size, } resp = api.legacy._request("GET", f"datasets/{dataset_hub_id}/repo/tree", params=params) data = api.legacy._json_data(resp) if isinstance(data, dict): return data.get("Files") or [] return data if isinstance(data, list) else [] # ------------------------------------------------------------------ # Datahub-specific methods (raw HTTP via legacy client) # ------------------------------------------------------------------ _dataset_id_type_cache: dict = {} def get_dataset_id_and_type( self, dataset_name: str, namespace: str, endpoint: str | None = None, token: str | None = None, ) -> tuple: """Get the dataset hub-internal id and type.""" api = self._api if token: api = HubApi(endpoint=endpoint or self._endpoint, token=token) cache_key = (namespace, dataset_name, endpoint or self._endpoint) cached = LegacyHubApi._dataset_id_type_cache.get(cache_key) if cached is not None: return cached data = api.legacy.get_repo_info(f"{namespace}/{dataset_name}", RepoType.DATASET) dataset_id = data["Id"] dataset_type = data["Type"] LegacyHubApi._dataset_id_type_cache[cache_key] = (dataset_id, dataset_type) return dataset_id, dataset_type def get_dataset_meta_file_list( self, dataset_name: str, namespace: str, dataset_id: str, revision: str, endpoint: str | None = None, token: str | None = None, ) -> list: """Get the meta file-list of the dataset.""" api = self._api if token: api = HubApi(endpoint=endpoint or self._endpoint, token=token) params = {"Revision": revision} resp = api.legacy._request("GET", f"datasets/{dataset_id}/repo/tree", params=params) data = api.legacy._json_data(resp) if data is None: raise NotExistError( f"The modelscope dataset [dataset_name = {dataset_name}, " f"namespace = {namespace}, version = {revision}] does not exist" ) file_list = data.get("Files") if isinstance(data, dict) else data if file_list is None: raise NotExistError( f"The modelscope dataset [dataset_name = {dataset_name}, " f"namespace = {namespace}, version = {revision}] does not exist" ) return file_list @staticmethod def dump_datatype_file(dataset_type: int, meta_cache_dir: str) -> None: """Dump dataset type marker file for offline formation detection.""" from modelscope.utils.constant import DatasetFormations ext = DatasetFormations.formation_mark_ext.value dataset_type_file_path = os.path.join(meta_cache_dir, f"{str(dataset_type)}{ext}") with open(dataset_type_file_path, "w") as fp: fp.write("*** Automatically-generated file, do not modify ***") def get_dataset_meta_files_local_paths( self, dataset_name: str, namespace: str, revision: str, meta_cache_dir: str, dataset_type: int, file_list: list, endpoint: str | None = None, token: str | None = None, ) -> tuple: """Download meta files and return local paths grouped by extension.""" from modelscope.utils.constant import DatasetFormations, DatasetMetaFormats api = self._api if token: api = HubApi(endpoint=endpoint or self._endpoint, token=token) local_paths: dict[str, list] = defaultdict(list) dataset_formation = DatasetFormations(dataset_type) dataset_meta_format = DatasetMetaFormats[dataset_formation] self.dump_datatype_file(dataset_type=dataset_type, meta_cache_dir=meta_cache_dir) for file_info in file_list: file_path = file_info["Path"] extension = os.path.splitext(file_path)[-1] if extension not in dataset_meta_format: continue resp = api.legacy._request( "GET", f"datasets/{namespace}/{dataset_name}/repo", params={"Revision": revision, "FilePath": file_path}, ) local_path = os.path.join(meta_cache_dir, file_path) os.makedirs(os.path.dirname(local_path), exist_ok=True) if os.path.exists(local_path): local_paths[extension].append(local_path) continue with open(local_path, "wb") as f: f.write(resp.content) local_paths[extension].append(local_path) return local_paths, dataset_formation def get_file_base_path( self, repo_id: str, endpoint: str | None = None, ) -> str: """Return the base URL prefix for dataset file downloads.""" if "/" not in repo_id: raise ValueError(f"Invalid repo_id format, expected 'namespace/name': {repo_id!r}") namespace, dataset_name = repo_id.split("/", 1) ep = endpoint or self._endpoint or self._api._config.endpoint return f"{ep}/api/v1/datasets/{namespace}/{dataset_name}/repo?" def get_dataset_file_url( self, file_name: str, dataset_name: str, namespace: str, revision: str | None = DEFAULT_DATASET_REVISION, view: bool = False, extension_filter: bool = True, endpoint: str | None = None, ) -> str: """Construct the download URL for a dataset file.""" if not file_name or not dataset_name or not namespace: raise ValueError("Args (file_name, dataset_name, namespace) cannot be empty!") ep = endpoint or self._endpoint or self._api._config.endpoint params = urlencode( { "Source": "SDK", "Revision": revision, "FilePath": file_name, "View": view, } ) return f"{ep}/api/v1/datasets/{namespace}/{dataset_name}/repo?{params}" def get_dataset_file_url_origin( self, file_name: str, dataset_name: str, namespace: str, revision: str | None = DEFAULT_DATASET_REVISION, endpoint: str | None = None, ) -> str: """Get dataset file URL, resolving meta files to API URLs.""" ep = endpoint or self._endpoint or self._api._config.endpoint if file_name and os.path.splitext(file_name)[-1] in META_FILES_FORMAT: file_name = f"{ep}/api/v1/datasets/{namespace}/{dataset_name}/repo?Revision={revision}&FilePath={file_name}" return file_name def get_dataset_access_config( self, dataset_name: str, namespace: str, revision: str | None = DEFAULT_DATASET_REVISION, endpoint: str | None = None, token: str | None = None, ) -> dict: """Get STS token config for dataset OSS access.""" api = self._api if token: api = HubApi(endpoint=endpoint or self._endpoint, token=token) resp = api.legacy._request( "GET", f"datasets/{namespace}/{dataset_name}/ststoken", params={"Revision": revision}, ) return api.legacy._json_data(resp) def get_dataset_access_config_session( self, dataset_name: str, namespace: str, check_cookie: bool = False, revision: str | None = DEFAULT_DATASET_REVISION, endpoint: str | None = None, token: str | None = None, ) -> dict: """Get STS token config with session-based auth.""" api = self._api if token: api = HubApi(endpoint=endpoint or self._endpoint, token=token) resp = api.legacy._request( "GET", f"datasets/{namespace}/{dataset_name}/ststoken", params={"Revision": revision}, ) return api.legacy._json_data(resp) def get_dataset_access_config_for_unzipped( self, dataset_name: str, namespace: str, revision: str, zip_file_name: str, endpoint: str | None = None, token: str | None = None, ) -> dict: """Get STS config for unzipped dataset files.""" api = self._api if token: api = HubApi(endpoint=endpoint or self._endpoint, token=token) # Get visibility data = api.legacy.get_repo_info(f"{namespace}/{dataset_name}", RepoType.DATASET) visibility_map = {1: "private", 3: "internal", 5: "public"} visibility = visibility_map.get(data.get("Visibility", 5), "public") # Get STS token resp = api.legacy._request( "GET", f"datasets/{namespace}/{dataset_name}/ststoken", params={"Revision": revision}, ) data_sts = api.legacy._json_data(resp) file_dir = f"{visibility}-unzipped/{namespace}_{dataset_name}_{zip_file_name}" data_sts["Dir"] = file_dir return data_sts def dataset_download_statistics( self, dataset_name: str, namespace: str, use_streaming: bool = False, endpoint: str | None = None, token: str | None = None, ) -> None: """Report dataset download for statistics.""" is_ci_test = os.getenv("CI_TEST") == "True" if not dataset_name or not namespace or is_ci_test or use_streaming: return try: api = self._api if token: api = HubApi(endpoint=endpoint or self._endpoint, token=token) api.legacy._request( "POST", f"datasets/{namespace}/{dataset_name}/download/increase", ) except Exception: pass _LEGACY_KEY_MAP: dict[str, str] = { "id": "Id", "owner": "Owner", "name": "Name", "repo_type": "RepoType", "visibility": "Visibility", "description": "Description", "downloads": "Downloads", "likes": "Likes", "created_at": "CreatedAt", "updated_at": "UpdatedAt", # backward compat if manually constructed "last_modified": "UpdatedAt", "license": "License", "tags": "Tags", } def _repo_info_to_dict(info: Any) -> dict: """Convert a RepoInfo to a plain dict with legacy PascalCase keys.""" if hasattr(info, "__dataclass_fields__"): from dataclasses import asdict raw = asdict(info) elif hasattr(info, "__dict__"): raw = {k: v for k, v in info.__dict__.items() if not k.startswith("_")} else: return {} return {_LEGACY_KEY_MAP.get(k, k): v for k, v in raw.items()} def _is_auth_related(exc: Exception) -> bool: """Return True if the exception is authentication/login related.""" if isinstance(exc, AuthenticationError): return True msg = str(exc).lower() return any(kw in msg for kw in ("login", "logged", "token", "unauthorized", "未登录"))