Python进度条用于Git克隆

3

我正在使用GitPython在程序中克隆一个repo。我已经找到了如何使用clone_from命令显示克隆的状态,但我希望状态看起来更像tqdm进度条。我尝试使用requests库获取文件的大小,但我仍然不确定如何实现它。我尝试像下面这样做,但它没有效果。任何帮助都将不胜感激,谢谢。

url = 'git@github.com:somegithubrepo/repo.git'
r = requests.get(url, stream=True)
total_length = r.headers.get('content-length')

for i in tqdm(range(len(total_length??))):
    git.Git(pathName).clone(url)
4个回答

6

这是另一个答案的改进版。当 CloneProgress 类被初始化时,进度条只会被创建一次。在更新时,它会将进度条设置为正确的数量。

import git
from git import RemoteProgress
from tqdm import tqdm

class CloneProgress(RemoteProgress):
    def __init__(self):
        super().__init__()
        self.pbar = tqdm()

    def update(self, op_code, cur_count, max_count=None, message=''):
        self.pbar.total = max_count
        self.pbar.n = cur_count
        self.pbar.refresh()

git.Repo.clone_from(project_url, repo_dir, branch='master', progress=CloneProgress()

2

本答案提供了一个基于rich进度条的工作示例,每个克隆过程的步骤都有一个进度条和一个新任务被分派。这个答案与我的另一个答案非常相似,但我认为将它们放在两个不同的帖子中可以提高可读性。主要区别在于,由于rich允许在一个进度条上下文中有多个任务,因此只需要进入和退出一次上下文-而不是每个阶段都需要。

from __future__ import annotations

import git
from rich import console, progress


class GitRemoteProgress(git.RemoteProgress):
    OP_CODES = [
        "BEGIN",
        "CHECKING_OUT",
        "COMPRESSING",
        "COUNTING",
        "END",
        "FINDING_SOURCES",
        "RECEIVING",
        "RESOLVING",
        "WRITING",
    ]
    OP_CODE_MAP = {
        getattr(git.RemoteProgress, _op_code): _op_code for _op_code in OP_CODES
    }

    def __init__(self) -> None:
        super().__init__()
        self.progressbar = progress.Progress(
            progress.SpinnerColumn(),
            # *progress.Progress.get_default_columns(),
            progress.TextColumn("[progress.description]{task.description}"),
            progress.BarColumn(),
            progress.TextColumn("[progress.percentage]{task.percentage:>3.0f}%"),
            "eta",
            progress.TimeRemainingColumn(),
            progress.TextColumn("{task.fields[message]}"),
            console=console.Console(),
            transient=False,
        )
        self.progressbar.start()
        self.active_task = None

    def __del__(self) -> None:
        # logger.info("Destroying bar...")
        self.progressbar.stop()

    @classmethod
    def get_curr_op(cls, op_code: int) -> str:
        """Get OP name from OP code."""
        # Remove BEGIN- and END-flag and get op name
        op_code_masked = op_code & cls.OP_MASK
        return cls.OP_CODE_MAP.get(op_code_masked, "?").title()

    def update(
        self,
        op_code: int,
        cur_count: str | float,
        max_count: str | float | None = None,
        message: str | None = "",
    ) -> None:
        # Start new bar on each BEGIN-flag
        if op_code & self.BEGIN:
            self.curr_op = self.get_curr_op(op_code)
            # logger.info("Next: %s", self.curr_op)
            self.active_task = self.progressbar.add_task(
                description=self.curr_op,
                total=max_count,
                message=message,
            )

        self.progressbar.update(
            task_id=self.active_task,
            completed=cur_count,
            message=message,
        )

        # End progress monitoring on each END-flag
        if op_code & self.END:
            # logger.info("Done: %s", self.curr_op)
            self.progressbar.update(
                task_id=self.active_task,
                message=f"[bright_black]{message}",
            )

使用它 - 完全克隆:

project_url = "https://github.com/u-boot/u-boot"

print("Cloning Git Repository 'u-boot' ('master' branch)...")
git.Repo.clone_from(
    url=project_url, 
    to_path="u-boot",
    progress=GitRemoteProgress(),
)
print("Done.")

Progress bar in action - full clone

使用浅克隆:

project_url = "https://github.com/u-boot/u-boot"

print("Cloning Git Repository 'u-boot' ('master' branch)...")
git.Repo.clone_from(
    url=project_url, 
    to_path="u-boot",
    depth=1,
    progress=GitRemoteProgress(),
)
print("Done.")

Progress bar in action - shallow clone


PS: 由于其较大的大小和可跟踪的克隆进度,使用u-boot仓库


这个例子已经不能运行了。TypeError: 不支持的操作数类型:'float' 和 'str' - Andrea Ricchi
@AndreaRicchi 你用的是哪个操作系统和哪个Python版本? - lcnittl
@AndreaRicchi 错误发生在哪一行? - lcnittl

1
您可以尝试类似以下的内容:
    import git
    from git import RemoteProgress
    from tqdm import tqdm
    
    
    class CloneProgress(RemoteProgress):
        def update(self, op_code, cur_count, max_count=None, message=''):
            pbar = tqdm(total=max_count)
            pbar.update(cur_count)
    
    git.Repo.clone_from(project_url, repo_dir, branch='master', progress=CloneProgress()

0

我知道这个问题似乎主要关注使用tqdm,但是,如果想使用alive-progress进度条,这里有一个正常工作的示例,它会为克隆过程的每个步骤分派一个进度条:

from __future__ import annotations

import git
from alive_progress import alive_bar


class GitRemoteProgress(git.RemoteProgress):
    OP_CODES = [
        "BEGIN",
        "CHECKING_OUT",
        "COMPRESSING",
        "COUNTING",
        "END",
        "FINDING_SOURCES",
        "RECEIVING",
        "RESOLVING",
        "WRITING",
    ]
    OP_CODE_MAP = {
        getattr(git.RemoteProgress, _op_code): _op_code for _op_code in OP_CODES
    }

    def __init__(self) -> None:
        super().__init__()
        self.alive_bar_instance = None

    @classmethod
    def get_curr_op(cls, op_code: int) -> str:
        """Get OP name from OP code."""
        # Remove BEGIN- and END-flag and get op name
        op_code_masked = op_code & cls.OP_MASK
        return cls.OP_CODE_MAP.get(op_code_masked, "?").title()

    def update(
        self,
        op_code: int,
        cur_count: str | float,
        max_count: str | float | None = None,
        message: str | None = "",
    ) -> None:
        cur_count = float(cur_count)
        max_count = float(max_count)

        # Start new bar on each BEGIN-flag
        if op_code & self.BEGIN:
            self.curr_op = self.get_curr_op(op_code)
            self._dispatch_bar(title=self.curr_op)

        self.bar(cur_count / max_count)
        self.bar.text(message)

        # End progress monitoring on each END-flag
        if op_code & git.RemoteProgress.END:
            self._destroy_bar()

    def _dispatch_bar(self, title: str | None = "") -> None:
        """Create a new progress bar"""
        self.alive_bar_instance = alive_bar(manual=True, title=title)
        self.bar = self.alive_bar_instance.__enter__()

    def _destroy_bar(self) -> None:
        """Destroy an existing progress bar"""
        self.alive_bar_instance.__exit__(None, None, None)

使用完整克隆:

project_url = "https://github.com/u-boot/u-boot"

print("Cloning Git Repository 'u-boot' ('master' branch)...")
git.Repo.clone_from(
    url=project_url, 
    to_path="u-boot",
    progress=GitRemoteProgress(),
)
print("Done.")

Progress bar in action - full clone

使用浅克隆:

project_url = "https://github.com/u-boot/u-boot"

print("Cloning Git Repository 'u-boot' ('master' branch)...")
git.Repo.clone_from(
    url=project_url, 
    to_path="u-boot",
    depth=1,
    progress=GitRemoteProgress(),
)
print("Done.")

Progress bar in action - shallow clone


附注:u-boot仓库由于其较大的大小,因此可以跟踪克隆进度
附言:录制使用richprint,因此有漂亮的颜色 :-)


这个例子已经不能工作了。TypeError: 不支持的操作数类型:'float' 和 'str' - Andrea Ricchi
@AndreaRicchi 你用的是哪个操作系统和哪个Python版本? - lcnittl
所以这是错误:File "gitremoteprogress.py", line 46, in update self.bar(cur_count / max_count) TypeError: 不支持的操作数类型:'float' 和 'str'在 Ubuntu 20.04 上使用 Python 3.8.10。 - Andrea Ricchi
有趣。对我来说,它在 Windows 11 上运行,并支持 Python 3.7–3.10,在 Ubuntu 20.04 上支持 Python 3.8。无论如何,你能试试更新版本吗? - lcnittl
在你的情况下,max_count 的值是多少会很有趣,以至于它不被视为浮点数。 - lcnittl

网页内容由stack overflow 提供, 点击上面的
可以查看英文原文,
原文链接