import json import os from datetime import datetime from re import findall from string import capwords from colorama import Fore from packaging.version import Version from common import ( exec_command, get_commit_tag_as_semver, get_env_variable, get_env_variable_required, get_homologx_manifest_project, is_frontend, is_semantic_released, is_semver, print_message, repository_is_flex, set_env_variable, unshallow_repo, ) from common.exceptions import ( homologx_service_not_found, homologx_unreachable, homologx_version_inferior, ) from common.getopt_helper import GetOptBuilder, Param from common.graphql_client import GraphqlClient from common.seniorx_helper import SeniorXHelper from common.sre_platform_client import SrePlatformClient from scripts.create_sre_issue.create_sre_mr import CreateSREMergeRequest from scripts.create_sre_issue.helpers import ( Backend, ChangelogReader, Frontend, Repository, find_changed_files, ) from scripts.create_sre_issue.jira_manager import JiraManager class CreateSreIssue(GetOptBuilder): def __init__(self, argv): builder = super() builder.__init__(argv) self.client_graphql = GraphqlClient() self.sre_platform = None self.seniorx_frontend_helper = None self.latest_deployed_version = None self.is_semantic_released = is_semantic_released() self.issues_to_be_linked = [] self.sci_debug = get_env_variable("SCI_DEBUG") if is_frontend(): self.sre_platform = SrePlatformClient() """ Set environment variables for SeniorXHelper for support compatibility with projects with outdated .gitlab-ci.yml """ set_env_variable("SCI_SENIORX_ENVIRONMENT_HOST", "platform.senior.com.br") set_env_variable("PROD", "true") self.seniorx_frontend_helper = SeniorXHelper() # Deprecated (Only backend remuneration projects use) repository_param = Param( "repository", default_value=get_env_variable_required("CI_PROJECT_NAME") ) module_param = Param( "module", default_value=get_env_variable_required("SCI_TEAM_MODULE") ) # Deprecated changelog_base_path_param = Param("changelog_base_path", default_value=".") sre_team_param = Param( "sre_team", default_value=get_env_variable_required("SCI_SRE_TEAM") ) translations_path_param = Param("translations_path", only_verbose=True) builder.add_params( [ repository_param, module_param, changelog_base_path_param, sre_team_param, translations_path_param, ] ) builder.build() self.changelog_base_path = changelog_base_path_param.get_param_value() self.module = module_param.get_param_value().upper() self.sre_team = sre_team_param.get_param_value().upper() self.translations_path = get_env_variable("SCI_TRANSLATIONS_DIR") if not self.translations_path: self.translations_path = translations_path_param.get_param_value() self.jira_manager = JiraManager() self.sre_issue = None self.tag_semver = get_commit_tag_as_semver() self.changelog_reader = ChangelogReader( self.changelog_base_path, self.jira_manager ) repository_name = repository_param.get_param_value() self.repository: Repository = ( Frontend(self.tag_semver, repository_name) if is_frontend() else Backend(self.tag_semver, repository_name) ) self.project_id = int(get_env_variable_required("CI_PROJECT_ID")) self.is_rollback = False self.data = { "gitlabProjectId": self.project_id, } def exec(self): """Reads the issues from a given version in changelog and creates an 'deploy issue, and stores deploy data'""" unshallow_repo() self._validate_homologx_version() if not self.is_semantic_released: changelog_issues = self.changelog_reader.read_issues() if not changelog_issues: print_message( f"Nenhuma task encontrada no CHANGELOG para a tag {self.tag_semver}", Fore.YELLOW, ) sre_issue_type = ( "Bug" if any( issue.fields.issuetype.name == "Bug" for issue in changelog_issues ) else "Story" ) else: sre_issue_type = "Story" self.sre_issue = self._create_sre_issue(sre_issue_type) self._add_issue_description() if self.is_rollback: self.jira_manager.update_issue_summary( self.sre_issue.key, f"Rollback {self.repository.deployment}" ) sre_issue_link = self.jira_manager.get_issue_url(self.sre_issue.key) success_message = ( "Task criada para o SRE: " f"{sre_issue_link} - {self.sre_issue.fields.summary} ({self.tag_semver})" ) print_message(success_message, Fore.GREEN) for issue in self.issues_to_be_linked: try: if self.jira_manager.create_link( "Dependência entre atividades", issue, self.sre_issue ): print_message( f"Link criado: {self.jira_manager.get_issue_url(issue.key)} - {self.sre_issue.key}", Fore.GREEN, ) else: print_message( f"⚠️ Falha ao criar link para: {issue.key} -> {self.sre_issue.key}", Fore.YELLOW, ) except Exception as error: print_message( f"❌ Erro ao tentar criar link entre {issue.key} e {self.sre_issue.key}", Fore.RED, ) print_message(f"Motivo: {type(error).__name__}: {error}", Fore.RED) def _validate_homologx_version(self): if get_env_variable("SKIP_HOMOLOGX_VALIDATION"): print_message( " Validação de HomologX desabilitada via SKIP_HOMOLOGX_VALIDATION. Ignorando...", Fore.YELLOW, ) return if is_frontend(): print_message( " Validação de HomologX não aplicável para projetos frontend. Ignorando...", Fore.YELLOW, ) return print_message( f"Validando pré-requisito: versão {self.tag_semver} deve estar em HomologX...", Fore.CYAN, ) create_sre_mr = CreateSREMergeRequest( self.repository, self.sre_team, jira_key="DRY-RUN", ) try: homologx_version = create_sre_mr.get_current_version( get_homologx_manifest_project() ) except Exception as error: homologx_unreachable(error, debug=bool(self.sci_debug)) if not homologx_version: homologx_service_not_found(self.repository.deployment) homologx_semver = homologx_version.lstrip("v").replace("-", ".") requested_semver = self.tag_semver.lstrip("v").replace("-", ".") if not is_semver(homologx_semver) or not is_semver(requested_semver): if homologx_semver != requested_semver: self._raise_homologx_version_error(homologx_version) print_message( f"Versão em HomologX ({homologx_version}) validada com sucesso (comparação literal).", Fore.GREEN, ) return if Version(homologx_semver) < Version(requested_semver): homologx_version_inferior(homologx_version, self.tag_semver) print_message( f" Versão em HomologX ({homologx_version}) >= versão solicitada ({self.tag_semver}). " "Pré-requisito satisfeito.", Fore.GREEN, ) def _raise_homologx_version_error(self, homologx_version: str): homologx_version_inferior(homologx_version, self.tag_semver) def _create_sre_issue(self, sre_issue_type): jira_issue = { "project": {"key": self.sre_team}, "summary": f"Deploy {self.repository.deployment}", "issuetype": {"name": "Deploy"}, "labels": [sre_issue_type, self.module], self.repository.deploy_name_custom_field: self.repository.deploy_info, "customfield_11100": [{"value": "Produção - platform.senior.com.br"}], } return self.jira_manager.create_issue(jira_issue) def _add_issue_description(self): latest_deployed_version, deploy_data = self._create_deploy_approval() self.latest_deployed_version = latest_deployed_version ci_commit_tag = get_env_variable_required("CI_COMMIT_TAG") if latest_deployed_version: description = "h4.Atualização\n" description += f"A última versão atualizada do serviço foi: {latest_deployed_version}\n" description += f"A versão solicitada para atualização é: {ci_commit_tag}\n" else: description = "h4.Criação\n" description += ( f"A versão solicitada para criação do serviço é: {ci_commit_tag}\n" ) description += "\n\n" description += self._add_changelog(latest_deployed_version, ci_commit_tag) translations_path = ( self.translations_path if self.translations_path else self.repository.translations_path ) changed_files = find_changed_files(translations_path, latest_deployed_version) translations = "" project_url = get_env_variable_required("CI_PROJECT_URL") for translation_file in changed_files: filename = os.path.basename(translation_file) if filename: translations += f"[{filename}|{project_url}/-/raw/{ci_commit_tag}/{translation_file}]\n" if translations: description += "h4.Tradução\n" description += "Por favor atualizar os arquivos de traduções:\n" description += translations description += "\n\n" try: mrs_data = self._get_mrs_data(latest_deployed_version) if mrs_data: description += "h4.Merges Requests\n" description += mrs_data description += "\n\n" except Exception as error: print_message( "Não foi possível coletar as informações dos MRs", Fore.YELLOW ) if "503" in str(error): raise error if self.sci_debug: print_message(f"Motivo: {error}", Fore.YELLOW) if deploy_data: description += "h4.Deploy\n" description += deploy_data self.jira_manager.add_description(self.sre_issue.key, description) def _create_deploy_approval(self): if is_frontend(): ( latest_deployed_version, deploy_data, ) = self._get_frontend_additional_description() else: ( latest_deployed_version, deploy_data, ) = self._create_mr_and_additional_description() if latest_deployed_version and is_semver(latest_deployed_version): self.is_rollback = Version(self.tag_semver) < Version( latest_deployed_version ) latest_deployed_version = self._convert_semver_to_senior_version( latest_deployed_version ) return latest_deployed_version, deploy_data def _get_mrs_data(self, latest_deployed_version): table_data = "" table_head = ( "||Título|||Validação do review|||Aberto por|||Aprovado por|||Fechado por" "|||Pipeline|||Aberto em|||Fechado em||\n" ) table_data += table_head merge_requests_data = [] merge_requests = self._get_merges_between_tags(latest_deployed_version) for merge_request in merge_requests: title = merge_request["title"].replace("[", "").replace("]", "") author = capwords(merge_request["author"].replace(".", " ")) merged_by = capwords(merge_request["mergedBy"].replace(".", " ")) mr_title = f"*[{title}|{merge_request['url']}]*" author_name = ( f"[{author}|https://git.senior.com.br/{merge_request['author']}]" ) closed_by_name = ( f"[{merged_by}|https://git.senior.com.br/{merge_request['mergedBy']}]" ) pipeline_status = ( "(/)" if merge_request["pipelineStatus"] == "success" else "(x)" ) review = "(/)" merge_requests_data.append( { "author": author, "reviewer": merged_by, "createdAt": merge_request["createdAt"], "mergedAt": merge_request["mergedAt"], "url": merge_request["url"], "pipelineSuccess": merge_request["pipelineStatus"] == "success", } ) approvers_list = [ f"[{capwords(name.replace('.', ' '))}|https://git.senior.com.br/{name}]" for name in merge_request["approvers"] ] ( author_name, approvers_list, closed_by_name, review, ) = self._add_flags_of_mr_data( author_name, approvers_list, closed_by_name, review, merge_request ) approvers_list = ", ".join(approvers_list) if approvers_list else "-" table_row = ( f"|{mr_title}||{review}|{author_name}|{approvers_list}|{closed_by_name}|" f"|{pipeline_status}|{merge_request['createdAt']}|{merge_request['mergedAt']}|\n" ) table_data += table_row table_subtitle = ( "~O ícone (flag) indica se o Merge Request foi " "aberto e fechado (ou aprovado) pela mesma pessoa.~\n" ) table_data += table_subtitle self.data.update( { "mergeRequests": merge_requests_data, } ) return table_data if merge_requests else None @staticmethod def _add_flags_of_mr_data( author_name, approvers_list, closed_by_name, review, merge_request ): flag_icon = " (flag)" if ( merge_request["author"] in merge_request["approvers"] and len(merge_request["approvers"]) <= 1 and merge_request["author"] == merge_request["mergedBy"] ) or ( not merge_request["approvers"] and merge_request["author"] == merge_request["mergedBy"] ): author_name += flag_icon if merge_request["approvers"]: approvers_list = [ ( f"{approver}{flag_icon}" if approver.replace("]", "").split("/")[-1] == merge_request["author"] else approver ) for approver in approvers_list ] else: closed_by_name += flag_icon review = "(x)" return author_name, approvers_list, closed_by_name, review def _get_tags_between_tags(self, latest_deployed_version): ci_commit_tag = get_env_variable_required("CI_COMMIT_TAG") if latest_deployed_version == ci_commit_tag: latest_deployed_version = None ranged_version = ci_commit_tag if latest_deployed_version: ranged_version = f"{ci_commit_tag}...{latest_deployed_version}" tags_command = exec_command( f"git log --format='%D' --simplify-by-decoration '{ranged_version}'", print_output=False, ) tags = findall(r"tag:\s*(.+?)[,\n]", tags_command.output) if self.is_rollback: tags.append(ci_commit_tag) elif latest_deployed_version: tags.append(latest_deployed_version) else: tags = [tags[0], tags[1]] tags = tags[::-1] return tags def _get_merges_between_tags(self, latest_deployed_version): merge_requests_data = [] tags = self._get_tags_between_tags(latest_deployed_version) for index, tag in enumerate(tags): if index == 0: continue prior_tag = tags[index - 1] query = """ query ($projectId: Int!, $referenceTag: String!, $initialTag: String) { mergesBetweenTags(projectId: $projectId, referenceTag: $referenceTag, initialTag: $initialTag) { mergeRequestsData { url title createdAt mergedAt pipelineStatus author approvers mergedBy } } } """ params = { "projectId": self.project_id, "referenceTag": tag, "initialTag": prior_tag, } try: request = self.client_graphql.call(query, params) except Exception as error: if "504" in str(error): print_message( "Lista de MRs entre as tags é muito grande.", Fore.YELLOW, ) raise error if request: merge_requests_data = ( merge_requests_data + request["mergesBetweenTags"]["mergeRequestsData"] ) for merge_request_data in merge_requests_data: merge_request_data["createdAt"] = datetime.strftime( datetime.fromisoformat(merge_request_data["createdAt"]), "%d/%m/%y %H:%M", ) merge_request_data["mergedAt"] = datetime.strftime( datetime.fromisoformat(merge_request_data["mergedAt"]), "%d/%m/%y %H:%M" ) return merge_requests_data def _get_releases_changelog(self, latest_deployed_version, ci_commit_tag): query = """ query ($projectId: Int!, $referenceTag: String!, $initialTag: String) { getReleasesChangelog(projectId: $projectId, referenceTag: $referenceTag, initialTag: $initialTag) { releaseChangelogUrlList { name url issues } } } """ params = { "projectId": self.project_id, "referenceTag": ci_commit_tag, "initialTag": latest_deployed_version, } return self.client_graphql.call(query, params)["getReleasesChangelog"][ "releaseChangelogUrlList" ] def _create_sre_platform_url(self): package_json = None with open("package.json", "r", encoding="utf-8") as package_json_file: package_json = json.load(package_json_file) name = package_json["project"]["app"] domain = package_json["project"]["domain"] service = package_json["project"]["service"] version = get_env_variable_required("CI_COMMIT_TAG").replace("-", ".") self.data.update({"deployType": "FRONTEND"}) self.data.update({"serviceName": f"{name}/{domain}/{service}"}) self.data.update({"deployDataId": self._create_deploy_data()}) query_params = f"deployDataId={self.data['deployDataId']}" query_params += f"&name={name}" query_params += f"&domain={domain}" query_params += f"&service={service}" query_params += f"&version={version}" query_params += "&type=FRONTEND" return f"https://sre-platform.senior.com.br/deployment?{query_params}" def _create_deploy_data(self): self.data.update( {"taskPipelineAuthor": get_env_variable_required("GITLAB_USER_NAME")} ) response = self.sre_platform.make_post_call("deployData/create", self.data) color = Fore.GREEN if response["status"] == "success" else Fore.RED print_message(response["message"], color) return response["deployDataId"] def _insert_jira_task(self): data = { "deployDataId": self.data["deployDataId"], "jiraTask": self.sre_issue.key, } response = self.sre_platform.make_post_call("deployData/insertJiraTask", data) color = Fore.GREEN if response["status"] == "success" else Fore.RED print_message(response["message"], color) @staticmethod def _convert_semver_to_senior_version(semver_version: str): senior_version = f"v{semver_version.replace('.', '-').replace('v', '')}" tag_exists = exec_command( f"git tag --list {senior_version}", print_output=False ).output return senior_version if tag_exists else f"v{semver_version}" def _get_frontend_additional_description(self): frontend_project_types = ["ANGULAR_SENIORX", "ANGULAR_GENERATED"] latest_deployed_version = None if get_env_variable_required("SCI_PROJECT_TYPE") in frontend_project_types: latest_deployed_version = ( self.seniorx_frontend_helper.get_current_deployed_version( self.repository.identifier ) ) deploy_data = f"[Deploy pelo SRE Platform|{self._create_sre_platform_url()}]" self._insert_jira_task() return latest_deployed_version, deploy_data def _create_mr_and_additional_description(self): deploy_data = None latest_deployed_version = None create_sre_merge_request = CreateSREMergeRequest( self.repository, self.sre_team, self.sre_issue.key ) merge_request, old_versions = create_sre_merge_request.create() if merge_request: deploy_data = "Merge request para atualização do manifesto:\n" deploy_data += merge_request.web_url if repository_is_flex(): merge_request_flex, old_versions = create_sre_merge_request.create(True) if merge_request_flex and merge_request: deploy_data = "Merge requests para atualização dos manifestos:\n" deploy_data += merge_request.web_url deploy_data += "\n" deploy_data += merge_request_flex.web_url elif merge_request_flex: deploy_data = "Merge request para atualização do manifesto:\n" deploy_data += merge_request_flex.web_url for old_version in old_versions: if self.repository.deployment in old_version: _, latest_deployed_version = old_version.split(":") return latest_deployed_version, deploy_data def _add_changelog(self, latest_deployed_version, ci_commit_tag) -> str: changelog_text = "h4.Changelog\n" if self.is_semantic_released: try: releases = self._get_releases_changelog( latest_deployed_version, ci_commit_tag ) except Exception as error: print_message( "Não foi possível coletar as informações dos releases", Fore.YELLOW ) if "503" in str(error): raise error print_message(f"Motivo: {type(error).__name__}: {error}", Fore.YELLOW) releases = [] for release in releases: changelog_text += f"# Release [{release['name']}|{release['url']}]\n" self.issues_to_be_linked += release["issues"] self._get_jira_issues_from_keys() else: changelog_text += self.changelog_reader.changelog_url self.issues_to_be_linked = self.changelog_reader.read_issues( latest_deployed_version ) changelog_text += "\n\n" return changelog_text def _get_jira_issues_from_keys(self): self.issues_to_be_linked = list(set(self.issues_to_be_linked)) valid_issues = [] for key in self.issues_to_be_linked: try: issue = self.jira_manager.get_issue(key) valid_issues.append(issue) except Exception as error: print_message( f"Issue {key} não encontrada no Jira. Será ignorada.", Fore.YELLOW, ) if self.sci_debug: print_message( f"Detalhes: {type(error).__name__}: {error}", Fore.YELLOW, ) self.issues_to_be_linked = valid_issues