summaryrefslogtreecommitdiff
path: root/ATRI/plugins
diff options
context:
space:
mode:
Diffstat (limited to 'ATRI/plugins')
-rw-r--r--ATRI/plugins/console/__init__.py171
-rw-r--r--ATRI/plugins/console/data_source.py35
-rw-r--r--ATRI/plugins/console/driver/__init__.py11
-rw-r--r--ATRI/plugins/console/models.py2
4 files changed, 67 insertions, 152 deletions
diff --git a/ATRI/plugins/console/__init__.py b/ATRI/plugins/console/__init__.py
index e4fec29..2bd14c8 100644
--- a/ATRI/plugins/console/__init__.py
+++ b/ATRI/plugins/console/__init__.py
@@ -4,7 +4,8 @@ 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 ATRI.exceptions import WriteFileError
+
from .data_source import Console, CONSOLE_DIR
from .models import AuthData
@@ -14,52 +15,29 @@ 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")):
+ data_path = CONSOLE_DIR / "data.json"
+ if not data_path.is_file:
+ with open(data_path, "w", encoding="utf-8") as w:
+ w.write(json.dumps(dict()))
+
if is_pub_n != "y":
- ip = str(await Console().get_host_ip(False))
+ host = 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({}))
+ host = str(await Console().get_host_ip(True))
- d = json.loads(df.read_bytes())
+ port = BotSelfConfig.port
+ token = Console().get_random_str(20)
- ca = d.get("data", None)
- if ca:
- # 此处原本想用 matcher.finish 但这是在 try 里啊！
- await gen_console_key.send("咱已经告诉你了嗷！啊！忘了.../con.load 获取吧")
- return
+ data = json.loads(data_path.read_bytes())
+ data["data"] = AuthData(token=token).dict()
+ with open(data_path, "w", encoding="utf-8") as w:
+ w.write(json.dumps(data))
- 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 Exception:
- msg = f"""
- 哦吼！写入文件失败了...还请自行记下哦...
- IP: {ip}
- PORT: {p}
- TOKEN: {rs}
- 一定要保管好哦！切勿告诉他人哦！
- 入口: atri-console.imki.moe
- """.strip()
- await gen_console_key.send(msg)
-
- raise WriteFileError("Writing file: " + str(df) + " failed!")
-
- msg = f"""
- 该信息已保存！可通过 /gauth 获取~
- IP: {ip}
- PORT: {p}
- TOKEN: {rs}
- 一定要保管好哦！切勿告诉他人哦！
+ msg = f"""控制台信息已生成！
+ 请访问: {host}:{port}
+ Token: {token}
+ 该 token 有效时间为 15min
""".strip()
await gen_console_key.finish(msg)
@@ -69,37 +47,6 @@ 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("你还没有问咱索要奥！/con.auth 以获取")
-
- try:
- d = json.loads(df.read_bytes())
- except Exception:
- 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", "销毁进入网页后台的凭证")
@@ -126,82 +73,10 @@ async def _(is_sure: str = ArgPlainText("is_sure_d")):
await del_console_key.finish("销毁成功！如需再次获取: /con.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("你还没向咱索取凭证呢.../con.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 Exception:
- 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("咱已经告诉你了嗷！啊！忘了.../con.load 获取吧")
- 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 Exception:
- msg = f"""
- 哦吼！写入文件失败了...还请自行记下哦...
- IP: {ip}
- PORT: {p}
- TOKEN: {rs}
- 一定要保管好哦！切勿告诉他人哦！
- """.strip()
- await res_console_key.send(msg)
-
- raise WriteFileError("Writing file: " + str(df) + " failed!")
-
- msg = f"""
- 该信息已保存！可通过 /con.load 获取~
- IP: {ip}
- PORT: {p}
- TOKEN: {rs}
- 一定要保管好哦！切勿告诉他人哦！
- """.strip()
- await res_console_key.finish(msg)
-
-
from ATRI import driver as dr
-from .driver import init
+from .data_source import init_resource
+from .driver import init_driver
-dr().on_startup(init)
+dr().on_startup(init_resource)
+dr().on_startup(init_driver)
diff --git a/ATRI/plugins/console/data_source.py b/ATRI/plugins/console/data_source.py
index eee862c..a0e5ef0 100644
--- a/ATRI/plugins/console/data_source.py
+++ b/ATRI/plugins/console/data_source.py
@@ -1,6 +1,7 @@
import json
import socket
import string
+import zipfile
from random import sample
from pathlib import Path
@@ -8,7 +9,9 @@ from nonebot.permission import SUPERUSER
from ATRI.service import Service
from ATRI.utils import request
+from ATRI.rule import is_in_service
from ATRI.exceptions import WriteFileError
+from ATRI.log import logger as log
CONSOLE_DIR = Path(".") / "data" / "plugins" / "console"
@@ -18,7 +21,13 @@ CONSOLE_DIR.mkdir(parents=True, exist_ok=True)
class Console(Service):
def __init__(self):
Service.__init__(
- self, "控制台", "前端管理页面", True, main_cmd="/con", permission=SUPERUSER
+ self,
+ "控制台",
+ "前端管理页面",
+ True,
+ is_in_service("控制台"),
+ main_cmd="/con",
+ permission=SUPERUSER,
)
@staticmethod
@@ -56,3 +65,27 @@ class Console(Service):
if not data:
return {"data": None}
return data
+
+
+FRONTEND_DIR = CONSOLE_DIR / "frontend"
+FRONTEND_DIR.mkdir(parents=True, exist_ok=True)
+CONSOLE_RESOURCE_URL = "https://jsd.imki.moe/gh/kyomotoi/Project-ATRI-Console@main/archive/dist.zip"
+
+
+async def init_resource():
+ log.info("控制台初始化中...")
+
+ try:
+ resp = await request.get(CONSOLE_RESOURCE_URL)
+ except Exception:
+ log.error("控制台资源装载失败, 将无法访问管理界面")
+ return
+ print(len(resp.read()))
+ file_path = CONSOLE_DIR / "dist.zip"
+ with open(file_path, "wb") as w:
+ w.write(resp.read())
+
+ with zipfile.ZipFile(file_path, "r") as zr:
+ zr.extractall(FRONTEND_DIR)
+
+ log.success("控制台初始化完成")
diff --git a/ATRI/plugins/console/driver/__init__.py b/ATRI/plugins/console/driver/__init__.py
index 22afd4a..04c714d 100644
--- a/ATRI/plugins/console/driver/__init__.py
+++ b/ATRI/plugins/console/driver/__init__.py
@@ -1,7 +1,9 @@
from nonebot.drivers.fastapi import Driver
+from fastapi.staticfiles import StaticFiles
from fastapi.middleware.cors import CORSMiddleware
+from ATRI.plugins.console.data_source import FRONTEND_DIR
from .view import (
handle_auther,
handle_base_uri,
@@ -49,8 +51,15 @@ def register_routes(driver: Driver):
app.get(CONSOLE_API_BLOCK_LIST_URI)(handle_get_block_list)
app.get(CONSOLE_API_BLOCK_EDIT_URI)(handle_edit_block)
+ static_path = str(FRONTEND_DIR)
+ app.mount(
+ "/",
+ StaticFiles(directory=static_path, html=True),
+ name="atri-console",
+ )
+
-def init():
+def init_driver():
from ATRI import driver
register_routes(driver()) # type: ignore
diff --git a/ATRI/plugins/console/models.py b/ATRI/plugins/console/models.py
index 973d05d..acbf916 100644
--- a/ATRI/plugins/console/models.py
+++ b/ATRI/plugins/console/models.py
@@ -2,8 +2,6 @@ from pydantic import BaseModel
class AuthData(BaseModel):
- ip: str
- port: str
token: str
Powered by cgit, Theme cgit-dark created by Ted Yin, updated by Mole Shang.