-
Notifications
You must be signed in to change notification settings - Fork 1.2k
Expand file tree
/
Copy pathflags2_common.py
More file actions
155 lines (136 loc) · 5.49 KB
/
flags2_common.py
File metadata and controls
155 lines (136 loc) · 5.49 KB
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
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
"""Utilities for second set of flag examples.
"""
import argparse
import string
import sys
import time
from collections import Counter
from enum import Enum
from pathlib import Path
DownloadStatus = Enum('DownloadStatus', 'OK NOT_FOUND ERROR')
POP20_CC = ('CN IN US ID BR PK NG BD RU JP '
'MX PH VN ET EG DE IR TR CD FR').split()
DEFAULT_CONCUR_REQ = 1
MAX_CONCUR_REQ = 1
SERVERS = {
'REMOTE': 'https://www.fluentpython.com/data/flags',
'LOCAL': 'http://localhost:8000/flags',
'DELAY': 'http://localhost:8001/flags',
'ERROR': 'http://localhost:8002/flags',
}
DEFAULT_SERVER = 'LOCAL'
DEST_DIR = Path('downloaded')
COUNTRY_CODES_FILE = Path('country_codes.txt')
def save_flag(img: bytes, filename: str) -> None:
(DEST_DIR / filename).write_bytes(img)
def initial_report(cc_list: list[str],
actual_req: int,
server_label: str) -> None:
if len(cc_list) <= 10:
cc_msg = ', '.join(cc_list)
else:
cc_msg = f'from {cc_list[0]} to {cc_list[-1]}'
print(f'{server_label} site: {SERVERS[server_label]}')
plural = 's' if len(cc_list) != 1 else ''
print(f'Searching for {len(cc_list)} flag{plural}: {cc_msg}')
if actual_req == 1:
print('1 connection will be used.')
else:
print(f'{actual_req} concurrent connections will be used.')
def final_report(cc_list: list[str],
counter: Counter[DownloadStatus],
start_time: float) -> None:
elapsed = time.perf_counter() - start_time
print('-' * 20)
plural = 's' if counter[DownloadStatus.OK] != 1 else ''
print(f'{counter[DownloadStatus.OK]:3} flag{plural} downloaded.')
if counter[DownloadStatus.NOT_FOUND]:
print(f'{counter[DownloadStatus.NOT_FOUND]:3} not found.')
if counter[DownloadStatus.ERROR]:
plural = 's' if counter[DownloadStatus.ERROR] != 1 else ''
print(f'{counter[DownloadStatus.ERROR]:3} error{plural}.')
print(f'Elapsed time: {elapsed:.2f}s')
def expand_cc_args(every_cc: bool,
all_cc: bool,
cc_args: list[str],
limit: int) -> list[str]:
codes: set[str] = set()
A_Z = string.ascii_uppercase
if every_cc:
codes.update(a+b for a in A_Z for b in A_Z)
elif all_cc:
text = COUNTRY_CODES_FILE.read_text()
codes.update(text.split())
else:
for cc in (c.upper() for c in cc_args):
if len(cc) == 1 and cc in A_Z:
codes.update(cc + c for c in A_Z)
elif len(cc) == 2 and all(c in A_Z for c in cc):
codes.add(cc)
else:
raise ValueError('*** Usage error: each CC argument '
'must be A to Z or AA to ZZ.')
return sorted(codes)[:limit]
def process_args(default_concur_req):
server_options = ', '.join(sorted(SERVERS))
parser = argparse.ArgumentParser(
description='Download flags for country codes. '
'Default: top 20 countries by population.')
parser.add_argument(
'cc', metavar='CC', nargs='*',
help='country code or 1st letter (eg. B for BA...BZ)')
parser.add_argument(
'-a', '--all', action='store_true',
help='get all available flags (AD to ZW)')
parser.add_argument(
'-e', '--every', action='store_true',
help='get flags for every possible code (AA...ZZ)')
parser.add_argument(
'-l', '--limit', metavar='N', type=int, help='limit to N first codes',
default=sys.maxsize)
parser.add_argument(
'-m', '--max_req', metavar='CONCURRENT', type=int,
default=default_concur_req,
help=f'maximum concurrent requests (default={default_concur_req})')
parser.add_argument(
'-s', '--server', metavar='LABEL', default=DEFAULT_SERVER,
help=f'Server to hit; one of {server_options} '
f'(default={DEFAULT_SERVER})')
parser.add_argument(
'-v', '--verbose', action='store_true',
help='output detailed progress info')
args = parser.parse_args()
if args.max_req < 1:
print('*** Usage error: --max_req CONCURRENT must be >= 1')
parser.print_usage()
# "standard" exit status codes:
# https://stackoverflow.com/questions/1101957/are-there-any-standard-exit-status-codes-in-linux/40484670#40484670
sys.exit(2) # command line usage error
if args.limit < 1:
print('*** Usage error: --limit N must be >= 1')
parser.print_usage()
sys.exit(2) # command line usage error
args.server = args.server.upper()
if args.server not in SERVERS:
print(f'*** Usage error: --server LABEL '
f'must be one of {server_options}')
parser.print_usage()
sys.exit(2) # command line usage error
try:
cc_list = expand_cc_args(args.every, args.all, args.cc, args.limit)
except ValueError as exc:
print(exc.args[0])
parser.print_usage()
sys.exit(2) # command line usage error
if not cc_list:
cc_list = sorted(POP20_CC)[:args.limit]
return args, cc_list
def main(download_many, default_concur_req, max_concur_req):
args, cc_list = process_args(default_concur_req)
actual_req = min(args.max_req, max_concur_req, len(cc_list))
initial_report(cc_list, actual_req, args.server)
base_url = SERVERS[args.server]
DEST_DIR.mkdir(exist_ok=True)
t0 = time.perf_counter()
counter = download_many(cc_list, base_url, args.verbose, actual_req)
final_report(cc_list, counter, t0)