Compare commits
24 Commits
v0.9.22
...
fx_storage
| Author | SHA1 | Date | |
|---|---|---|---|
| 54fae88914 | |||
| eed21e6223 | |||
| bf5c10b7a7 | |||
| 274ca0fa9a | |||
| c72cdd6a6b | |||
|
16b0451133
|
|||
|
cb34813c4b
|
|||
| 2de3be271e | |||
| f7d2168dac | |||
| 40be5ce335 | |||
| 8e6131473d | |||
|
26e10be4ec
|
|||
|
78bda5fc0a
|
|||
|
97658a6c56
|
|||
|
3fedc685a9
|
|||
|
d1a3e44c45
|
|||
|
f637778173
|
|||
|
145bfedf67
|
|||
|
61b9d733a5
|
|||
| ae59c20e2f | |||
| 0b7d21aeb0 | |||
|
d6ede3e6cd
|
|||
|
07ace8e6e9
|
|||
|
6f08c22b5b
|
6
.gitignore
vendored
6
.gitignore
vendored
@ -1,5 +1,11 @@
|
||||
# 基本的数据文件,以及环境用文件
|
||||
/.env
|
||||
/data
|
||||
/pyrightconfig.json
|
||||
/pyrightconfig.toml
|
||||
|
||||
# 缓存文件
|
||||
__pycache__
|
||||
|
||||
# 可能会偶然生成的 diff 文件
|
||||
/*.diff
|
||||
|
||||
90
konabot/common/apis/ali_content_safety.py
Normal file
90
konabot/common/apis/ali_content_safety.py
Normal file
@ -0,0 +1,90 @@
|
||||
import asyncio
|
||||
import json
|
||||
|
||||
from alibabacloud_green20220302.client import Client as AlibabaGreenClient
|
||||
from alibabacloud_green20220302.models import TextModerationPlusRequest
|
||||
from alibabacloud_tea_openapi.models import Config as AlibabaTeaConfig
|
||||
from loguru import logger
|
||||
from pydantic import BaseModel
|
||||
|
||||
import nonebot
|
||||
|
||||
|
||||
class AlibabaGreenPluginConfig(BaseModel):
|
||||
module_aligreen_enable: bool = False
|
||||
module_aligreen_access_key_id: str = ""
|
||||
module_aligreen_access_key_secret: str = ""
|
||||
module_aligreen_region_id: str = "cn-shenzhen"
|
||||
module_aligreen_endpoint: str = "green-cip.cn-shenzhen.aliyuncs.com"
|
||||
module_aligreen_service: str = "llm_query_moderation"
|
||||
|
||||
|
||||
class AlibabaGreen:
|
||||
_client: AlibabaGreenClient | None = None
|
||||
_config: AlibabaGreenPluginConfig | None = None
|
||||
|
||||
@staticmethod
|
||||
def get_client() -> AlibabaGreenClient:
|
||||
assert AlibabaGreen._client is not None
|
||||
return AlibabaGreen._client
|
||||
|
||||
@staticmethod
|
||||
def get_config() -> AlibabaGreenPluginConfig:
|
||||
assert AlibabaGreen._config is not None
|
||||
return AlibabaGreen._config
|
||||
|
||||
@staticmethod
|
||||
def init():
|
||||
config = nonebot.get_plugin_config(AlibabaGreenPluginConfig)
|
||||
AlibabaGreen._config = config
|
||||
if not config.module_aligreen_enable:
|
||||
logger.info("该环境未启用阿里内容审查,跳过初始化")
|
||||
return
|
||||
AlibabaGreen._client = AlibabaGreenClient(AlibabaTeaConfig(
|
||||
access_key_id=config.module_aligreen_access_key_id,
|
||||
access_key_secret=config.module_aligreen_access_key_secret,
|
||||
connect_timeout=10000,
|
||||
read_timeout=3000,
|
||||
region_id=config.module_aligreen_region_id,
|
||||
endpoint=config.module_aligreen_endpoint,
|
||||
))
|
||||
|
||||
@staticmethod
|
||||
def _detect_sync(content: str) -> bool:
|
||||
if not AlibabaGreen.get_config().module_aligreen_enable:
|
||||
logger.debug("该环境未启用阿里内容审查,直接跳过")
|
||||
return True
|
||||
|
||||
client = AlibabaGreen.get_client()
|
||||
try:
|
||||
response = client.text_moderation_plus(TextModerationPlusRequest(
|
||||
service=AlibabaGreen.get_config().module_aligreen_service,
|
||||
service_parameters=json.dumps({
|
||||
"content": content,
|
||||
}),
|
||||
))
|
||||
if response.status_code == 200:
|
||||
result = response.body
|
||||
logger.info(f"检测违规内容 API 调用成功:{result}")
|
||||
risk_level: str = result.data.risk_level or "none"
|
||||
if risk_level == "high":
|
||||
return False
|
||||
return True
|
||||
logger.error(f"检测违规内容 API 调用失败:{response}")
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error("检测违规内容 API 调用失败")
|
||||
logger.exception(e)
|
||||
return True
|
||||
|
||||
@staticmethod
|
||||
async def detect(content: str) -> bool:
|
||||
return await asyncio.to_thread(AlibabaGreen._detect_sync, content)
|
||||
|
||||
|
||||
driver = nonebot.get_driver()
|
||||
|
||||
@driver.on_startup
|
||||
async def _():
|
||||
AlibabaGreen.init()
|
||||
|
||||
@ -59,6 +59,9 @@ def get_llm(llm_model: str | None = None):
|
||||
if llm_model is None:
|
||||
llm_model = llm_config.default_llm
|
||||
if llm_model not in llm_config.llms:
|
||||
raise NotImplementedError("LLM 未配置,该功能无法使用")
|
||||
if llm_config.default_llm in llm_config.llms:
|
||||
logger.warning(f"[LLM] 需求的 LLM 不存在,回退到默认模型 REQUIRED={llm_model}")
|
||||
return llm_config.llms[llm_config.default_llm]
|
||||
raise NotImplementedError("[LLM] LLM 未配置,该功能无法使用")
|
||||
return llm_config.llms[llm_model]
|
||||
|
||||
|
||||
@ -208,5 +208,8 @@ async def _ext_img(
|
||||
return None
|
||||
|
||||
|
||||
|
||||
DepImageBytes = Annotated[bytes, nonebot.params.Depends(_ext_img_data)]
|
||||
DepPILImage = Annotated[PIL.Image.Image, nonebot.params.Depends(_ext_img)]
|
||||
|
||||
DepImageBytesOrNone = Annotated[bytes | None, nonebot.params.Depends(_ext_img_data)]
|
||||
|
||||
3
konabot/common/ptimeparse/README.md
Normal file
3
konabot/common/ptimeparse/README.md
Normal file
@ -0,0 +1,3 @@
|
||||
# 已废弃
|
||||
|
||||
坏枪用简单的 LLM + 提示词工程,完成了这 200 块的 `qwen3-coder-plus` 都搞不定的 nb 功能
|
||||
@ -9,7 +9,6 @@ import datetime
|
||||
from typing import Optional
|
||||
|
||||
from .expression import TimeExpression
|
||||
from .err import TokenUnhandledException, MultipleSpecificationException
|
||||
|
||||
|
||||
def parse(text: str, now: Optional[datetime.datetime] = None) -> datetime.datetime:
|
||||
@ -56,19 +55,4 @@ class Parser:
|
||||
TokenUnhandledException: If the input cannot be parsed
|
||||
"""
|
||||
return TimeExpression.parse(text, self.now)
|
||||
|
||||
def digest_chinese_number(self, text: str) -> tuple[str, int]:
|
||||
"""
|
||||
Parse a Chinese number from the beginning of text and return the rest and the parsed number.
|
||||
|
||||
This matches the interface of the original digest_chinese_number method.
|
||||
|
||||
Args:
|
||||
text: Text that may start with a Chinese number
|
||||
|
||||
Returns:
|
||||
Tuple of (remaining_text, parsed_number)
|
||||
"""
|
||||
from .chinese_number import ChineseNumberParser
|
||||
parser = ChineseNumberParser()
|
||||
return parser.digest(text)
|
||||
|
||||
|
||||
@ -2,10 +2,9 @@
|
||||
Abstract Syntax Tree (AST) nodes for the time expression parser.
|
||||
"""
|
||||
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import Optional, List
|
||||
from abc import ABC
|
||||
from typing import Optional
|
||||
from dataclasses import dataclass
|
||||
import datetime
|
||||
|
||||
|
||||
@dataclass
|
||||
@ -69,4 +68,4 @@ class TimeExpressionNode(ASTNode):
|
||||
time: Optional[TimeNode] = None
|
||||
relative_date: Optional[RelativeDateNode] = None
|
||||
relative_time: Optional[RelativeTimeNode] = None
|
||||
weekday: Optional[WeekdayNode] = None
|
||||
weekday: Optional[WeekdayNode] = None
|
||||
|
||||
@ -10,7 +10,7 @@ from .ptime_ast import (
|
||||
TimeExpressionNode, DateNode, TimeNode,
|
||||
RelativeDateNode, RelativeTimeNode, WeekdayNode, NumberNode
|
||||
)
|
||||
from .err import TokenUnhandledException, MultipleSpecificationException
|
||||
from .err import TokenUnhandledException
|
||||
|
||||
|
||||
class SemanticAnalyzer:
|
||||
@ -366,4 +366,4 @@ class SemanticAnalyzer:
|
||||
smart_time = self.infer_smart_time(time.hour, time.minute, time.second, base_time=result)
|
||||
result = smart_time
|
||||
|
||||
return result
|
||||
return result
|
||||
|
||||
79
konabot/docs/user/fx.txt
Normal file
79
konabot/docs/user/fx.txt
Normal file
@ -0,0 +1,79 @@
|
||||
## 指令介绍
|
||||
|
||||
`fx` - 用于对图片应用各种滤镜效果的指令
|
||||
|
||||
## 格式
|
||||
|
||||
```
|
||||
fx [滤镜名称] <参数1> <参数2> ...
|
||||
```
|
||||
|
||||
## 示例
|
||||
|
||||
- `fx 模糊`
|
||||
- `fx 阈值 150`
|
||||
- `fx 缩放 2.0`
|
||||
- `fx 色彩 1.8`
|
||||
- `fx 色键 rgb(0,255,0) 50`
|
||||
|
||||
## 可用滤镜列表
|
||||
|
||||
### 基础滤镜
|
||||
* ```fx 轮廓```
|
||||
* ```fx 锐化```
|
||||
* ```fx 边缘增强```
|
||||
* ```fx 浮雕```
|
||||
* ```fx 查找边缘```
|
||||
* ```fx 平滑```
|
||||
* ```fx 暗角 <半径=1.5>```
|
||||
* ```fx 发光 <强度=0.5> <模糊半径=15>```
|
||||
* ```fx 噪点 <数量=0.05>```
|
||||
* ```fx 素描```
|
||||
* ```fx 阴影 <x偏移量=10> <y偏移量=10> <模糊量=10> <不透明度=0.5> <阴影颜色=black>```
|
||||
|
||||
### 模糊滤镜
|
||||
* ```fx 模糊 <半径=10>```
|
||||
* ```fx 马赛克 <像素大小=10>```
|
||||
* ```fx 径向模糊 <强度=3.0> <采样量=6>```
|
||||
* ```fx 旋转模糊 <强度=30.0> <采样量=6>```
|
||||
* ```fx 方向模糊 <角度=0.0> <距离=20> <采样量=6>```
|
||||
* ```fx 缩放模糊 <强度=0.1> <采样量=6>```
|
||||
* ```fx 边缘模糊 <半径=10.0>```
|
||||
|
||||
### 色彩处理滤镜
|
||||
* ```fx 反色```
|
||||
* ```fx 黑白```
|
||||
* ```fx 阈值 <阈值=128>```
|
||||
* ```fx 对比度 <因子=1.5>```
|
||||
* ```fx 亮度 <因子=1.5>```
|
||||
* ```fx 色彩 <因子=1.5>```
|
||||
* ```fx 色调 <颜色="rgb(255,0,0)">```
|
||||
* ```fx RGB分离 <偏移量=5>```
|
||||
* ```fx 叠加颜色 <颜色列表=[rgb(255,0,0)|(0,0),rgb(0,255,0)|(0,100),rgb(0,0,255)|(50,100)]> <叠加模式=overlay>```
|
||||
* ```fx 像素抖动 <最大偏移量=2>```
|
||||
|
||||
### 几何变换滤镜
|
||||
* ```fx 平移 <x偏移量=10> <y偏移量=10>```
|
||||
* ```fx 缩放 <比例(X)=1.5> <比例Y=None>```
|
||||
* ```fx 旋转 <角度=45>```
|
||||
* ```fx 透视变换 <变换矩阵>```
|
||||
* ```fx 裁剪 <左=0> <上=0> <右=100> <下=100>(百分比)```
|
||||
* ```fx 拓展边缘 <拓展量=10>```
|
||||
* ```fx 波纹 <振幅=5> <波长=20>```
|
||||
* ```fx 光学补偿 <数量=100> <反转=false>```
|
||||
* ```fx 球面化 <强度=0.5>```
|
||||
* ```fx 镜像 <角度=90>```
|
||||
* ```fx 水平翻转```
|
||||
* ```fx 垂直翻转```
|
||||
* ```fx 复制 <目标位置=(100,100)> <缩放=1.0> <源区域=(0,0,100,100)>(百分比)```
|
||||
|
||||
### 特殊效果滤镜
|
||||
* ```fx 色键 <目标颜色="rgb(255,0,0)"> <容差=60>```
|
||||
* ```fx 晃动 <最大偏移量=5> <运动模糊=False>```
|
||||
* ```fx 动图 <帧率=10>```
|
||||
|
||||
## 颜色名称支持
|
||||
- **基本颜色**:红、绿、蓝、黄、紫、黑、白、橙、粉、灰、青、靛、棕
|
||||
- **修饰词**:浅、深、亮、暗(可组合使用,如`浅红`、`深蓝`)
|
||||
- **RGB格式**:`rgb(255,0,0)`、`rgb(0,255,0)`、`(255,0,0)` 等
|
||||
- **HEX格式**:`#66ccff`等
|
||||
24
konabot/docs/user/tqszm.txt
Normal file
24
konabot/docs/user/tqszm.txt
Normal file
@ -0,0 +1,24 @@
|
||||
# tqszm
|
||||
|
||||
引用一条消息,让此方帮你提取首字母。
|
||||
|
||||
例子:
|
||||
|
||||
```
|
||||
John: 11-28 16:50:37
|
||||
谁来总结一下今天的工作?
|
||||
|
||||
Jack: 11-28 16:50:55
|
||||
[引用John的消息] @此方Bot tqszm
|
||||
|
||||
此方Bot: 11-28 16:50:56
|
||||
slzjyxjtdgz?
|
||||
```
|
||||
|
||||
或者,你也可以直接以正常指令的方式调用:
|
||||
|
||||
```
|
||||
@此方Bot 提取首字母 中山大学
|
||||
> zsdx
|
||||
```
|
||||
|
||||
@ -6,34 +6,25 @@ from nonebot_plugin_alconna import Reference, Reply, UniMsg
|
||||
from nonebot.adapters import Event
|
||||
|
||||
|
||||
matcher_fix = on_message()
|
||||
|
||||
pattern = (
|
||||
r"^(?:(?:av|cv)\d+|BV[a-zA-Z0-9]{10})|"
|
||||
r"(?:b23\.tv|bili(?:22|23|33|2233)\.cn|\.bilibili\.com|QQ小程序(?:&#93;|]|\])哔哩哔哩).{0,500}"
|
||||
)
|
||||
|
||||
|
||||
@matcher_fix.handle()
|
||||
async def _(msg: UniMsg, event: Event):
|
||||
def _rule(msg: UniMsg):
|
||||
to_search = msg.exclude(Reply, Reference).dump(json=True)
|
||||
to_search2 = msg.exclude(Reply, Reference).extract_plain_text()
|
||||
if not re.search(pattern, to_search) and not re.search(pattern, to_search2):
|
||||
return
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
matcher_fix = on_message(rule=_rule)
|
||||
|
||||
@matcher_fix.handle()
|
||||
async def _(event: Event):
|
||||
from nonebot_plugin_analysis_bilibili import handle_analysis
|
||||
|
||||
await handle_analysis(event)
|
||||
|
||||
# b_url: str
|
||||
# b_page: str | None
|
||||
# b_time: str | None
|
||||
#
|
||||
# from nonebot_plugin_analysis_bilibili.analysis_bilibili import extract as bilibili_extract
|
||||
#
|
||||
# b_url, b_page, b_time = bilibili_extract(to_search)
|
||||
# if b_url is None:
|
||||
# return
|
||||
#
|
||||
# await matcher_fix.send(await UniMessage().text(b_url).export())
|
||||
|
||||
|
||||
215
konabot/plugins/fx_process/__init__.py
Normal file
215
konabot/plugins/fx_process/__init__.py
Normal file
@ -0,0 +1,215 @@
|
||||
import asyncio as asynkio
|
||||
from dataclasses import dataclass
|
||||
from io import BytesIO
|
||||
|
||||
from inspect import signature
|
||||
|
||||
from konabot.common.longtask import DepLongTaskTarget
|
||||
from konabot.common.nb.extract_image import DepImageBytesOrNone
|
||||
from nonebot.adapters import Event as BaseEvent
|
||||
from nonebot import on_message, logger
|
||||
from returns.result import Failure, Result, Success
|
||||
|
||||
from nonebot_plugin_alconna import (
|
||||
UniMessage,
|
||||
UniMsg
|
||||
)
|
||||
|
||||
from konabot.plugins.fx_process.fx_manager import ImageFilterManager
|
||||
|
||||
from PIL import Image, ImageSequence
|
||||
|
||||
from konabot.plugins.fx_process.types import FilterItem, ImageRequireSignal, ImagesListRequireSignal, SenderInfo
|
||||
|
||||
def prase_input_args(input_str: str, sender_info: SenderInfo = None) -> list[FilterItem]:
|
||||
# 按分号或换行符分割参数
|
||||
args = []
|
||||
for part in input_str.replace('\n', ';').split(';'):
|
||||
part = part.strip()
|
||||
if not part:
|
||||
continue
|
||||
split_part = part.split()
|
||||
filter_name = split_part[0]
|
||||
if not ImageFilterManager.has_filter(filter_name):
|
||||
continue
|
||||
filter_func = ImageFilterManager.get_filter(filter_name)
|
||||
input_filter_args = split_part[1:]
|
||||
# 获取函数最大参数数量
|
||||
sig = signature(filter_func)
|
||||
max_params = len(sig.parameters) - 1 # 减去第一个参数 image
|
||||
# 从 args 提取参数,并转换为适当类型
|
||||
func_args = []
|
||||
for i in range(0, min(len(input_filter_args), max_params)):
|
||||
# 尝试将参数转换为函数签名中对应的类型,并检测是不是 Image 类型,如果有则表示多个图像输入
|
||||
param = list(sig.parameters.values())[i + 1]
|
||||
param_type = param.annotation
|
||||
arg_value = input_filter_args[i]
|
||||
try:
|
||||
if param_type is float:
|
||||
converted_value = float(arg_value)
|
||||
elif param_type is int:
|
||||
converted_value = int(arg_value)
|
||||
elif param_type is bool:
|
||||
converted_value = arg_value.lower() in ['true', '1', 'yes', '是', '开']
|
||||
elif param_type is Image.Image:
|
||||
converted_value = ImageRequireSignal()
|
||||
elif param_type is SenderInfo:
|
||||
converted_value = sender_info
|
||||
elif param_type is list[Image.Image]:
|
||||
converted_value = ImagesListRequireSignal()
|
||||
else:
|
||||
converted_value = arg_value
|
||||
except Exception:
|
||||
converted_value = arg_value
|
||||
func_args.append(converted_value)
|
||||
args.append(FilterItem(name=filter_name,filter=filter_func, args=func_args))
|
||||
return args
|
||||
|
||||
def handle_filters_to_image(images: list[Image.Image], filters: list[FilterItem]) -> Image.Image:
|
||||
for filter_item in filters:
|
||||
filter_func = filter_item.filter
|
||||
func_args = filter_item.args
|
||||
# 检测参数中是否有 ImageRequireSignal,如果有则传入对应数量的图像列表
|
||||
if any(isinstance(arg, ImageRequireSignal) for arg in func_args):
|
||||
# 替换 ImageRequireSignal 为 images 对应索引的图像
|
||||
actual_args = []
|
||||
img_signal_count = 1 # 从 images[1] 开始取图像
|
||||
for arg in func_args:
|
||||
if isinstance(arg, ImageRequireSignal):
|
||||
actual_args.append(images[img_signal_count])
|
||||
img_signal_count += 1
|
||||
else:
|
||||
actual_args.append(arg)
|
||||
func_args = actual_args
|
||||
# 检测参数中是否有 ImagesListRequireSignal,如果有则传入整个图像列表
|
||||
if any(isinstance(arg, ImagesListRequireSignal) for arg in func_args):
|
||||
actual_args = []
|
||||
for arg in func_args:
|
||||
if isinstance(arg, ImagesListRequireSignal):
|
||||
actual_args.append(images)
|
||||
else:
|
||||
actual_args.append(arg)
|
||||
func_args = actual_args
|
||||
|
||||
images[0] = filter_func(images[0], *func_args)
|
||||
return images[0]
|
||||
|
||||
async def apply_filters_to_images(images: list[Image.Image], filters: list[FilterItem]) -> BytesIO:
|
||||
# 如果第一项是“加载图像”参数,那么就加载图像
|
||||
if filters and filters[0].name == "加载图像":
|
||||
load_filter = filters.pop(0)
|
||||
# 加载全部路径
|
||||
for path in load_filter.args:
|
||||
img = Image.open(path)
|
||||
images.append(img)
|
||||
|
||||
if len(images) <= 0:
|
||||
raise ValueError("没有提供任何图像进行处理")
|
||||
|
||||
# 检测是否需要将静态图视作动图处理
|
||||
frozen_to_move = any(
|
||||
filter_item.name == "动图"
|
||||
for filter_item in filters
|
||||
)
|
||||
static_fps = 10
|
||||
# 找到动图参数 fps
|
||||
if frozen_to_move:
|
||||
for filter_item in filters:
|
||||
if filter_item.name == "动图" and filter_item.args:
|
||||
try:
|
||||
static_fps = int(filter_item.args[0])
|
||||
except Exception:
|
||||
static_fps = 10
|
||||
break
|
||||
# 如果 image 是动图,则逐帧处理
|
||||
img = images[0]
|
||||
logger.debug("开始图像处理")
|
||||
output = BytesIO()
|
||||
if getattr(img, "is_animated", False) or frozen_to_move:
|
||||
frames = []
|
||||
all_frames = []
|
||||
if getattr(img, "is_animated", False):
|
||||
logger.debug("处理动图帧")
|
||||
for frame in ImageSequence.Iterator(img):
|
||||
frame_copy = frame.copy()
|
||||
all_frames.append(frame_copy)
|
||||
else:
|
||||
# 将静态图视作单帧动图处理,拷贝多份
|
||||
logger.debug("处理静态图为多帧动图")
|
||||
for _ in range(10): # 默认复制10帧
|
||||
all_frames.append(img.copy())
|
||||
img.info['duration'] = int(1000 / static_fps)
|
||||
|
||||
async def process_single_frame(frame: list[Image.Image], frame_idx: int) -> Image.Image:
|
||||
"""处理单帧的异步函数"""
|
||||
logger.debug(f"开始处理帧 {frame_idx}")
|
||||
result = await asynkio.to_thread(handle_filters_to_image, frame, images, filters)
|
||||
logger.debug(f"完成处理帧 {frame_idx}")
|
||||
return result[0]
|
||||
# 并发处理所有帧
|
||||
tasks = []
|
||||
for i, frame in enumerate(all_frames):
|
||||
task = process_single_frame(frame, i)
|
||||
tasks.append(task)
|
||||
|
||||
frames = await asynkio.gather(*tasks, return_exceptions=True)
|
||||
|
||||
# 检查是否有处理失败的帧
|
||||
for i, result in enumerate(frames):
|
||||
if isinstance(result, Exception):
|
||||
logger.error(f"帧 {i} 处理失败: {result}")
|
||||
# 使用原始帧作为回退
|
||||
frames[i] = all_frames[i]
|
||||
|
||||
logger.debug("保存动图")
|
||||
frames[0].save(
|
||||
output,
|
||||
format="GIF",
|
||||
save_all=True,
|
||||
append_images=frames[1:],
|
||||
loop=img.info.get("loop", 0),
|
||||
disposal=img.info.get("disposal", 2),
|
||||
duration=img.info.get("duration", 100),
|
||||
)
|
||||
logger.debug("Animated image saved")
|
||||
else:
|
||||
img = handle_filters_to_image(images=images, filters=filters)
|
||||
img.save(output, format="PNG")
|
||||
logger.debug("Image processing completed")
|
||||
output.seek(0)
|
||||
return output
|
||||
|
||||
def is_fx_mentioned(evt: BaseEvent, msg: UniMsg) -> bool:
|
||||
txt = msg.extract_plain_text()
|
||||
if "fx" not in txt[:3].lower():
|
||||
return False
|
||||
return True
|
||||
|
||||
fx_on = on_message(rule=is_fx_mentioned)
|
||||
|
||||
@fx_on.handle()
|
||||
async def _(msg: UniMsg, event: BaseEvent, target: DepLongTaskTarget, image_data: DepImageBytesOrNone = None):
|
||||
preload_imgs = []
|
||||
# 提取图像
|
||||
try:
|
||||
if image_data is not None:
|
||||
preload_imgs.append(Image.open(BytesIO(image_data)))
|
||||
logger.debug("Image extracted for FX processing.")
|
||||
except Exception:
|
||||
logger.info("No image found in message for FX processing.")
|
||||
args = msg.extract_plain_text().split()
|
||||
if len(args) < 2:
|
||||
return
|
||||
|
||||
sender_info = SenderInfo(
|
||||
group_id=target.channel_id,
|
||||
qq_id=target.target_id
|
||||
)
|
||||
|
||||
filters = prase_input_args(msg.extract_plain_text()[2:], sender_info=sender_info)
|
||||
if not filters:
|
||||
return
|
||||
output = await apply_filters_to_images(preload_imgs, filters)
|
||||
logger.debug("FX processing completed, sending result.")
|
||||
await fx_on.send(await UniMessage().image(raw=output).export())
|
||||
|
||||
50
konabot/plugins/fx_process/color_handle.py
Normal file
50
konabot/plugins/fx_process/color_handle.py
Normal file
@ -0,0 +1,50 @@
|
||||
from typing import Optional
|
||||
from PIL import ImageColor
|
||||
|
||||
class ColorHandle:
|
||||
color_name_map = {
|
||||
"红": (255, 0, 0),
|
||||
"绿": (0, 255, 0),
|
||||
"蓝": (0, 0, 255),
|
||||
"黄": (255, 255, 0),
|
||||
"紫": (128, 0, 128),
|
||||
"黑": (0, 0, 0),
|
||||
"白": (255, 255, 255),
|
||||
"橙": (255, 165, 0),
|
||||
"粉": (255, 192, 203),
|
||||
"灰": (128, 128, 128),
|
||||
"青": (0, 255, 255),
|
||||
"靛": (75, 0, 130),
|
||||
"棕": (165, 42, 42),
|
||||
"浅": (200, 200, 200),
|
||||
"深": (50, 50, 50),
|
||||
"亮": (255, 255, 224),
|
||||
"暗": (47, 79, 79),
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def set_or_blend_color(ori_color: Optional[tuple], target_color: tuple) -> tuple:
|
||||
# 如果没有指定初始颜色,返回目标颜色
|
||||
if ori_color is None:
|
||||
return target_color
|
||||
# 混合颜色,取平均值
|
||||
blended_color = tuple((o + t) // 2 for o, t in zip(ori_color, target_color))
|
||||
return blended_color
|
||||
|
||||
@staticmethod
|
||||
def parse_color(color_str: str) -> tuple:
|
||||
# 如果是纯括号,则加上前缀 rgb
|
||||
if color_str.startswith('(') and color_str.endswith(')'):
|
||||
color_str = 'rgb' + color_str
|
||||
try:
|
||||
return ImageColor.getrgb(color_str)
|
||||
except ValueError:
|
||||
pass
|
||||
base_color = None
|
||||
color_str = color_str.replace('色', '')
|
||||
for name, rgb in ColorHandle.color_name_map.items():
|
||||
if name in color_str:
|
||||
base_color = ColorHandle.set_or_blend_color(base_color, rgb)
|
||||
if base_color is not None:
|
||||
return base_color
|
||||
return (255, 255, 255) # 默认白色
|
||||
1111
konabot/plugins/fx_process/fx_handle.py
Normal file
1111
konabot/plugins/fx_process/fx_handle.py
Normal file
File diff suppressed because it is too large
Load Diff
63
konabot/plugins/fx_process/fx_manager.py
Normal file
63
konabot/plugins/fx_process/fx_manager.py
Normal file
@ -0,0 +1,63 @@
|
||||
from typing import Optional
|
||||
from konabot.plugins.fx_process.fx_handle import ImageFilterEmpty, ImageFilterImplement, ImageFilterStorage
|
||||
|
||||
class ImageFilterManager:
|
||||
filter_map = {
|
||||
"模糊": ImageFilterImplement.apply_blur,
|
||||
"马赛克": ImageFilterImplement.apply_mosaic,
|
||||
"轮廓": ImageFilterImplement.apply_contour,
|
||||
"锐化": ImageFilterImplement.apply_sharpen,
|
||||
"边缘增强": ImageFilterImplement.apply_edge_enhance,
|
||||
"浮雕": ImageFilterImplement.apply_emboss,
|
||||
"查找边缘": ImageFilterImplement.apply_find_edges,
|
||||
"平滑": ImageFilterImplement.apply_smooth,
|
||||
"反色": ImageFilterImplement.apply_invert,
|
||||
"黑白": ImageFilterImplement.apply_black_white,
|
||||
"阈值": ImageFilterImplement.apply_threshold,
|
||||
"对比度": ImageFilterImplement.apply_contrast,
|
||||
"亮度": ImageFilterImplement.apply_brightness,
|
||||
"色彩": ImageFilterImplement.apply_color,
|
||||
"色调": ImageFilterImplement.apply_to_color,
|
||||
"缩放": ImageFilterImplement.apply_resize,
|
||||
"波纹": ImageFilterImplement.apply_wave,
|
||||
"色键": ImageFilterImplement.apply_color_key,
|
||||
"暗角": ImageFilterImplement.apply_vignette,
|
||||
"发光": ImageFilterImplement.apply_glow,
|
||||
"RGB分离": ImageFilterImplement.apply_rgb_split,
|
||||
"光学补偿": ImageFilterImplement.apply_optical_compensation,
|
||||
"球面化": ImageFilterImplement.apply_spherize,
|
||||
"旋转": ImageFilterImplement.apply_rotate,
|
||||
"透视变换": ImageFilterImplement.apply_perspective_transform,
|
||||
"裁剪": ImageFilterImplement.apply_crop,
|
||||
"噪点": ImageFilterImplement.apply_noise,
|
||||
"平移": ImageFilterImplement.apply_translate,
|
||||
"拓展边缘": ImageFilterImplement.apply_expand_edges,
|
||||
"素描": ImageFilterImplement.apply_sketch,
|
||||
"叠加颜色": ImageFilterImplement.apply_gradient_overlay,
|
||||
"阴影": ImageFilterImplement.apply_shadow,
|
||||
"径向模糊": ImageFilterImplement.apply_radial_blur,
|
||||
"旋转模糊": ImageFilterImplement.apply_spin_blur,
|
||||
"方向模糊": ImageFilterImplement.apply_directional_blur,
|
||||
"边缘模糊": ImageFilterImplement.apply_focus_blur,
|
||||
"缩放模糊": ImageFilterImplement.apply_zoom_blur,
|
||||
"镜像": ImageFilterImplement.apply_mirror_half,
|
||||
"水平翻转": ImageFilterImplement.apply_flip_horizontal,
|
||||
"垂直翻转": ImageFilterImplement.apply_flip_vertical,
|
||||
"复制": ImageFilterImplement.copy_area,
|
||||
"晃动": ImageFilterImplement.apply_random_wiggle,
|
||||
"动图": ImageFilterEmpty.empty_filter_param,
|
||||
"像素抖动": ImageFilterImplement.apply_pixel_jitter,
|
||||
"存入图像": ImageFilterStorage.store_image,
|
||||
"读取图像": ImageFilterStorage.load_image,
|
||||
"暂存图像": ImageFilterStorage.temp_store_image,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def get_filter(cls, name: str) -> Optional[callable]:
|
||||
return cls.filter_map.get(name)
|
||||
|
||||
@classmethod
|
||||
def has_filter(cls, name: str) -> bool:
|
||||
return name in cls.filter_map
|
||||
|
||||
|
||||
341
konabot/plugins/fx_process/gradient.py
Normal file
341
konabot/plugins/fx_process/gradient.py
Normal file
@ -0,0 +1,341 @@
|
||||
import re
|
||||
from konabot.plugins.fx_process.color_handle import ColorHandle
|
||||
import numpy as np
|
||||
from PIL import Image, ImageDraw
|
||||
from typing import List, Tuple, Dict, Optional
|
||||
|
||||
class GradientGenerator:
|
||||
"""渐变生成器类"""
|
||||
|
||||
def __init__(self):
|
||||
self.has_numpy = hasattr(np, '__version__')
|
||||
|
||||
def parse_color_list(self, color_list_str: str) -> List[Dict]:
|
||||
"""解析渐变颜色列表字符串
|
||||
|
||||
Args:
|
||||
color_list_str: 格式如 '[rgb(255,0,0)|(0,0),rgb(0,255,0)|(0,100),rgb(0,0,255)|(50,100)]'
|
||||
|
||||
Returns:
|
||||
list: 包含颜色和位置信息的字典列表
|
||||
"""
|
||||
color_nodes = []
|
||||
color_list_str = color_list_str.strip('[] ').strip()
|
||||
pattern = r'([^|]+)\|\(([^)]+)\)'
|
||||
matches = re.findall(pattern, color_list_str)
|
||||
|
||||
for color_str, pos_str in matches:
|
||||
color = ColorHandle.parse_color(color_str.strip())
|
||||
|
||||
try:
|
||||
x_str, y_str = pos_str.split(',')
|
||||
x_percent = float(x_str.strip().replace('%', ''))
|
||||
y_percent = float(y_str.strip().replace('%', ''))
|
||||
x_percent = max(0, min(100, x_percent))
|
||||
y_percent = max(0, min(100, y_percent))
|
||||
except:
|
||||
x_percent = 0
|
||||
y_percent = 0
|
||||
|
||||
color_nodes.append({
|
||||
'color': color,
|
||||
'position': (x_percent / 100.0, y_percent / 100.0)
|
||||
})
|
||||
|
||||
if not color_nodes:
|
||||
color_nodes = [
|
||||
{'color': (255, 0, 0), 'position': (0, 0)},
|
||||
{'color': (0, 0, 255), 'position': (1, 1)}
|
||||
]
|
||||
|
||||
return color_nodes
|
||||
|
||||
def create_gradient(self, width: int, height: int, color_nodes: List[Dict]) -> Image.Image:
|
||||
"""创建渐变图像
|
||||
|
||||
Args:
|
||||
width: 图像宽度
|
||||
height: 图像高度
|
||||
color_nodes: 颜色节点列表
|
||||
|
||||
Returns:
|
||||
Image.Image: 渐变图像
|
||||
"""
|
||||
if len(color_nodes) == 1:
|
||||
return Image.new('RGB', (width, height), color_nodes[0]['color'])
|
||||
elif len(color_nodes) == 2:
|
||||
return self._create_linear_gradient(width, height, color_nodes)
|
||||
else:
|
||||
return self._create_radial_gradient(width, height, color_nodes)
|
||||
|
||||
def _create_linear_gradient(self, width: int, height: int, color_nodes: List[Dict]) -> Image.Image:
|
||||
"""创建线性渐变"""
|
||||
color1 = color_nodes[0]['color']
|
||||
color2 = color_nodes[1]['color']
|
||||
pos1 = color_nodes[0]['position']
|
||||
pos2 = color_nodes[1]['position']
|
||||
|
||||
if self.has_numpy:
|
||||
return self._create_linear_gradient_numpy(width, height, color1, color2, pos1, pos2)
|
||||
else:
|
||||
return self._create_linear_gradient_pil(width, height, color1, color2, pos1, pos2)
|
||||
|
||||
def _create_linear_gradient_numpy(self, width: int, height: int,
|
||||
color1: Tuple, color2: Tuple,
|
||||
pos1: Tuple, pos2: Tuple) -> Image.Image:
|
||||
"""使用numpy创建线性渐变"""
|
||||
# 创建坐标网格
|
||||
x = np.linspace(0, 1, width)
|
||||
y = np.linspace(0, 1, height)
|
||||
xx, yy = np.meshgrid(x, y)
|
||||
|
||||
# 计算渐变方向
|
||||
dx = pos2[0] - pos1[0]
|
||||
dy = pos2[1] - pos1[1]
|
||||
length_sq = dx * dx + dy * dy
|
||||
|
||||
if length_sq > 0:
|
||||
# 计算投影参数
|
||||
t = ((xx - pos1[0]) * dx + (yy - pos1[1]) * dy) / length_sq
|
||||
t = np.clip(t, 0, 1)
|
||||
else:
|
||||
t = np.zeros_like(xx)
|
||||
|
||||
# 插值颜色
|
||||
r = color1[0] + (color2[0] - color1[0]) * t
|
||||
g = color1[1] + (color2[1] - color1[1]) * t
|
||||
b = color1[2] + (color2[2] - color1[2]) * t
|
||||
|
||||
# 创建图像
|
||||
gradient_array = np.stack([r, g, b], axis=-1).astype(np.uint8)
|
||||
return Image.fromarray(gradient_array)
|
||||
|
||||
def _create_linear_gradient_pil(self, width: int, height: int,
|
||||
color1: Tuple, color2: Tuple,
|
||||
pos1: Tuple, pos2: Tuple) -> Image.Image:
|
||||
"""使用PIL创建线性渐变(没有numpy时使用)"""
|
||||
gradient = Image.new('RGB', (width, height))
|
||||
draw = ImageDraw.Draw(gradient)
|
||||
|
||||
# 判断渐变方向
|
||||
if abs(pos1[0] - pos2[0]) < 0.01: # 垂直渐变
|
||||
y1 = int(pos1[1] * (height - 1))
|
||||
y2 = int(pos2[1] * (height - 1))
|
||||
|
||||
if y2 < y1:
|
||||
y1, y2 = y2, y1
|
||||
color1, color2 = color2, color1
|
||||
|
||||
if y2 > y1:
|
||||
for y in range(height):
|
||||
if y <= y1:
|
||||
fill_color = color1
|
||||
elif y >= y2:
|
||||
fill_color = color2
|
||||
else:
|
||||
ratio = (y - y1) / (y2 - y1)
|
||||
r = int(color1[0] + (color2[0] - color1[0]) * ratio)
|
||||
g = int(color1[1] + (color2[1] - color1[1]) * ratio)
|
||||
b = int(color1[2] + (color2[2] - color1[2]) * ratio)
|
||||
fill_color = (r, g, b)
|
||||
|
||||
draw.line([(0, y), (width, y)], fill=fill_color)
|
||||
else:
|
||||
draw.rectangle([0, 0, width, height], fill=color1)
|
||||
|
||||
elif abs(pos1[1] - pos2[1]) < 0.01: # 水平渐变
|
||||
x1 = int(pos1[0] * (width - 1))
|
||||
x2 = int(pos2[0] * (width - 1))
|
||||
|
||||
if x2 < x1:
|
||||
x1, x2 = x2, x1
|
||||
color1, color2 = color2, color1
|
||||
|
||||
if x2 > x1:
|
||||
for x in range(width):
|
||||
if x <= x1:
|
||||
fill_color = color1
|
||||
elif x >= x2:
|
||||
fill_color = color2
|
||||
else:
|
||||
ratio = (x - x1) / (x2 - x1)
|
||||
r = int(color1[0] + (color2[0] - color1[0]) * ratio)
|
||||
g = int(color1[1] + (color2[1] - color1[1]) * ratio)
|
||||
b = int(color1[2] + (color2[2] - color1[2]) * ratio)
|
||||
fill_color = (r, g, b)
|
||||
|
||||
draw.line([(x, 0), (x, height)], fill=fill_color)
|
||||
else:
|
||||
draw.rectangle([0, 0, width, height], fill=color1)
|
||||
|
||||
else: # 对角渐变(简化处理为左上到右下)
|
||||
for y in range(height):
|
||||
for x in range(width):
|
||||
distance = (x/width + y/height) / 2
|
||||
r = int(color1[0] + (color2[0] - color1[0]) * distance)
|
||||
g = int(color1[1] + (color2[1] - color1[1]) * distance)
|
||||
b = int(color1[2] + (color2[2] - color1[2]) * distance)
|
||||
draw.point((x, y), fill=(r, g, b))
|
||||
|
||||
return gradient
|
||||
|
||||
def _create_radial_gradient(self, width: int, height: int, color_nodes: List[Dict]) -> Image.Image:
|
||||
"""创建径向渐变"""
|
||||
if self.has_numpy and len(color_nodes) > 2:
|
||||
return self._create_radial_gradient_numpy(width, height, color_nodes)
|
||||
else:
|
||||
return self._create_simple_gradient(width, height, color_nodes)
|
||||
|
||||
def _create_radial_gradient_numpy(self, width: int, height: int, color_nodes: List[Dict]) -> Image.Image:
|
||||
"""使用numpy创建径向渐变(多色)"""
|
||||
# 创建坐标网格
|
||||
x = np.linspace(0, 1, width)
|
||||
y = np.linspace(0, 1, height)
|
||||
xx, yy = np.meshgrid(x, y)
|
||||
|
||||
# 提取颜色和位置
|
||||
positions = np.array([node['position'] for node in color_nodes])
|
||||
colors = np.array([node['color'] for node in color_nodes])
|
||||
|
||||
# 计算每个点到所有节点的距离
|
||||
distances = np.sqrt((xx[:, :, np.newaxis] - positions[np.newaxis, np.newaxis, :, 0]) ** 2 +
|
||||
(yy[:, :, np.newaxis] - positions[np.newaxis, np.newaxis, :, 1]) ** 2)
|
||||
|
||||
# 找到最近的两个节点
|
||||
sorted_indices = np.argsort(distances, axis=2)
|
||||
nearest_idx = sorted_indices[:, :, 0]
|
||||
second_idx = sorted_indices[:, :, 1]
|
||||
|
||||
# 获取对应的颜色
|
||||
nearest_colors = colors[nearest_idx]
|
||||
second_colors = colors[second_idx]
|
||||
|
||||
# 获取距离并计算权重
|
||||
nearest_dist = np.take_along_axis(distances, np.expand_dims(nearest_idx, axis=2), axis=2)[:, :, 0]
|
||||
second_dist = np.take_along_axis(distances, np.expand_dims(second_idx, axis=2), axis=2)[:, :, 0]
|
||||
|
||||
total_dist = nearest_dist + second_dist
|
||||
mask = total_dist > 0
|
||||
weight1 = np.zeros_like(nearest_dist)
|
||||
weight1[mask] = second_dist[mask] / total_dist[mask]
|
||||
weight2 = 1 - weight1
|
||||
|
||||
# 插值颜色
|
||||
r = nearest_colors[:, :, 0] * weight1 + second_colors[:, :, 0] * weight2
|
||||
g = nearest_colors[:, :, 1] * weight1 + second_colors[:, :, 1] * weight2
|
||||
b = nearest_colors[:, :, 2] * weight1 + second_colors[:, :, 2] * weight2
|
||||
|
||||
gradient_array = np.stack([r, g, b], axis=-1).astype(np.uint8)
|
||||
return Image.fromarray(gradient_array)
|
||||
|
||||
def _create_simple_gradient(self, width: int, height: int, color_nodes: List[Dict]) -> Image.Image:
|
||||
"""创建简化渐变(没有numpy或多色时使用)"""
|
||||
gradient = Image.new('RGB', (width, height))
|
||||
draw = ImageDraw.Draw(gradient)
|
||||
|
||||
if len(color_nodes) >= 2:
|
||||
# 使用第一个和最后一个颜色创建简单渐变
|
||||
color1 = color_nodes[0]['color']
|
||||
color2 = color_nodes[-1]['color']
|
||||
|
||||
# 判断节点分布
|
||||
x_positions = [node['position'][0] for node in color_nodes]
|
||||
y_positions = [node['position'][1] for node in color_nodes]
|
||||
|
||||
if all(abs(x - x_positions[0]) < 0.01 for x in x_positions):
|
||||
# 垂直渐变
|
||||
for y in range(height):
|
||||
ratio = y / (height - 1) if height > 1 else 0
|
||||
r = int(color1[0] + (color2[0] - color1[0]) * ratio)
|
||||
g = int(color1[1] + (color2[1] - color1[1]) * ratio)
|
||||
b = int(color1[2] + (color2[2] - color1[2]) * ratio)
|
||||
draw.line([(0, y), (width, y)], fill=(r, g, b))
|
||||
else:
|
||||
# 水平渐变
|
||||
for x in range(width):
|
||||
ratio = x / (width - 1) if width > 1 else 0
|
||||
r = int(color1[0] + (color2[0] - color1[0]) * ratio)
|
||||
g = int(color1[1] + (color2[1] - color1[1]) * ratio)
|
||||
b = int(color1[2] + (color2[2] - color1[2]) * ratio)
|
||||
draw.line([(x, 0), (x, height)], fill=(r, g, b))
|
||||
else:
|
||||
# 单色
|
||||
draw.rectangle([0, 0, width, height], fill=color_nodes[0]['color'])
|
||||
|
||||
return gradient
|
||||
|
||||
def create_simple_gradient(self, width: int, height: int,
|
||||
start_color: Tuple, end_color: Tuple,
|
||||
direction: str = 'vertical') -> Image.Image:
|
||||
"""创建简单双色渐变
|
||||
|
||||
Args:
|
||||
width: 图像宽度
|
||||
height: 图像高度
|
||||
start_color: 起始颜色
|
||||
end_color: 结束颜色
|
||||
direction: 渐变方向 'vertical', 'horizontal', 'diagonal'
|
||||
|
||||
Returns:
|
||||
Image.Image: 渐变图像
|
||||
"""
|
||||
if direction == 'vertical':
|
||||
return self._create_vertical_gradient(width, height, start_color, end_color)
|
||||
elif direction == 'horizontal':
|
||||
return self._create_horizontal_gradient(width, height, start_color, end_color)
|
||||
else: # diagonal
|
||||
return self._create_diagonal_gradient(width, height, start_color, end_color)
|
||||
|
||||
def _create_vertical_gradient(self, width: int, height: int,
|
||||
color1: Tuple, color2: Tuple) -> Image.Image:
|
||||
"""创建垂直渐变"""
|
||||
gradient = Image.new('RGB', (width, height))
|
||||
draw = ImageDraw.Draw(gradient)
|
||||
|
||||
for y in range(height):
|
||||
ratio = y / (height - 1) if height > 1 else 0
|
||||
r = int(color1[0] + (color2[0] - color1[0]) * ratio)
|
||||
g = int(color1[1] + (color2[1] - color1[1]) * ratio)
|
||||
b = int(color1[2] + (color2[2] - color1[2]) * ratio)
|
||||
draw.line([(0, y), (width, y)], fill=(r, g, b))
|
||||
|
||||
return gradient
|
||||
|
||||
def _create_horizontal_gradient(self, width: int, height: int,
|
||||
color1: Tuple, color2: Tuple) -> Image.Image:
|
||||
"""创建水平渐变"""
|
||||
gradient = Image.new('RGB', (width, height))
|
||||
draw = ImageDraw.Draw(gradient)
|
||||
|
||||
for x in range(width):
|
||||
ratio = x / (width - 1) if width > 1 else 0
|
||||
r = int(color1[0] + (color2[0] - color1[0]) * ratio)
|
||||
g = int(color1[1] + (color2[1] - color1[1]) * ratio)
|
||||
b = int(color1[2] + (color2[2] - color1[2]) * ratio)
|
||||
draw.line([(x, 0), (x, height)], fill=(r, g, b))
|
||||
|
||||
return gradient
|
||||
|
||||
def _create_diagonal_gradient(self, width: int, height: int,
|
||||
color1: Tuple, color2: Tuple) -> Image.Image:
|
||||
"""创建对角渐变"""
|
||||
if self.has_numpy:
|
||||
return self._create_diagonal_gradient_numpy(width, height, color1, color2)
|
||||
else:
|
||||
return self._create_horizontal_gradient(width, height, color1, color2) # 降级为水平渐变
|
||||
|
||||
def _create_diagonal_gradient_numpy(self, width: int, height: int,
|
||||
color1: Tuple, color2: Tuple) -> Image.Image:
|
||||
"""使用numpy创建对角渐变"""
|
||||
x = np.linspace(0, 1, width)
|
||||
y = np.linspace(0, 1, height)
|
||||
xx, yy = np.meshgrid(x, y)
|
||||
|
||||
distance = (xx + yy) / 2.0
|
||||
|
||||
r = color1[0] + (color2[0] - color1[0]) * distance
|
||||
g = color1[1] + (color2[1] - color1[1]) * distance
|
||||
b = color1[2] + (color2[2] - color1[2]) * distance
|
||||
|
||||
gradient_array = np.stack([r, g, b], axis=-1).astype(np.uint8)
|
||||
return Image.fromarray(gradient_array)
|
||||
150
konabot/plugins/fx_process/image_storage.py
Normal file
150
konabot/plugins/fx_process/image_storage.py
Normal file
@ -0,0 +1,150 @@
|
||||
import asyncio
|
||||
from dataclasses import dataclass
|
||||
from hashlib import md5
|
||||
import time
|
||||
|
||||
from nonebot import logger
|
||||
from nonebot_plugin_apscheduler import driver
|
||||
from konabot.common.path import DATA_PATH
|
||||
import os
|
||||
from PIL import Image
|
||||
|
||||
IMAGE_PATH = DATA_PATH / "temp" / "images"
|
||||
|
||||
@dataclass
|
||||
class ImageResource:
|
||||
name: str
|
||||
expire: int
|
||||
|
||||
@dataclass
|
||||
class StorageImage:
|
||||
name: str
|
||||
resources: dict[str,
|
||||
dict[str,ImageResource]] # {群号: {QQ号: ImageResource}}
|
||||
|
||||
class ImageStorager:
|
||||
images_pool: dict[str,StorageImage] = {}
|
||||
|
||||
max_storage: int = 10 * 1024 * 1024 # 最大存储10MB
|
||||
max_image_count: int = 200 # 最大存储图片数量
|
||||
|
||||
@staticmethod
|
||||
def init():
|
||||
if not IMAGE_PATH.exists():
|
||||
IMAGE_PATH.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
@staticmethod
|
||||
def delete_path_image(name: str):
|
||||
resource_path = IMAGE_PATH / name
|
||||
if resource_path.exists():
|
||||
os.remove(resource_path)
|
||||
|
||||
@classmethod
|
||||
async def clear_expire_image(cls):
|
||||
# 清理过期的图片资源,将未被删除的放入列表中,如果超过最大数量则删除最早过期的
|
||||
remaining_images = []
|
||||
current_time = time.time()
|
||||
for name, storage_image in list(ImageStorager.images_pool.items()):
|
||||
for group_id, resources in list(storage_image.resources.items()):
|
||||
for qq_id, resource in list(resources.items()):
|
||||
if resource.expire < current_time:
|
||||
del storage_image.resources[group_id][qq_id]
|
||||
cls.delete_path_image(name)
|
||||
else:
|
||||
remaining_images.append((name, group_id, qq_id, resource.expire))
|
||||
if not storage_image.resources:
|
||||
del ImageStorager.images_pool[name]
|
||||
# 如果剩余图片超过最大数量,按过期时间排序并删除最早过期的
|
||||
if len(remaining_images) > ImageStorager.max_image_count:
|
||||
remaining_images.sort(key=lambda x: x[3]) # 按过期时间排序
|
||||
to_delete = len(remaining_images) - ImageStorager.max_image_count
|
||||
for i in range(to_delete):
|
||||
name, group_id, qq_id, _ = remaining_images[i]
|
||||
resource = ImageStorager.images_pool[name].resources[group_id][qq_id]
|
||||
del ImageStorager.images_pool[name].resources[group_id][qq_id]
|
||||
cls.delete_path_image(name)
|
||||
|
||||
@classmethod
|
||||
def _add_to_pool(cls, image: bytes, name: str, group_id: str, qq_id: str, expire: int = 36000):
|
||||
expire_time = time.time() + expire
|
||||
if name not in cls.images_pool:
|
||||
cls.images_pool[name] = StorageImage(name=name,resources={})
|
||||
if group_id not in cls.images_pool[name].resources:
|
||||
cls.images_pool[name].resources[group_id] = {}
|
||||
cls.images_pool[name].resources[group_id][qq_id] = ImageResource(name=name, expire=expire_time)
|
||||
|
||||
@classmethod
|
||||
def save_image(cls, image: bytes, name: str, group_id: str, qq_id: str) -> None:
|
||||
"""
|
||||
以哈希值命名保存图片,并返回图片资源信息
|
||||
"""
|
||||
# 检测图像大小,不得超过 10 MB
|
||||
if len(image) > cls.max_storage:
|
||||
raise ValueError("图片大小超过 10 MB 限制")
|
||||
hash_name = md5(image).hexdigest()
|
||||
ext = os.path.splitext(name)[1]
|
||||
file_name = f"{hash_name}{ext}"
|
||||
full_path = IMAGE_PATH / file_name
|
||||
with open(full_path, "wb") as f:
|
||||
f.write(image)
|
||||
# 将文件写入 images_pool
|
||||
cls._add_to_pool(image, file_name, group_id, qq_id)
|
||||
|
||||
@classmethod
|
||||
def load_image(cls, name: str, group_id: str, qq_id: str) -> Image:
|
||||
if name not in cls.images_pool:
|
||||
return None
|
||||
if group_id not in cls.images_pool[name].resources:
|
||||
return None
|
||||
# 寻找对应 QQ 号 的资源,如果没有就返回相同群下的第一个资源
|
||||
if qq_id not in cls.images_pool[name].resources[group_id]:
|
||||
first_qq_id = next(iter(cls.images_pool[name].resources[group_id]))
|
||||
qq_id = first_qq_id
|
||||
resource = cls.images_pool[name].resources[group_id][qq_id]
|
||||
resource_path = IMAGE_PATH / resource.name
|
||||
return Image.open(resource_path)
|
||||
|
||||
class ImageStoragerManager:
|
||||
def __init__(self, interval: int = 300): # 默认 5 分钟执行一次
|
||||
self.interval = interval
|
||||
self._clear_task = None
|
||||
self._running = False
|
||||
|
||||
async def start_auto_clear(self):
|
||||
"""启动自动任务"""
|
||||
self._running = True
|
||||
self._clear_task = asyncio.create_task(self._auto_clear_loop())
|
||||
|
||||
logger.info(f"自动清理任务已启动,间隔: {self.interval}秒")
|
||||
|
||||
async def stop_auto_clear(self):
|
||||
"""停止自动清理任务"""
|
||||
if self._clear_task:
|
||||
self._running = False
|
||||
self._clear_task.cancel()
|
||||
try:
|
||||
await self._clear_task
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
logger.info("自动清理任务已停止")
|
||||
else:
|
||||
logger.warning("没有正在运行的自动清理任务")
|
||||
|
||||
async def _auto_clear_loop(self):
|
||||
"""自动清理循环"""
|
||||
while self._running:
|
||||
try:
|
||||
await asyncio.sleep(self.interval)
|
||||
await ImageStorager.clear_expire_image()
|
||||
except asyncio.CancelledError:
|
||||
break
|
||||
except Exception as e:
|
||||
logger.error(f"定时清理失败: {e}")
|
||||
|
||||
image_manager = ImageStoragerManager(interval=300) # 每5分钟清理一次
|
||||
|
||||
@driver.on_startup
|
||||
async def init_image_storage():
|
||||
ImageStorager.init()
|
||||
# 启用定时任务清理过期图片
|
||||
await image_manager.start_auto_clear()
|
||||
20
konabot/plugins/fx_process/types.py
Normal file
20
konabot/plugins/fx_process/types.py
Normal file
@ -0,0 +1,20 @@
|
||||
from dataclasses import dataclass
|
||||
|
||||
|
||||
@dataclass
|
||||
class FilterItem:
|
||||
name: str
|
||||
filter: callable
|
||||
args: list
|
||||
|
||||
class ImageRequireSignal:
|
||||
pass
|
||||
|
||||
@dataclass
|
||||
class ImagesListRequireSignal:
|
||||
pass
|
||||
|
||||
@dataclass
|
||||
class SenderInfo:
|
||||
group_id: str
|
||||
qq_id: str
|
||||
@ -6,24 +6,11 @@ import PIL
|
||||
import PIL.Image
|
||||
import cv2
|
||||
import imageio.v3 as iio
|
||||
from nonebot import on_message
|
||||
from nonebot.adapters import Bot
|
||||
from nonebot_plugin_alconna import Alconna, Args, Image, Option, UniMessage, on_alconna
|
||||
import numpy
|
||||
|
||||
from konabot.common.nb.exc import BotExceptionMessage
|
||||
from konabot.common.nb.extract_image import DepImageBytes, DepPILImage
|
||||
from konabot.common.nb.match_keyword import match_keyword
|
||||
from konabot.common.nb.reply_image import reply_image
|
||||
|
||||
# 保持不变
|
||||
cmd_black_white = on_message(rule=match_keyword("黑白"))
|
||||
|
||||
|
||||
@cmd_black_white.handle()
|
||||
async def _(img: DepPILImage, bot: Bot):
|
||||
# 保持不变
|
||||
await reply_image(cmd_black_white, bot, img.convert("LA"))
|
||||
from konabot.common.nb.extract_image import DepImageBytes
|
||||
|
||||
|
||||
# 保持不变
|
||||
|
||||
@ -10,7 +10,9 @@ from nonebot_plugin_alconna import (Alconna, Args, UniMessage, UniMsg,
|
||||
on_alconna)
|
||||
from nonebot_plugin_apscheduler import scheduler
|
||||
|
||||
from konabot.common import username
|
||||
from konabot.common.longtask import DepLongTaskTarget
|
||||
from konabot.common.pager import PagerQuery
|
||||
from konabot.plugins.kona_ph.core.message import (get_daily_report,
|
||||
get_daily_report_v2,
|
||||
get_puzzle_description,
|
||||
@ -76,7 +78,7 @@ async def _(target: DepLongTaskTarget):
|
||||
|
||||
|
||||
cmd_history = on_alconna(Alconna(
|
||||
"历史题目",
|
||||
"re:历史(题目|谜题)",
|
||||
Args["page?", int],
|
||||
Args["index_id?", str],
|
||||
), rule=to_me())
|
||||
@ -118,6 +120,24 @@ async def _(target: DepLongTaskTarget, index_id: str = "", page: int = 1):
|
||||
await target.send_message(msg)
|
||||
|
||||
|
||||
cmd_leadboard = on_alconna(Alconna(
|
||||
"re:此方(解谜|谜题)排行榜",
|
||||
Args["page?", int],
|
||||
))
|
||||
|
||||
@cmd_leadboard.handle()
|
||||
async def _(target: DepLongTaskTarget, page: int = 1):
|
||||
async with puzzle_manager() as manager:
|
||||
result = manager.get_leadboard(PagerQuery(page, 10))
|
||||
await target.send_message(result.to_unimessage(
|
||||
title="此方解谜排行榜",
|
||||
formatter=lambda data: (
|
||||
f"✨ {data[1]} 已完成 | "
|
||||
f"{username.get_username(data[0])}"
|
||||
)
|
||||
))
|
||||
|
||||
|
||||
@scheduler.scheduled_job("cron", hour="8")
|
||||
async def _():
|
||||
async with puzzle_manager() as manager:
|
||||
|
||||
@ -17,7 +17,7 @@ class PuzzleImageManager:
|
||||
21,
|
||||
)
|
||||
img_name = f"{id}{suffix}"
|
||||
(KONAPH_IMAGE_BASE / img_name).write_bytes(data)
|
||||
_ = (KONAPH_IMAGE_BASE / img_name).write_bytes(data)
|
||||
return img_name
|
||||
|
||||
def remove_puzzle_image(self, img_name: str):
|
||||
|
||||
@ -33,7 +33,7 @@ def get_puzzle_description(puzzle: Puzzle, with_answer: bool = False) -> UniMess
|
||||
)
|
||||
|
||||
result = result.text(f"\n\n出题者:{get_username(puzzle.author_id)}")
|
||||
|
||||
|
||||
if with_answer:
|
||||
result = result.text(f"\n\n题目答案:{puzzle.flag}")
|
||||
else:
|
||||
|
||||
@ -1,5 +1,6 @@
|
||||
import asyncio
|
||||
import datetime
|
||||
import functools
|
||||
import random
|
||||
import re
|
||||
from contextlib import asynccontextmanager
|
||||
@ -7,6 +8,7 @@ from contextlib import asynccontextmanager
|
||||
import nanoid
|
||||
from pydantic import BaseModel, Field, ValidationError
|
||||
|
||||
from konabot.common.pager import PagerQuery
|
||||
from konabot.plugins.kona_ph.core.path import KONAPH_DATA_JSON
|
||||
|
||||
|
||||
@ -112,6 +114,10 @@ class PuzzleManager(BaseModel):
|
||||
|
||||
index_id_counter: int = 1
|
||||
submissions: dict[str, dict[str, list[PuzzleSubmission]]] = {}
|
||||
"""
|
||||
类型:{ [raw_id: str]: { [user_id: str]: PuzzleSubmission[] } }
|
||||
"""
|
||||
|
||||
last_checked_date: datetime.date = Field(
|
||||
default_factory=lambda: get_today_date() - datetime.timedelta(days=1)
|
||||
)
|
||||
@ -235,6 +241,20 @@ class PuzzleManager(BaseModel):
|
||||
if p.author_id == user
|
||||
], key=lambda p: p.created_at, reverse=True)
|
||||
|
||||
def get_leadboard(self, pager: PagerQuery):
|
||||
return pager.apply(sorted([
|
||||
(user, sum((
|
||||
len([
|
||||
s for s in sl.get(user, [])
|
||||
if s.success
|
||||
]) for sl in self.submissions.values()
|
||||
)))
|
||||
for user in functools.reduce(
|
||||
lambda x, y: x | y,
|
||||
(set(sl.keys()) for sl in self.submissions.values()),
|
||||
)
|
||||
], key=lambda t: t[1], reverse=True))
|
||||
|
||||
|
||||
lock = asyncio.Lock()
|
||||
|
||||
|
||||
@ -10,7 +10,6 @@ from pydantic import BaseModel
|
||||
from konabot.common.longtask import DepLongTaskTarget
|
||||
from konabot.common.nb.exc import BotExceptionMessage
|
||||
from konabot.common.nb.extract_image import download_image_bytes
|
||||
from konabot.common.nb.qq_broadcast import qq_broadcast
|
||||
from konabot.common.username import get_username
|
||||
from konabot.plugins.kona_ph.core.image import get_image_manager
|
||||
from konabot.plugins.kona_ph.core.message import (get_puzzle_description, get_puzzle_hint_list,
|
||||
@ -19,6 +18,7 @@ from konabot.plugins.kona_ph.core.message import (get_puzzle_description, get_pu
|
||||
from konabot.plugins.kona_ph.core.storage import (Puzzle, PuzzleHint, PuzzleManager,
|
||||
get_today_date,
|
||||
puzzle_manager)
|
||||
from konabot.plugins.poster.service import broadcast
|
||||
|
||||
PUZZLE_PAGE_SIZE = 10
|
||||
|
||||
@ -344,7 +344,7 @@ def create_admin_commands():
|
||||
p = manager.get_today_puzzle(strong=True)
|
||||
if p is None:
|
||||
return await target.send_message("上架失败了orz,可能是没题了")
|
||||
await qq_broadcast(config.plugin_puzzle_playgroup, get_puzzle_description(p))
|
||||
await broadcast("每日谜题", get_puzzle_description(p))
|
||||
return await target.send_message("Ok!")
|
||||
|
||||
@cmd_admin.assign("preview")
|
||||
|
||||
26
konabot/plugins/notice_ui/__init__.py
Normal file
26
konabot/plugins/notice_ui/__init__.py
Normal file
@ -0,0 +1,26 @@
|
||||
from loguru import logger
|
||||
import nonebot
|
||||
from nonebot.adapters import Event as BaseEvent
|
||||
from nonebot.adapters.discord.event import MessageEvent as DiscordMessageEvent
|
||||
from nonebot_plugin_alconna import (
|
||||
UniMessage,
|
||||
UniMsg
|
||||
)
|
||||
from konabot.plugins.notice_ui.notice import NoticeUI
|
||||
from nonebot_plugin_alconna import on_alconna, Alconna, Args
|
||||
|
||||
evt = on_alconna(Alconna(
|
||||
"notice",
|
||||
Args["title", str],
|
||||
Args["message", str]
|
||||
),
|
||||
use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=True
|
||||
)
|
||||
|
||||
@evt.handle()
|
||||
async def _(title: str, message: str, msg: UniMsg, event: BaseEvent):
|
||||
logger.debug(f"Received notice command with title: {title}, message: {message}")
|
||||
|
||||
out = await NoticeUI.render_notice(title, message)
|
||||
|
||||
await evt.send(await UniMessage().image(raw=out).export())
|
||||
77
konabot/plugins/notice_ui/notice.py
Normal file
77
konabot/plugins/notice_ui/notice.py
Normal file
@ -0,0 +1,77 @@
|
||||
from io import BytesIO
|
||||
from PIL import Image
|
||||
from konabot.common.web_render import konaweb
|
||||
from konabot.common.web_render.core import WebRenderer
|
||||
|
||||
from playwright.async_api import Page
|
||||
|
||||
class NoticeUI:
|
||||
@staticmethod
|
||||
async def render_notice(title: str, message: str) -> bytes:
|
||||
"""
|
||||
渲染一个通知图片,包含标题和消息内容。
|
||||
"""
|
||||
async def page_function(page: Page):
|
||||
# 直到 setMaskMode 函数加载完成
|
||||
await page.wait_for_function("typeof setMaskMode === 'function'", timeout=1000)
|
||||
await page.evaluate('setMaskMode(false)')
|
||||
# 直到 setContent 函数加载完成
|
||||
await page.wait_for_function("typeof setContent === 'function'", timeout=1000)
|
||||
# 设置标题和消息内容
|
||||
await page.evaluate(f'setContent("{title}", "{message}")')
|
||||
|
||||
async def mask_function(page: Page):
|
||||
# 直到 setContent 函数加载完成
|
||||
await page.wait_for_function("typeof setContent === 'function'", timeout=1000)
|
||||
# 设置标题和消息内容
|
||||
await page.evaluate(f'setContent("{title}", "{message}")')
|
||||
# 直到 setMaskMode 函数加载完成
|
||||
await page.wait_for_function("typeof setMaskMode === 'function'", timeout=1000)
|
||||
await page.evaluate('setMaskMode(true)')
|
||||
|
||||
image_bytes = await WebRenderer.render_with_persistent_page(
|
||||
"notice_renderer",
|
||||
konaweb('notice'),
|
||||
target='#main',
|
||||
other_function=page_function,
|
||||
)
|
||||
|
||||
mask_bytes = await WebRenderer.render_with_persistent_page(
|
||||
"notice_renderer",
|
||||
konaweb('notice'),
|
||||
target='#main',
|
||||
other_function=mask_function)
|
||||
|
||||
image = Image.open(BytesIO(image_bytes)).convert("RGBA")
|
||||
mask = Image.open(BytesIO(mask_bytes)).convert("L")
|
||||
|
||||
# 使用mask作为alpha通道
|
||||
r, g, b, _ = image.split()
|
||||
transparent_image = Image.merge("RGBA", (r, g, b, mask))
|
||||
|
||||
# 先创建一个纯白色背景,然后粘贴透明图像
|
||||
background = Image.new("RGBA", transparent_image.size, (255, 255, 255, 255))
|
||||
composite = Image.alpha_composite(background, transparent_image)
|
||||
|
||||
palette_img = composite.convert("RGB").convert(
|
||||
"P",
|
||||
palette=Image.Palette.WEB,
|
||||
colors=256,
|
||||
dither=Image.Dither.NONE
|
||||
)
|
||||
|
||||
# 将alpha值小于128的设为透明
|
||||
alpha_mask = mask.point(lambda x: 0 if x < 128 else 255)
|
||||
|
||||
# 保存为GIF
|
||||
output_buffer = BytesIO()
|
||||
palette_img.save(
|
||||
output_buffer,
|
||||
format="GIF",
|
||||
transparency=0, # 将索引0设为透明
|
||||
disposal=2,
|
||||
loop=0
|
||||
)
|
||||
|
||||
output_buffer.seek(0)
|
||||
return output_buffer.getvalue()
|
||||
@ -1,9 +1,10 @@
|
||||
from nonebot import on_message
|
||||
from nonebot_plugin_alconna import UniMessage, UniMsg
|
||||
from nonebot_plugin_alconna import UniMessage
|
||||
|
||||
evt = on_message()
|
||||
from konabot.common.nb.match_keyword import match_keyword
|
||||
|
||||
evt = on_message(rule=match_keyword("喵"))
|
||||
|
||||
@evt.handle()
|
||||
async def _(msg: UniMsg):
|
||||
if msg.extract_plain_text() == "喵":
|
||||
await evt.send(await UniMessage().text("喵").export())
|
||||
async def _():
|
||||
await evt.send(await UniMessage().text("喵").export())
|
||||
|
||||
@ -1,11 +1,12 @@
|
||||
import re
|
||||
import aiohttp
|
||||
import asyncio as asynkio
|
||||
from math import ceil
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
import datetime
|
||||
|
||||
import nanoid
|
||||
from konabot.plugins.notice_ui.notice import NoticeUI
|
||||
import nonebot
|
||||
from loguru import logger
|
||||
from nonebot import get_plugin_config, on_message
|
||||
@ -14,9 +15,10 @@ from nonebot_plugin_alconna import Alconna, Args, Subcommand, UniMessage, UniMsg
|
||||
from pydantic import BaseModel
|
||||
|
||||
from konabot.common.longtask import DepLongTaskTarget, LongTask, create_longtask, handle_long_task, longtask_data
|
||||
from konabot.common.ptimeparse import parse
|
||||
from konabot.common.nb.match_keyword import match_keyword
|
||||
from konabot.plugins.simple_notify.ask_llm import ask_ai
|
||||
|
||||
evt = on_message()
|
||||
evt = on_message(rule=match_keyword(re.compile("^.+提醒我.+$")))
|
||||
|
||||
(Path(__file__).parent.parent.parent.parent / "data").mkdir(exist_ok=True)
|
||||
DATA_FILE_PATH = Path(__file__).parent.parent.parent.parent / "data" / "notify.json"
|
||||
@ -76,21 +78,12 @@ async def _(msg: UniMsg, mEvt: Event, target: DepLongTaskTarget):
|
||||
return
|
||||
|
||||
text = msg.extract_plain_text()
|
||||
if "提醒我" not in text:
|
||||
return
|
||||
|
||||
segments = text.split("提醒我", maxsplit=1)
|
||||
if len(segments) != 2:
|
||||
return
|
||||
|
||||
notify_time, notify_text = segments
|
||||
try:
|
||||
target_time = parse(notify_time)
|
||||
logger.info(f"从 {notify_time} 解析出了时间:{target_time}")
|
||||
except Exception:
|
||||
logger.info(f"无法从 {notify_time} 中解析出时间")
|
||||
return
|
||||
if not notify_text:
|
||||
target_time, notify_text = await ask_ai(text)
|
||||
if target_time is None:
|
||||
return
|
||||
|
||||
await create_longtask(
|
||||
@ -115,6 +108,11 @@ async def _(task: LongTask):
|
||||
await task.target.send_message(
|
||||
UniMessage().text(f"代办提醒:{message}")
|
||||
)
|
||||
notice_bytes = await NoticeUI.render_notice("代办提醒", message)
|
||||
await task.target.send_message(
|
||||
UniMessage().image(raw=notice_bytes),
|
||||
at=False
|
||||
)
|
||||
async with DATA_FILE_LOCK:
|
||||
data = load_notify_config()
|
||||
if (chan := data.notify_channels.get(task.target.target_id)) is not None:
|
||||
|
||||
158
konabot/plugins/simple_notify/ask_llm.py
Normal file
158
konabot/plugins/simple_notify/ask_llm.py
Normal file
@ -0,0 +1,158 @@
|
||||
import datetime
|
||||
import json
|
||||
import re
|
||||
|
||||
from loguru import logger
|
||||
|
||||
from konabot.common.apis.ali_content_safety import AlibabaGreen
|
||||
from konabot.common.llm import get_llm
|
||||
|
||||
|
||||
SYSTEM_PROMPT = """你是一个专门解析提醒请求的助手。请分析用户输入,识别其中是否包含提醒信息,并输出标准化的JSON格式结果。
|
||||
|
||||
输入格式通常是:"xxxx提醒我yyyy",其中:
|
||||
- xxxx 是用户提供的时间信息
|
||||
- yyyy 是提醒内容。有些时候用户会有一些需求。你可以在合理的范围进行衍生。
|
||||
|
||||
输出要求:
|
||||
- 必须是有效的JSON对象
|
||||
- 包含以下字段:
|
||||
* datetime: 如果是绝对时间,填入ISO 8601格式的日期时间字符串;否则为null
|
||||
* datetime_delta: 如果是相对时间,填入ISO 8601持续时间格式;否则为null
|
||||
* datetime_delta_minus: 如果时间偏移量是负数,则此项为 true,否则为 false
|
||||
* content: 提醒内容的字符串
|
||||
* is_notice: 布尔值,表示这是否是真正的提醒请求
|
||||
|
||||
时间处理规则:
|
||||
- 绝对时间示例:如果 xxxx 输入了非常明确的时间点,如"2024年12月25日" → 转换为具体datetime
|
||||
- 相对时间示例:如果 xxxx 没有输入非常明确的时间点,如"10分钟后"、"2小时后"、"3天后" → 转换为datetime_delta
|
||||
- 如果用户输入了需要计算的时间,你需要计算出正确的结果,如"10分钟后的8分钟前" → 转换为 “PT2M”
|
||||
- zzzz 是系统提供的时间,每句话肯定都有,这不是你判断相对或绝对时间的依据,需严格按照 xxx 来判断
|
||||
- datetime和datetime_delta有且仅有一个不为null
|
||||
|
||||
时间格式要求:
|
||||
- datetime: "YYYY-MM-DDTHH:MM:SS" (ISO 8601)
|
||||
- datetime_delta: "PxYxMxDTxHxMxS" 格式 (如"PT1H30M"表示1小时30分钟,"P3DT4H"表示三天四小时,"P5MT2M"表示五个月两分钟)
|
||||
|
||||
判断标准:
|
||||
- is_notice=true: 明确包含时间+提醒内容的请求
|
||||
- is_notice=false: 闲聊、疑问句、或不符合提醒格式的内容
|
||||
|
||||
示例:
|
||||
用户:"明天下午2点提醒我开会"
|
||||
输出:{"datetime": "2024-01-16T14:00:00", "datetime_delta": null,
|
||||
"datetime_delta_minus": false, "content": "开会", "is_notice": true}
|
||||
|
||||
用户:"5分钟后提醒我关火"
|
||||
输出:{"datetime": null, "datetime_delta": "PT5M", "datetime_delta_minus": false, "content": "关火", "is_notice": true}
|
||||
|
||||
用户:"5分钟前提醒我关火"
|
||||
输出:{"datetime": null, "datetime_delta": "PT5M", "datetime_delta_minus": true, "content": "关火", "is_notice": true}
|
||||
|
||||
用户:"五百年后提醒我关火"
|
||||
输出:{"datetime": null, "datetime_delta": "P500Y", "datetime_delta_minus": false, "content": "关火", "is_notice": true}
|
||||
|
||||
用户:"昨天提醒我关火"
|
||||
输出:{"datetime": null, "datetime_delta": "P1D", "datetime_delta_minus": true, "content": "关火", "is_notice": true}
|
||||
|
||||
用户:"什么是提醒功能?"
|
||||
输出:{"datetime": null, "datetime_delta": null, "datetime_delta_minus": false, "content": "", "is_notice": false}
|
||||
|
||||
用户:"过一会会,用可爱的语气提醒我该睡觉了"
|
||||
输出:{"datetime": null, "datetime_delta": "PT10M", "datetime_delta_minus": false, "content": "呼呼!该睡觉了哦!ヾ(•ω•`)o", "is_notice": true}
|
||||
|
||||
请严格按照上述格式输出JSON,不要添加任何其他文字说明。现在是 DATETIME"""
|
||||
|
||||
pt_pattern = re.compile(
|
||||
r"^P"
|
||||
r"((?P<year>\d+)Y)?"
|
||||
r"((?P<month>\d+)M)?"
|
||||
r"((?P<day>\d+)D)?"
|
||||
r"(T((?P<hour>\d+)H)?"
|
||||
r"((?P<minute>\d+)M)?"
|
||||
r"((?P<second>\d+)S)?)?$"
|
||||
)
|
||||
|
||||
|
||||
def tryint(s: str | None):
|
||||
if s:
|
||||
if re.match(r"^\d+$", s):
|
||||
return int(s)
|
||||
return 0
|
||||
|
||||
|
||||
async def ask_ai(expression: str, now: datetime.datetime | None = None) -> tuple[datetime.datetime | None, str]:
|
||||
if now is None:
|
||||
now = datetime.datetime.now()
|
||||
prompt = SYSTEM_PROMPT.replace("DATETIME", f"{now}, 星期 {now.weekday() + 1}")
|
||||
|
||||
is_safe = await AlibabaGreen.detect(expression)
|
||||
if not is_safe:
|
||||
logger.info(f"提醒功能:消息被阿里绿网拦截 message={expression}")
|
||||
return None, ""
|
||||
|
||||
llm = get_llm("qwen3-max")
|
||||
message = await llm.chat([
|
||||
{ "role": "system", "content": prompt },
|
||||
{ "role": "user", "content": expression },
|
||||
])
|
||||
result = message.content
|
||||
if result is None:
|
||||
return (None, "")
|
||||
|
||||
try:
|
||||
data = json.loads(result)
|
||||
except json.JSONDecodeError:
|
||||
logger.info(f"提醒功能:解析 AI 返回值时出现问题 raw={result}")
|
||||
return (None, "")
|
||||
|
||||
datetime_absolute = data.get("datetime", None)
|
||||
datetime_delta = data.get("datetime_delta", None)
|
||||
is_minus = data.get("datetime_delta_minus", False)
|
||||
content = data.get("content", "")
|
||||
is_notice = data.get("is_notice", False)
|
||||
|
||||
if not is_notice:
|
||||
return (None, "")
|
||||
if datetime_absolute:
|
||||
try:
|
||||
res = datetime.datetime.strptime(datetime_absolute, "%Y-%m-%dT%H:%M:%S"), content
|
||||
logger.info(f"提醒功能:使用绝对时间解析 AI 返回值 raw={result} target={res[0]}")
|
||||
return res
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
if datetime_delta and (match := pt_pattern.match(datetime_delta)):
|
||||
years = tryint(match.group("year"))
|
||||
months = tryint(match.group("month"))
|
||||
days = tryint(match.group("day"))
|
||||
hours = tryint(match.group("hour"))
|
||||
minutes = tryint(match.group("minute"))
|
||||
seconds = tryint(match.group("second"))
|
||||
|
||||
dt = datetime.timedelta(days=days, hours=hours, minutes=minutes, seconds=seconds)
|
||||
if is_minus:
|
||||
dt = -dt
|
||||
if is_minus:
|
||||
now2 = now.replace(year=now.year - years)
|
||||
m = now2.month
|
||||
if (months - m) >= 0:
|
||||
neg_months = -(m - months - 1)
|
||||
neg_years = (neg_months + 11) // 12
|
||||
target_month = 12 - ((neg_months - 1) % 12)
|
||||
now2 = now2.replace(year=now2.year - neg_years, month=target_month)
|
||||
else:
|
||||
now2 = now2.replace(month=m - months)
|
||||
else:
|
||||
now2 = now.replace(year=now.year + years)
|
||||
m = now2.month
|
||||
now2 = now2.replace(
|
||||
year=now2.year + (m + months - 1) // 12,
|
||||
month=(m + months - 1) % 12 + 1
|
||||
)
|
||||
logger.info(f"提醒功能:使用相对时间解析 AI 返回值 raw={result} target={now2+dt}")
|
||||
return (now2 + dt, content)
|
||||
|
||||
logger.warning(f"提醒功能:解析 AI 返回值时没有找到解析方法 raw={result}")
|
||||
return (None, "")
|
||||
|
||||
71
konabot/plugins/tqszm/__init__.py
Normal file
71
konabot/plugins/tqszm/__init__.py
Normal file
@ -0,0 +1,71 @@
|
||||
"""
|
||||
提取首字母
|
||||
|
||||
...谁需要啊!!!
|
||||
"""
|
||||
|
||||
import functools
|
||||
|
||||
from loguru import logger
|
||||
from nonebot import on_message
|
||||
from nonebot.rule import KeywordsRule, Rule, ToMeRule
|
||||
from nonebot.adapters.onebot.v11.bot import Bot as OBBot
|
||||
from nonebot.adapters.onebot.v11.event import MessageEvent as OBMessageEvent
|
||||
from nonebot.adapters.onebot.v11.message import MessageSegment as OBMessageSegment
|
||||
from nonebot_plugin_alconna import Text, UniMsg
|
||||
from nonebot_plugin_alconna.uniseg.adapters.onebot11.builder import Onebot11MessageBuilder
|
||||
from pypinyin import pinyin, Style as PinyinStyle
|
||||
|
||||
from konabot.common.longtask import DepLongTaskTarget
|
||||
from konabot.common.apis.ali_content_safety import AlibabaGreen
|
||||
|
||||
|
||||
keywords = ("szmtq", "tqszm", "提取首字母", "首字母提取", )
|
||||
|
||||
|
||||
cmd_tqszm = on_message(rule=Rule(ToMeRule(), KeywordsRule(*keywords)))
|
||||
|
||||
@cmd_tqszm.handle()
|
||||
async def _(target: DepLongTaskTarget, msg: UniMsg, evt: OBMessageEvent | None = None, bot: OBBot | None = None):
|
||||
texts = ""
|
||||
command_occurred = False
|
||||
is_reply_mode = False
|
||||
|
||||
if evt is not None and bot is not None:
|
||||
reply = await Onebot11MessageBuilder().extract_reply(evt, bot)
|
||||
if reply is not None:
|
||||
is_reply_mode = True
|
||||
for seg in reply.data.get('msg', []):
|
||||
if isinstance(seg, OBMessageSegment) and seg.type == 'text':
|
||||
segtext: str = seg.data.get('text', '')
|
||||
texts += segtext
|
||||
|
||||
if is_reply_mode:
|
||||
for seg in msg:
|
||||
if isinstance(seg, Text):
|
||||
if all(seg.text.strip() != k for k in keywords):
|
||||
return
|
||||
else:
|
||||
for seg in msg:
|
||||
if isinstance(seg, Text):
|
||||
if command_occurred:
|
||||
texts += seg.text
|
||||
continue
|
||||
if any(seg.text.startswith(w) for w in keywords):
|
||||
command_occurred = True
|
||||
texts += next(
|
||||
seg.text.removeprefix(w) for w in keywords
|
||||
if seg.text.startswith(w)
|
||||
).removeprefix(" ")
|
||||
|
||||
result = pinyin(texts, style=PinyinStyle.FIRST_LETTER, errors=lambda x: x)
|
||||
print(result)
|
||||
result_text = functools.reduce(lambda x, y: x + y, [
|
||||
p[0] for p in result if len(p) > 0
|
||||
], "")
|
||||
|
||||
if not await AlibabaGreen.detect(result_text):
|
||||
logger.info(f"对首字母序列的检测未通过安全测试 RAW={texts} FORMATTED={result_text}")
|
||||
|
||||
await target.send_message(result_text, at=False)
|
||||
|
||||
304
poetry.lock
generated
304
poetry.lock
generated
@ -20,6 +20,23 @@ type = "legacy"
|
||||
url = "https://pypi.tuna.tsinghua.edu.cn/simple"
|
||||
reference = "mirrors"
|
||||
|
||||
[[package]]
|
||||
name = "aiofiles"
|
||||
version = "24.1.0"
|
||||
description = "File support for asyncio."
|
||||
optional = false
|
||||
python-versions = ">=3.8"
|
||||
groups = ["main"]
|
||||
files = [
|
||||
{file = "aiofiles-24.1.0-py3-none-any.whl", hash = "sha256:b4ec55f4195e3eb5d7abd1bf7e061763e864dd4954231fb8539a0ef8bb8260e5"},
|
||||
{file = "aiofiles-24.1.0.tar.gz", hash = "sha256:22a075c9e5a3810f0c2e48f3008c94d68c65d763b9b03857924c99e57355166c"},
|
||||
]
|
||||
|
||||
[package.source]
|
||||
type = "legacy"
|
||||
url = "https://pypi.tuna.tsinghua.edu.cn/simple"
|
||||
reference = "mirrors"
|
||||
|
||||
[[package]]
|
||||
name = "aiohappyeyeballs"
|
||||
version = "2.6.1"
|
||||
@ -233,6 +250,187 @@ type = "legacy"
|
||||
url = "https://pypi.tuna.tsinghua.edu.cn/simple"
|
||||
reference = "mirrors"
|
||||
|
||||
[[package]]
|
||||
name = "alibabacloud-credentials"
|
||||
version = "1.0.3"
|
||||
description = "The alibabacloud credentials module of alibabaCloud Python SDK."
|
||||
optional = false
|
||||
python-versions = ">=3.7"
|
||||
groups = ["main"]
|
||||
files = [
|
||||
{file = "alibabacloud-credentials-1.0.3.tar.gz", hash = "sha256:9d8707e96afc6f348e23f5677ed15a21c2dfce7cfe6669776548ee4c80e1dfaf"},
|
||||
{file = "alibabacloud_credentials-1.0.3-py3-none-any.whl", hash = "sha256:30c8302f204b663c655d97e1c283ee9f9f84a6257d7901b931477d6cf34445a8"},
|
||||
]
|
||||
|
||||
[package.dependencies]
|
||||
aiofiles = ">=22.1.0,<25.0.0"
|
||||
alibabacloud-credentials-api = ">=1.0.0,<2.0.0"
|
||||
alibabacloud-tea = ">=0.4.0"
|
||||
APScheduler = ">=3.10.0,<4.0.0"
|
||||
|
||||
[package.source]
|
||||
type = "legacy"
|
||||
url = "https://pypi.tuna.tsinghua.edu.cn/simple"
|
||||
reference = "mirrors"
|
||||
|
||||
[[package]]
|
||||
name = "alibabacloud-credentials-api"
|
||||
version = "1.0.0"
|
||||
description = "Alibaba Cloud Gateway SPI SDK Library for Python"
|
||||
optional = false
|
||||
python-versions = ">=3.7"
|
||||
groups = ["main"]
|
||||
files = [
|
||||
{file = "alibabacloud-credentials-api-1.0.0.tar.gz", hash = "sha256:8c340038d904f0218d7214a8f4088c31912bfcf279af2cbc7d9be4897a97dd2f"},
|
||||
]
|
||||
|
||||
[package.source]
|
||||
type = "legacy"
|
||||
url = "https://pypi.tuna.tsinghua.edu.cn/simple"
|
||||
reference = "mirrors"
|
||||
|
||||
[[package]]
|
||||
name = "alibabacloud-endpoint-util"
|
||||
version = "0.0.4"
|
||||
description = "The endpoint-util module of alibabaCloud Python SDK."
|
||||
optional = false
|
||||
python-versions = "*"
|
||||
groups = ["main"]
|
||||
files = [
|
||||
{file = "alibabacloud_endpoint_util-0.0.4.tar.gz", hash = "sha256:a593eb8ddd8168d5dc2216cd33111b144f9189fcd6e9ca20e48f358a739bbf90"},
|
||||
]
|
||||
|
||||
[package.source]
|
||||
type = "legacy"
|
||||
url = "https://pypi.tuna.tsinghua.edu.cn/simple"
|
||||
reference = "mirrors"
|
||||
|
||||
[[package]]
|
||||
name = "alibabacloud-gateway-spi"
|
||||
version = "0.0.3"
|
||||
description = "Alibaba Cloud Gateway SPI SDK Library for Python"
|
||||
optional = false
|
||||
python-versions = ">=3.6"
|
||||
groups = ["main"]
|
||||
files = [
|
||||
{file = "alibabacloud_gateway_spi-0.0.3.tar.gz", hash = "sha256:10d1c53a3fc5f87915fbd6b4985b98338a776e9b44a0263f56643c5048223b8b"},
|
||||
]
|
||||
|
||||
[package.dependencies]
|
||||
alibabacloud_credentials = ">=0.3.4"
|
||||
|
||||
[package.source]
|
||||
type = "legacy"
|
||||
url = "https://pypi.tuna.tsinghua.edu.cn/simple"
|
||||
reference = "mirrors"
|
||||
|
||||
[[package]]
|
||||
name = "alibabacloud-green20220302"
|
||||
version = "3.0.1"
|
||||
description = "Alibaba Cloud Green (20220302) SDK Library for Python"
|
||||
optional = false
|
||||
python-versions = ">=3.6"
|
||||
groups = ["main"]
|
||||
files = [
|
||||
{file = "alibabacloud_green20220302-3.0.1-py3-none-any.whl", hash = "sha256:01cd4a402f8de45423cf5c52c59fc696255ef5f291c3e1e10ac4985033e3e1d9"},
|
||||
{file = "alibabacloud_green20220302-3.0.1.tar.gz", hash = "sha256:083ad03e7c553ec127b69ad5b039f42aaac10730784df319ead2f8a0d8ee928a"},
|
||||
]
|
||||
|
||||
[package.dependencies]
|
||||
alibabacloud-endpoint-util = ">=0.0.4,<1.0.0"
|
||||
alibabacloud-openapi-util = ">=0.2.2,<1.0.0"
|
||||
alibabacloud-tea-openapi = ">=0.3.16,<1.0.0"
|
||||
alibabacloud-tea-util = ">=0.3.13,<1.0.0"
|
||||
|
||||
[package.source]
|
||||
type = "legacy"
|
||||
url = "https://pypi.tuna.tsinghua.edu.cn/simple"
|
||||
reference = "mirrors"
|
||||
|
||||
[[package]]
|
||||
name = "alibabacloud-openapi-util"
|
||||
version = "0.2.2"
|
||||
description = "Aliyun Tea OpenApi Library for Python"
|
||||
optional = false
|
||||
python-versions = "*"
|
||||
groups = ["main"]
|
||||
files = [
|
||||
{file = "alibabacloud_openapi_util-0.2.2.tar.gz", hash = "sha256:ebbc3906f554cb4bf8f513e43e8a33e8b6a3d4a0ef13617a0e14c3dda8ef52a8"},
|
||||
]
|
||||
|
||||
[package.dependencies]
|
||||
alibabacloud_tea_util = ">=0.0.2"
|
||||
cryptography = ">=3.0.0"
|
||||
|
||||
[package.source]
|
||||
type = "legacy"
|
||||
url = "https://pypi.tuna.tsinghua.edu.cn/simple"
|
||||
reference = "mirrors"
|
||||
|
||||
[[package]]
|
||||
name = "alibabacloud-tea"
|
||||
version = "0.4.3"
|
||||
description = "The tea module of alibabaCloud Python SDK."
|
||||
optional = false
|
||||
python-versions = ">=3.7"
|
||||
groups = ["main"]
|
||||
files = [
|
||||
{file = "alibabacloud-tea-0.4.3.tar.gz", hash = "sha256:ec8053d0aa8d43ebe1deb632d5c5404339b39ec9a18a0707d57765838418504a"},
|
||||
]
|
||||
|
||||
[package.dependencies]
|
||||
aiohttp = ">=3.7.0,<4.0.0"
|
||||
requests = ">=2.21.0,<3.0.0"
|
||||
|
||||
[package.source]
|
||||
type = "legacy"
|
||||
url = "https://pypi.tuna.tsinghua.edu.cn/simple"
|
||||
reference = "mirrors"
|
||||
|
||||
[[package]]
|
||||
name = "alibabacloud-tea-openapi"
|
||||
version = "0.4.2"
|
||||
description = "Alibaba Cloud openapi SDK Library for Python"
|
||||
optional = false
|
||||
python-versions = ">=3.7"
|
||||
groups = ["main"]
|
||||
files = [
|
||||
{file = "alibabacloud_tea_openapi-0.4.2-py3-none-any.whl", hash = "sha256:c498065a297fd1972ed7709ef935c9ce1f9757f267f68933de6e63853f37366f"},
|
||||
{file = "alibabacloud_tea_openapi-0.4.2.tar.gz", hash = "sha256:0a8d79374ca692469472355a125969c8a22cc5fb08328c75c26663ccf5c8b168"},
|
||||
]
|
||||
|
||||
[package.dependencies]
|
||||
alibabacloud-credentials = ">=1.0.2,<2.0.0"
|
||||
alibabacloud-gateway-spi = ">=0.0.2,<1.0.0"
|
||||
alibabacloud-tea-util = ">=0.3.13,<1.0.0"
|
||||
cryptography = ">=3.0.0,<45.0.0"
|
||||
darabonba-core = ">=1.0.3,<2.0.0"
|
||||
|
||||
[package.source]
|
||||
type = "legacy"
|
||||
url = "https://pypi.tuna.tsinghua.edu.cn/simple"
|
||||
reference = "mirrors"
|
||||
|
||||
[[package]]
|
||||
name = "alibabacloud-tea-util"
|
||||
version = "0.3.14"
|
||||
description = "The tea-util module of alibabaCloud Python SDK."
|
||||
optional = false
|
||||
python-versions = ">=3.6"
|
||||
groups = ["main"]
|
||||
files = [
|
||||
{file = "alibabacloud_tea_util-0.3.14-py3-none-any.whl", hash = "sha256:10d3e5c340d8f7ec69dd27345eb2fc5a1dab07875742525edf07bbe86db93bfe"},
|
||||
{file = "alibabacloud_tea_util-0.3.14.tar.gz", hash = "sha256:708e7c9f64641a3c9e0e566365d2f23675f8d7c2a3e2971d9402ceede0408cdb"},
|
||||
]
|
||||
|
||||
[package.dependencies]
|
||||
alibabacloud-tea = ">=0.3.3"
|
||||
|
||||
[package.source]
|
||||
type = "legacy"
|
||||
url = "https://pypi.tuna.tsinghua.edu.cn/simple"
|
||||
reference = "mirrors"
|
||||
|
||||
[[package]]
|
||||
name = "annotated-doc"
|
||||
version = "0.0.3"
|
||||
@ -982,6 +1180,93 @@ type = "legacy"
|
||||
url = "https://pypi.tuna.tsinghua.edu.cn/simple"
|
||||
reference = "mirrors"
|
||||
|
||||
[[package]]
|
||||
name = "cryptography"
|
||||
version = "44.0.3"
|
||||
description = "cryptography is a package which provides cryptographic recipes and primitives to Python developers."
|
||||
optional = false
|
||||
python-versions = ">=3.7, !=3.9.0, !=3.9.1"
|
||||
groups = ["main"]
|
||||
files = [
|
||||
{file = "cryptography-44.0.3-cp37-abi3-macosx_10_9_universal2.whl", hash = "sha256:962bc30480a08d133e631e8dfd4783ab71cc9e33d5d7c1e192f0b7c06397bb88"},
|
||||
{file = "cryptography-44.0.3-cp37-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:4ffc61e8f3bf5b60346d89cd3d37231019c17a081208dfbbd6e1605ba03fa137"},
|
||||
{file = "cryptography-44.0.3-cp37-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:58968d331425a6f9eedcee087f77fd3c927c88f55368f43ff7e0a19891f2642c"},
|
||||
{file = "cryptography-44.0.3-cp37-abi3-manylinux_2_28_aarch64.whl", hash = "sha256:e28d62e59a4dbd1d22e747f57d4f00c459af22181f0b2f787ea83f5a876d7c76"},
|
||||
{file = "cryptography-44.0.3-cp37-abi3-manylinux_2_28_armv7l.manylinux_2_31_armv7l.whl", hash = "sha256:af653022a0c25ef2e3ffb2c673a50e5a0d02fecc41608f4954176f1933b12359"},
|
||||
{file = "cryptography-44.0.3-cp37-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:157f1f3b8d941c2bd8f3ffee0af9b049c9665c39d3da9db2dc338feca5e98a43"},
|
||||
{file = "cryptography-44.0.3-cp37-abi3-manylinux_2_34_aarch64.whl", hash = "sha256:c6cd67722619e4d55fdb42ead64ed8843d64638e9c07f4011163e46bc512cf01"},
|
||||
{file = "cryptography-44.0.3-cp37-abi3-manylinux_2_34_x86_64.whl", hash = "sha256:b424563394c369a804ecbee9b06dfb34997f19d00b3518e39f83a5642618397d"},
|
||||
{file = "cryptography-44.0.3-cp37-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:c91fc8e8fd78af553f98bc7f2a1d8db977334e4eea302a4bfd75b9461c2d8904"},
|
||||
{file = "cryptography-44.0.3-cp37-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:25cd194c39fa5a0aa4169125ee27d1172097857b27109a45fadc59653ec06f44"},
|
||||
{file = "cryptography-44.0.3-cp37-abi3-win32.whl", hash = "sha256:3be3f649d91cb182c3a6bd336de8b61a0a71965bd13d1a04a0e15b39c3d5809d"},
|
||||
{file = "cryptography-44.0.3-cp37-abi3-win_amd64.whl", hash = "sha256:3883076d5c4cc56dbef0b898a74eb6992fdac29a7b9013870b34efe4ddb39a0d"},
|
||||
{file = "cryptography-44.0.3-cp39-abi3-macosx_10_9_universal2.whl", hash = "sha256:5639c2b16764c6f76eedf722dbad9a0914960d3489c0cc38694ddf9464f1bb2f"},
|
||||
{file = "cryptography-44.0.3-cp39-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:f3ffef566ac88f75967d7abd852ed5f182da252d23fac11b4766da3957766759"},
|
||||
{file = "cryptography-44.0.3-cp39-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:192ed30fac1728f7587c6f4613c29c584abdc565d7417c13904708db10206645"},
|
||||
{file = "cryptography-44.0.3-cp39-abi3-manylinux_2_28_aarch64.whl", hash = "sha256:7d5fe7195c27c32a64955740b949070f21cba664604291c298518d2e255931d2"},
|
||||
{file = "cryptography-44.0.3-cp39-abi3-manylinux_2_28_armv7l.manylinux_2_31_armv7l.whl", hash = "sha256:3f07943aa4d7dad689e3bb1638ddc4944cc5e0921e3c227486daae0e31a05e54"},
|
||||
{file = "cryptography-44.0.3-cp39-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:cb90f60e03d563ca2445099edf605c16ed1d5b15182d21831f58460c48bffb93"},
|
||||
{file = "cryptography-44.0.3-cp39-abi3-manylinux_2_34_aarch64.whl", hash = "sha256:ab0b005721cc0039e885ac3503825661bd9810b15d4f374e473f8c89b7d5460c"},
|
||||
{file = "cryptography-44.0.3-cp39-abi3-manylinux_2_34_x86_64.whl", hash = "sha256:3bb0847e6363c037df8f6ede57d88eaf3410ca2267fb12275370a76f85786a6f"},
|
||||
{file = "cryptography-44.0.3-cp39-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:b0cc66c74c797e1db750aaa842ad5b8b78e14805a9b5d1348dc603612d3e3ff5"},
|
||||
{file = "cryptography-44.0.3-cp39-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:6866df152b581f9429020320e5eb9794c8780e90f7ccb021940d7f50ee00ae0b"},
|
||||
{file = "cryptography-44.0.3-cp39-abi3-win32.whl", hash = "sha256:c138abae3a12a94c75c10499f1cbae81294a6f983b3af066390adee73f433028"},
|
||||
{file = "cryptography-44.0.3-cp39-abi3-win_amd64.whl", hash = "sha256:5d186f32e52e66994dce4f766884bcb9c68b8da62d61d9d215bfe5fb56d21334"},
|
||||
{file = "cryptography-44.0.3-pp310-pypy310_pp73-macosx_10_9_x86_64.whl", hash = "sha256:cad399780053fb383dc067475135e41c9fe7d901a97dd5d9c5dfb5611afc0d7d"},
|
||||
{file = "cryptography-44.0.3-pp310-pypy310_pp73-manylinux_2_28_aarch64.whl", hash = "sha256:21a83f6f35b9cc656d71b5de8d519f566df01e660ac2578805ab245ffd8523f8"},
|
||||
{file = "cryptography-44.0.3-pp310-pypy310_pp73-manylinux_2_28_x86_64.whl", hash = "sha256:fc3c9babc1e1faefd62704bb46a69f359a9819eb0292e40df3fb6e3574715cd4"},
|
||||
{file = "cryptography-44.0.3-pp310-pypy310_pp73-manylinux_2_34_aarch64.whl", hash = "sha256:e909df4053064a97f1e6565153ff8bb389af12c5c8d29c343308760890560aff"},
|
||||
{file = "cryptography-44.0.3-pp310-pypy310_pp73-manylinux_2_34_x86_64.whl", hash = "sha256:dad80b45c22e05b259e33ddd458e9e2ba099c86ccf4e88db7bbab4b747b18d06"},
|
||||
{file = "cryptography-44.0.3-pp310-pypy310_pp73-win_amd64.whl", hash = "sha256:479d92908277bed6e1a1c69b277734a7771c2b78633c224445b5c60a9f4bc1d9"},
|
||||
{file = "cryptography-44.0.3-pp311-pypy311_pp73-macosx_10_9_x86_64.whl", hash = "sha256:896530bc9107b226f265effa7ef3f21270f18a2026bc09fed1ebd7b66ddf6375"},
|
||||
{file = "cryptography-44.0.3-pp311-pypy311_pp73-manylinux_2_28_aarch64.whl", hash = "sha256:9b4d4a5dbee05a2c390bf212e78b99434efec37b17a4bff42f50285c5c8c9647"},
|
||||
{file = "cryptography-44.0.3-pp311-pypy311_pp73-manylinux_2_28_x86_64.whl", hash = "sha256:02f55fb4f8b79c1221b0961488eaae21015b69b210e18c386b69de182ebb1259"},
|
||||
{file = "cryptography-44.0.3-pp311-pypy311_pp73-manylinux_2_34_aarch64.whl", hash = "sha256:dd3db61b8fe5be220eee484a17233287d0be6932d056cf5738225b9c05ef4fff"},
|
||||
{file = "cryptography-44.0.3-pp311-pypy311_pp73-manylinux_2_34_x86_64.whl", hash = "sha256:978631ec51a6bbc0b7e58f23b68a8ce9e5f09721940933e9c217068388789fe5"},
|
||||
{file = "cryptography-44.0.3-pp311-pypy311_pp73-win_amd64.whl", hash = "sha256:5d20cc348cca3a8aa7312f42ab953a56e15323800ca3ab0706b8cd452a3a056c"},
|
||||
{file = "cryptography-44.0.3.tar.gz", hash = "sha256:fe19d8bc5536a91a24a8133328880a41831b6c5df54599a8417b62fe015d3053"},
|
||||
]
|
||||
|
||||
[package.dependencies]
|
||||
cffi = {version = ">=1.12", markers = "platform_python_implementation != \"PyPy\""}
|
||||
|
||||
[package.extras]
|
||||
docs = ["sphinx (>=5.3.0)", "sphinx-rtd-theme (>=3.0.0) ; python_version >= \"3.8\""]
|
||||
docstest = ["pyenchant (>=3)", "readme-renderer (>=30.0)", "sphinxcontrib-spelling (>=7.3.1)"]
|
||||
nox = ["nox (>=2024.4.15)", "nox[uv] (>=2024.3.2) ; python_version >= \"3.8\""]
|
||||
pep8test = ["check-sdist ; python_version >= \"3.8\"", "click (>=8.0.1)", "mypy (>=1.4)", "ruff (>=0.3.6)"]
|
||||
sdist = ["build (>=1.0.0)"]
|
||||
ssh = ["bcrypt (>=3.1.5)"]
|
||||
test = ["certifi (>=2024)", "cryptography-vectors (==44.0.3)", "pretend (>=0.7)", "pytest (>=7.4.0)", "pytest-benchmark (>=4.0)", "pytest-cov (>=2.10.1)", "pytest-xdist (>=3.5.0)"]
|
||||
test-randomorder = ["pytest-randomly"]
|
||||
|
||||
[package.source]
|
||||
type = "legacy"
|
||||
url = "https://pypi.tuna.tsinghua.edu.cn/simple"
|
||||
reference = "mirrors"
|
||||
|
||||
[[package]]
|
||||
name = "darabonba-core"
|
||||
version = "1.0.4"
|
||||
description = "The darabonba module of alibabaCloud Python SDK."
|
||||
optional = false
|
||||
python-versions = ">=3.7"
|
||||
groups = ["main"]
|
||||
files = [
|
||||
{file = "darabonba_core-1.0.4-py3-none-any.whl", hash = "sha256:4c3bc1d76d5af1087297b6afde8e960ea2f54f93e725e2df8453f0b4bb27dd24"},
|
||||
{file = "darabonba_core-1.0.4.tar.gz", hash = "sha256:6ede4e9bfd458148bab19ab2331716ae9b5c226ba5f6d221de6f88ee65704137"},
|
||||
]
|
||||
|
||||
[package.dependencies]
|
||||
aiohttp = ">=3.7.0,<4.0.0"
|
||||
alibabacloud-tea = "*"
|
||||
requests = ">=2.21.0,<3.0.0"
|
||||
|
||||
[package.source]
|
||||
type = "legacy"
|
||||
url = "https://pypi.tuna.tsinghua.edu.cn/simple"
|
||||
reference = "mirrors"
|
||||
|
||||
[[package]]
|
||||
name = "distro"
|
||||
version = "1.9.0"
|
||||
@ -3454,6 +3739,23 @@ type = "legacy"
|
||||
url = "https://pypi.tuna.tsinghua.edu.cn/simple"
|
||||
reference = "mirrors"
|
||||
|
||||
[[package]]
|
||||
name = "pypinyin"
|
||||
version = "0.55.0"
|
||||
description = "汉字拼音转换模块/工具."
|
||||
optional = false
|
||||
python-versions = ">=2.6, !=3.0.*, !=3.1.*, !=3.2.*, <4"
|
||||
groups = ["main"]
|
||||
files = [
|
||||
{file = "pypinyin-0.55.0-py2.py3-none-any.whl", hash = "sha256:d53b1e8ad2cdb815fb2cb604ed3123372f5a28c6f447571244aca36fc62a286f"},
|
||||
{file = "pypinyin-0.55.0.tar.gz", hash = "sha256:b5711b3a0c6f76e67408ec6b2e3c4987a3a806b7c528076e7c7b86fcf0eaa66b"},
|
||||
]
|
||||
|
||||
[package.source]
|
||||
type = "legacy"
|
||||
url = "https://pypi.tuna.tsinghua.edu.cn/simple"
|
||||
reference = "mirrors"
|
||||
|
||||
[[package]]
|
||||
name = "pytest"
|
||||
version = "9.0.1"
|
||||
@ -4681,4 +4983,4 @@ reference = "mirrors"
|
||||
[metadata]
|
||||
lock-version = "2.1"
|
||||
python-versions = ">=3.12,<4.0"
|
||||
content-hash = "2c341fdc0d5b29ad3b24516c46e036b2eff4c11e244047d114971039255c2ac4"
|
||||
content-hash = "59498c038a603c90f051d2f360cb9226ec0fc4470942c0a7cf34f832701f0ce7"
|
||||
|
||||
@ -29,6 +29,8 @@ dependencies = [
|
||||
"imageio (>=2.37.2,<3.0.0)",
|
||||
"aiosqlite (>=0.20.0,<1.0.0)",
|
||||
"sqlparse (>=0.5.0,<1.0.0)",
|
||||
"alibabacloud-green20220302 (>=3.0.1,<4.0.0)",
|
||||
"pypinyin (>=0.55.0,<0.56.0)",
|
||||
]
|
||||
|
||||
[tool.poetry]
|
||||
|
||||
Reference in New Issue
Block a user