blob: 2ea3f95794be0b3e3124a1610fc8823dcd82f273 (
plain)
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
|
import json
from pathlib import Path
from typing import Optional
from ATRI.utils.file import write_file
from . import SERVICE_DIR
class Limit:
@staticmethod
def _get_file(group: Optional[int] = None) -> Path:
file_name = f"{group}.service.json"
LIMIT_DIR = SERVICE_DIR / "limit"
path = LIMIT_DIR / file_name
if not LIMIT_DIR.exists():
LIMIT_DIR.mkdir()
return path
@classmethod
def _read_file(cls, group: Optional[int] = None) -> dict:
try:
data = json.loads(cls._get_file(group).read_bytes())
except:
data = {}
return data
@classmethod
async def auth_service(cls, service: str, group: Optional[int] = None) -> bool:
data = cls._read_file(group)
if service not in data:
data[service] = True
await write_file(cls._get_file(group), json.dumps(data))
if data[service]:
return True
else:
return False
@classmethod
async def control_service(
cls,
service: str,
status: bool,
group: Optional[int] = None
) -> None:
data = cls._read_file(group)
if service not in data:
data[service] = True
await write_file(cls._get_file(group), json.dumps(data))
data[service] = status
await write_file(cls._get_file(group), json.dumps(data))
|
