chenxu233/nonebot_plugin_dialectlist
1
0
Fork 0
mirror of https://github.com/ChenXu233/nonebot_plugin_dialectlist.git synced 2025-02-17 16:20:31 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

🔀 Merge branch 'master' of https://github.com/ChenXu233/nonebot_plugin_dialectlist

Browse Source
This commit is contained in:
Chen_Xu233 2024-06-27 22:48:47 +08:00
commit 155d2c6de0
7 changed files with 107 additions and 288 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
nonebot_plugin_dialectlist/__init__.py
View File

@ -56,7 +56,7 @@ from .utils import (
got_rank, got_rank,
msg_counter, msg_counter,
persist_id2user_id, persist_id2user_id,
parse_datetime parse_datetime,
) )
__plugin_meta__ = PluginMetadata( __plugin_meta__ = PluginMetadata(
@ -81,18 +81,20 @@ class SameTime(ArparmaBehavior):
if type is None and time: if type is None and time:
interface.behave_fail() interface.behave_fail()
def wrapper(slot: Union[int, str], content: Optional[str]) -> str: def wrapper(slot: Union[int, str], content: Optional[str]) -> str:
if slot == "type" and content: if slot == "type" and content:
return content return content
return "" # pragma: no cover return "" # pragma: no cover
rank_cmd = on_alconna( rank_cmd = on_alconna(
Alconna( Alconna(
"B话榜", "B话榜",
Args["type?", ["今日", "昨日", "本周", "上周", "本月", "上月", "年度", "历史"]][ Args["type?", ["今日", "昨日", "本周", "上周", "本月", "上月", "年度", "历史"]][
"time?", str, "time?",
"group_id?", int str,
], ]["group_id?", str],
behaviors=[SameTime()], behaviors=[SameTime()],
), ),
aliases={"废话榜"}, aliases={"废话榜"},
@ -110,6 +112,7 @@ rank_cmd.shortcut(
}, },
) )
# 这段函数完全抄的词云 # 这段函数完全抄的词云
@rank_cmd.handle() @rank_cmd.handle()
async def _group_message( async def _group_message(
@ -117,13 +120,15 @@ async def _group_message(
session: Session = Depends(extract_session), session: Session = Depends(extract_session),
type: Optional[str] = None, type: Optional[str] = None,
time: Optional[str] = None, time: Optional[str] = None,
group_id: Optional[int] = None, group_id: Optional[str] = None,
): ):
dt = get_datetime_now_with_timezone() dt = get_datetime_now_with_timezone()
if not group_id: if group_id is None:
state["group_id"] = session.id2 group_id = session.id2
state["group_id"] = group_id
if not type: if not type:
await rank_cmd.finish(__plugin_meta__.usage) await rank_cmd.finish(__plugin_meta__.usage)
@ -191,15 +196,25 @@ async def _group_message(
prompt="请输入你要查询的结束日期（如 2022-02-22）", prompt="请输入你要查询的结束日期（如 2022-02-22）",
parameterless=[Depends(parse_datetime("stop"))], parameterless=[Depends(parse_datetime("stop"))],
) )
@rank_cmd.got("group_id", prompt="请输入你要查询的群号。")
async def handle_rank( async def handle_rank(
bot: Bot, bot: Bot,
event: Event, event: Event,
session: Session = Depends(extract_session), session: Session = Depends(extract_session),
start: datetime = Arg(), start: datetime = Arg(),
stop: datetime = Arg(), stop: datetime = Arg(),
group_id: str = Arg(),
): ):
if group_id:
id = group_id
else:
id = session.id2
if not id:
await saa.Text("没有指定群哦").finish()
messages = await get_message_records( messages = await get_message_records(
session=session, id2s=[id],
id_type=SessionIdType.GROUP, id_type=SessionIdType.GROUP,
include_bot_id=False, include_bot_id=False,
include_bot_type=False, include_bot_type=False,
@ -240,10 +255,10 @@ async def handle_rank(
bar = Bar(init_opts=opts.InitOpts(theme=ThemeType.LIGHT)) bar = Bar(init_opts=opts.InitOpts(theme=ThemeType.LIGHT))
bar.add_xaxis(nicknames) bar.add_xaxis(nicknames)
bar.add_yaxis("B话数量", [i[1] for i in rank]) # type: ignore bar.add_yaxis("B话数量", [i[1] for i in rank]) # type: ignore
bar.render(str(get_cache_file("nonebot_plugin_dialectlist","cache.html"))) bar.render(str(get_cache_file("nonebot_plugin_dialectlist", "cache.html")))
with open(get_cache_file("nonebot_plugin_dialectlist","cache.html")) as f: with open(get_cache_file("nonebot_plugin_dialectlist", "cache.html")) as f:
a = f.read() a = f.read()
image = await html_to_pic(a,device_scale_factor=3.2) image = await html_to_pic(a, device_scale_factor=3.2)
await (saa.Text(string)+saa.Image(image)).finish(reply=True) await (saa.Text(string) + saa.Image(image)).finish(reply=True)

2
nonebot_plugin_dialectlist/render.py
View File

@ -1 +1 @@
# TODO 更好的图片渲染，支持自定义模板渲染。 # TODO 更好的图片渲染，支持自定义模板渲染。

BIN
nonebot_plugin_dialectlist/source/img/background.jpg Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 4.3 MiB

61
nonebot_plugin_dialectlist/source/rank.html Normal file
View File

@ -0,0 +1,61 @@
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<title>
在 title 里夹一点私货应该不会被发现吧？——ddl 真是太厉害啦！
</title>
<style>
.container {
position: relative;
width: "auto";
height: "auto";
}
.blur-image {
width: 100%;
height: 100%;
/*filter: blur(0.6px); /* 模糊程度可以根据需要进行调整 */
}
.container1 {
position: absolute;
}
.card {
position: absolute;
top: 1%;
left: 5%;
width: 20%;
height: auto;
padding: 2rem;
border-radius: 2rem;
background: rgba(255, 255, 255, .2);
-webkit-backdrop-filter: blur(10px);
backdrop-filter: blur(10px);
}
</style>
</head>
<body>
<div
class="container"
backdrop-filter: blur(50px);
>
<img src="./img/background.jpg" class="blur-image">
</div>
<div
class="container1"
backdrop-filter: blur(50px);
>
<p>
hhhhhhh
</p>
</div>
<div
class="card"
top: .5rem;
left: .5rem;
>
<p>nicknames：你好</p>
</div>
</body>
</html>

2
nonebot_plugin_dialectlist/storage.py
View File

@ -1 +1 @@
# TODO 使用计数缓存进行数据库查询优化，避免一次性查询过多消息导致内存爆炸。 # TODO 使用计数缓存进行数据库查询优化，避免一次性查询过多消息导致内存爆炸。

2
nonebot_plugin_dialectlist/time.py
View File

@ -1 +1 @@
# TODO 时间处理模块，用于处理时间相关操作。 # TODO 时间处理模块，用于处理时间相关操作。

287
nonebot_plugin_dialectlist/utils.py
View File

@ -19,9 +19,9 @@ from nonebot_plugin_chatrecorder import MessageRecord
from nonebot_plugin_alconna import AlconnaMatcher from nonebot_plugin_alconna import AlconnaMatcher
from .config import plugin_config from .config import plugin_config
def parse_datetime(key: str): def parse_datetime(key: str):
"""解析数字，并将结果存入 state 中""" """解析数字，并将结果存入 state 中"""
@ -117,6 +117,18 @@ async def persist_id2user_id(ids: List) -> List[str]:
return [i.id1 for i in records] return [i.id1 for i in records]
async def user_id2persist_id(id: str) -> int:
whereclause: List[ColumnElement[bool]] = []
whereclause.append(or_(*[SessionModel.id2 == id]))
statement = (
select(SessionModel).where(*whereclause)
# .join(SessionModel, SessionModel.id == MessageRecord.session_persist_id)
)
async with get_session() as db_session:
records = (await db_session.scalars(statement)).all()
return records[0].id
def msg_counter(msg_list: List[MessageRecord]) -> Dict[str, int]: def msg_counter(msg_list: List[MessageRecord]) -> Dict[str, int]:
"""### 计算每个人的消息量 """### 计算每个人的消息量
@ -132,13 +144,12 @@ def msg_counter(msg_list: List[MessageRecord]) -> Dict[str, int]:
logger.info("wow , there are {} msgs to count !!!".format(msg_len)) logger.info("wow , there are {} msgs to count !!!".format(msg_len))
for i in msg_list: for i in msg_list:
logger.debug(i.session_persist_id)
try: try:
lst[str(i.session_persist_id)] += 1 lst[str(i.session_persist_id)] += 1
except KeyError: except KeyError:
lst[str(i.session_persist_id)] = 1 lst[str(i.session_persist_id)] = 1
logger.debug(lst) logger.debug(f"finish counting, result is {lst}")
return lst return lst
@ -168,39 +179,6 @@ def got_rank(msg_dict: Dict[str, int]) -> List[List[Union[str, int]]]:
return rank return rank
# import abc
# import pygal
# import unicodedata
# import requests
# from datetime import datetime
# from typing import List, Dict, Union
# from pygal.style import Style
# from nonebot import require
# from nonebot.log import logger
# from nonebot.params import Arg
# from nonebot.typing import T_State
# from nonebot.matcher import Matcher
# from nonebot.adapters import Bot, Message
# from nonebot.adapters.onebot import V11Bot, V12Bot, V11Message, V12Message, V11MessageSegment, V12MessageSegment # type: ignore
# from nonebot.exception import ActionFailed
# try:
# from zoneinfo import ZoneInfo
# except ImportError:
# from backports.zoneinfo import ZoneInfo # type: ignore
# require("nonebot_plugin_chatrecorder")
# from nonebot_plugin_chatrecorder import get_message_records
# from nonebot_plugin_chatrecorder.model import MessageRecord
# from .config import plugin_config
# style = Style(font_family=plugin_config.dialectlist_font)
# def remove_control_characters(string: str) -> str: # def remove_control_characters(string: str) -> str:
# """### 将字符串中的控制符去除 # """### 将字符串中的控制符去除
@ -213,143 +191,6 @@ def got_rank(msg_dict: Dict[str, int]) -> List[List[Union[str, int]]]:
# return "".join(ch for ch in string if unicodedata.category(ch)[0] != "C") # return "".join(ch for ch in string if unicodedata.category(ch)[0] != "C")
# def parse_datetime(key: str):
# """解析数字，并将结果存入 state 中"""
# async def _key_parser(
# matcher: Matcher,
# state: T_State,
# input: Union[datetime, Union[V11Message, V12Message]] = 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:
# """从 iso8601 格式字符串中获取时间，并包含时区信息"""
# if plugin_config.timezone:
# return datetime.fromisoformat(date_string).astimezone(
# ZoneInfo(plugin_config.timezone)
# )
# else:
# return datetime.fromisoformat(date_string).astimezone()
# def msg_counter(msg_list: List[MessageRecord]) -> Dict[str, int]:
# """### 计算每个人的消息量
# Args:
# msg_list (list[MessageRecord]): 需要处理的消息列表
# Returns:
# (dict[str,int]): 处理后的消息数量字典,键为用户,值为消息数量
# """
# lst: Dict[str, int] = {}
# msg_len = len(msg_list)
# logger.info("wow , there are {} msgs to count !!!".format(msg_len))
# for i in msg_list:
# try:
# lst[i.user_id] += 1
# except KeyError:
# lst[i.user_id] = 1
# logger.debug(lst)
# return lst
# def got_rank(msg_dict: Dict[str, int]) -> List[List[Union[str, int]]]:
# """### 获得排行榜
# Args:
# msg_dict (Dict[str,int]): 要处理的字典
# Returns:
# List[Tuple[str,int]]: 排行榜列表(已排序)
# """
# rank = []
# while len(rank) < plugin_config.dialectlist_get_num:
# try:
# max_key = max(msg_dict.items(), key=lambda x: x[1])
# rank.append(list(max_key))
# msg_dict.pop(max_key[0])
# except ValueError:
# rank.append(["null", 0])
# continue
# return rank
# class MsgProcesser(abc.ABC):
# def __init__(self, bot: Bot, gid: str, msg_list: List[MessageRecord]) -> None:
# if isinstance(bot, Bot):
# self.bot = bot
# else:
# self.bot = None
# self.gid = gid
# self.rank = got_rank(msg_counter(msg_list))
# @abc.abstractmethod
# async def get_nickname_list(self) -> List:
# """
# ### 获得昵称
# #### 抽象原因
# 要对onebot协议不同版本进行适配
# """
# raise NotImplementedError
# @abc.abstractmethod
# def get_head_portrait_urls(self) -> List:
# raise NotImplementedError
# @abc.abstractmethod
# async def get_send_msg(self) -> Message:
# raise NotImplementedError
# async def get_msg(self) -> List[Union[str, bytes, None]]:
# str_msg = await self.render_template_msg()
# pic_msg = None
# if plugin_config.dialectlist_visualization:
# try:
# pic_msg = await self.render_template_pic()
# except OSError:
# plugin_config.dialectlist_visualization = False
# str_msg += "\n\n无法发送可视化图片，请检查是否安装GTK+，详细安装教程可见github\nhttps://github.com/tschoonj/GTK-for-Windows-Runtime-Environment-Installer \n若不想安装这个软件，再次使用这个指令不会显示这个提示"
# return [str_msg, pic_msg]
# async def render_template_msg(self) -> str:
# """渲染文字"""
# string: str = ""
# rank: List = self.rank
# nicknames: List = await self.get_nickname_list()
# for i in range(len(rank)):
# index = i + 1
# nickname, chatdatanum = nicknames[i], rank[i][1]
# str_example = plugin_config.dialectlist_string_format.format(
# index=index, nickname=nickname, chatdatanum=chatdatanum
# )
# string += str_example
# return string
# async def render_template_pic(self) -> bytes: # async def render_template_pic(self) -> bytes:
# if plugin_config.dialectlist_visualization_type == "圆环图": # if plugin_config.dialectlist_visualization_type == "圆环图":
# view = pygal.Pie(inner_radius=0.6, style=style) # view = pygal.Pie(inner_radius=0.6, style=style)
@ -364,102 +205,4 @@ def got_rank(msg_dict: Dict[str, int]) -> List[List[Union[str, int]]]:
# png: bytes = view.render_to_png() # type: ignore # png: bytes = view.render_to_png() # type: ignore
# self.img = png # self.img = png
# return png # return png
# class V11GroupMsgProcesser(MsgProcesser):
# def __init__(self, bot: V11Bot, gid: str, msg_list: List[MessageRecord]) -> None:
# super().__init__(bot, gid, msg_list)
# self.bot = bot
# async def get_nickname_list(self) -> List:
# nicknames = []
# for i in range(len(self.rank)):
# try:
# member_info = await self.bot.get_group_member_info(
# group_id=int(self.gid), user_id=int(self.rank[i][0]), no_cache=True
# )
# nickname=(
# member_info["nickname"]
# if not member_info["card"]
# else member_info["card"]
# )
# nicknames.append(remove_control_characters(nickname))
# except (ActionFailed,ValueError) as e:
# nicknames.append("{}这家伙不在群里了".format(self.rank[i][0]))
# return nicknames
# def get_head_portrait_urls(self) -> List:
# self.portrait_urls = [
# "http://q2.qlogo.cn/headimg_dl?dst_uin={}&spec=640".format(i[0])
# for i in self.rank
# ]
# return self.portrait_urls
# async def get_send_msg(self) -> V11Message:
# msgs: List = await self.get_msg()
# msg = V11Message()
# msg += V11MessageSegment.text(msgs[0]) # type: ignore
# msg += V11MessageSegment.image(msgs[1]) # type: ignore
# return msg
# class V12MsgProcesser(MsgProcesser):
# def __init__(self, bot: V12Bot, gid: str, msg_list: List[MessageRecord]) -> None:
# super().__init__(bot, gid, msg_list)
# self.bot = bot
# async def get_send_msg(self) -> V12Message:
# msgs: List = await self.get_msg()
# msg = V12Message()
# msg += V12MessageSegment.text(msgs[0]) # type: ignore
# msg += V12MessageSegment.image(msgs[1]) # type: ignore
# return msg
# def get_head_portrait_urls(self) -> List:
# return super().get_head_portrait_urls()
# class V12GroupMsgProcesser(V12MsgProcesser):
# def __init__(self, bot: V12Bot, gid: str, msg_list: List[MessageRecord]) -> None:
# super().__init__(bot, gid, msg_list)
# async def get_nickname_list(self) -> List:
# nicknames = []
# for i in range(len(self.rank)):
# try:
# member_info = await self.bot.get_group_member_info(
# group_id=str(self.gid), user_id=str(self.rank[i][0]), no_cache=True
# )
# nickname=(
# member_info["user_displayname"]
# if member_info["user_displayname"]
# else member_info["user_name"]
# )
# nicknames.append(remove_control_characters(nickname))
# except ActionFailed as e:
# nicknames.append("{}这家伙不在群里了".format(self.rank[i][0]))
# return nicknames
# class V12GuildMsgProcesser(V12MsgProcesser):
# def __init__(self, bot: V12Bot, gid: str, msg_list: List[MessageRecord]) -> None:
# super().__init__(bot, gid, msg_list)
# async def get_nickname_list(self) -> List:
# nicknames = []
# for i in range(len(self.rank)):
# try:
# member_info = await self.bot.get_guild_member_info(
# guild_id=str(self.gid), user_id=str(self.rank[i][0]), no_cache=True
# )
# nickname=(
# member_info["user_displayname"]
# if member_info["user_displayname"]
# else member_info["user_name"]
# )
# nicknames.append(remove_control_characters(nickname))
# except ActionFailed as e:
# nicknames.append("{}这家伙不在群里了".format(self.rank[i][0]))
# return nicknames