-
Notifications
You must be signed in to change notification settings - Fork 0
/
shared.py
113 lines (91 loc) · 3.21 KB
/
shared.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
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
import json
import requests
import sys
import time
def check_instance(domain_name):
'''
Returns the instance software as a string.
Parameter:
domain_name (str): domain name.
Returns:
instance_software (str): Server software.
In case of error:
Returns -1 (int).
'''
url = f'https://{domain_name}/api/v1/instance'
json_instance, status_code = fetch(url)
# If status_code == 404, this is probably a Misskey instance.
if status_code == 404:
url = f'https://{domain_name}/api/meta'
json_instance, status_code = fetch(url, http_method='POST')
if status_code == 200:
return 'Misskey'
else:
return -1
if json_instance == -1:
return -1
json_instance = json_instance.text
dict_instance = load_json(json_instance)
if not isinstance(dict_instance, dict):
sys.stderr.write('Error parsing instance JSON.\n')
return -1
if 'pleroma' in dict_instance.keys():
return 'Pleroma'
else:
return 'Mastodon'
sys.stderr.write('Server software not recognized.\n')
return -1
def fetch(url, tries=5, http_method='GET', payload={}):
'''
Returns an HTTP response (requests.models.Response) and a status code (int).
Parameters:
url (str): URL to fetch (GET request).
tries (int): How many requests before abandoning (Optional).
http_method (str): HTTP method GET/POST (Optional)
payload (dict): Request payload (Optional)
Returns:
r (requests.models.Response), status_code (int): HTTP response, status code.
In case of error:
Returns -1 (int), -1 (int).
'''
SLEEP = 5
status_code = -1
for i in range(tries):
try:
if http_method == 'GET':
r = requests.get(url, data=json.dumps(payload))
else:
r = requests.post(url, data=json.dumps(payload))
status_code = r.status_code
if status_code == 200 or status_code == 404:
time.sleep(SLEEP)
return (r, status_code)
except requests.exceptions.ConnectionError as e:
sys.stderr.write(f'{e}\n')
except requests.exceptions.MissingSchema as e:
sys.stderr.write(f'{e}\n')
return -1, -1
except requests.exceptions.InvalidURL as e:
sys.stderr.write(f'{e}\n')
return -1, -1
except UnicodeError as e:
sys.stderr.write(f'{e}\n')
return -1, -1
time.sleep(SLEEP)
sys.stderr.write(f'Connection failed after {tries} tries.\n')
return -1, -1
def load_json(json_data):
'''
Returns a parsed JSON (list or dict).
Parameter:
json_data (str): JSON.
Returns:
List or dictionary.
In case of error:
Returns -1 (int).
'''
try:
return json.loads(json_data)
except json.decoder.JSONDecodeError:
sys.stderr.write('json.decoder.JSONDecodeError\n')
return -1