-
Notifications
You must be signed in to change notification settings - Fork 2
/
collect_status.py
86 lines (63 loc) · 2.14 KB
/
collect_status.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
#!/usr/bin/env python3
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this file,
# You can obtain one at http://mozilla.org/MPL/2.0/.
import datetime
import json
import pathlib
import re
import subprocess
import sys
import urllib.request
class Logger:
@classmethod
def info(cls, s):
print("[INFO]", s)
# Flush to make it apeear immediately in automation log.
sys.stdout.flush()
@classmethod
def fetch(cls, url):
cls.info(f"Fetching {url}")
class TaskCluster:
API_URL = "https://firefox-ci-tc.services.mozilla.com/api/"
@classmethod
def url(cls, route, artifact):
return f"{cls.API_URL}index/v1/task/{route}/artifacts/{artifact}"
@classmethod
def call(cls, route, artifact):
url = cls.url(route, artifact)
Logger.fetch(url)
req = urllib.request.Request(url, None, {
"User-Agent": "areweesmifiedyet",
})
response = urllib.request.urlopen(req)
return response.read()
@classmethod
def call_json(cls, route, artifact):
return json.loads(cls.call(route, artifact))
class StatusUpdater:
def run():
job_name = "are-we-esmified-yet-check"
route_name = "gecko.v2.mozilla-central.latest.firefox.are-we-esmified-yet"
artifact_name = "public/are-we-esmified-yet.json"
Logger.info("Loading log")
log_file = "./log.json"
if pathlib.Path(log_file).exists():
with open(log_file, "r") as f:
log_data = json.loads(f.read())
else:
log_data = []
if len(log_data) > 0:
last_date = log_data[-1]["date"]
else:
last_date = None
artifact = TaskCluster.call_json(route_name, artifact_name)
date = artifact["date"]
if last_date == date:
Logger.info("The latest artifact is from the same date")
return
Logger.info("Saving log")
log_data.append(artifact)
with open(log_file, "w") as f:
f.write(json.dumps(log_data))
StatusUpdater.run()