-
Notifications
You must be signed in to change notification settings - Fork 0
/
__main__.py
84 lines (71 loc) · 2.61 KB
/
__main__.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
'''
This is the entry point for the VPN Switcher application.
'''
from threading import Thread
import argparse
from sys import exit as end
from time import sleep
from src.models.user_switches import UserSwitches
from src.models.vpn_model import abstract_vpn_model
from src.services.vpn_parser_service import VpnDataParserService
PROMPT: str = 'Run in Connect (c), Disconnect (d), or be in Always-Connected mode (w)'
DEFAULT_VPN_DATA_PATH: str = './vpn_data.json'
def get_user_switches() -> UserSwitches:
'''
Get user switches from command line arguments or prompt user for input.
Returns:
UserSwitches: User switches in a structured object
'''
parser: argparse.ArgumentParser = argparse.ArgumentParser()
parser.add_argument('-a', '--action', help=PROMPT, type=str)
parser.add_argument(
'-p',
'--path',
help='Path to VPN data JSON file',
type=str,
required=False
)
parser.add_argument(
'-v',
'--verbose',
help='Whether to run in verbose mode. DEFAULT false',
type=bool,
default=False,
required=False
)
args: argparse.Namespace = parser.parse_args()
if args.action is None and args.path is None and args.verbose is False:
parser.print_help()
end(0)
return UserSwitches(
args.action if args.action else input(f'{PROMPT}: ').lower(),
args.path if args.path else DEFAULT_VPN_DATA_PATH,
args.verbose if args.verbose else False
)
if __name__ == '__main__':
# Get and Validate user switch to know whether to connect or disconnect VPNs
user_switches: UserSwitches = get_user_switches()
if user_switches.action not in ['w', 'c', 'd']:
raise ValueError(f'Invalid action switch. {PROMPT}!')
# List all VPNs
vpn_data_list: list[abstract_vpn_model.AbstractVpnModel] = []
with open(user_switches.vpn_data_json_path, 'r', encoding='utf-8') as f:
vpn_parser_service: VpnDataParserService = VpnDataParserService()
vpn_data_list = vpn_parser_service.parse_vpn_data(f.read())
while True:
# Perform action on all VPNs
threads: set[Thread] = set()
for vpn in vpn_data_list:
t: Thread = Thread(
target=vpn.disconnect if user_switches.action == 'd' else vpn.connect,
kwargs={'verbose': user_switches.verbose}
)
t.start()
threads.add(t)
# Ensure all threads complete
for t in threads:
t.join()
# If not in watch mode, exit
if user_switches.action != 'w':
break
sleep(5)