fz-mirrors/AstrBot
1
0
Fork 0
mirror of https://github.com/AstrBotDevs/AstrBot synced 2026-07-19 18:47:41 +08:00
Code Issues Packages Projects Releases Wiki Activity

chore(core.utils): 🚨 修正错误Lint

Browse Source
This commit is contained in:
Dt8333
2025-11-01 17:01:57 +08:00
parent 0b7fc29ac4
commit b26d8d711b
7 changed files with 57 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
astrbot/core/utils/io.py
View File

@@ -49,7 +49,7 @@ def port_checker(port: int, host: str = "localhost"):
return False
def save_temp_img(img: Image.Image | str) -> str:
def save_temp_img(img: Image.Image | bytes) -> str:
temp_dir = os.path.join(get_astrbot_data_path(), "temp")
# 获得文件创建时间，清除超过 12 小时的
try:
@@ -77,7 +77,7 @@ def save_temp_img(img: Image.Image | str) -> str:
async def download_image_by_url(
url: str,
post: bool = False,
post_data: dict = None,
post_data: dict | None = None,
path=None,
) -> str:
"""下载图片, 返回 path"""

8
astrbot/core/utils/pip_installer.py
View File

@@ -6,15 +6,15 @@ logger = logging.getLogger("astrbot")
class PipInstaller:
def __init__(self, pip_install_arg: str, pypi_index_url: str = None):
def __init__(self, pip_install_arg: str, pypi_index_url: str | None = None):
self.pip_install_arg = pip_install_arg
self.pypi_index_url = pypi_index_url
async def install(
self,
package_name: str = None,
requirements_path: str = None,
mirror: str = None,
package_name: str | None = None,
requirements_path: str | None = None,
mirror: str | None = None,
):
args = ["install"]
if package_name:

27
astrbot/core/utils/session_waiter.py
View File

