Last active
October 24, 2022 04:37
-
-
Save phucngta/90c291f13582748059a62820866a615c to your computer and use it in GitHub Desktop.
Update Merge Staging - Production v2
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/usr/bin/env python3 | |
| import re | |
| import odoorpc | |
| import gitlab | |
| import os | |
| import ast | |
| from datetime import date, datetime | |
| STAGE_CODES = { | |
| 'to_deploy_integration': 'TDI', | |
| 'ready_for_integration': 'RFT', | |
| 'ready_for_staging': 'TDS', | |
| 'ready_for_production': 'TDP', | |
| 'done': 'done', | |
| } | |
| PROJECT_DEVELOP_BRANCH = os.getenv('PROJECT_DEVELOP_BRANCH', 'develop') | |
| TARGET_BRANCHS = os.getenv('TARGET_BRANCHS', False) and ast.literal_eval(os.getenv('TARGET_BRANCHS')) or ['master', PROJECT_DEVELOP_BRANCH, 'staging', 'production'] | |
| MIRROR_SRC_BRANCHS = os.getenv('MIRROR_SRC_BRANCHS', False) and ast.literal_eval(os.getenv('MIRROR_SRC_BRANCHS')) or [] | |
| CI_SERVER_HOST = os.environ['CI_SERVER_HOST'] | |
| SERVER_URL = f'https://{CI_SERVER_HOST}' | |
| GITLAB_PERSONAL_TOKEN = os.environ['GITLAB_PERSONAL_TOKEN'] | |
| PROJECT_ID = os.environ['CI_PROJECT_ID'] | |
| gl = gitlab.Gitlab(SERVER_URL, private_token=GITLAB_PERSONAL_TOKEN) | |
| project = gl.projects.get(PROJECT_ID) | |
| CI_JOB_STATUS = os.environ['CI_JOB_STATUS'] | |
| CI_JOB_STAGE = os.environ['CI_JOB_STAGE'] | |
| CI_JOB_ID = os.environ['CI_JOB_ID'] | |
| CI_JOB_URL = os.environ['CI_JOB_URL'] | |
| CI_COMMIT_REF_NAME = os.environ['CI_COMMIT_REF_NAME'] | |
| TASK_MS_URL = os.getenv('TASK_MS_URL', 'rms.reach.com.vn') | |
| TASK_MS_DB = os.getenv('TASK_MS_DB', 'rms-template') | |
| TASK_MS_PORT = os.getenv('TASK_MS_PORT', 7069) | |
| TASK_MS_LOGIN_USER = os.getenv('TASK_MS_LOGIN_USER', False) | |
| TASK_MS_LOGIN_PWD = os.getenv('TASK_MS_LOGIN_PWD', False) | |
| # print(TASK_MS_URL, TASK_MS_DB, TASK_MS_LOGIN_USER, TASK_MS_LOGIN_PWD) | |
| odoo = odoorpc.ODOO(TASK_MS_URL, port=TASK_MS_PORT) | |
| odoo.login(TASK_MS_DB, TASK_MS_LOGIN_USER, TASK_MS_LOGIN_PWD) | |
| TASK_OBJECT = odoo.env['project.task'] | |
| def get_latest_merge_date(): | |
| jobs = project.jobs.list( | |
| query_parameters={'scope': 'success'}, as_list=False) | |
| last_job = project.jobs.get(CI_JOB_ID) | |
| for job in jobs: | |
| if job._attrs['ref'] == CI_COMMIT_REF_NAME \ | |
| and job._attrs['stage'] == 'post-merge' \ | |
| and job._attrs['id'] != CI_JOB_ID: | |
| last_job = job | |
| break | |
| latest_merge_date = last_job._attrs['created_at'] | |
| # create_date = datetime.datetime.strptime(create_date, '%Y-%m-%dT%H:%M:%S.%f%z') | |
| print("Latest Successful Merge Job", last_job._attrs['id']) | |
| print("Latest Merge Date", latest_merge_date) | |
| return latest_merge_date | |
| def update_merge_state(): | |
| """ | |
| Update merged-date for merge-request are manually merged | |
| """ | |
| if CI_JOB_STAGE == "post-merge" and CI_COMMIT_REF_NAME in ['staging', 'production']: | |
| print("========================= Update merged date for task ==========================") | |
| latest_merge_date = get_latest_merge_date() | |
| mrs = project.mergerequests.list( | |
| query_parameters={ | |
| 'target_branch': CI_COMMIT_REF_NAME, | |
| 'state': 'merged', | |
| 'updated_after': latest_merge_date | |
| }, as_list=False) | |
| for mr in mrs: | |
| if re.match('T\d{5}', mr._attrs['source_branch']): | |
| print(f"search task bms for MR {mr._attrs['source_branch']}- updated_at {mr._attrs['updated_at']} - merged_at {mr._attrs['merged_at']}") | |
| task_id = TASK_OBJECT.search_read( | |
| [('reference', '=', mr._attrs['source_branch'])], ['id']) | |
| if task_id and task_id[0].get('id'): | |
| task_bms_id = task_id[0].get('id') | |
| task = TASK_OBJECT.browse(task_bms_id) | |
| if CI_COMMIT_REF_NAME == 'staging' and not task.merge_staging_date and task.stage_id.code == STAGE_CODES['ready_for_staging']: | |
| task.merge_staging_date = datetime.now() | |
| _output_log(task, f"Update merge staging date for task {task.reference} by {CI_JOB_URL}" ) | |
| if CI_COMMIT_REF_NAME == 'production' and not task.merge_production_date and task.stage_id.code == STAGE_CODES['ready_for_production']: | |
| task.merge_production_date = datetime.now() | |
| _output_log(task, f"Update merge production date for task {task.reference} by {CI_JOB_URL}") | |
| # Update Ngày Merge Staging/Production cho tất cả task con khi Task Cha được Merge | |
| # sub_tasks = task.child_ids | |
| # for sub_task in sub_tasks: | |
| # if CI_COMMIT_REF_NAME == 'staging' and not sub_task.merge_staging_date: | |
| # print( | |
| # f">>> Update merge staging date {date.today()} for child task {sub_task.reference}") | |
| # sub_task.merge_staging_date = date.today() | |
| # if CI_COMMIT_REF_NAME == 'production' and not sub_task.merge_production_date: | |
| # print( | |
| # f">>> Update merge production date {date.today()} for child task {sub_task.reference}") | |
| # sub_task.merge_production_date = date.today() | |
| def automation_merge_task(): | |
| """ | |
| Automation merge merge-requests and update merged-date to task | |
| """ | |
| if CI_JOB_STAGE == "merge": | |
| print("======================= Start automation merge task ==========================") | |
| mrs = [] | |
| for TARGET_BRANCH in TARGET_BRANCHS + MIRROR_SRC_BRANCHS: | |
| mrs += project.mergerequests.list( | |
| query_parameters={ | |
| 'target_branch': TARGET_BRANCH, | |
| 'state': 'opened', | |
| }, as_list=True) | |
| for mr in mrs: | |
| if re.match('T\d{5}', mr._attrs['source_branch']): | |
| print(f"search task for MR {mr._attrs['source_branch']}") | |
| task_id = TASK_OBJECT.search_read( | |
| [('reference', '=', mr._attrs['source_branch'])], ['id']) | |
| if task_id and task_id[0].get('id'): | |
| task_bms_id = task_id[0].get('id') | |
| task = TASK_OBJECT.browse(task_bms_id) | |
| if mr._attrs['target_branch'] == PROJECT_DEVELOP_BRANCH and task.stage_id.code: | |
| result = _merge_task(mr, task, merge_from_fork=True) | |
| if mr._attrs['target_branch'] == 'staging' and task.stage_id.code in [STAGE_CODES['ready_for_staging'], STAGE_CODES['ready_for_production']]: | |
| result = _merge_task(mr, task, merge_from_fork=True) | |
| if result: | |
| task.merge_staging_date = datetime.now() | |
| _output_log(task, f"Update date merge staging for task {task.reference} by {CI_JOB_URL}") | |
| if mr._attrs['target_branch'] == 'production' and task.stage_id.code == STAGE_CODES['ready_for_production']: | |
| result = _merge_task(mr, task) | |
| if result: | |
| task.merge_production_date = datetime.now() | |
| _output_log(task, f"Update date merge production for task {task.reference} by {CI_JOB_URL}") | |
| # (Notes: Các Merge Request vào Mirror Branch chỉ nằm ở Project Gốc) | |
| # Tự động merge task branch vào mirror branch để đồng bộ code giữa hai project (Project Gốc và Project Mirror) | |
| # nếu | |
| # 1. task của dự án fork ở trạng thái Done | |
| # 2. task của dự án hiện tại ở trạng thái To deploy Production | |
| if mr._attrs['target_branch'] in MIRROR_SRC_BRANCHS: | |
| if mr._attrs['source_project_id'] == mr._attrs['target_project_id'] and task.stage_id.code == STAGE_CODES['ready_for_production']: | |
| result = _merge_task(mr, task) | |
| if mr._attrs['source_project_id'] != mr._attrs['target_project_id'] and task.stage_id.code in [STAGE_CODES['ready_for_staging']]: | |
| result = _merge_task(mr, task, merge_from_fork=True) | |
| def _merge_task(mr, task, merge_from_fork=False): | |
| if mr._attrs['source_project_id'] != mr._attrs['target_project_id'] and not merge_from_fork: | |
| return False | |
| print("Is MR conflicts", mr.has_conflicts) | |
| if mr.has_conflicts: | |
| _output_log(task, f"{mr._attrs['web_url']} has conflicts") | |
| return False | |
| if mr.draft: | |
| if task.review_before_merge: | |
| _output_log(task, f"Need review for task {task.reference}") | |
| return False | |
| else: | |
| mr.title = mr.title[7:] | |
| mr.save() | |
| try: | |
| print("Debug Merge Request", mr._attrs) | |
| mr.merge() | |
| except Exception as e: | |
| _output_log(task, f"Error when merge task {task.reference} with error {e}") | |
| return False | |
| _output_log(task, f"Merged to {mr.target_branch} merge-request {mr._attrs['web_url']} by job {CI_JOB_URL}") | |
| return True | |
| def _output_log(task, message): | |
| logging = f">>> {datetime.now().strftime('%m/%d/%Y %H:%M:%S')} {message}" | |
| task.merge_deploy_note = f"{logging}\n{task.merge_deploy_note or ''}" | |
| print(logging) | |
| if __name__ == "__main__": | |
| automation_merge_task() | |
| update_merge_state() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment