102 lines
2.7 KiB
Python
Executable File
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())
|