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
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
|
import json
from nonebot.params import ArgPlainText
from nonebot.adapters.onebot.v11 import PrivateMessageEvent, GroupMessageEvent
from ATRI.config import BotSelfConfig
from ATRI.exceptions import WriteFileError, ReadFileError
from .data_source import Console, CONSOLE_DIR
from .models import AuthData
gen_console_key = Console().cmd_as_group("auth", "获取进入网页后台的凭证")
@gen_console_key.got("is_pub_n", "咱的运行环境是否有公网(y/n)")
async def _(event: PrivateMessageEvent, is_pub_n: str = ArgPlainText("is_pub_n")):
if is_pub_n != "y":
ip = str(await Console().get_host_ip(False))
await gen_console_key.send("没有公网吗...嗯知道了")
else:
ip = str(await Console().get_host_ip(True))
p = BotSelfConfig.port
rs = Console().get_random_str(20)
df = CONSOLE_DIR / "data.json"
try:
if not df.is_file():
with open(df, "w", encoding="utf-8") as w:
w.write(json.dumps({}))
d = json.loads(df.read_bytes())
ca = d.get("data", None)
if ca:
# 此处原本想用 matcher.finish 但这是在 try 里啊！
await gen_console_key.send("咱已经告诉你了嗷！啊！忘了.../gauth 获取吧")
return
d["data"] = AuthData(ip=ip, port=str(p), token=rs).dict()
with open(df, "w", encoding="utf-8") as w:
w.write(json.dumps(d))
except WriteFileError:
msg = f"""
哦吼！写入文件失败了...还请自行记下哦...
IP: {ip}
PORT: {p}
TOKEN: {rs}
一定要保管好哦！切勿告诉他人哦！
""".strip()
await gen_console_key.send(msg)
raise WriteFileError("Writing file: " + str(df) + " failed!")
msg = f"""
该信息已保存！可通过 /gauth 获取~
IP: {ip}
PORT: {p}
TOKEN: {rs}
一定要保管好哦！切勿告诉他人哦！
""".strip()
await gen_console_key.finish(msg)
@gen_console_key.handle()
async def _(event: GroupMessageEvent):
await gen_console_key.finish("请私戳咱获取（")
load_console_key = Console().cmd_as_group("load", "获取已生成的后台凭证")
@load_console_key.handle()
async def _(event: PrivateMessageEvent):
df = CONSOLE_DIR / "data.json"
if not df.is_file():
await load_console_key.finish("你还没有问咱索要奥！/auth 以获取")
try:
d = json.loads(df.read_bytes())
except ReadFileError:
await load_console_key.send("获取数据失败了...请自行打开文件查看吧:\n" + str(df))
raise ReadFileError("Reading file: " + str(df) + " failed!")
data = d["data"]
msg = f"""
诶嘿嘿嘿——凭证信息来咯！
IP: {data['ip']}
PORT: {data['port']}
TOKEN: {data['token']}
切记！不要告诉他人！！
""".strip()
await load_console_key.finish(msg)
@load_console_key.handle()
async def _(event: GroupMessageEvent):
await load_console_key.finish("请私戳咱获取（")
del_console_key = Console().cmd_as_group("del", "销毁进入网页后台的凭证")
@del_console_key.got("is_sure_d", "...你确定吗(y/n)")
async def _(is_sure: str = ArgPlainText("is_sure_d")):
if is_sure != "y":
await del_console_key.finish("反悔了呢...")
df = CONSOLE_DIR / "data.json"
if not df.is_file():
await del_console_key.finish("你还没向咱索取凭证呢.../auth 以获取")
try:
data: dict = json.loads(df.read_bytes())
del data["data"]
with open(df, "w", encoding="utf-8") as w:
w.write(json.dumps(data))
except WriteFileError:
await del_console_key.send("销毁失败了...请至此处自行删除文件:\n" + str(df))
raise WriteFileError("Writing / Reading file: " + str(df) + " failed!")
await del_console_key.finish("销毁成功！如需再次获取: /auth")
res_console_key = Console().cmd_as_group("reauth", "重置进入网页后台的凭证")
@res_console_key.got("is_sure_r", "...你确定吗(y/n)")
async def _(is_sure: str = ArgPlainText("is_sure_r")):
if is_sure != "y":
await res_console_key.finish("反悔了呢...")
df = CONSOLE_DIR / "data.json"
if not df.is_file():
await del_console_key.finish("你还没向咱索取凭证呢.../auth 以获取")
try:
data: dict = json.loads(df.read_bytes())
del data["data"]
with open(df, "w", encoding="utf-8") as w:
w.write(json.dumps(data))
except WriteFileError:
await del_console_key.send("销毁失败了...请至此处自行删除文件:\n" + str(df))
raise WriteFileError("Writing / Reading file: " + str(df) + " failed!")
@res_console_key.got("is_pub_r_n", "咱的运行环境是否有公网(y/n)")
async def _(event: PrivateMessageEvent, is_pub_n: str = ArgPlainText("is_pub_n")):
if is_pub_n != "y":
ip = str(await Console().get_host_ip(False))
await res_console_key.send("没有公网吗...嗯知道了")
else:
ip = str(await Console().get_host_ip(True))
p = BotSelfConfig.port
rs = Console().get_random_str(20)
df = CONSOLE_DIR / "data.json"
try:
if not df.is_file():
with open(df, "w", encoding="utf-8") as w:
w.write(json.dumps({}))
d = json.loads(df.read_bytes())
ca = d.get("data", None)
if ca:
await res_console_key.send("咱已经告诉你了嗷！啊！忘了.../gauth 获取吧")
return
d["data"] = AuthData(ip=ip, port=str(p), token=rs).dict()
with open(df, "w", encoding="utf-8") as w:
w.write(json.dumps(d))
except WriteFileError:
msg = f"""
哦吼！写入文件失败了...还请自行记下哦...
IP: {ip}
PORT: {p}
TOKEN: {rs}
一定要保管好哦！切勿告诉他人哦！
""".strip()
await res_console_key.send(msg)
raise WriteFileError("Writing file: " + str(df) + " failed!")
msg = f"""
该信息已保存！可通过 /gauth 获取~
IP: {ip}
PORT: {p}
TOKEN: {rs}
一定要保管好哦！切勿告诉他人哦！
""".strip()
await res_console_key.finish(msg)
|
