229 lines
6.5 KiB
Python
229 lines
6.5 KiB
Python
# 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 xxx@xxx.xxx \
|
|
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') as f:
|
|
return f.read().strip()
|
|
|
|
|
|
def _send_email(proj, latest_ver, recipient, upgrade_log):
|
|
print('Sending email for {}: {}'.format(proj, latest_ver))
|
|
msg = ""
|
|
match = CHANGE_URL_RE.search(upgrade_log)
|
|
if match is not None:
|
|
subject = "[Succeeded]"
|
|
msg = 'An upgrade change is generated at:\n{}'.format(
|
|
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 = """Failed to send email for {} ({}).
|
|
stdout: {}
|
|
stderr: {}""".format(proj, latest_ver, err.stdout, 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') as f:
|
|
results = json.load(f)
|
|
history = {}
|
|
try:
|
|
with open(args.history, 'r') as f:
|
|
history = json.load(f)
|
|
except (FileNotFoundError, json.decoder.JSONDecodeError):
|
|
pass
|
|
|
|
_process_results(args, history, results)
|
|
|
|
with open(args.history, 'w') 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',
|
|
'--branch_and_commit', '--push_change', proj
|
|
],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
cwd=_get_android_top())
|
|
stdout = out.stdout.decode('utf-8')
|
|
stderr = out.stderr.decode('utf-8')
|
|
return """
|
|
====================
|
|
| Debug Info |
|
|
====================
|
|
-=-=-=-=stdout=-=-=-=-
|
|
{}
|
|
|
|
-=-=-=-=stderr=-=-=-=-
|
|
{}
|
|
""".format(stdout, 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()
|