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
|
#!/usr/bin/env python
import argparse
import asyncio
import logging
import os
import aiohttp
logger = logging.getLogger()
logger.setLevel(logging.DEBUG)
def build_data(args: argparse.Namespace, action: str) -> dict[str, str]:
return {"token": args.token, "arch": args.arch, "action": action}
# source: https://stackoverflow.com/a/1094933
def sizeof_fmt(num: int, suffix: str = "B"):
for unit in ["", "Ki", "Mi", "Gi", "Ti", "Pi", "Ei", "Zi"]:
if abs(num) < 1024.0:
return f"{num:3.1f}{unit}{suffix}"
num /= 1024.0
return f"{num:.1f}Yi{suffix}"
async def send_file(
session: aiohttp.ClientSession, args: argparse.Namespace, file: str
) -> None:
data = aiohttp.FormData()
data.add_field("file", open(file, "rb"), filename=file)
data.add_field("token", args.token)
data.add_field("arch", args.arch)
async with session.post(args.remote_address, data=data):
pass
async def main(args: argparse.Namespace) -> int:
pending_upload = []
for file in os.listdir("."):
if ".pkg" in file:
pending_upload.append(file)
if not len(pending_upload):
logger.debug("Could not find any packaged packages, skip upload")
return 0
async with aiohttp.ClientSession(
raise_for_status=True, timeout=aiohttp.ClientTimeout(60)
) as session:
async with session.post(
args.remote_address, data=build_data(args, "REQUIRE_CLEAN")
):
pass
for file in pending_upload:
logger.info("Start upload %s (%s)", sizeof_fmt(os.stat(file).st_size))
await send_file(session, args, file)
async with session.post(args.remote_address, data=build_data(args, "UPLOADED")):
pass
if __name__ == "__main__":
parser = argparse.ArgumentParser("upload.py")
parser.add_argument("remote_path")
parser.add_argument("token")
parser.add_argument("arch")
parser.add_argument("--directory")
args_ = parser.parse_args()
try:
import coloredlogs
coloredlogs.install(
logging.DEBUG,
fmt="%(asctime)s - %(levelname)s - %(funcName)s - %(lineno)d - %(message)s",
)
except ModuleNotFoundError:
logging.basicConfig(
level=logging.DEBUG,
format="%(asctime)s - %(levelname)s - %(funcName)s - %(lineno)d - %(message)s",
)
if args_.directory is not None:
os.chdir(args_.directory)
exit(asyncio.run(main(args_)))
|
