#!/usr/bin/python # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or # implied. # See the License for the specific language governing permissions and # limitations under the License. import argparse import logging import os import pathlib import requests import subprocess import warnings from git import exc from git import Repo from git import SymbolicReference from jinja2 import PackageLoader from jinja2 import Environment from jinja2 import select_autoescape import otc_metadata.services data = otc_metadata.services.Services() api_session = requests.Session() def process_repositories(args, service): """Checkout repositories """ logging.debug(f"Processing service {service}") workdir = pathlib.Path(args.work_dir) workdir.mkdir(exist_ok=True) copy_to = None repo_to = None url_to = None target_repo = None git_fqdn = None env = Environment( loader=PackageLoader("otc_metadata"), autoescape=select_autoescape()) conf_py_template = env.get_template("conf.py.j2") for repo in service["repositories"]: logging.debug(f"Processing repository {repo}") repo_dir = pathlib.Path( workdir, repo["type"], repo["repo"] ) if repo["environment"] == args.target_environment: copy_to = repo_dir else: logging.debug(f"Skipping repository {repo}") continue repo_dir.mkdir(parents=True, exist_ok=True) if repo["type"] == 'gitea': repo_url = ( f"ssh://git@gitea.eco.tsi-dev.otc-service.com:2222/" f"{repo['repo']}" ) git_fqdn = "gitea.eco.tsi-dev.otc-service.com" elif repo["type"] == 'github': repo_url = f"git@github.com:/{repo['repo']}" else: logging.error(f"Repository type {repo['type']} is not supported") exit(1) if repo_dir.exists(): logging.debug(f"Repository {repo} already checked out") try: git_repo = Repo(repo_dir) git_repo.remotes.origin.fetch() git_repo.heads.main.checkout() git_repo.remotes.origin.pull() except exc.InvalidGitRepositoryError: logging.error("Existing repository checkout is bad") repo_dir.rmdir() if not repo_dir.exists(): try: git_repo = Repo.clone_from( repo_url, repo_dir, branch='main') except Exception: logging.error(f"Error cloning repository {repo_url}") return if repo["environment"] == args.target_environment: url_to = repo_url repo_to = git_repo target_repo = repo break if not target_repo: logging.info( f"No repository service {service['service_title']}" f"for environment {args.target_environment}") return branch_name = f"{args.branch_name}" if args.branch_force: logging.debug("Dropping current branch") try: repo_to.delete_head(branch_name) except exc.GitCommandError: pass try: new_branch = repo_to.create_head(branch_name, 'main') except Exception: logging.warning(f"Skipping service {service}") return new_branch.checkout() for doc in data.docs_by_service_type(service["service_type"]): logging.debug(f"Analyzing document {doc}") conf_py_path = pathlib.Path(copy_to, doc['rst_location'], 'conf.py') if not conf_py_path.exists(): logging.info(f"Path for document {doc['title']} does not exist") conf_py_path.parent.mkdir(parents=True, exist_ok=True) context = dict( repo_name=target_repo["repo"], project=service["service_title"], pdf_name=doc["pdf_name"], title=f"{service['service_title']} - {doc['title']}" ) if git_fqdn: context["git_fqdn"] = git_fqdn if target_repo.get("type") != "github": context["git_type"] = target_repo["type"] conf_py_content = conf_py_template.render(**context) with open(conf_py_path, 'w', encoding='utf-8', newline='') as out: logging.debug(f"Generating {conf_py_path} from template...") out.write(conf_py_content) repo_to.index.add([doc["rst_location"]]) repo_to.index.commit( ( f"Update conf.py file\n\n" "Performed-by: gitea/infra/otc-metadata/tools/generate_doc_confpy.py" ) ) push_args = ("--set-upstream", "origin", branch_name) repo_to.git.push(*push_args) if 'github' in url_to: subprocess.run( args=['gh', 'pr', 'create', '-f'], cwd=copy_to, check=True ) elif 'gitea' in url_to and args.token: open_pr( args, repo["repo"], dict( head=branch_name, ), ) def open_pr(args, repository, pr_data): req = dict( base=pr_data.get("base", "main"), head=pr_data["head"], ) if "title" in pr_data: req["title"] = pr_data["title"] if "body" in pr_data: req["body"] = pr_data["body"].replace("\\n", "\n") if "assignees" in pr_data: req["assignees"] = pr_data["assignees"] if "labels" in pr_data: req["labels"] = pr_data["labels"] rsp = api_session.post( f"{args.api_url}/repos/{repository}/pulls", json=req ) if rsp.status_code != 201: print(rsp.text) print(f"Going to open PR with title {pr_data['title']} in {repository}") def main(): parser = argparse.ArgumentParser( description='Update conf.py file in repositories.') parser.add_argument( '--target-environment', required=True, help='Environment to be used as a source' ) parser.add_argument( '--service-type', help='Service to update' ) parser.add_argument( '--work-dir', required=True, help='Working directory to use for repository checkout.' ) parser.add_argument( '--branch-name', default='confpy', help='Branch name to be used for synchronizing.' ) parser.add_argument( '--branch-force', action='store_true', help='Whether to force branch recreation.' ) parser.add_argument("--token", metavar="token", help="API token") parser.add_argument("--api-url", help="API base url of the Git hoster") args = parser.parse_args() logging.basicConfig(level=logging.DEBUG) services = [] if args.service_type: services = [data.service_dict.get(args.service_type)] else: services = data.all_services for service in services: process_repositories(args, service) if __name__ == '__main__': main()