zulip/tools/merge-api-changelogs

#!/usr/bin/env python3
import glob
import os
import re
import subprocess
import sys
from pathlib import Path


def get_changelog_files_list() -> list[str]:
    dir_path = Path("api_docs/unmerged.d")
    if os.path.exists(dir_path):
        return [os.path.basename(path) for path in glob.glob(f"{dir_path}/ZF-??????")]

    return []


def get_unmerged_changelogs() -> str:
    changelogs = ""
    dir_path = Path("api_docs/unmerged.d")
    changelog_files_list = get_changelog_files_list()
    if changelog_files_list:
        print(f"Unmerged changelog files: {changelog_files_list}")
    else:
        print("No unmerged changelog files found.")

    for file_name in changelog_files_list:
        file_path = Path(f"{dir_path}/{file_name}")
        with open(file_path) as f:
            changelogs += f.read().strip("\n") + "\n"

    return changelogs


def increment_and_get_feature_level() -> int:
    new_feature_level = None
    version_file_path = Path("version.py")

    with open(version_file_path) as file:
        lines = file.readlines()

    new_feature_level = None

    with open(version_file_path, "w") as file:
        for line in lines:
            if line.startswith("API_FEATURE_LEVEL = "):
                match = re.search(r"\d+", line)
                if match:
                    new_feature_level = int(match.group()) + 1
                    file.write(f"API_FEATURE_LEVEL = {new_feature_level}\n")
                    continue

            file.write(line)

    assert new_feature_level is not None
    print(f"Updated API feature level: {new_feature_level - 1} -> {new_feature_level}")
    return new_feature_level


def merge_changelogs(changelogs: str, new_feature_level: int) -> None:
    changelogs_merged = False
    changelog_path = Path("api_docs/changelog.md")

    with open(changelog_path) as file:
        lines = file.readlines()

    changelogs_merged = False

    with open(changelog_path, "w") as file:
        for line in lines:
            file.write(line)
            if changelogs_merged:
                continue
            if re.fullmatch(r"## Changes in Zulip \d+\.\d+\n", line):
                changelogs_merged = True
                file.write(f"\n**Feature level {new_feature_level}**\n")
                file.write(f"\n{changelogs}")

    print(f"Changelogs merged to {changelog_path}.")


def update_feature_level_in_api_docs(new_feature_level: int) -> None:
    changelog_files_list = get_changelog_files_list()
    num_replaces = 0
    api_docs_path = Path("zerver/openapi/zulip.yaml")

    with open(api_docs_path) as file:
        lines = file.readlines()

    num_replaces = 0

    with open(api_docs_path, "w") as file:
        for line in lines:
            old_line = line
            for file_name in changelog_files_list:
                temporary_feature_level = file_name
                line = line.replace(temporary_feature_level, str(new_feature_level))

            if old_line != line:
                num_replaces += 1

            file.write(line)

    if num_replaces:
        print(f"Updated {api_docs_path}; {num_replaces} replaces were made.")
    else:
        print(f"No changes were made to {api_docs_path}; no matching file name found.")


def remove_unmerged_changelog_files() -> None:
    changelog_files_list = get_changelog_files_list()
    for file_name in changelog_files_list:
        os.remove(Path(f"api_docs/unmerged.d/{file_name}"))

    if changelog_files_list:
        print("Removed all the unmerged changelog files.")


if __name__ == "__main__":
    ZULIP_PATH = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    os.chdir(ZULIP_PATH)

    changelogs = get_unmerged_changelogs()
    if changelogs:
        new_feature_level = increment_and_get_feature_level()
        merge_changelogs(changelogs, new_feature_level)
        update_feature_level_in_api_docs(new_feature_level)
        remove_unmerged_changelog_files()

        commit_message = f"api: Increment API feature level to {new_feature_level}."

        try:
            subprocess.run(["git", "add", "-u"], check=True)
            subprocess.run(["git", "commit", "-m", commit_message], check=True)
        except subprocess.CalledProcessError as e:
            print(e)
            sys.exit(1)