tnot/konabot
1
0
Fork 0
forked from mttu-developers/konabot
Code Pull Requests Activity

空调调温优化与排行榜，浏览器添加本地HTML支持

Browse Source
This commit is contained in:
MixBadGun
2025-10-27 00:04:27 +08:00
parent b28f8f85a2
commit ce2b7fd6f6
5 changed files with 211 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
assets/webpage/ac/assets/background.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 1.1 MiB

76
assets/webpage/ac/index.html Normal file
View File

@ -0,0 +1,76 @@
<html>
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>空调炸炸排行榜</title>
</head>
<body>
<div class="box">
<div class="text">位居全球第 <span id="ranking" class="ranking">200</span>！</div>
<div class="text-2">您的群总共坏了 <span id="number" class="number">200</span> 台空调</div>
<img class="background" src="./assets/background.png" alt="空调炸炸排行榜">
</div>
</body>
<style>
.box {
position: relative;
width: 1024px;
}
.number {
font-size: 2em;
color: #ffdd00;
text-shadow: 3px 3px 6px rgba(0, 0, 0, 0.7);
font-weight: bold;
font-stretch: 50%;
max-width: 520px;
word-wrap: break-word;
line-height: 0.8em;
}
.background {
width: 1024px;
}
.text {
position: absolute;
top: 125px;
width: 100%;
font-size: 72px;
color: white;
text-shadow: 2px 2px 4px rgba(0, 0, 0, 0.7);
font-weight: bolder;
display: flex;
align-items: baseline;
justify-content: center;
}
.text-2 {
position: absolute;
top: 50px;
width: 100%;
font-size: 48px;
color: white;
text-shadow: 2px 2px 4px rgba(0, 0, 0, 0.7);
font-weight: bolder;
display: flex;
align-items: baseline;
justify-content: center;
}
.ranking {
font-size: 2em;
color: #ff0000;
-webkit-text-stroke: #ffffff 2px;
text-shadow: 3px 3px 6px rgba(0, 0, 0, 0.7);
font-weight: bold;
font-stretch: 50%;
}
</style>
<script>
// 从 URL 参数中获取 number 的值
const urlParams = new URLSearchParams(window.location.search);
const number = urlParams.get('number');
// 将 number 显示在页面上
document.getElementById('number').textContent = number;
// 从 URL 参数中获取 ranking 的值
const ranking = urlParams.get('ranking');
// 将 ranking 显示在页面上
document.getElementById('ranking').textContent = ranking;
</script>
</html>

28
konabot/common/web_render/__init__.py
View File

@ -80,6 +80,30 @@ class WebRenderer:
page = cls.page_pool[page_id] page = cls.page_pool[page_id]
return await instance.render_with_page(page, url, target, params=params, other_function=other_function, timeout=timeout) return await instance.render_with_page(page, url, target, params=params, other_function=other_function, timeout=timeout)
@classmethod
async def render_file(
cls,
file_path: str,
target: str,
params: dict = {},
other_function: PageFunction | None = None,
timeout: int = 30,
) -> bytes:
'''
访问指定本地文件URL并返回截图
:param file_path: 目标文件路径
:param target: 渲染目标，如 ".box"、"#main" 等CSS选择器
:param timeout: 页面加载超时时间，单位秒
:param params: URL键值对参数
:param other_function: 其他自定义操作函数，接受page参数
:return: 截图的字节数据
'''
instance = await cls.get_browser_instance()
logger.debug(f"Using WebRendererInstance {id(instance)} to render file {file_path} targeting {target}")
return await instance.render_file(file_path, target, params=params, other_function=other_function, timeout=timeout)
@classmethod @classmethod
async def close_persistent_page(cls, page_id: str) -> None: async def close_persistent_page(cls, page_id: str) -> None:
''' '''
@ -154,6 +178,10 @@ class WebRendererInstance:
async with self.lock: async with self.lock:
screenshot = await self.inner_render(page, url, target, index, params, other_function, timeout) screenshot = await self.inner_render(page, url, target, index, params, other_function, timeout)
return screenshot return screenshot
async def render_file(self, file_path: str, target: str, index: int = 0, params: dict = {}, other_function: callable = None, timeout: int = 30) -> bytes:
file_path = "file:///" + str(file_path).replace("\\", "/")
return await self.render(file_path, target, index, params, other_function, timeout)
async def inner_render(self, page: Page, url: str, target: str, index: int = 0, params: dict = {}, other_function: callable = None, timeout: int = 30) -> bytes: async def inner_render(self, page: Page, url: str, target: str, index: int = 0, params: dict = {}, other_function: callable = None, timeout: int = 30) -> bytes:
logger.debug(f"Navigating to {url} with timeout {timeout}") logger.debug(f"Navigating to {url} with timeout {timeout}")

55
konabot/plugins/air_conditioner/__init__.py
View File

