otc-metadata-swiss/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 pathlib
import requests
import subprocess

from git import exc
from git import Repo

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")
    tox_ini_template = env.get_template("tox.ini.j2")
    index_sbv_template = env.get_template("index_sbv.rst.j2")
    doc_requirements_template = env.get_template("doc_requirements.txt.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, force=True)
            repo_to.delete_head(f"refs/heads/{args.branch_name}", force=True)
        except exc.GitCommandError as e:
            print(e)
            pass
    try:
        new_branch = repo_to.create_head(branch_name, "main")
    except Exception as e:
        logging.warning(f"Skipping service {service} due to {e}")
        return
    new_branch.checkout()

    service_docs = list(data.docs_by_service_type(service["service_type"]))

    for doc in service_docs:
        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 "pdf_name" in doc:
            context["pdf_name"] = doc["pdf_name"]
        if git_fqdn:
            context["git_fqdn"] = git_fqdn
        if target_repo.get("type") != "github":
            context["git_type"] = target_repo["type"]
        if args.target_environment == "internal":
            context["html_options"] = dict(
                disable_search=True,
                site_name="Internal Documentation Portal",
                logo_url="https://docs-int.sc.otc.t-systems.com",
            )
        else:
            context["html_options"] = dict(
                site_name="Swiss Open Telekom Cloud Docs",
                logo_url="https://docs-beta.sc.otc.t-systems.com",
            )
        context["doc_environment"] = args.target_environment
        if doc['link']:
            context["doc_link"] = doc['link']
        else:
            context["doc_link"] = (
                '/'
                + service['service_uri']
                + '/'
                + doc['type']
                + '/'
            )
        context["doc_title"] = doc['title']
        context["doc_type"] = doc['type']
        context["service_category"] = service['service_category']
        context["service_title"] = service['service_title']
        context["service_type"] = service['service_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"]])

    if args.update_sbv:
        """Add or update service-based-view"""
        copy_path = pathlib.Path(copy_to, 'doc', 'source')
        context = dict(
            repo_name=target_repo["repo"],
            project=service["service_title"],
            # pdf_name=doc["pdf_name"],
            title=f"{service['service_title']} - Service Based View",
            service_type=service["service_type"]
        )
        context["service_category"] = service['service_category']
        context["service_title"] = service['service_title']
        if not copy_path.exists():
            logging.info("Path for sbv does not exist")
            copy_path.mkdir(parents=True, exist_ok=True)
        context["otc_sbv"] = True
        if git_fqdn:
            context["git_fqdn"] = git_fqdn
        if target_repo.get("type") != "github":
            context["git_type"] = target_repo["type"]
        if args.target_environment == "internal":
            context["html_options"] = dict(
                disable_search=True,
                site_name="Internal Documentation Portal",
                logo_url="https://docs-int.sc.otc.t-systems.com",
            )
        else:
            context["html_options"] = dict(
                site_name="Open Telekom Cloud Swiss Documentation Portal",
                logo_url="https://docs-beta.sc.otc.t-systems.com",
            )
        sbv_title = (service["service_title"] + "\n"
                     + ('=' * len(service["service_title"])))
        context["sbv_title"] = sbv_title
        conf_py_content = conf_py_template.render(**context)
        index_sbv_content = index_sbv_template.render(**context)
        with open(
                pathlib.Path(copy_path, "conf.py"),
                "w",
                encoding="utf-8") as out:
            out.write(conf_py_content)
        repo_to.index.add(pathlib.Path(copy_path, "conf.py"))

        if (not args.overwrite_index_sbv
                and pathlib.Path(copy_path, "index.rst").exists()):
            logging.info("File index.rst for sbv exists. Skipping")
        else:
            with open(
                    pathlib.Path(copy_path, "index.rst"),
                    "w",
                    encoding="utf-8") as out:
                out.write(index_sbv_content)
            repo_to.index.add(pathlib.Path(copy_path, "index.rst"))

        placeholder_path = pathlib.Path(copy_path, "_static")
        if not pathlib.Path(placeholder_path, "placeholder").exists():
            placeholder_path.mkdir(parents=True, exist_ok=True)
            open(pathlib.Path(placeholder_path, "placeholder"), 'a').close()
            repo_to.index.add(pathlib.Path(placeholder_path, "placeholder"))

    if args.update_tox:
        """Update tox.ini"""
        context = dict(docs=[])
        for doc in service_docs:
            if doc["type"] == "dev":
                doc["type"] = "dev-guide"
            context["docs"].append(doc)

        tox_ini_content = tox_ini_template.render(**context)
        tox_ini_path = pathlib.Path(copy_to, "tox.ini")
        doc_requirements_content = doc_requirements_template.render(**context)
        doc_requirements_path = pathlib.Path(
            copy_to, "doc", "requirements.txt"
        )
        doc_requirements_path.parent.mkdir(parents=True, exist_ok=True)
        with open(tox_ini_path, "w", encoding="utf-8", newline="") as out:
            logging.debug(f"Generating {tox_ini_path} from template...")
            out.write(tox_ini_content)
        repo_to.index.add(["tox.ini"])
        with open(
            doc_requirements_path, "w", encoding="utf-8", newline=""
        ) as out:
            logging.debug(
                f"Generating {doc_requirements_path} from template..."
            )
            out.write(doc_requirements_content)
        repo_to.index.add(["doc/requirements.txt"])

    if len(repo_to.index.diff("HEAD")) == 0:
        # Nothing to commit
        logging.debug(
            "No changes for service %s required" % service["service_type"]
        )
        return
    repo_to.index.commit(
        args.commit_description
    )
    push_args = ["--set-upstream", "origin", branch_name]
    if args.force_push:
        push_args.append("--force")
    repo_to.git.push(*push_args)
    if "github" in url_to:
        subprocess.run(
            args=["gh", "pr", "create", "-f"], cwd=copy_to, check=False
        )
    elif "gitea" in url_to and args.token:
        open_pr(
            args,
            repo["repo"],
            dict(
                title="Update Docs configuration",
                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(
        "--update-tox", action="store_true", help="Whether to update tox.ini."
    )
    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")
    parser.add_argument(
        "--update-sbv",
        action="store_true",
        help="Whether to update service-based-view"
    )
    parser.add_argument(
        "--overwrite-index-sbv",
        action="store_true",
        help=("Whether to overwrite index.rst for service-based-view."
              + "\nCan only be used if --update-sbv is also specified")
    )
    parser.add_argument(
        "--force-push",
        action="store_true",
        help="Whether to force push the commit"
    )
    parser.add_argument(
        "--commit-description",
        default=(
            "Update tox.ini && conf.py file\n\n"
            "Performed-by: gitea/infra/otc-metadata-swiss/"
            "tools/generate_doc_confpy.py"
        ),
        help="Commit description for the commit",
    )

    args = parser.parse_args()
    logging.basicConfig(level=logging.DEBUG)
    services = []
    if args.overwrite_index_sbv and not args.update_sbv:
        logging.error(
            "Cannot overwrite index.rst for service-based-view"
            + " without updating service-based-view"
        )
        exit(1)
    if args.service_type:
        services = [data.service_dict.get(args.service_type)]
    else:
        services = data.all_services

    if args.token:
        api_session.headers.update({"Authorization": f"token {args.token}"})

    for service in services:
        process_repositories(args, service)


if __name__ == "__main__":
    main()