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
|
import json
import aiofiles
from ATRI.exceptions import WriteError
from . import SERVICE_DIR
class Limit:
file_name = "limit.service.json"
path = SERVICE_DIR / file_name
path.parent.mkdir(exist_ok=True, parents=True)
try:
data = json.loads(path.read_bytes())
except:
data = {}
@classmethod
def _filling_service(cls, service, group: int = None):
# Determine whether the service exists in global variables.
if service not in cls.data["global"]:
cls.data["global"][service] = True
# Similary, for group.
if service not in cls.data[group]:
cls.data[group][service] = True
@classmethod
def get_service(cls) -> dict:
return cls.data
@classmethod
async def auth_service(
cls, service: str, group: int = None) -> bool:
cls.data.setdefault("global", {})
cls.data.setdefault(group, {})
cls._filling_service(service, group)
try:
async with aiofiles.open(
cls.path, 'w', encoding='utf-8') as target:
await target.write(
json.dumps(
cls.data
)
)
except WriteError:
raise WriteError("Writing file failed!")
if cls.data["global"][service]:
return True if cls.data[group][service] else False
else:
return False
@classmethod
async def control_service(
cls,
service: str,
status: bool,
group: int = None
) -> None:
cls._filling_service(service, group)
if group:
cls.data[group][service] = status
else:
cls.data["global"][service] = status
try:
async with aiofiles.open(
cls.path, 'w', encoding='utf-8') as target:
await target.write(
json.dumps(
cls.data
)
)
except WriteError:
raise WriteError("Writing file failed!")
|
