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
|
import argparse
import os
import aiohttp
import asyncio
import json
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from .typing import *
class Cli(object):
parser = argparse.ArgumentParser(description="水系终端脚手架")
def __init__(self):
self.parser.add_argument(
"-i",
"--install",
dest="command",
help="安装规则包、插件与模型",
action="store_const",
const="install_package",
)
self.parser.add_argument(
"-T",
"--template",
dest="command",
help="选择模板快速创建Bot实例",
action="store_const",
const="build_template",
)
self.parser.add_argument(
"-S",
"--search",
dest="command",
help="在指定镜像源查找规则包、插件与模型",
action="store_const",
const="search_package",
)
self.parser.add_argument(
"-c",
"--config",
dest="command",
help="配置管理",
action="store_const",
const="config",
)
self.args = self.parser.parse_args()
def get_args(self):
return self.args
def get_help(self):
return self.parser.format_help()
async def install_packages(self):
package_name = input("请输入要安装的包名：")
url = f"https://pypi.org/pypi/{package_name}/json"
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
if response.status == 200:
data = await response.json()
await self._extract_package(data, package_name)
else:
print(f"找不到包：{package_name}")
async def _extract_package(self, data, package_name):
latest_version = data["info"]["version"]
download_url = data["releases"][latest_version][0]["url"]
plugins_dir = "plugins"
if not os.path.exists(plugins_dir):
os.mkdir(plugins_dir)
file_name = download_url.split("/")[-1]
file_path = os.path.join(plugins_dir, file_name)
async with aiohttp.ClientSession() as session:
async with session.get(download_url) as response:
if response.status == 200:
with open(file_path, "wb") as file:
file.write(await response.read())
print(f"成功安装包：{package_name}")
else:
print(f"下载包时出错：{package_name}")
def build_template(self):
template = input("请选择应用模板（输入数字）:\n" "1. 创建轻量应用\n" "2. 创建标准应用\n" "3. 创建开发应用\n")
if template == "1":
print("选择了轻量应用模板")
elif template == "2":
print("选择了标准应用模板")
elif template == "3":
print("选择了开发应用模板")
else:
print("无效的模板选择")
async def search_package(self):
search_term = input("请输入要搜索的包名关键字：")
url = f"https://pypi.org/search/?q={search_term}"
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
if response.status == 200:
data: dict = response.json() # type: ignore[dict]
packages = data.get("results", [])
for package in packages:
name = package["name"]
topics = package.get("topics", [])
if (
search_term.lower() in name.lower()
and "HydroRoll" in topics
):
print(f"包名：{name}")
else:
print("搜索失败")
def config(self):
config_dir = os.path.expanduser("~/.hydroroll")
if not os.path.exists(config_dir):
os.makedirs(config_dir)
config_file = os.path.join(config_dir, "config.json")
subcommand = input("请输入子命令（add/delete）：")
if subcommand == "add":
key = input("请输入要添加的键名：")
value = input("请输入要添加的键值：")
with open(config_file, "r+") as file:
try:
config_data = json.load(file)
except json.JSONDecodeError:
config_data = {}
config_data[key] = value
self._extracted_from_config_21(file, config_data)
print(f"成功添加配置项：{key}={value}")
elif subcommand == "delete":
key = input("请输入要删除的键名：")
with open(config_file, "r+") as file:
try:
config_data = json.load(file)
except json.JSONDecodeError:
config_data = {}
if key in config_data:
del config_data[key]
self._extracted_from_config_21(file, config_data)
print(f"成功删除配置项：{key}")
else:
print(f"配置项不存在：{key}")
else:
print("无效的子命令选择")
# TODO Rename this here and in `config`
def _extracted_from_config_21(self, file, config_data):
file.seek(0)
json.dump(config_data, file, indent=4)
file.truncate()
cli = Cli()
if cli.get_args().command == "install_package":
asyncio.run(cli.install_packages())
elif cli.get_args().command == "build_template":
cli.build_template()
elif cli.get_args().command == "search_package":
asyncio.run(cli.search_package())
elif cli.get_args().command == "config":
cli.config()
else:
print(cli.get_help())
|