git-batch/gitbatch/Cli.py

133 lines
4.9 KiB
Python
Raw Normal View History

2019-12-02 08:53:35 +00:00
#!/usr/bin/env python3
"""Main program."""
import argparse
import os
from collections import defaultdict
from urllib.parse import urlparse
import git
from gitbatch import __version__
from gitbatch.Logging import SingleLog
2019-12-02 08:57:38 +00:00
from gitbatch.Utils import normalize_path
from gitbatch.Utils import to_bool
2019-12-02 08:53:35 +00:00
class GitBatch:
2020-04-11 11:17:01 +00:00
"""Handles git operations."""
2019-12-02 08:53:35 +00:00
def __init__(self):
self.log = SingleLog()
self.logger = self.log.logger
self.args = self._cli_args()
self.config = self._config()
self.run()
def _cli_args(self):
parser = argparse.ArgumentParser(
2020-04-11 11:17:01 +00:00
description=("Clone single branch from all repositories listed in a file")
)
parser.add_argument(
"--version", action="version", version="%(prog)s {}".format(__version__)
)
2020-10-02 09:41:32 +00:00
parser.add_argument(
"-v", dest="logging.level", action="append_const", const=-1, help="increase log level"
)
parser.add_argument(
"-q", dest="logging.level", action="append_const", const=1, help="decrease log level"
)
2019-12-02 08:53:35 +00:00
return parser.parse_args()
def _config(self):
config = defaultdict(dict)
2020-10-02 09:41:32 +00:00
# Override correct log level from argparse
levels = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
log_level = levels.index("ERROR")
tmp_dict = self.args.__dict__
if tmp_dict.get("logging.level"):
for adjustment in tmp_dict["logging.level"]:
log_level = min(len(levels) - 1, max(log_level + adjustment, 0))
config["logging"]["level"] = levels[log_level]
2019-12-02 09:01:21 +00:00
input_file_raw = os.environ.get("GIT_BATCH_INPUT_FILE", ".batchfile")
2019-12-02 08:53:35 +00:00
config["input_file"] = normalize_path(input_file_raw)
config["ignore_existing"] = to_bool(os.environ.get("GIT_BATCH_IGNORE_EXISTING_REPO", True))
config["ignore_missing"] = to_bool(os.environ.get("GIT_BATCH_IGNORE_MISSING_REMOTE", True))
return config
def _repos_from_file(self, src):
repos = []
with open(src, "r") as f:
for num, line in enumerate(f, start=1):
repo = {}
line = line.strip()
if line and not line.startswith("#"):
try:
url, branch, dest = [x.strip() for x in line.split(";")]
except ValueError as e:
2020-04-11 11:17:01 +00:00
self.log.sysexit_with_message(
"Wrong numer of delimiters in line {line_num}: {exp}".format(
line_num=num, exp=e
)
)
2019-12-02 08:53:35 +00:00
if url:
url_parts = urlparse(url)
repo["url"] = url
repo["branch"] = branch or "master"
repo["name"] = os.path.basename(url_parts.path)
2020-10-02 09:41:32 +00:00
repo["rel_dest"] = dest
2020-04-11 11:17:01 +00:00
repo["dest"] = normalize_path(dest) or normalize_path(
"./{}".format(repo["name"])
)
2019-12-02 08:53:35 +00:00
repos.append(repo)
else:
2020-04-11 11:17:01 +00:00
self.log.sysexit_with_message(
"Repository Url is not set on line {line_num}".format(line_num=num)
)
2019-12-02 08:53:35 +00:00
return repos
def _repos_clone(self, repos, ignore_existing):
for repo in repos:
try:
2020-04-11 11:17:01 +00:00
options = ["--branch={}".format(repo["branch"]), "--single-branch"]
2019-12-02 08:53:35 +00:00
git.Repo.clone_from(repo["url"], repo["dest"], multi_options=options)
except git.exc.GitCommandError as e:
passed = False
err_raw = e.stderr.strip().splitlines()[:-1]
2020-10-02 09:41:32 +00:00
err = [
x.split(":", 1)[1].strip().replace(repo["dest"], repo["rel_dest"])
for x in err_raw
]
2019-12-02 08:53:35 +00:00
if any(["already exists and is not an empty directory" in item for item in err]):
if self.config["ignore_existing"]:
2020-10-02 09:41:32 +00:00
self.logger.warn("Git error: {}".format("\n".join(err)))
2019-12-02 08:53:35 +00:00
passed = True
if any(["Could not find remote branch" in item for item in err]):
if self.config["ignore_missing"]:
passed = True
if not passed:
2020-10-02 09:41:32 +00:00
self.log.sysexit_with_message("Git error: {}".format("\n".join(err)))
2019-12-02 08:53:35 +00:00
def run(self):
2020-10-02 09:41:32 +00:00
self.log.set_level(self.config["logging"]["level"])
2019-12-02 08:53:35 +00:00
if os.path.isfile(self.config["input_file"]):
repos = self._repos_from_file(self.config["input_file"])
self._repos_clone(repos, self.config["ignore_existing"])
else:
2020-04-11 11:17:01 +00:00
self.log.sysexit_with_message(
"The given batch file at '{}' does not exist".format(
os.path.relpath(os.path.join("./", self.config["input_file"]))
)
)