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
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
|
#!/usr/bin/env python
"""Download packages from github actions artifacts
"""
import os
import re
import sys
import logging
import datetime as dt
from pathlib import Path
import requests
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
API_URL = "https://ci.appveyor.com/api"
REPOS = "psycopg/psycopg2"
WORKFLOW_NAME = "Build packages"
class ScriptError(Exception):
"""Controlled exception raised by the script."""
def main():
try:
token = os.environ["APPVEYOR_TOKEN"]
except KeyError:
raise ScriptError("please set a APPVEYOR_TOKEN to download artifacts")
s = requests.Session()
s.headers["Content-Type"] = "application/json"
s.headers["Authorization"] = f"Bearer {token}"
logger.info("fetching last run")
resp = s.get(f"{API_URL}/projects/{REPOS}/")
resp.raise_for_status()
data = resp.json()
updated_at = dt.datetime.fromisoformat(
re.sub(r"\.\d+", "", data["build"]["finished"])
)
now = dt.datetime.now(dt.timezone.utc)
age = now - updated_at
logger.info(
f"found build {data['build']['version']} updated {pretty_interval(age)} ago"
)
if age > dt.timedelta(hours=6):
logger.warning("maybe it's a bit old?")
jobs = data["build"]["jobs"]
for job in jobs:
if job["status"] != "success":
raise ScriptError("status for job {job['jobId']} is {job['status']}")
logger.info(f"fetching artifacts info for {job['name']}")
resp = s.get(f"{API_URL}/buildjobs/{job['jobId']}/artifacts/")
resp.raise_for_status()
afs = resp.json()
for af in afs:
fn = af["fileName"]
if fn.startswith("dist/"):
fn = fn.split("/", 1)[1]
dest = Path("packages") / fn
logger.info(f"downloading {dest}")
resp = s.get(
f"{API_URL}/buildjobs/{job['jobId']}/artifacts/{af['fileName']}"
)
resp.raise_for_status()
if not dest.parent.exists():
dest.parent.mkdir()
with dest.open("wb") as f:
f.write(resp.content)
logger.info("now you can run: 'twine upload -s packages/*'")
def pretty_interval(td):
secs = td.total_seconds()
mins, secs = divmod(secs, 60)
hours, mins = divmod(mins, 60)
days, hours = divmod(hours, 24)
if days:
return f"{int(days)} days, {int(hours)} hours, {int(mins)} minutes"
elif hours:
return f"{int(hours)} hours, {int(mins)} minutes"
else:
return f"{int(mins)} minutes"
if __name__ == "__main__":
try:
sys.exit(main())
except ScriptError as e:
logger.error("%s", e)
sys.exit(1)
except KeyboardInterrupt:
logger.info("user interrupt")
sys.exit(1)
|
