2

我使用 GitPython 在我的程序中克隆一个 repo。我想出了如何使用 clone_from 命令显示克隆的状态，但我希望状态看起来更像一个 tqdm 进度条。我尝试使用请求库来获取文件的大小，但我仍然不确定如何实现它。尝试在下面做这样的事情，但它不起作用。任何帮助表示赞赏，谢谢。

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

4 回答 4

2

这是另一个答案的改进版本。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()
于 2021-01-05T09:24:14.017 回答
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()
于 2020-06-28T13:56:26.113 回答
0

我知道这个问题似乎主要集中在 using 上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:
        # 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.")

进度条在行动 - 完整克隆

使用它 - 浅克隆：

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.")

进度条在行动 - 浅克隆

PS：u-boot使用 repo 是因为它的大小很大，因此克隆进度可遵循
PPS：录音使用rich's print，因此颜色很漂亮:-)

于 2022-02-10T13:13:34.690 回答
0

这个答案提供了一个基于rich's progress bar的工作示例，其中有一个进度条和一个为克隆过程的每个步骤分派的新任务。这个答案与我的另一个答案非常相似，但我认为将它们放在 2 个单独的帖子中有助于提高可读性。主要区别在于 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_instace = 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 = self.progressbar_instace.__enter__()
        self.active_task = None

    def __del__(self) -> None:
        # logger.info("Destroying bar...")
        self.progressbar_instace.__exit__(None, None, 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:
        # 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.")

进度条在行动 - 完整克隆

使用它 - 浅克隆：

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.")

进度条在行动 - 浅克隆

PS：u-boot使用repo，因为它的大小很大，因此可以跟踪克隆进度

于 2022-02-27T14:36:07.723 回答