话痨榜的图片渲染!

This commit is contained in:
Chenric
2024-07-08 13:42:58 +08:00
parent e08e1d2392
commit cff3a951b4
17 changed files with 482 additions and 263 deletions

View File

@@ -1,6 +1,8 @@
from datetime import datetime, time, tzinfo
import os
import unicodedata
from typing import Optional, Dict, List, Union
from zoneinfo import ZoneInfo
from pathlib import Path
from sqlalchemy import or_, select
from sqlalchemy.sql import ColumnElement
@@ -14,79 +16,17 @@ from nonebot_plugin_orm import get_session
from nonebot_plugin_session import Session, SessionLevel, extract_session
from nonebot_plugin_session_orm import SessionModel
from nonebot_plugin_userinfo import EventUserInfo, UserInfo
from nonebot_plugin_htmlrender import html_to_pic,template_to_pic
from nonebot_plugin_apscheduler import scheduler
from nonebot_plugin_chatrecorder import MessageRecord
from nonebot_plugin_localstore import get_cache_dir
from nonebot_plugin_alconna import AlconnaMatcher
from .config import plugin_config
from .model import UserRankInfo
def parse_datetime(key: str):
"""解析数字,并将结果存入 state 中"""
async def _key_parser(
matcher: AlconnaMatcher,
state: T_State,
input: Union[datetime, Message] = Arg(key),
):
if isinstance(input, datetime):
return
plaintext = input.extract_plain_text()
try:
state[key] = get_datetime_fromisoformat_with_timezone(plaintext)
except ValueError:
await matcher.reject_arg(key, "请输入正确的日期,不然我没法理解呢!")
return _key_parser
def get_datetime_now_with_timezone() -> datetime:
"""获取当前时间,并包含时区信息"""
if plugin_config.timezone:
return datetime.now(ZoneInfo(plugin_config.timezone))
else:
return datetime.now().astimezone()
def get_datetime_fromisoformat_with_timezone(date_string: str) -> datetime:
"""从 ISO-8601 格式字符串中获取时间,并包含时区信息"""
if not plugin_config.timezone:
return datetime.fromisoformat(date_string).astimezone()
raw = datetime.fromisoformat(date_string)
return (
raw.astimezone(ZoneInfo(plugin_config.timezone))
if raw.tzinfo
else raw.replace(tzinfo=ZoneInfo(plugin_config.timezone))
)
def time_astimezone(time: time, tz: Optional[tzinfo] = None) -> time:
"""将 time 对象转换为指定时区的 time 对象
如果 tz 为 None则转换为本地时区
"""
local_time = datetime.combine(datetime.today(), time)
return local_time.astimezone(tz).timetz()
def get_time_fromisoformat_with_timezone(time_string: str) -> time:
"""从 iso8601 格式字符串中获取时间,并包含时区信息"""
if not plugin_config.timezone:
return time_astimezone(time.fromisoformat(time_string))
raw = time.fromisoformat(time_string)
return (
time_astimezone(raw, ZoneInfo(plugin_config.timezone))
if raw.tzinfo
else raw.replace(tzinfo=ZoneInfo(plugin_config.timezone))
)
def get_time_with_scheduler_timezone(time: time) -> time:
"""获取转换到 APScheduler 时区的时间"""
return time_astimezone(time, scheduler.timezone)
cache_path = get_cache_dir("nonebot_plugin_dialectlist")
# 暂时不做考虑
# def admin_permission():
@@ -154,7 +94,7 @@ def msg_counter(msg_list: List[MessageRecord]) -> Dict[str, int]:
return lst
def got_rank(msg_dict: Dict[str, int]) -> List[List[Union[str, int]]]:
def got_rank(msg_dict: Dict[str, int]) -> List:
"""### 获得排行榜
Args:
@@ -175,34 +115,37 @@ def got_rank(msg_dict: Dict[str, int]) -> List[List[Union[str, int]]]:
plugin_config.get_num, len(rank)
)
)
break
break
return rank
# def remove_control_characters(string: str) -> str:
# """### 将字符串中的控制符去除
def remove_control_characters(string: str) -> str:
"""### 将字符串中的控制符去除
# Args:
# string (str): 需要去除的字符串
Args:
string (str): 需要去除的字符串
# Returns:
# (str): 经过处理的字符串
# """
# return "".join(ch for ch in string if unicodedata.category(ch)[0] != "C")
Returns:
(str): 经过处理的字符串
"""
return "".join(ch for ch in string if unicodedata.category(ch)[0] != "C")
# async def render_template_pic(self) -> bytes:
# if plugin_config.dialectlist_visualization_type == "圆环图":
# view = pygal.Pie(inner_radius=0.6, style=style)
# elif plugin_config.dialectlist_visualization_type == "饼图":
# view = pygal.Pie(style=style)
# else:
# view = pygal.Bar(style=style)
# view.title = "消息可视化"
# for i, j in zip(self.rank, await self.get_nickname_list()): # type: ignore
# view.add(str(j), int(i[1]))
# png: bytes = view.render_to_png() # type: ignore
# self.img = png
# return png
async def get_rank_image(rank: List[UserRankInfo]) -> bytes:
for i in rank:
if i.user_avatar:
try:
user_avatar = i.user_avatar_bytes
except NotImplementedError:
user_avatar = open(os.path.dirname(os.path.abspath(__file__))+"/template/avatar/default.jpg", "rb").read()
# if not os.path.exists(cache_path / str(i.user_id)):
with open(cache_path / (str(i.user_id) + ".jpg"), "wb") as f:
f.write(user_avatar)
if plugin_config.template_path[:2] == './':
path = os.path.dirname(os.path.abspath(__file__)) + plugin_config.template_path[1:]
else:
path = plugin_config.template_path
path_dir, filename = os.path.split(path)
logger.debug(os.path.dirname(os.path.abspath(__file__)) + plugin_config.template_path[1:])
return await template_to_pic(path_dir,filename,{'users': rank, 'cache_path': cache_path, 'file_path': os.path.dirname(os.path.abspath(__file__))})