Joe Julian
96 lines
3.1 KiB
Python
96 lines
3.1 KiB
Python
#!/usr/bin/env python3
|
|
import argparse
|
|
import json
|
|
import mimetypes
|
|
import pathlib
|
|
import sys
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
|
|
def request_json(method: str, url: str, token: str, payload: dict | None = None) -> dict:
|
|
data = None
|
|
headers = {"Authorization": f"token {token}", "Accept": "application/json"}
|
|
if payload is not None:
|
|
data = json.dumps(payload).encode()
|
|
headers["Content-Type"] = "application/json"
|
|
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
|
with urllib.request.urlopen(req) as resp:
|
|
return json.loads(resp.read().decode())
|
|
|
|
|
|
def request_no_content(method: str, url: str, token: str) -> None:
|
|
req = urllib.request.Request(
|
|
url,
|
|
headers={"Authorization": f"token {token}", "Accept": "application/json"},
|
|
method=method,
|
|
)
|
|
with urllib.request.urlopen(req):
|
|
return
|
|
|
|
|
|
def get_or_create_release(server: str, repo: str, token: str, tag: str) -> dict:
|
|
base = f"{server.rstrip('/')}/api/v1/repos/{repo}"
|
|
tag_url = f"{base}/releases/tags/{urllib.parse.quote(tag, safe='')}"
|
|
try:
|
|
return request_json("GET", tag_url, token)
|
|
except urllib.error.HTTPError as err:
|
|
if err.code != 404:
|
|
raise
|
|
payload = {
|
|
"tag_name": tag,
|
|
"name": tag,
|
|
"draft": False,
|
|
"prerelease": False,
|
|
}
|
|
return request_json("POST", f"{base}/releases", token, payload)
|
|
|
|
|
|
def upload_asset(server: str, repo: str, token: str, release: dict, path: pathlib.Path) -> None:
|
|
base = f"{server.rstrip('/')}/api/v1/repos/{repo}"
|
|
assets = release.get("assets", [])
|
|
for asset in assets:
|
|
if asset.get("name") == path.name:
|
|
request_no_content("DELETE", f"{base}/releases/{release['id']}/assets/{asset['id']}", token)
|
|
query = urllib.parse.urlencode({"name": path.name})
|
|
url = f"{base}/releases/{release['id']}/assets?{query}"
|
|
content_type = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
|
|
req = urllib.request.Request(
|
|
url,
|
|
data=path.read_bytes(),
|
|
headers={
|
|
"Authorization": f"token {token}",
|
|
"Accept": "application/json",
|
|
"Content-Type": content_type,
|
|
},
|
|
method="POST",
|
|
)
|
|
with urllib.request.urlopen(req):
|
|
return
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--server", required=True)
|
|
parser.add_argument("--repo", required=True)
|
|
parser.add_argument("--token", required=True)
|
|
parser.add_argument("--tag", required=True)
|
|
parser.add_argument("artifacts", nargs="+")
|
|
args = parser.parse_args()
|
|
|
|
paths = [pathlib.Path(p) for p in args.artifacts]
|
|
missing = [str(p) for p in paths if not p.is_file()]
|
|
if missing:
|
|
print(f"missing artifacts: {', '.join(missing)}", file=sys.stderr)
|
|
return 1
|
|
|
|
release = get_or_create_release(args.server, args.repo, args.token, args.tag)
|
|
for path in paths:
|
|
upload_asset(args.server, args.repo, args.token, release, path)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|