xref: /aosp_15_r20/tools/external_updater/notifier.py (revision 3c875a214f382db1236d28570d1304ce57138f32)

# Copyright (C) 2018 The Android Open Source Project
#
# 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.
"""Send notification email if new version is found.

Example usage:
external_updater_notifier \
    --history ~/updater/history \
    --generate_change \
    --recipients [email protected] \
    googletest
"""

from datetime import timedelta, datetime
import argparse
import json
import os
import re
import subprocess
import time

# pylint: disable=invalid-name

def parse_args():
    """Parses commandline arguments."""

    parser = argparse.ArgumentParser(
        description='Check updates for third party projects in external/.')
    parser.add_argument('--history',
                        help='Path of history file. If doesn'
                        't exist, a new one will be created.')
    parser.add_argument(
        '--recipients',
        help='Comma separated recipients of notification email.')
    parser.add_argument(
        '--generate_change',
        help='If set, an upgrade change will be uploaded to Gerrit.',
        action='store_true',
        required=False)
    parser.add_argument('paths', nargs='*', help='Paths of the project.')
    parser.add_argument('--all',
                        action='store_true',
                        help='Checks all projects.')

    return parser.parse_args()


def _get_android_top():
    return os.environ['ANDROID_BUILD_TOP']


CHANGE_URL_PATTERN = r'(https:\/\/[^\s]*android-review[^\s]*) Upgrade'
CHANGE_URL_RE = re.compile(CHANGE_URL_PATTERN)


def _read_owner_file(proj):
    owner_file = os.path.join(_get_android_top(), 'external', proj, 'OWNERS')
    if not os.path.isfile(owner_file):
        return None
    with open(owner_file, 'r', encoding='utf-8') as f:
        return f.read().strip()


def _send_email(proj, latest_ver, recipient, upgrade_log):
    print(f'Sending email for {proj}: {latest_ver}')
    msg = ""
    match = CHANGE_URL_RE.search(upgrade_log)
    if match is not None:
        subject = "[Succeeded]"
        msg = f'An upgrade change is generated at:\n{match.group(1)}'
    else:
        subject = "[Failed]"
        msg = 'Failed to generate upgrade change. See logs below for details.'

    subject += f" {proj} {latest_ver}"
    owners = _read_owner_file(proj)
    if owners:
        msg += '\n\nOWNERS file: \n'
        msg += owners

    msg += '\n\n'
    msg += upgrade_log

    cc_recipient = ''
    for line in owners.splitlines():
        line = line.strip()
        if line.endswith('@google.com'):
            cc_recipient += line
            cc_recipient += ','

    subprocess.run(['sendgmr',
                    f'--to={recipient}',
                    f'--cc={cc_recipient}',
                    f'--subject={subject}'],
                   check=True,
                   stdout=subprocess.PIPE,
                   stderr=subprocess.PIPE,
                   input=msg,
                   encoding='ascii')


COMMIT_PATTERN = r'^[a-f0-9]{40}$'
COMMIT_RE = re.compile(COMMIT_PATTERN)


def is_commit(commit: str) -> bool:
    """Whether a string looks like a SHA1 hash."""
    return bool(COMMIT_RE.match(commit))


NOTIFIED_TIME_KEY_NAME = 'latest_notified_time'


def _should_notify(latest_ver, proj_history):
    if latest_ver in proj_history:
        # Processed this version before.
        return False

    timestamp = proj_history.get(NOTIFIED_TIME_KEY_NAME, 0)
    time_diff = datetime.today() - datetime.fromtimestamp(timestamp)
    if is_commit(latest_ver) and time_diff <= timedelta(days=30):
        return False

    return True


def _process_results(args, history, results):
    for proj, res in results.items():
        if 'latest' not in res:
            continue
        latest_ver = res['latest']
        current_ver = res['current']
        if latest_ver == current_ver:
            continue
        proj_history = history.setdefault(proj, {})
        if _should_notify(latest_ver, proj_history):
            upgrade_log = _upgrade(proj) if args.generate_change else ""
            try:
                _send_email(proj, latest_ver, args.recipients, upgrade_log)
                proj_history[latest_ver] = int(time.time())
                proj_history[NOTIFIED_TIME_KEY_NAME] = int(time.time())
            except subprocess.CalledProcessError as err:
                msg = f"""Failed to send email for {proj} ({latest_ver}).
stdout: {err.stdout}
stderr: {err.stderr}"""
                print(msg)


RESULT_FILE_PATH = '/tmp/update_check_result.json'


def send_notification(args):
    """Compare results and send notification."""
    results = {}
    with open(RESULT_FILE_PATH, 'r', encoding='utf-8') as f:
        results = json.load(f)
    history = {}
    try:
        with open(args.history, 'r', encoding='utf-8') as f:
            history = json.load(f)
    except (FileNotFoundError, json.decoder.JSONDecodeError):
        pass

    _process_results(args, history, results)

    with open(args.history, 'w', encoding='utf-8') as f:
        json.dump(history, f, sort_keys=True, indent=4)


def _upgrade(proj):
    # pylint: disable=subprocess-run-check
    out = subprocess.run([
        'out/soong/host/linux-x86/bin/external_updater', 'update', proj
    ],
                         stdout=subprocess.PIPE,
                         stderr=subprocess.PIPE,
                         cwd=_get_android_top())
    stdout = out.stdout.decode('utf-8')
    stderr = out.stderr.decode('utf-8')
    return f"""
====================
|    Debug Info    |
====================
-=-=-=-=stdout=-=-=-=-
{stdout}

-=-=-=-=stderr=-=-=-=-
{stderr}
"""


def _check_updates(args):
    params = [
        'out/soong/host/linux-x86/bin/external_updater', 'check',
        '--json_output', RESULT_FILE_PATH, '--delay', '30'
    ]
    if args.all:
        params.append('--all')
    else:
        params += args.paths

    print(_get_android_top())
    # pylint: disable=subprocess-run-check
    subprocess.run(params, cwd=_get_android_top())


def main():
    """The main entry."""

    args = parse_args()
    _check_updates(args)
    send_notification(args)


if __name__ == '__main__':
    main()