init
This commit is contained in:
tamakyi
@@ -0,0 +1,25 @@
|
||||
from nonebot import require
|
||||
from nonebot.plugin import PluginMetadata, inherit_supported_adapters
|
||||
|
||||
require("nonebot_plugin_alconna")
|
||||
require("nonebot_plugin_localstore")
|
||||
require("nonebot_plugin_apscheduler")
|
||||
|
||||
from . import check_on_statup as _
|
||||
from . import matcher as _ # noqa
|
||||
|
||||
__plugin_meta__ = PluginMetadata(
|
||||
name="今天吃什么（图片版）",
|
||||
description="随机发送吃的或者喝的图片",
|
||||
usage="""
|
||||
今天吃什么
|
||||
今天喝什么
|
||||
今天蜜什么
|
||||
添加菜单
|
||||
删除菜单
|
||||
查看菜单
|
||||
""",
|
||||
type="application",
|
||||
homepage="https://github.com/tamakyi/nonebot-plugin-whateat-pic",
|
||||
supported_adapters=inherit_supported_adapters("nonebot_plugin_alconna"),
|
||||
)
|
||||
@@ -0,0 +1,119 @@
|
||||
import asyncio
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import httpx
|
||||
from nonebot import get_driver
|
||||
from nonebot.log import logger
|
||||
from rich.progress import Progress
|
||||
|
||||
from .config import config
|
||||
|
||||
available_urls = [
|
||||
"https://raw.githubusercontent.com/tamakyi/nonebot-plugin-whateat-pic/",
|
||||
"https://fastly.jsdelivr.net/gh/tamakyi/nonebot-plugin-whateat-pic@",
|
||||
"https://raw.gitmirror.com/tamakyi/nonebot-plugin-whateat-pic/",
|
||||
"https://ghproxy.cfd/https:/raw.githubusercontent.com/tamakyi/nonebot-plugin-whateat-pic/",
|
||||
"https://ghfast.top/https:/raw.githubusercontent.com/tamakyi/nonebot-plugin-whateat-pic/",
|
||||
]
|
||||
|
||||
|
||||
async def check_resource():
|
||||
semaphore = asyncio.Semaphore(10)
|
||||
|
||||
async def _download(client: httpx.AsyncClient, name: str):
|
||||
async with semaphore:
|
||||
for base_url in available_urls:
|
||||
url = base_url + "refs/heads/main/res/" + name
|
||||
try:
|
||||
resp = await client.get(url, timeout=20, follow_redirects=True)
|
||||
if resp.raise_for_status():
|
||||
logger.debug(f"{url} download success!")
|
||||
return resp.content
|
||||
except httpx.HTTPError:
|
||||
pass
|
||||
logger.warning(f"{url} download failed!")
|
||||
return None
|
||||
|
||||
# 下载资源列表
|
||||
async with httpx.AsyncClient() as client:
|
||||
try:
|
||||
if content := await _download(client, "download_list.json"):
|
||||
resource_list = json.loads(content.decode("utf-8"))
|
||||
else:
|
||||
logger.warning("download_list.json not found!")
|
||||
return
|
||||
except json.decoder.JSONDecodeError:
|
||||
logger.warning("download_list.json is not a valid json file!")
|
||||
return
|
||||
|
||||
# 检查资源目录下文件是否完整
|
||||
drink_pic_path = Path(config.whatpic_res_path) / "drink_pic" # 本地资源目录
|
||||
eat_pic_path = Path(config.whatpic_res_path) / "eat_pic" # 本地资源目录
|
||||
me_pic_path = Path(config.whatpic_res_path) / "me_pic"
|
||||
# TODO: 检查文件是否完整
|
||||
|
||||
# 创建下载任务列表
|
||||
download_list: list[tuple[Path, str]] = []
|
||||
|
||||
# 添加下载任务
|
||||
for item in resource_list["drink_pic"]:
|
||||
path_name = (
|
||||
"drink_pic/" + item["name"]
|
||||
) # 在线资源路径: github.com/Cvandia/nonebot-plugin-whateat-pic/main/res/+ path_name
|
||||
download_list.append((drink_pic_path / item["name"], path_name))
|
||||
for item in resource_list["eat_pic"]:
|
||||
path_name = "eat_pic/" + item["name"] # 在线资源路径
|
||||
download_list.append((eat_pic_path / item["name"], path_name))
|
||||
for item in resource_list["me_pic"]:
|
||||
path_name = "me_pic/" + item["name"] # 在线资源路径
|
||||
download_list.append((me_pic_path / item["name"], path_name))
|
||||
|
||||
if download_list:
|
||||
logger.info("Downloading images ...")
|
||||
else:
|
||||
return
|
||||
|
||||
# 下载资源
|
||||
async with httpx.AsyncClient() as client:
|
||||
exist_count = 0
|
||||
|
||||
async def download_image(file_path: Path, file_name: str):
|
||||
if file_path.exists():
|
||||
logger.debug(f"{file_path} already exists, skipping download.")
|
||||
nonlocal exist_count
|
||||
exist_count += 1
|
||||
return
|
||||
if content := await _download(client, file_name):
|
||||
file_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with file_path.open("wb") as f:
|
||||
f.write(content)
|
||||
|
||||
with Progress(
|
||||
*Progress.get_default_columns(), "[yellow]{task.completed}/{task.total}"
|
||||
) as progress:
|
||||
progress_task = progress.add_task(
|
||||
"[green]Downloading pic...", total=len(download_list)
|
||||
)
|
||||
tasks = [
|
||||
download_image(file_path, file_name)
|
||||
for file_path, file_name in download_list
|
||||
]
|
||||
for task in asyncio.as_completed(tasks):
|
||||
await task
|
||||
progress.update(progress_task, advance=1)
|
||||
|
||||
logger.info(f"Downloaded {len(download_list) - exist_count} images.")
|
||||
|
||||
|
||||
driver = get_driver()
|
||||
|
||||
|
||||
@driver.on_startup
|
||||
async def on_startup():
|
||||
logger.info("Checking resources...")
|
||||
task = asyncio.create_task(check_resource())
|
||||
await task
|
||||
|
||||
|
||||
# 测试代码
|
||||
@@ -0,0 +1,46 @@
|
||||
import time
|
||||
|
||||
from nonebot.adapters import Event
|
||||
|
||||
from .config import config
|
||||
|
||||
cd = config.whateat_cd
|
||||
max_count = config.whateat_max
|
||||
|
||||
|
||||
def check_iscd(last_time: float) -> tuple[bool, float, float]:
|
||||
"""
|
||||
判断是否在冷却时间内
|
||||
|
||||
Args:
|
||||
last_time(float): 上次使用时间
|
||||
|
||||
Returns:
|
||||
- tuple[bool, float, float]: 是否在冷却时间内, 剩余冷却时间, 当前时间
|
||||
"""
|
||||
now_time = time.time()
|
||||
if now_time - last_time < cd:
|
||||
return True, cd - (now_time - last_time), last_time
|
||||
return False, 0, now_time
|
||||
|
||||
|
||||
def check_ismax(message: Event, user_count: dict) -> tuple[bool, dict]:
|
||||
"""
|
||||
判断是否达到最大次数
|
||||
|
||||
Args:
|
||||
message(Event): 消息事件
|
||||
user_count(dict): 用户使用次数记录
|
||||
|
||||
Returns:
|
||||
- tuple[bool, dict]: 是否达到最大次数, 用户使用次数记录
|
||||
"""
|
||||
user_id = message.get_user_id()
|
||||
if max_count == 0:
|
||||
return False, {}
|
||||
if user_id not in user_count:
|
||||
user_count[f"{user_id}"] = 0
|
||||
if user_count[f"{user_id}"] < max_count:
|
||||
user_count[f"{user_id}"] += 1
|
||||
return False, user_count
|
||||
return True, user_count
|
||||
@@ -0,0 +1,12 @@
|
||||
import nonebot_plugin_localstore as store
|
||||
from nonebot.plugin import get_plugin_config
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
|
||||
class Config(BaseModel):
|
||||
whateat_cd: int = Field(default=10)
|
||||
whateat_max: int = Field(default=0)
|
||||
whatpic_res_path: str = Field(default=str(store.get_plugin_cache_dir()))
|
||||
|
||||
|
||||
config = get_plugin_config(Config)
|
||||
@@ -0,0 +1,33 @@
|
||||
from pathlib import Path
|
||||
|
||||
from nonebot.log import logger
|
||||
|
||||
from .config import config
|
||||
|
||||
|
||||
def save_pic(img: bytes, img_type: str, name: str) -> None:
|
||||
"""
|
||||
保存图片
|
||||
"""
|
||||
save_path = Path(config.whatpic_res_path) / f"{img_type}_pic" / (name + ".jpg")
|
||||
if isinstance(img, bytes):
|
||||
with Path.open(save_path, "wb") as f:
|
||||
f.write(img)
|
||||
else:
|
||||
logger.error(f"img must be bytes, but got {type(img)}")
|
||||
raise TypeError("img must be bytes")
|
||||
|
||||
|
||||
def delete_pic(img_type: str, name: str) -> None:
|
||||
"""
|
||||
删除图片
|
||||
"""
|
||||
delete_path = Path(config.whatpic_res_path) / f"{img_type}_pic" / (name + ".jpg")
|
||||
if delete_path.exists():
|
||||
try:
|
||||
delete_path.unlink()
|
||||
except OSError as e:
|
||||
logger.error(f"Error: {e}")
|
||||
raise
|
||||
else:
|
||||
raise FileNotFoundError(f"{delete_path} not found")
|
||||
@@ -0,0 +1,246 @@
|
||||
import io
|
||||
import secrets
|
||||
|
||||
import aiofiles
|
||||
from nonebot import get_driver
|
||||
from nonebot.adapters import Event
|
||||
from nonebot.log import logger
|
||||
from nonebot.permission import SUPERUSER
|
||||
from nonebot_plugin_alconna import Alconna, Args, Match, on_alconna
|
||||
from nonebot_plugin_alconna.uniseg import Image, Text, UniMessage
|
||||
from nonebot_plugin_alconna.uniseg.tools import image_fetch
|
||||
from nonebot_plugin_apscheduler import scheduler
|
||||
|
||||
from .check_user_data import check_iscd, check_ismax
|
||||
from .files import delete_pic, save_pic
|
||||
from .menu import Menu
|
||||
from .random_pic import random_pic
|
||||
|
||||
NICKNAME = list(get_driver().config.nickname)
|
||||
BOT_NAME = NICKNAME[0] if NICKNAME else "脑积水"
|
||||
|
||||
TIME = 0.0
|
||||
USER_DATA = {}
|
||||
MAX_MSG = [
|
||||
"你今天吃的够多了！不许再吃了(´-ωก`)",
|
||||
"吃吃吃，就知道吃，你都吃饱了！明天再来(▼皿▼#)",
|
||||
"(*｀へ´*)你猜我会不会再给你发好吃的图片",
|
||||
f"没得吃的了，{BOT_NAME}的食物都被你这坏蛋吃光了！",
|
||||
"你在等我给你发好吃的？做梦哦！你都吃那么多了，不许再吃了！ヽ(≧Д≦)ノ",
|
||||
]
|
||||
|
||||
eat_pic_matcher = on_alconna(
|
||||
Alconna("今天吃什么"),
|
||||
use_cmd_start=True,
|
||||
)
|
||||
|
||||
drink_pic_matcher = on_alconna(
|
||||
Alconna("今天喝什么"),
|
||||
use_cmd_start=True,
|
||||
)
|
||||
|
||||
me_pic_matcher = on_alconna(
|
||||
Alconna("今天蜜什么"),
|
||||
use_cmd_start=True,
|
||||
)
|
||||
|
||||
view_menu_matcher = on_alconna(
|
||||
Alconna("全部菜单", Args["img_type?", str]),
|
||||
use_cmd_start=True,
|
||||
aliases=("查看菜单", "查看菜品"),
|
||||
)
|
||||
|
||||
add_menu_matcher = on_alconna(
|
||||
Alconna(
|
||||
"添加菜单", Args["name?", str], Args["img_type?", str], Args["img?", Image]
|
||||
),
|
||||
use_cmd_start=True,
|
||||
permission=SUPERUSER,
|
||||
)
|
||||
|
||||
del_menu_matcher = on_alconna(
|
||||
Alconna("删除菜单", Args["name?", str], Args["img_type?", str]),
|
||||
use_cmd_start=True,
|
||||
permission=SUPERUSER,
|
||||
)
|
||||
|
||||
|
||||
eat_pic_matcher.shortcut(
|
||||
r"^[今|明|后]?[天|日]?(早|中|晚)?(上|午|餐|饭|夜宵|宵夜|早|晚)吃(什么|啥|点啥)$",
|
||||
fuzzy=False,
|
||||
prefix=True,
|
||||
)
|
||||
drink_pic_matcher.shortcut(
|
||||
r"^[今|明|后]?[天|日]?(早|中|晚)?(上|午|餐|饭|夜宵|宵夜|早|晚)喝(什么|啥|点啥)$",
|
||||
fuzzy=False,
|
||||
prefix=True,
|
||||
)
|
||||
me_pic_matcher.shortcut(
|
||||
r"^(我要蜜)?[今|明|后]?[天|日]?(早|中|晚)?(上|午|餐|饭|夜宵|宵夜|早|晚)蜜(什么|啥|点啥)$",
|
||||
fuzzy=False,
|
||||
prefix=True,
|
||||
)
|
||||
|
||||
|
||||
@eat_pic_matcher.handle()
|
||||
async def handle_eat_pic(event: Event):
|
||||
global TIME
|
||||
global USER_DATA
|
||||
check_max_result, USER_DATA = check_ismax(event, USER_DATA)
|
||||
check_result, remain_time, TIME = check_iscd(TIME)
|
||||
if check_max_result:
|
||||
await UniMessage.text(secrets.choice(MAX_MSG)).finish()
|
||||
elif check_result:
|
||||
await UniMessage.text(f"cd冷却中,还有{remain_time:.2f}秒").finish()
|
||||
else:
|
||||
pic_path, pic_name = random_pic("eat")
|
||||
send_msg: UniMessage = UniMessage(Text(f"🎉{BOT_NAME}建议你吃🎉\n{pic_name}"))
|
||||
async with aiofiles.open(pic_path, "rb") as f:
|
||||
send_msg.append(Image(raw=await f.read()))
|
||||
await send_msg.finish()
|
||||
|
||||
|
||||
@drink_pic_matcher.handle()
|
||||
async def handle_drink_pic(event: Event):
|
||||
global TIME
|
||||
global USER_DATA
|
||||
check_max_result, USER_DATA = check_ismax(event, USER_DATA)
|
||||
check_result, remain_time, TIME = check_iscd(TIME)
|
||||
if check_max_result:
|
||||
await UniMessage.text(secrets.choice(MAX_MSG)).finish()
|
||||
elif check_result:
|
||||
await UniMessage.text(f"cd冷却中,还有{remain_time:.2f}秒").finish()
|
||||
else:
|
||||
pic_path, pic_name = random_pic("drink")
|
||||
send_msg: UniMessage = UniMessage(Text(f"🎉{BOT_NAME}建议你喝🎉\n{pic_name}"))
|
||||
async with aiofiles.open(pic_path, "rb") as f:
|
||||
send_msg.append(Image(raw=await f.read()))
|
||||
await send_msg.finish()
|
||||
|
||||
|
||||
@me_pic_matcher.handle()
|
||||
async def handle_me_pic(event: Event):
|
||||
global TIME
|
||||
global USER_DATA
|
||||
check_max_result, USER_DATA = check_ismax(event, USER_DATA)
|
||||
check_result, remain_time, TIME = check_iscd(TIME)
|
||||
if check_max_result:
|
||||
await UniMessage.text(secrets.choice(MAX_MSG)).finish()
|
||||
elif check_result:
|
||||
await UniMessage.text(f"cd冷却中,还有{remain_time:.2f}秒").finish()
|
||||
else:
|
||||
pic_path, pic_name = random_pic("me")
|
||||
send_msg = UniMessage(Text(f"🎉{BOT_NAME}建议你喝🎉\n{pic_name}"))
|
||||
send_msg.append(Image(path=pic_path))
|
||||
await send_msg.finish()
|
||||
|
||||
|
||||
@view_menu_matcher.handle()
|
||||
async def handle_view_menu(img_type: Match[str]):
|
||||
if img_type.available:
|
||||
view_menu_matcher.set_path_arg("img_type", img_type.result)
|
||||
|
||||
|
||||
@view_menu_matcher.got_path("img_type", prompt=f"请告诉{BOT_NAME}具体菜单类型吧")
|
||||
async def _(img_type: str):
|
||||
menu_type = img_type.strip()
|
||||
if menu_type in ["菜单", "菜品"]:
|
||||
menu_type = "eat"
|
||||
elif menu_type in ["饮料", "饮品"]:
|
||||
menu_type = "drink"
|
||||
elif menu_type in ["蜜雪", "蜜", "蜜雪冰城"]:
|
||||
menu_type = "me"
|
||||
else:
|
||||
await UniMessage.text("菜单类型错误，请重新输入").finish()
|
||||
|
||||
try:
|
||||
menu = Menu(menu_type)
|
||||
send_msg_list = UniMessage(Text("菜单如下："))
|
||||
for img in menu.draw_menu():
|
||||
img_bytesio = io.BytesIO()
|
||||
img.save(img_bytesio, format="JPEG")
|
||||
send_msg_list.append(Image(raw=img_bytesio)) # type: ignore
|
||||
await send_msg_list.finish()
|
||||
except OSError as e:
|
||||
await UniMessage.text(f"没有找到菜单，请稍后重试\n{e}").finish()
|
||||
|
||||
|
||||
@add_menu_matcher.handle()
|
||||
async def _(name: Match[str], img_type: Match[str]):
|
||||
if name.available:
|
||||
add_menu_matcher.set_path_arg("name", name.result)
|
||||
if img_type.available:
|
||||
add_menu_matcher.set_path_arg("img_type", img_type.result)
|
||||
|
||||
|
||||
@add_menu_matcher.got_path("name", prompt=f"请告诉{BOT_NAME}具体菜名或者饮品名吧")
|
||||
async def _(name: str):
|
||||
if not name:
|
||||
await UniMessage.text("菜名不能为空，请重新输入").finish()
|
||||
|
||||
|
||||
@add_menu_matcher.got_path("img_type", prompt=f"请告诉{BOT_NAME}具体菜单类型吧")
|
||||
async def _(img_type: str):
|
||||
if img_type in ["菜品", "菜单"]:
|
||||
add_menu_matcher.set_path_arg("img_type", "eat")
|
||||
elif img_type in ["饮料", "饮品"]:
|
||||
add_menu_matcher.set_path_arg("img_type", "drink")
|
||||
elif img_type in ["蜜雪", "蜜", "蜜雪冰城"]:
|
||||
add_menu_matcher.set_path_arg("img_type", "me")
|
||||
else:
|
||||
await UniMessage.text("菜单类型错误，请重新输入").finish()
|
||||
|
||||
|
||||
@add_menu_matcher.got_path(
|
||||
"img", prompt=f"请告诉{BOT_NAME}图片吧", middleware=image_fetch
|
||||
)
|
||||
async def _(
|
||||
name: str,
|
||||
img_type: str,
|
||||
img: bytes,
|
||||
):
|
||||
if not img:
|
||||
await UniMessage.text("图片不能为空，请重新输入").finish()
|
||||
try:
|
||||
save_pic(img, img_type=img_type, name=name)
|
||||
await UniMessage.text(f"成功添加{name}").finish()
|
||||
except OSError as e:
|
||||
await UniMessage.text(f"添加失败，请稍后重试\n{e}").finish()
|
||||
|
||||
|
||||
@del_menu_matcher.handle()
|
||||
async def _(name: Match[str], img_type: Match[str]):
|
||||
if name.available:
|
||||
del_menu_matcher.set_path_arg("name", name.result)
|
||||
if img_type.available:
|
||||
del_menu_matcher.set_path_arg("img_type", img_type.result)
|
||||
|
||||
|
||||
@del_menu_matcher.got_path("name", prompt=f"请告诉{BOT_NAME}具体菜名或者饮品名吧")
|
||||
async def _(name: str):
|
||||
if not name:
|
||||
await UniMessage.text("菜名不能为空，请重新输入").finish()
|
||||
|
||||
|
||||
@del_menu_matcher.got_path("img_type", prompt=f"请告诉{BOT_NAME}具体菜单类型吧")
|
||||
async def _(img_type: str, name: str):
|
||||
if img_type in ["菜品", "菜单"]:
|
||||
img_type = "eat"
|
||||
elif img_type in ["饮料", "饮品"]:
|
||||
img_type = "drink"
|
||||
elif img_type in ["蜜雪", "蜜", "蜜雪冰城"]:
|
||||
img_type = "me"
|
||||
else:
|
||||
await UniMessage.text("菜单类型错误，请重新输入").finish()
|
||||
try:
|
||||
delete_pic(img_type, name) # type: ignore
|
||||
await UniMessage.text(f"成功删除{name}").finish()
|
||||
except FileNotFoundError as e:
|
||||
await UniMessage.text(f"删除失败, {e}").finish()
|
||||
|
||||
|
||||
# 每日8点清空用户数据
|
||||
@scheduler.scheduled_job("cron", hour=8)
|
||||
async def _():
|
||||
USER_DATA.clear()
|
||||
logger.info("已清空用户数据")
|
||||
@@ -0,0 +1,69 @@
|
||||
from collections.abc import Iterator
|
||||
from pathlib import Path
|
||||
from typing import Literal
|
||||
|
||||
from PIL import Image, ImageDraw, ImageFont
|
||||
|
||||
from .config import config
|
||||
|
||||
|
||||
class Menu:
|
||||
"""
|
||||
生成菜单的类
|
||||
"""
|
||||
|
||||
dish_path: Path # 菜品图片的路径
|
||||
menu_background: Image.Image # 菜单背景图
|
||||
menu_font: ImageFont.FreeTypeFont # 菜单字体
|
||||
|
||||
def __init__(self, menu_type: Literal["drink", "eat", "me"]) -> None:
|
||||
# 菜单图片的路径
|
||||
self.dish_path = Path(config.whatpic_res_path) / f"{menu_type}_pic"
|
||||
# 获取所有菜品的名字
|
||||
self.all_dish_name = [i.stem for i in self.dish_path.iterdir()]
|
||||
# 菜单背景图
|
||||
self.menu_background = Image.open(
|
||||
Path(__file__).parent / "menu_res" / "menu_bg.jpg"
|
||||
)
|
||||
# 字体大小
|
||||
self.font_size = 30
|
||||
# 加载字体
|
||||
self.menu_font = ImageFont.truetype(
|
||||
str(Path(__file__).parent / "menu_res" / "FZSJ-QINGCRJ.TTF"), self.font_size
|
||||
)
|
||||
|
||||
@property
|
||||
def menu_bg_size(self) -> tuple:
|
||||
"""
|
||||
获取菜单背景图的大小
|
||||
"""
|
||||
return self.menu_background.size
|
||||
|
||||
# 生成迭代器用于for循环遍历所有的生成的菜单
|
||||
def draw_menu(self) -> Iterator[Image.Image]:
|
||||
# 每个菜单的行数: 背景图高度 - 200(上下边距) // （字体大小+10(行间距)） = 菜单行数
|
||||
line_num = (self.menu_bg_size[1] - 150) // (self.font_size + 10)
|
||||
# 总共需要合成的图片数量
|
||||
img_num = len(self.all_dish_name) // line_num + 1
|
||||
# 生成所有菜单的图片
|
||||
for i in range(img_num):
|
||||
# 生成一张背景图的副本
|
||||
menu_img = self.menu_background.copy()
|
||||
# 生成draw对象
|
||||
draw = ImageDraw.Draw(menu_img)
|
||||
# 生成菜单的名字
|
||||
for j in range(line_num):
|
||||
# 如果菜单名字已经全部生成完毕
|
||||
if i * line_num + j >= len(self.all_dish_name):
|
||||
break
|
||||
# 生成菜单序号+名字，（居中显示）
|
||||
draw.text(
|
||||
((self.menu_bg_size[0] - 300) // 2, 75 + j * (self.font_size + 10)),
|
||||
f"{i * line_num + j + 1}.{self.all_dish_name[i * line_num + j]}",
|
||||
font=self.menu_font,
|
||||
fill="black",
|
||||
)
|
||||
yield menu_img
|
||||
|
||||
|
||||
# 测试程序
|
||||
Binary file not shown.
Reference in New Issue
Block a user