from typing import Dict, Optional from enum import Enum from actions_logging.app_logging import logger from env_files.constants import DEFAULT_ENV_FILES_PATH, ONLY_VERSION_SUFFIX, VERSION_SUFFIX from github.env import exit_on_error_and_write_summary, get_required_env_var from aws.constants import ENVS from github.get_folder_from_github import get_folder_contents_from_github from env_files.get_diff_and_calculate_version import get_severity_from_dicts_diff from master.calculate_version import get_git_high_tags from time import time import os import json import shutil class MergedEnvVarsChangeType(Enum): MINOR = "Minor" PATCH = "Patch" def load_jsons_from_env_files_dir(env_files_path, path_separator=DEFAULT_ENV_FILES_PATH) -> tuple[dict, dict]: """iterate recursively over all env files in the env_files_path and load them into dict""" try: env_jsons_data = {} global_jsons_data = {} for root, dirs, files in os.walk(env_files_path): for file in files: if file.endswith(".json"): file_path = os.path.join(root, file) with open(file_path, 'r') as f: try: json_data = json.load(f) except Exception as e: logger.error(f"error in load_jsons_from_env_files_dir for file: {file_path} - {e}") continue file_path_key = file_path.split(path_separator)[-1] if file == "global.json": global_jsons_data[file_path_key] = json_data else: env_name = file.split(".json")[0] if env_name not in ENVS: logger.warning(f"file {file_path} of env {env_name} is not in ENVS, skipping") env_jsons_data[file_path_key] = json_data return env_jsons_data, global_jsons_data except Exception as e: exit_on_error_and_write_summary(f"error in load_jsons_from_env_files_dir: {e}") def merge_env_to_global(global_jsons_data: dict[str, dict[str, str]], env_jsons_data: dict[str, dict[str, str]]): """take 2 dicts and make them 1. the global dict is the base and the env data for every env will override the global and add it to new dict of merged dict""" try: merged_jsons_dict = {} globals_files = list(global_jsons_data.keys()) for global_file_path in globals_files: global_dir = global_file_path.split("global.json")[0] for file_path in env_jsons_data: if file_path.startswith(global_dir): merged_jsons_dict[file_path] = {**global_jsons_data[global_file_path], **env_jsons_data[file_path]} logger.debug(f"merged_jsons_dict: {merged_jsons_dict}") return merged_jsons_dict except Exception as e: raise RuntimeError(f"error in merge_env_to_global: {e}") def get_merged_json_data(ref: str = None, env_files_path: str = DEFAULT_ENV_FILES_PATH) -> Dict[str, Dict[str, str]]: """ get the merged json data from file system or from github api return dict of env_name as key and json data of global and env as value """ try: files_location = DEFAULT_ENV_FILES_PATH if not ref: logger.info(f"getting json data from default github env file path {env_files_path}") if ref: location_prefix = f"old_env_files_{int(time())}" files_location = f"{location_prefix}/{files_location}" repo_name = get_required_env_var("GITHUB_REPOSITORY") get_folder_contents_from_github(repo_name=repo_name, folder_path=env_files_path, ref=ref, output_folder=files_location) logger.debug(f"got env files from github for ref {ref} into {env_files_path}") env_jsons_data, global_jsons_data = load_jsons_from_env_files_dir(files_location) if ref: shutil.rmtree(files_location) merged_json_data = merge_env_to_global(global_jsons_data, env_jsons_data) return merged_json_data except Exception as e: raise RuntimeError(f"error in get_merged_json_data: {e}") def compute_semver_for_merged_json(old_merged_json_data, new_merged_json_data) -> dict: """ compute the semver for the merged json data :param old_merged_json_data: :param new_merged_json_data: :return: merged_json_semver: dict with env file path as key and MergeEnvVarsChangeType as value (or None) """ try: merged_json_semver = {} for file_path in new_merged_json_data: logger.debug(f"computing semver for file in new_merged_json_data: {file_path}") new_merged_env_dict = new_merged_json_data[file_path] old_merged_env_dict = old_merged_json_data.get(file_path, {}) semver_level = get_severity_from_dicts_diff(old_merged_env_dict, new_merged_env_dict) logger.info(f"semver for env {file_path} is {semver_level}") file_key = f"{DEFAULT_ENV_FILES_PATH}{file_path}" if semver_level: merged_json_semver[file_key] = MergedEnvVarsChangeType(semver_level) else: merged_json_semver[file_key] = None return merged_json_semver except Exception as e: raise RuntimeError(f"error in compute_semver_for_merged_json: {e}") def get_merged_env_vars_changes(env_files_path=DEFAULT_ENV_FILES_PATH) -> Dict[str, Optional[MergedEnvVarsChangeType]]: """return dict with env file path as key and MergeEnvVarsChangeType as value""" try: logger.info("getting merged env vars changes. will iterate over the env.jsons merge them to global.json") logger.info("will do it twice. one for the current merged pr will named 'new' and one for the last git tag and name it 'old'") repo_name = get_required_env_var("GITHUB_REPOSITORY") logger.debug(f"getting merged env vars changes for repo {repo_name}") high_env_version = get_git_high_tags(repo_name, ONLY_VERSION_SUFFIX) logger.debug(f"high env version is {high_env_version}") if high_env_version: high_env_tag = f"{high_env_version}{VERSION_SUFFIX}" else: logger.info(f"no high env tag found, continue with new same as the old status") # this is the first time will act like the old same as the new high_env_tag = None old_merged_json_data = get_merged_json_data(ref=high_env_tag, env_files_path=env_files_path) logger.debug(f"old_merged_json_data len: {len(old_merged_json_data)}") new_merged_json_data = get_merged_json_data() logger.debug(f"new_merged_json_data len: {len(new_merged_json_data)}") merged_json_semver = compute_semver_for_merged_json(old_merged_json_data, new_merged_json_data) return merged_json_semver except Exception as e: raise RuntimeError(f"error in get_merged_env_vars_changes: {e}")