@ -1,13 +1,16 @@
from io import BytesIO from io import BytesIO
from typing import Optional, Union from typing import Optional, Union
import cv2
from nonebot.adapters import Event as BaseEvent from nonebot.adapters import Event as BaseEvent
from nonebot.adapters.console.event import MessageEvent as ConsoleMessageEvent from nonebot.adapters.console.event import MessageEvent as ConsoleMessageEvent
from nonebot.adapters.discord.event import MessageEvent as DiscordMessageEvent from nonebot.adapters.discord.event import MessageEvent as DiscordMessageEvent
from nonebot_plugin_alconna import Alconna, AlconnaMatcher, Args, UniMessage, on_alconna from nonebot_plugin_alconna import Alconna, AlconnaMatcher, Args, UniMessage, on_alconna
from PIL import Image from PIL import Image
import numpy as np
from konabot.common.longtask import DepLongTaskTarget from konabot.common.longtask import DepLongTaskTarget
from konabot.common.path import ASSETS_PATH from konabot.common.path import ASSETS_PATH
from konabot.plugins.air_conditioner.ac import AirConditioner, generate_ac_image from konabot.common.web_render import WebRenderer
from konabot.plugins.air_conditioner.ac import AirConditioner, CrashType, generate_ac_image, wiggle_transform
import random import random
import math import math
@ -73,17 +76,20 @@ async def _(event: BaseEvent, target: DepLongTaskTarget):
await send_ac_image(evt, ac) await send_ac_image(evt, ac)
evt = on_alconna(Alconna( evt = on_alconna(Alconna(
"空调升温" "空调升温",
Args["temp?", Optional[Union[int, float]]] # 可选参数，升温的度数，默认为1
), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=True) ), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=True)
@evt.handle() @evt.handle()
async def _(event: BaseEvent, target: DepLongTaskTarget): async def _(event: BaseEvent, target: DepLongTaskTarget, temp: Optional[Union[int, float]] = 1):
if temp <= 0:
return
id = target.channel_id id = target.channel_id
ac = get_ac(id) ac = get_ac(id)
if not ac.on or ac.burnt == True or ac.frozen == True: if not ac.on or ac.burnt == True or ac.frozen == True:
await send_ac_image(evt, ac) await send_ac_image(evt, ac)
return return
ac.temperature += 1 ac.temperature += temp
if ac.temperature > 40: if ac.temperature > 40:
# 根据温度随机出是否爆炸，40度开始，呈指数增长 # 根据温度随机出是否爆炸，40度开始，呈指数增长
possibility = -math.e ** ((40-ac.temperature) / 50) + 1 possibility = -math.e ** ((40-ac.temperature) / 50) + 1
@ -91,30 +97,37 @@ async def _(event: BaseEvent, target: DepLongTaskTarget):
# 打开爆炸图片 # 打开爆炸图片
with open(ASSETS_PATH / "img" / "other" / "boom.jpg", "rb") as f: with open(ASSETS_PATH / "img" / "other" / "boom.jpg", "rb") as f:
output = BytesIO() output = BytesIO()
Image.open(f).save(output, format="GIF") # 爆炸抖动
frames = wiggle_transform(np.array(Image.open(f)), intensity=5)
pil_frames = [Image.fromarray(frame) for frame in frames]
pil_frames[0].save(output, format="GIF", save_all=True, append_images=pil_frames[1:], loop=0, duration=35, disposal=2)
output.seek(0)
await evt.send(await UniMessage().image(raw=output).export()) await evt.send(await UniMessage().image(raw=output).export())
ac.burnt = True ac.broke_ac(CrashType.BURNT)
await evt.send("太热啦，空调炸了！") await evt.send("太热啦，空调炸了！")
return return
await send_ac_image(evt, ac) await send_ac_image(evt, ac)
evt = on_alconna(Alconna( evt = on_alconna(Alconna(
"空调降温" "空调降温",
Args["temp?", Optional[Union[int, float]]] # 可选参数，降温的度数，默认为1
), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=True) ), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=True)
@evt.handle() @evt.handle()
async def _(event: BaseEvent, target: DepLongTaskTarget): async def _(event: BaseEvent, target: DepLongTaskTarget, temp: Optional[Union[int, float]] = 1):
if temp <= 0:
return
id = target.channel_id id = target.channel_id
ac = get_ac(id) ac = get_ac(id)
if not ac.on or ac.burnt == True or ac.frozen == True: if not ac.on or ac.burnt == True or ac.frozen == True:
await send_ac_image(evt, ac) await send_ac_image(evt, ac)
return return
ac.temperature -= 1 ac.temperature -= temp
if ac.temperature < 0: if ac.temperature < 0:
# 根据温度随机出是否冻结，0度开始，呈指数增长 # 根据温度随机出是否冻结，0度开始，呈指数增长
possibility = -math.e ** (ac.temperature / 50) + 1 possibility = -math.e ** (ac.temperature / 50) + 1
if random.random() < possibility: if random.random() < possibility:
ac.frozen = True ac.broke_ac(CrashType.FROZEN)
await send_ac_image(evt, ac) await send_ac_image(evt, ac)
evt = on_alconna(Alconna( evt = on_alconna(Alconna(
@ -126,4 +139,24 @@ async def _(event: BaseEvent, target: DepLongTaskTarget):
id = target.channel_id id = target.channel_id
ac = get_ac(id) ac = get_ac(id)
ac.change_ac() ac.change_ac()
await send_ac_image(evt, ac) await send_ac_image(evt, ac)
evt = on_alconna(Alconna(
"空调炸炸排行榜",
), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=True)
@evt.handle()
async def _(event: BaseEvent, target: DepLongTaskTarget):
id = target.channel_id
ac = get_ac(id)
number, ranking = ac.get_crashes_and_ranking()
params = {
"number": number,
"ranking": ranking
}
image = await WebRenderer.render_file(
file_path=ASSETS_PATH / "webpage" / "ac" / "index.html",
target=".box",
params=params
)
await evt.send(await UniMessage().image(raw=image).export())

65
konabot/plugins/air_conditioner/ac.py
View File

@ -1,3 +1,4 @@
from enum import Enum
from io import BytesIO from io import BytesIO
import cv2 import cv2
@ -5,6 +6,12 @@ import numpy as np
from PIL import Image, ImageDraw, ImageFont from PIL import Image, ImageDraw, ImageFont
from konabot.common.path import ASSETS_PATH, FONTS_PATH from konabot.common.path import ASSETS_PATH, FONTS_PATH
from konabot.common.path import DATA_PATH
import json
class CrashType(Enum):
BURNT = 0
FROZEN = 1
class AirConditioner: class AirConditioner:
air_conditioners: dict[str, "AirConditioner"] = {} air_conditioners: dict[str, "AirConditioner"] = {}
@ -23,6 +30,60 @@ class AirConditioner:
self.on = False self.on = False
self.temperature = 24 # 重置为默认温度 self.temperature = 24 # 重置为默认温度
def broke_ac(self, crash_type: CrashType):
'''
让空调坏掉，并保存数据
:param crash_type: CrashType 枚举，表示空调坏掉的类型
'''
match crash_type:
case CrashType.BURNT:
self.burnt = True
case CrashType.FROZEN:
self.frozen = True
self.save_crash_data(crash_type)
def save_crash_data(self, crash_type: CrashType):
'''
如果空调爆炸了，就往本地的 ac_crash_data.json 里该 id 的记录加一
'''
data_file = DATA_PATH / "ac_crash_data.json"
crash_data = {}
if data_file.exists():
with open(data_file, "r", encoding="utf-8") as f:
crash_data = json.load(f)
if self.id not in crash_data:
crash_data[self.id] = {"burnt": 0, "frozen": 0}
match crash_type:
case CrashType.BURNT:
crash_data[self.id]["burnt"] += 1
case CrashType.FROZEN:
crash_data[self.id]["frozen"] += 1
with open(data_file, "w", encoding="utf-8") as f:
json.dump(crash_data, f, ensure_ascii=False, indent=4)
def get_crashes_and_ranking(self) -> tuple[int, int]:
'''
获取该群在全国空调损坏的数量与排行榜的位置
'''
data_file = DATA_PATH / "ac_crash_data.json"
if not data_file.exists():
return 0, 1
with open(data_file, "r", encoding="utf-8") as f:
crash_data = json.load(f)
ranking_list = []
for gid, record in crash_data.items():
total = record.get("burnt", 0) + record.get("frozen", 0)
ranking_list.append((gid, total))
ranking_list.sort(key=lambda x: x[1], reverse=True)
total_crashes = crash_data.get(self.id, {}).get("burnt", 0) + crash_data.get(self.id, {}).get("frozen", 0)
rank = 1
for gid, total in ranking_list:
if gid == self.id:
break
rank += 1
return total_crashes, rank
def text_to_transparent_image(text, font_size=40, padding=0, text_color=(0, 0, 0)): def text_to_transparent_image(text, font_size=40, padding=0, text_color=(0, 0, 0)):
""" """
将文本转换为带透明背景的图像，图像大小刚好包含文本 将文本转换为带透明背景的图像，图像大小刚好包含文本
@ -193,7 +254,7 @@ async def generate_ac_image(ac: AirConditioner) -> BytesIO:
return output return output
# 根据生成温度文本图像 # 根据生成温度文本图像
text = f"{ac.temperature}°C" text = f"{round(ac.temperature, 1)}°C"
text_image = text_to_transparent_image( text_image = text_to_transparent_image(
text, text,
font_size=60, font_size=60,
@ -218,7 +279,7 @@ async def generate_ac_image(ac: AirConditioner) -> BytesIO:
final_image_simple = blend_with_transparency(ac_image, transformed_text, (0, 0)) final_image_simple = blend_with_transparency(ac_image, transformed_text, (0, 0))
intensity = max(2, abs(ac.temperature - 24) // 2) intensity = max(2, abs(int(ac.temperature) - 24) // 2)
frames = wiggle_transform(final_image_simple, intensity=intensity) frames = wiggle_transform(final_image_simple, intensity=intensity)
pil_frames = [Image.fromarray(frame) for frame in frames] pil_frames = [Image.fromarray(frame) for frame in frames]