diff --git a/astrbot/core/utils/io.py b/astrbot/core/utils/io.py index bd0bea920..384a995f6 100644 --- a/astrbot/core/utils/io.py +++ b/astrbot/core/utils/io.py @@ -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""" diff --git a/astrbot/core/utils/pip_installer.py b/astrbot/core/utils/pip_installer.py index abe247146..6076a114a 100644 --- a/astrbot/core/utils/pip_installer.py +++ b/astrbot/core/utils/pip_installer.py @@ -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: diff --git a/astrbot/core/utils/session_waiter.py b/astrbot/core/utils/session_waiter.py index 33b7cb17a..e1f2fbef7 100644 --- a/astrbot/core/utils/session_waiter.py +++ b/astrbot/core/utils/session_waiter.py @@ -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, ): diff --git a/astrbot/core/utils/t2i/__init__.py b/astrbot/core/utils/t2i/__init__.py index 5038a46f7..e4112c354 100644 --- a/astrbot/core/utils/t2i/__init__.py +++ b/astrbot/core/utils/t2i/__init__.py @@ -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, diff --git a/astrbot/core/utils/t2i/local_strategy.py b/astrbot/core/utils/t2i/local_strategy.py index 19eab2efe..2fa235129 100644 --- a/astrbot/core/utils/t2i/local_strategy.py +++ b/astrbot/core/utils/t2i/local_strategy.py @@ -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 diff --git a/astrbot/core/utils/tencent_record_helper.py b/astrbot/core/utils/tencent_record_helper.py index 9cc36571e..08347f922 100644 --- a/astrbot/core/utils/tencent_record_helper.py +++ b/astrbot/core/utils/tencent_record_helper.py @@ -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 命令行进行转换") diff --git a/astrbot/core/utils/version_comparator.py b/astrbot/core/utils/version_comparator.py index e3bf74951..4ad2da10e 100644 --- a/astrbot/core/utils/version_comparator.py +++ b/astrbot/core/utils/version_comparator.py @@ -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: