repo-cloner/repo_cloner/process_repository_dir.py
Václav Valíček 1fef7bc404
New behavior, new tests, added wizzard and processor
Signed-off-by: Václav Valíček <valicek1994@gmail.com>
2022-08-07 21:44:31 +02:00

102 lines
2.7 KiB
Python
Executable File

#!/usr/bin/env python3
from repo_cloner.lib.logger_setup import log
import argparse
import logging
import os
import subprocess
import base64
from repo_cloner.lib.repo_dir_structure import RepoDirStructure
from repo_cloner.lib import Cloner, DetectedCommit, prepare_git_auth, init_gh_token
def detector_executor(commit: DetectedCommit):
message = base64.b64encode(commit.log.encode()).decode()
env = {
"sha": commit.commit,
"abbrev": commit.abbrev,
"author": commit.author,
"has_tags": commit.is_tag,
"has_branches": commit.is_branch,
"tags": commit.tags,
"branches": commit.branches,
"date": commit.date,
"log": message,
}
arg_list = ["/bin/echo", "laminarc", "queue", commit.project]
for key, val in env.items():
arg_list.append(f"COMMIT_{key.upper()}={val}")
subprocess.run(arg_list)
def clone_or_fetch(base_dir: str, clone_init: bool = False, detector_init: bool = False):
log.info(f"Started processing git group in folder: {base_dir}")
dirs = RepoDirStructure(base_dir)
log.debug(f"Patching XDG_CONFIG_HOME to mock up git config")
os.environ['XDG_CONFIG_HOME'] = dirs.conf_dir
init_gh_token()
# check dir existence
try:
assert dirs.dirs_exist
except Exception as e:
log.critical(f"Dir structure problem: {e.__str__()}")
return 1
# does config exist?
try:
assert dirs.has_config
except Exception as e:
log.critical(f"Config file not found!?")
log.critical(e.__str__())
return 1
# check if there is project name & git url supplied
config = dirs.config
if len(config.cloner_repo_url) == 0:
log.critical("Config directive cloner_repo_url is missing/empty! Cannot continue!")
return 1
if len(config.cloner_project_name) == 0:
log.warning("Config directive cloner_project_name should not be omitted!")
cloner = Cloner(dirs)
prepare_git_auth(dirs.conf_dir)
if clone_init:
log.info(f"Initial cloning of repositories")
if not cloner.clone():
return 1
if detector_init:
cloner.detector_init()
return 0
# regular run
if not cloner.sync():
log.warning(f"Repo sync did not succeed")
if cloner.detector_enabled:
cloner.detector_run(detector_executor)
return 0
def main() -> int:
# parse input arguments
parser = argparse.ArgumentParser(description = "repo-cloner entering script")
parser.add_argument('--base-dir', help = 'path to directory containing whole cloner structure', required = True,
default = None, type = str)
parser.add_argument('--debug', '-d', help = "enable debug output", action = 'store_true')
args = parser.parse_args()
if args.debug:
log.setLevel(logging.DEBUG)
return clone_or_fetch(args.base_dir)
if __name__ == "__main__":
exit(main())