notifier.py

# 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 \
    --result_file ~/updater/new_result \
    --history ~/updater/history \
    --recipients xxx@xxx.xxx
"""

import argparse
import json
import os
import subprocess
import time


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

    parser = argparse.ArgumentParser(
        description='Check updates for third party projects in external/.')
    parser.add_argument(
        '--result_file',
        help='Json check result generated by external updater.')
    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.')

    return parser.parse_args()


def _send_email(proj, latest_ver, recipient):
    print('Sending email for {}: {}'.format(proj, latest_ver))
    msg = """New version: {}

To upgrade:
    tools/external_updater/updater.sh update {}""".format(
        latest_ver, proj)
    subprocess.run(['sendgmr', '--to=' + recipient,
                    '--subject=' + proj], check=True,
                   stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                   input=msg, encoding='ascii')


def _process_results(history, results, recipient):
    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 latest_ver not in proj_history:
            try:
                _send_email(proj, latest_ver, recipient)
                proj_history[latest_ver] = int(time.time())
            except subprocess.CalledProcessError as err:
                msg = """Failed to send email for {} ({}).
stdout: {}
stderr: {}""".format(proj, latest_ver, err.stdout, err.stderr)
                print(msg)


def send_notification(args):
    """Compare results and send notification."""
    results = {}
    with open(args.result_file, 'r') as f:
        results = json.load(f)
    history = {}
    try:
        with open(args.history, 'r') as f:
            history = json.load(f)
    except FileNotFoundError:
        pass

    _process_results(history, results, args.recipients)

    with open(args.history, 'w') as f:
        json.dump(history, f, sort_keys=True)


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

    args = parse_args()
    send_notification(args)


if __name__ == '__main__':
    main()