@@ -20,16 +20,16 @@ class SessionController:
def __init__(self):
self.future = asyncio.Future()
self.current_event: asyncio.Event = None
self.current_event: asyncio.Event | None = None
"""当前正在等待的所用的异步事件"""
self.ts: float = None
self.ts: float | None = None
"""上次保持(keep)开始时的时间"""
self.timeout: float | int = None
self.timeout: float | int | None = None
"""上次保持(keep)开始时的超时时间"""
self.history_chains: list[list[Comp.BaseMessageComponent]] = []
def stop(self, error: Exception = None):
def stop(self, error: Exception | None = None):
"""立即结束这个会话"""
if not self.future.done():
if error:
@@ -53,6 +53,8 @@ class SessionController:
self.stop()
return
else:
assert self.timeout is not None
assert self.ts is not None
left_timeout = self.timeout - (new_ts - self.ts)
timeout = left_timeout + timeout
if timeout <= 0:
@@ -69,7 +71,7 @@ class SessionController:
asyncio.create_task(self._holding(new_event, timeout)) # 开始新的 keep
async def _holding(self, event: asyncio.Event, timeout: int):
async def _holding(self, event: asyncio.Event, timeout: float):
"""等待事件结束或超时"""
try:
await asyncio.wait_for(event.wait(), timeout)
@@ -108,7 +110,9 @@ class SessionWaiter:
):
self.session_id = session_id
self.session_filter = session_filter
self.handler: Callable[[str], Awaitable[Any]] | None = None # 处理函数
self.handler: (
Callable[[SessionController, AstrMessageEvent], Awaitable[Any]] | None
) = None # 处理函数
self.session_controller = SessionController()
self.record_history_chains = record_history_chains
@@ -119,7 +123,7 @@ class SessionWaiter:
async def register_wait(
self,
handler: Callable[[str], Awaitable[Any]],
handler: Callable[[SessionController, AstrMessageEvent], Awaitable[Any]],
timeout: int = 30,
) -> Any:
"""等待外部输入并处理"""
@@ -137,7 +141,7 @@ class SessionWaiter:
finally:
self._cleanup()
def _cleanup(self, error: Exception = None):
def _cleanup(self, error: Exception | None = None):
"""清理会话"""
USER_SESSIONS.pop(self.session_id, None)
try:
@@ -161,6 +165,7 @@ class SessionWaiter:
)
try:
# TODO: 这里使用 create_task，跟踪 task，防止超时后这里 handler 仍然在执行
assert session.handler is not None
await session.handler(session.session_controller, event)
except Exception as e:
session.session_controller.stop(e)
@@ -173,11 +178,13 @@ def session_waiter(timeout: int = 30, record_history_chains: bool = False):
:param record_history_chain: 是否自动记录历史消息链。可以通过 controller.get_history_chains() 获取。深拷贝。
"""
def decorator(func: Callable[[str], Awaitable[Any]]):
def decorator(
func: Callable[[SessionController, AstrMessageEvent], Awaitable[Any]],
):
@functools.wraps(func)
async def wrapper(
event: AstrMessageEvent,
session_filter: SessionFilter = None,
session_filter: SessionFilter | None = None,
*args,
**kwargs,
):

4
astrbot/core/utils/t2i/__init__.py
View File

@@ -3,11 +3,11 @@ from abc import ABC, abstractmethod
class RenderStrategy(ABC):
@abstractmethod
def render(self, text: str, return_url: bool) -> str:
async def render(self, text: str, return_url: bool) -> str:
pass
@abstractmethod
def render_custom_template(
async def render_custom_template(
self,
tmpl_str: str,
tmpl_data: dict,

56
astrbot/core/utils/t2i/local_strategy.py
View File

@@ -20,7 +20,7 @@ class FontManager:
_font_cache = {}
@classmethod
def get_font(cls, size: int) -> ImageFont.FreeTypeFont:
def get_font(cls, size: int) -> ImageFont.FreeTypeFont|ImageFont.ImageFont:
"""获取指定大小的字体，优先从缓存获取"""
if size in cls._font_cache:
return cls._font_cache[size]
@@ -66,23 +66,17 @@ class TextMeasurer:
"""测量文本尺寸的工具类"""
@staticmethod
def get_text_size(text: str, font: ImageFont.FreeTypeFont) -> Tuple[int, int]:
def get_text_size(text: str, font: ImageFont.FreeTypeFont|ImageFont.ImageFont) -> tuple[int, int]:
"""获取文本的尺寸"""
try:
# PIL 9.0.0 以上版本
return (
font.getbbox(text)[2:]
if hasattr(font, "getbbox")
else font.getsize(text)
)
except Exception:
# 兼容旧版本
return font.getsize(text)
# 依赖库Pillow>=11.2.1，不再需要考虑<9.0.0
left, top, right, bottom = font.getbbox("Hello world")
return int(right - left), int(bottom - top)
@staticmethod
def split_text_to_fit_width(
text: str, font: ImageFont.FreeTypeFont, max_width: int
) -> List[str]:
text: str, font: ImageFont.FreeTypeFont|ImageFont.ImageFont, max_width: int
) -> list[str]:
"""将文本拆分为多行，确保每行不超过指定宽度"""
lines = []
if not text:
@@ -126,7 +120,7 @@ class MarkdownElement(ABC):
def render(
self,
image: Image.Image,
draw: ImageDraw.Draw,
draw: ImageDraw.ImageDraw,
x: int,
y: int,
image_width: int,
@@ -152,7 +146,7 @@ class TextElement(MarkdownElement):
def render(
self,
image: Image.Image,
draw: ImageDraw.Draw,
draw: ImageDraw.ImageDraw,
x: int,
y: int,
image_width: int,
@@ -186,7 +180,7 @@ class BoldTextElement(MarkdownElement):
def render(
self,
image: Image.Image,
draw: ImageDraw.Draw,
draw: ImageDraw.ImageDraw,
x: int,
y: int,
image_width: int,
@@ -251,7 +245,7 @@ class ItalicTextElement(MarkdownElement):
def render(
self,
image: Image.Image,
draw: ImageDraw.Draw,
draw: ImageDraw.ImageDraw,
x: int,
y: int,
image_width: int,
@@ -299,7 +293,7 @@ class ItalicTextElement(MarkdownElement):
# 倾斜变换，使用仿射变换实现斜体效果
# 变换矩阵: [1, 0.2, 0, 0, 1, 0]
italic_img = text_img.transform(
text_img.size, Image.AFFINE, (1, 0.2, 0, 0, 1, 0), Image.BICUBIC
text_img.size, Image.Transform.AFFINE, (1, 0.2, 0, 0, 1, 0), Image.Resampling.BICUBIC
)
# 粘贴到原图像
@@ -331,7 +325,7 @@ class UnderlineTextElement(MarkdownElement):
def render(
self,
image: Image.Image,
draw: ImageDraw.Draw,
draw: ImageDraw.ImageDraw,
x: int,
y: int,
image_width: int,
@@ -371,7 +365,7 @@ class StrikethroughTextElement(MarkdownElement):
def render(
self,
image: Image.Image,
draw: ImageDraw.Draw,
draw: ImageDraw.ImageDraw,
x: int,
y: int,
image_width: int,
@@ -422,7 +416,7 @@ class HeaderElement(MarkdownElement):
def render(
self,
image: Image.Image,
draw: ImageDraw.Draw,
draw: ImageDraw.ImageDraw,
x: int,
y: int,
image_width: int,
@@ -458,7 +452,7 @@ class QuoteElement(MarkdownElement):
def render(
self,
image: Image.Image,
draw: ImageDraw.Draw,
draw: ImageDraw.ImageDraw,
x: int,
y: int,
image_width: int,
@@ -502,7 +496,7 @@ class ListItemElement(MarkdownElement):
def render(
self,
image: Image.Image,
draw: ImageDraw.Draw,
draw: ImageDraw.ImageDraw,
x: int,
y: int,
image_width: int,
@@ -532,7 +526,7 @@ class ListItemElement(MarkdownElement):
class CodeBlockElement(MarkdownElement):
"""代码块元素"""
def __init__(self, content: List[str]):
def __init__(self, content: list[str]):
super().__init__("\n".join(content))
def calculate_height(self, image_width: int, font_size: int) -> int:
@@ -552,7 +546,7 @@ class CodeBlockElement(MarkdownElement):
def render(
self,
image: Image.Image,
draw: ImageDraw.Draw,
draw: ImageDraw.ImageDraw,
x: int,
y: int,
image_width: int,
@@ -595,7 +589,7 @@ class InlineCodeElement(MarkdownElement):
def render(
self,
image: Image.Image,
draw: ImageDraw.Draw,
draw: ImageDraw.ImageDraw,
x: int,
y: int,
image_width: int,
@@ -667,7 +661,7 @@ class ImageElement(MarkdownElement):
def render(
self,
image: Image.Image,
draw: ImageDraw.Draw,
draw: ImageDraw.ImageDraw,
x: int,
y: int,
image_width: int,
@@ -686,7 +680,7 @@ class ImageElement(MarkdownElement):
if pasted_image.width > max_width:
ratio = max_width / pasted_image.width
new_size = (int(max_width), int(pasted_image.height * ratio))
pasted_image = pasted_image.resize(new_size, Image.LANCZOS)
pasted_image = pasted_image.resize(new_size, Image.Resampling.LANCZOS)
# 计算居中位置
paste_x = x + (image_width - pasted_image.width) // 2 - 10
@@ -705,7 +699,7 @@ class MarkdownParser:
"""Markdown解析器，将文本解析为元素"""
@staticmethod
async def parse(text: str) -> List[MarkdownElement]:
async def parse(text: str) -> list[MarkdownElement]:
elements = []
lines = text.split("\n")
@@ -847,7 +841,7 @@ class MarkdownRenderer:
self,
font_size: int = 26,
width: int = 800,
bg_color: Tuple[int, int, int] = (255, 255, 255),
bg_color: tuple[int, int, int] = (255, 255, 255),
):
self.font_size = font_size
self.width = width

2
astrbot/core/utils/tencent_record_helper.py
View File

@@ -68,7 +68,7 @@ async def convert_to_pcm_wav(input_path: str, output_path: str) -> str:
from pyffmpeg import FFmpeg
ff = FFmpeg()
ff.convert(input=input_path, output=output_path)
ff.convert(input_file=input_path, output_file=output_path)
except Exception as e:
logger.debug(f"pyffmpeg 转换失败: {e}, 尝试使用 ffmpeg 命令行进行转换")

9
astrbot/core/utils/version_comparator.py
View File

@@ -60,9 +60,12 @@ class VersionComparator:
return -1
if isinstance(p1, str) and isinstance(p2, int):
return 1
if (isinstance(p1, int) and isinstance(p2, int)) or (
isinstance(p1, str) and isinstance(p2, str)
):
if isinstance(p1, int) and isinstance(p2, int):
if p1 > p2:
return 1
if p1 < p2:
return -1
if isinstance(p1, str) and isinstance(p2, str):
if p1 > p2:
return 1
if p1 < p2: