otc-metadata/tools/generate_doc_confpy.py

#!/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 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()


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.warn(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
        )


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


    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()