modelscope/modelscope/hub/git.py

# Copyright (c) Alibaba, Inc. and its affiliates.
"""Git wrapper — shim delegating to ``modelscope_hub._git``.

Preserves the legacy ``GitCommandWrapper`` Singleton interface used
throughout the SDK while routing primitive Git operations through
:class:`modelscope_hub._git.GitCommand`.
"""
from __future__ import annotations
import os
from pathlib import Path
from typing import List, Optional

from modelscope_hub._git import GitCommand as _GitCommand

from modelscope.hub.errors import GitError
from modelscope.utils.constant import MASTER_MODEL_BRANCH
from modelscope.utils.logger import get_logger

logger = get_logger()

__all__ = ['GitError', 'GitCommandWrapper', 'Singleton']


class Singleton(type):
    """Metaclass enforcing one instance per class — preserved for parity."""

    _instances: dict = {}

    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            cls._instances[cls] = super().__call__(*args, **kwargs)
        return cls._instances[cls]


class GitCommandWrapper(metaclass=Singleton):
    """Backward-compatible Git wrapper.

    Wraps :class:`modelscope_hub._git.GitCommand` to expose the legacy
    instance-method API (``clone``, ``push``, ``pull``, ``tag``…) used
    by callers that pre-date the SDK refactor.
    """

    default_git_path = 'git'

    def __init__(self, path: Optional[str] = None):
        self.git_path = path or self.default_git_path
        _GitCommand.set_git_path(self.git_path)

    # ------------------------------------------------------------------
    # Low-level subprocess passthrough (legacy contract)
    # ------------------------------------------------------------------
    def _run_git_command(self, *args):
        """Run a git subcommand, raising :class:`GitError` on failure."""
        try:
            return _GitCommand._run(*[a for a in args if a])
        except Exception as exc:  # _git.GitError → legacy GitError
            raise GitError(str(exc)) from exc

    # ------------------------------------------------------------------
    # URL / token helpers
    # ------------------------------------------------------------------
    def _add_git_token(self, git_token: str, url: str) -> str:
        return _GitCommand._inject_token(url, git_token)

    def remove_token_from_url(self, url: str) -> str:
        return _GitCommand.strip_token_from_url(url)

    # ------------------------------------------------------------------
    # LFS
    # ------------------------------------------------------------------
    def is_lfs_installed(self) -> bool:
        return _GitCommand.is_lfs_available()

    def git_lfs_install(self, repo_dir: str) -> bool:
        try:
            _GitCommand.lfs_install(Path(repo_dir))
            return True
        except Exception:
            return False

    def list_lfs_files(self, repo_dir: str) -> List[str]:
        rsp = self._run_git_command('-C', repo_dir, 'lfs', 'ls-files')
        return [
            line.split(' ')[-1] for line in rsp.stdout.strip().splitlines()
            if line
        ]

    # ------------------------------------------------------------------
    # Auth / user config
    # ------------------------------------------------------------------
    def config_git_token(self, repo_dir: str, git_token: str) -> None:
        url = self.get_repo_remote_url(repo_dir)
        if '//oauth2' in url:
            return
        auth_url = self._add_git_token(git_token, url)
        self._run_git_command('-C', repo_dir, 'remote', 'set-url', 'origin',
                              auth_url)

    def add_user_info(self, repo_base_dir: str, repo_name: str) -> None:
        from modelscope.hub.api import ModelScopeConfig
        user_name, user_email = ModelScopeConfig.get_user_info()
        if not (user_name and user_email):
            return
        repo_dir = os.path.join(repo_base_dir, repo_name)
        self._run_git_command('-C', repo_dir, 'config', 'user.name', user_name)
        self._run_git_command('-C', repo_dir, 'config', 'user.email',
                              user_email)

    # ------------------------------------------------------------------
    # Clone / pull / push
    # ------------------------------------------------------------------
    def clone(self,
              repo_base_dir: str,
              git_token: Optional[str],
              url: str,
              repo_name: str,
              branch: Optional[str] = None):
        target = Path(repo_base_dir) / repo_name
        try:
            _GitCommand.clone(
                url=url, target_dir=target, branch=branch, token=git_token)
        except Exception as exc:
            if (target / '.git').is_dir():
                logger.warning(
                    'git clone exited non-zero but repository was cloned '
                    'at %s. Likely a post-clone hook. Continuing.', target)
                return None
            raise GitError(str(exc)) from exc

    def pull(self,
             repo_dir: str,
             remote: str = 'origin',
             branch: str = 'master'):
        return self._run_git_command('-C', repo_dir, 'pull', remote, branch)

    def push(self,
             repo_dir: str,
             git_token: str,
             url: str,
             local_branch: str,
             remote_branch: str,
             force: bool = False):
        auth_url = self._add_git_token(git_token, url)
        args = [
            '-C', repo_dir, 'push', auth_url, f'{local_branch}:{remote_branch}'
        ]
        if force:
            args.append('-f')
        return self._run_git_command(*args)

    # ------------------------------------------------------------------
    # Add / commit / branch / checkout
    # ------------------------------------------------------------------
    def add(self,
            repo_dir: str,
            files: Optional[List[str]] = None,
            all_files: bool = False):
        if all_files:
            return self._run_git_command('-C', repo_dir, 'add', '-A')
        return self._run_git_command('-C', repo_dir, 'add', *(files or []))

    def commit(self, repo_dir: str, message: str):
        return self._run_git_command('-C', repo_dir, 'commit', '-m',
                                     f"'{message}'")

    def checkout(self, repo_dir: str, revision: str):
        return self._run_git_command('-C', repo_dir, 'checkout', revision)

    def new_branch(self, repo_dir: str, revision: str):
        return self._run_git_command('-C', repo_dir, 'checkout', '-b',
                                     revision)

    def get_remote_branches(self, repo_dir: str) -> List[str]:
        rsp = self._run_git_command('-C', repo_dir, 'branch', '-r')
        info = [
            line.strip() for line in rsp.stdout.strip().splitlines() if line
        ]
        if len(info) <= 1:
            return ['/'.join(info[0].split('/')[1:])] if info else []
        return ['/'.join(line.split('/')[1:]) for line in info[1:]]

    def get_repo_remote_url(self, repo_dir: str) -> str:
        rsp = self._run_git_command('-C', repo_dir, 'config', '--get',
                                    'remote.origin.url')
        return rsp.stdout.strip()

    # ------------------------------------------------------------------
    # Tags
    # ------------------------------------------------------------------
    def tag(self,
            repo_dir: str,
            tag_name: str,
            message: str,
            ref: str = MASTER_MODEL_BRANCH):
        return self._run_git_command('-C', repo_dir, 'tag', tag_name, '-m',
                                     f'"{message}"', ref)

    def push_tag(self, repo_dir: str, tag_name: str):
        return self._run_git_command('-C', repo_dir, 'push', 'origin',
                                     tag_name)