mirror of
http://shenjack.top:5100/shenjack/icalingua-python-bot.git
synced 2024-11-23 12:41:05 +08:00
seperate code into different modules
NOTE: UNTESTED CODE!
This commit is contained in:
dongdigua
parent
666678ef23
commit
e635ff143d
61
bmcl.py
Normal file
61
bmcl.py
Normal file
|
|
@ -0,0 +1,61 @@
|
|||
import time
|
||||
import json
|
||||
import asyncio
|
||||
import aiohttp
|
||||
import socketio
|
||||
|
||||
async def bmcl(sio):
|
||||
await asyncio.sleep(0.1)
|
||||
await sio.emit(
|
||||
"sendMessage", reply_msg.to_content("请求数据中……").to_json()
|
||||
)
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(
|
||||
"https://bd.bangbang93.com/openbmclapi/metric/dashboard"
|
||||
) as response:
|
||||
if not response.status == 200 or response.reason != "OK":
|
||||
await sio.emit(
|
||||
"sendMessage",
|
||||
reply_msg.to_content(
|
||||
f"请求数据失败\n{response.status}"
|
||||
).to_json(),
|
||||
)
|
||||
logger.warn(
|
||||
f"数据请求失败, 请检查网络\n{response.status}",
|
||||
tag="bmclapi_dashboard",
|
||||
)
|
||||
return
|
||||
raw_data = await response.text()
|
||||
try:
|
||||
data = json.loads(raw_data)
|
||||
data_bytes: int = data["bytes"]
|
||||
data_hits: int = data["hits"]
|
||||
data_bandwidth: float = data["currentBandwidth"]
|
||||
load_str: float = data["load"] * 100
|
||||
online_node: int = data["currentNodes"]
|
||||
online_bandwidth: int = data["bandwidth"]
|
||||
data_lens = ["B", "KB", "MB", "GB", "TB"]
|
||||
for i in range(5):
|
||||
if data_bytes < 1024:
|
||||
data_bytes = round(data_bytes, 5)
|
||||
data_bytes = f"{data_bytes}{data_lens[i]}"
|
||||
break
|
||||
data_bytes /= 1024
|
||||
|
||||
report_msg = (
|
||||
"OpenBMCLAPI 状态:\n"
|
||||
f"在线节点: {online_node} 带宽: {online_bandwidth}Mbps\n"
|
||||
f"实时负载带宽: {data_bandwidth:.5f}Mbps 负载: {load_str:.3f}%\n"
|
||||
f"当日 总请求: {data_hits} 总数据量: {data_bytes}"
|
||||
)
|
||||
await sio.emit(
|
||||
"sendMessage",
|
||||
reply_msg.to_content(report_msg).to_json()
|
||||
)
|
||||
|
||||
except (json.JSONDecodeError, AttributeError, ValueError) as e:
|
||||
await sio.emit(
|
||||
"sendMessage",
|
||||
reply_msg.to_content(f"返回数据解析错误\n{e}").to_json(),
|
||||
)
|
||||
logger.warn(f"返回数据解析错误\n{e}", tag="bmclapi_dashboard")
|
||||
155
main.py
155
main.py
|
|
@ -93,69 +93,6 @@ def update_room(data: Dict[str, Any]):
|
|||
logger.info(f"{Fore.CYAN}update_room: {data}")
|
||||
|
||||
|
||||
def safe_eval(code: str) -> str:
|
||||
try:
|
||||
# code = code.replace('help', '坏东西！\n')
|
||||
# code = code.replace('bytes', '坏东西！\n')
|
||||
# code = code.replace('encode', '坏东西！\n')
|
||||
# code = code.replace('decode', '坏东西！\n')
|
||||
# code = code.replace('compile', '屑的！\n')
|
||||
# code = code.replace('globals', '拿不到！\n')
|
||||
code = code.replace("os", "坏东西！\n")
|
||||
code = code.replace("sys", "坏东西！\n")
|
||||
# code = code.replace('input', '坏东西！\n')
|
||||
# code = code.replace('__', '啊哈！\n')
|
||||
# code = code.replace('import', '很坏！\n')
|
||||
code = code.replace(" kill", "别跑！\n")
|
||||
code = code.replace(" rm ", "别跑！\n")
|
||||
code = code.replace("exit", "好坏！\n")
|
||||
code = code.replace("eval", "啊哈！\n")
|
||||
code = code.replace("exec", "抓住！\n")
|
||||
start_time = time.time()
|
||||
try:
|
||||
import os
|
||||
import math
|
||||
import decimal
|
||||
|
||||
global_val = {
|
||||
"time": time,
|
||||
"math": math,
|
||||
"decimal": decimal,
|
||||
"random": random,
|
||||
"__import__": "<built-in function __import__>",
|
||||
"globals": "也别惦记你那个 globals 了",
|
||||
"compile": "想得美",
|
||||
"help": "虽然但是 help 也不行",
|
||||
"exit": "不许 exit",
|
||||
"input": "你想干嘛",
|
||||
"return": "别惦记你那个 return 了",
|
||||
"getattr": "<built-in function getattr>",
|
||||
"setattr": "<built-in function setattr>",
|
||||
}
|
||||
os.system = "不许"
|
||||
result = str(eval(code, global_val, {}))
|
||||
limit = 500
|
||||
if len(result) > limit:
|
||||
result = result[:limit]
|
||||
except:
|
||||
result = traceback.format_exc()
|
||||
end_time = time.time()
|
||||
result = result.replace(BOTCONFIG.private_key, "***")
|
||||
result = result.replace(BOTCONFIG.host, "***")
|
||||
|
||||
logger.info(f"{Fore.MAGENTA}safe_eval: {result}")
|
||||
|
||||
if result == "6" or result == 6:
|
||||
result = "他确实等于 6"
|
||||
|
||||
result = f"{code}\neval result:\n{result}\n耗时: {end_time - start_time} s"
|
||||
return result
|
||||
except:
|
||||
error = traceback.format_exc()
|
||||
result = f"error:\n{error}"
|
||||
return result
|
||||
|
||||
|
||||
@sio.on("addMessage") # type: ignore
|
||||
async def add_message(data: Dict[str, Any]):
|
||||
logger.info(f"{Fore.MAGENTA}add_message: {data}")
|
||||
|
|
@ -169,97 +106,7 @@ async def add_message(data: Dict[str, Any]):
|
|||
|
||||
reply_msg = Message(content="", room_id=room_id, reply_to=ReplyMessage(id=msg_id))
|
||||
if not is_self:
|
||||
if content == "/bot":
|
||||
message = reply_msg.to_content(f"icalingua bot pong v{_version_}")
|
||||
await sio.emit("sendMessage", message.to_json())
|
||||
elif content.startswith("=="):
|
||||
evals: str = content[2:]
|
||||
|
||||
result = safe_eval(evals)
|
||||
# whitelist = ['0', '1', '2', '3', '4', '5', '6', '7', '8', '9', ' ', '.', '+', '-', '*', '/', '(', ')', '<',
|
||||
# '>', '=']
|
||||
# evals = evals.replace('**', '')
|
||||
# express = ''
|
||||
# for text in evals:
|
||||
# if text in whitelist:
|
||||
# express += text
|
||||
# if express == '':
|
||||
# result = '你在干嘛'
|
||||
# else:
|
||||
# result = str(eval(express))
|
||||
message = reply_msg.to_content(result)
|
||||
|
||||
await asyncio.sleep(random.random() * 2)
|
||||
await sio.emit("sendMessage", message.to_json())
|
||||
elif content == "!!jrrp":
|
||||
randomer = random.Random(
|
||||
f'{sender_id}-{data["message"]["date"]}-jrrp-{_version_}'
|
||||
)
|
||||
result = randomer.randint(0, 50) + randomer.randint(0, 50)
|
||||
logger.info(f"{sender_name} 今日人品值为 {result}")
|
||||
message = reply_msg.to_content(f"{sender_name} 今日人品为 {result}")
|
||||
await asyncio.sleep(0.5)
|
||||
await sio.emit("sendMessage", message.to_json())
|
||||
elif content == "/bmcl":
|
||||
await asyncio.sleep(0.1)
|
||||
await sio.emit(
|
||||
"sendMessage", reply_msg.to_content("请求数据中……").to_json()
|
||||
)
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(
|
||||
"https://bd.bangbang93.com/openbmclapi/metric/dashboard"
|
||||
) as response:
|
||||
if not response.status == 200 or response.reason != "OK":
|
||||
await sio.emit(
|
||||
"sendMessage",
|
||||
reply_msg.to_content(
|
||||
f"请求数据失败\n{response.status}"
|
||||
).to_json(),
|
||||
)
|
||||
logger.warn(
|
||||
f"数据请求失败, 请检查网络\n{response.status}",
|
||||
tag="bmclapi_dashboard",
|
||||
)
|
||||
return
|
||||
raw_data = await response.text()
|
||||
try:
|
||||
data = json.loads(raw_data)
|
||||
data_bytes: int = data["bytes"]
|
||||
data_hits: int = data["hits"]
|
||||
data_bandwidth: float = data["currentBandwidth"]
|
||||
load_str: float = data["load"] * 100
|
||||
online_node: int = data["currentNodes"]
|
||||
online_bandwidth: int = data["bandwidth"]
|
||||
data_lens = ["B", "KB", "MB", "GB", "TB"]
|
||||
for i in range(5):
|
||||
if data_bytes < 1024:
|
||||
data_bytes = round(data_bytes, 5)
|
||||
data_bytes = f"{data_bytes}{data_lens[i]}"
|
||||
break
|
||||
data_bytes /= 1024
|
||||
|
||||
report_msg = (
|
||||
"OpenBMCLAPI 状态:\n"
|
||||
f"在线节点: {online_node} 带宽: {online_bandwidth}Mbps\n"
|
||||
f"实时负载带宽: {data_bandwidth:.5f}Mbps 负载: {load_str:.3f}%\n"
|
||||
f"当日 总请求: {data_hits} 总数据量: {data_bytes}"
|
||||
)
|
||||
await sio.emit(
|
||||
"sendMessage",
|
||||
reply_msg.to_content(report_msg).to_json()
|
||||
)
|
||||
|
||||
except (json.JSONDecodeError, AttributeError, ValueError) as e:
|
||||
await sio.emit(
|
||||
"sendMessage",
|
||||
reply_msg.to_content(f"返回数据解析错误\n{e}").to_json(),
|
||||
)
|
||||
logger.warn(f"返回数据解析错误\n{e}", tag="bmclapi_dashboard")
|
||||
"""
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get('http://httpbin.org/get') as resp:
|
||||
print(resp.status)
|
||||
print(await resp.text())"""
|
||||
route(content, sio)
|
||||
|
||||
|
||||
@sio.on("deleteMessage") # type: ignore
|
||||
|
|
|
|||
43
router.py
Normal file
43
router.py
Normal file
|
|
@ -0,0 +1,43 @@
|
|||
from safe_eval import safe_eval
|
||||
import bmcl
|
||||
import yw
|
||||
|
||||
async def route(content, sio):
|
||||
if content == "/bot":
|
||||
message = reply_msg.to_content(f"icalingua bot pong v{_version_}")
|
||||
await sio.emit("sendMessage", message.to_json())
|
||||
elif content.startswith("=="):
|
||||
evals: str = content[2:]
|
||||
|
||||
result = safe_eval(evals)
|
||||
# whitelist = ['0', '1', '2', '3', '4', '5', '6', '7', '8', '9', ' ', '.', '+', '-', '*', '/', '(', ')', '<',
|
||||
# '>', '=']
|
||||
# evals = evals.replace('**', '')
|
||||
# express = ''
|
||||
# for text in evals:
|
||||
# if text in whitelist:
|
||||
# express += text
|
||||
# if express == '':
|
||||
# result = '你在干嘛'
|
||||
# else:
|
||||
# result = str(eval(express))
|
||||
message = reply_msg.to_content(result)
|
||||
|
||||
await asyncio.sleep(random.random() * 2)
|
||||
await sio.emit("sendMessage", message.to_json())
|
||||
elif content == "!!jrrp":
|
||||
randomer = random.Random(
|
||||
f'{sender_id}-{data["message"]["date"]}-jrrp-{_version_}'
|
||||
)
|
||||
result = randomer.randint(0, 50) + randomer.randint(0, 50)
|
||||
logger.info(f"{sender_name} 今日人品值为 {result}")
|
||||
message = reply_msg.to_content(f"{sender_name} 今日人品为 {result}")
|
||||
await asyncio.sleep(0.5)
|
||||
await sio.emit("sendMessage", message.to_json())
|
||||
elif content == "/bmcl":
|
||||
await bmcl.bmcl(sio)
|
||||
elif content == "/yw":
|
||||
message = yw.yw()
|
||||
await asyncio.sleep(random.random() * 2)
|
||||
await sio.emit("sendMessage", message.to_json())
|
||||
|
||||
61
safe_eval.py
Normal file
61
safe_eval.py
Normal file
|
|
@ -0,0 +1,61 @@
|
|||
def safe_eval(code: str) -> str:
|
||||
try:
|
||||
# code = code.replace('help', '坏东西！\n')
|
||||
# code = code.replace('bytes', '坏东西！\n')
|
||||
# code = code.replace('encode', '坏东西！\n')
|
||||
# code = code.replace('decode', '坏东西！\n')
|
||||
# code = code.replace('compile', '屑的！\n')
|
||||
# code = code.replace('globals', '拿不到！\n')
|
||||
code = code.replace("os", "坏东西！\n")
|
||||
code = code.replace("sys", "坏东西！\n")
|
||||
# code = code.replace('input', '坏东西！\n')
|
||||
# code = code.replace('__', '啊哈！\n')
|
||||
# code = code.replace('import', '很坏！\n')
|
||||
code = code.replace(" kill", "别跑！\n")
|
||||
code = code.replace(" rm ", "别跑！\n")
|
||||
code = code.replace("exit", "好坏！\n")
|
||||
code = code.replace("eval", "啊哈！\n")
|
||||
code = code.replace("exec", "抓住！\n")
|
||||
start_time = time.time()
|
||||
try:
|
||||
import os
|
||||
import math
|
||||
import decimal
|
||||
|
||||
global_val = {
|
||||
"time": time,
|
||||
"math": math,
|
||||
"decimal": decimal,
|
||||
"random": random,
|
||||
"__import__": "<built-in function __import__>",
|
||||
"globals": "也别惦记你那个 globals 了",
|
||||
"compile": "想得美",
|
||||
"help": "虽然但是 help 也不行",
|
||||
"exit": "不许 exit",
|
||||
"input": "你想干嘛",
|
||||
"return": "别惦记你那个 return 了",
|
||||
"getattr": "<built-in function getattr>",
|
||||
"setattr": "<built-in function setattr>",
|
||||
}
|
||||
os.system = "不许"
|
||||
result = str(eval(code, global_val, {}))
|
||||
limit = 500
|
||||
if len(result) > limit:
|
||||
result = result[:limit]
|
||||
except:
|
||||
result = traceback.format_exc()
|
||||
end_time = time.time()
|
||||
result = result.replace(BOTCONFIG.private_key, "***")
|
||||
result = result.replace(BOTCONFIG.host, "***")
|
||||
|
||||
logger.info(f"{Fore.MAGENTA}safe_eval: {result}")
|
||||
|
||||
if result == "6" or result == 6:
|
||||
result = "他确实等于 6"
|
||||
|
||||
result = f"{code}\neval result:\n{result}\n耗时: {end_time - start_time} s"
|
||||
return result
|
||||
except:
|
||||
error = traceback.format_exc()
|
||||
result = f"error:\n{error}"
|
||||
return result
|
||||
Loading…
Reference in New Issue
Block a user