forked from infra/otc-metadata
286 lines
9.3 KiB
Python
286 lines
9.3 KiB
Python
#!/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")
|
|
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.otc-service.com",
|
|
)
|
|
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_tox:
|
|
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(
|
|
(
|
|
"Update tox.ini && conf.py file\n\n"
|
|
"Performed-by: gitea/infra/otc-metadata/"
|
|
"tools/generate_doc_confpy.py"
|
|
)
|
|
)
|
|
push_args = ["--set-upstream", "origin", branch_name]
|
|
if args.branch_force:
|
|
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")
|
|
|
|
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
|
|
|
|
if args.token:
|
|
api_session.headers.update({"Authorization": f"token {args.token}"})
|
|
|
|
for service in services:
|
|
process_repositories(args, service)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|