diff --git a/packages/astrbot/commands/conversation.py b/packages/astrbot/commands/conversation.py index 9538d8f53..64f22e223 100644 --- a/packages/astrbot/commands/conversation.py +++ b/packages/astrbot/commands/conversation.py @@ -1,4 +1,5 @@ import datetime +from typing import cast from astrbot.api import logger, sp, star from astrbot.api.event import AstrMessageEvent, MessageEventResult @@ -26,6 +27,7 @@ class ConversationCommands: session_id, curr, ) + assert conv is not None return conv.persona_id def ltm_enabled(self, event: AstrMessageEvent): @@ -287,7 +289,7 @@ class ConversationCommands: if sid: session = str( MessageSesion( - platform_name=message.platform_meta.id, + platform_name=cast(str, message.platform_meta.id), message_type=MessageType("GroupMessage"), session_id=sid, ), diff --git a/packages/astrbot/long_term_memory.py b/packages/astrbot/long_term_memory.py index ceca60ef7..5d40262e6 100644 --- a/packages/astrbot/long_term_memory.py +++ b/packages/astrbot/long_term_memory.py @@ -172,8 +172,9 @@ class LongTermMemory: if event.unified_msg_origin not in self.session_chats: return - if event.get_result() and event.get_result().is_llm_result(): - final_message = f"[You/{datetime.datetime.now().strftime('%H:%M:%S')}]: {event.get_result().get_plain_text()}" + result = event.get_result() + if result is not None and result.is_llm_result(): + final_message = f"[You/{datetime.datetime.now().strftime('%H:%M:%S')}]: {result.get_plain_text()}" logger.debug(f"ltm | {event.unified_msg_origin} | {final_message}") self.session_chats[event.unified_msg_origin].append(final_message) cfg = self.cfg(event) diff --git a/packages/astrbot/process_llm_request.py b/packages/astrbot/process_llm_request.py index 6d8c896f4..093a6dab8 100644 --- a/packages/astrbot/process_llm_request.py +++ b/packages/astrbot/process_llm_request.py @@ -129,6 +129,7 @@ class ProcessLLMRequest: # group name identifier if cfg.get("group_name_display") and event.message_obj.group_id: + assert event.message_obj.group is not None group_name = event.message_obj.group.group_name if group_name: req.system_prompt += f"\nGroup name: {group_name}\n" diff --git a/packages/python_interpreter/main.py b/packages/python_interpreter/main.py index 35a2f2698..1e8e4efa0 100644 --- a/packages/python_interpreter/main.py +++ b/packages/python_interpreter/main.py @@ -14,6 +14,7 @@ from astrbot.api import llm_tool, logger, star from astrbot.api.event import AstrMessageEvent, MessageEventResult, filter from astrbot.api.message_components import File, Image from astrbot.api.provider import ProviderRequest +from astrbot.core.message.components import BaseMessageComponent from astrbot.core.utils.astrbot_path import get_astrbot_data_path from astrbot.core.utils.io import download_file, download_image_by_url @@ -224,6 +225,7 @@ class Main(star.Star): del self.user_waiting[uid] elif isinstance(comp, Image): image_url = comp.url if comp.url else comp.file + assert image_url is not None if image_url.startswith("http"): image_path = await download_image_by_url(image_url) elif image_url.startswith("file:///"): @@ -477,7 +479,9 @@ class Main(star.Star): # file_s3_url = await self.file_upload(file_path) # logger.info(f"文件上传到 AstrBot 云节点: {file_s3_url}") file_name = os.path.basename(file_path) - chain = [File(name=file_name, file=file_path)] + chain: list[BaseMessageComponent] = [ + File(name=file_name, file=file_path) + ] yield event.set_result(MessageEventResult(chain=chain)) elif "Traceback (most recent call last)" in log or "[Error]: " in log: diff --git a/packages/reminder/main.py b/packages/reminder/main.py index eaeec8d73..688f20aa8 100644 --- a/packages/reminder/main.py +++ b/packages/reminder/main.py @@ -5,6 +5,7 @@ import uuid import zoneinfo from apscheduler.schedulers.asyncio import AsyncIOScheduler +from apscheduler.triggers.cron import CronTrigger from astrbot.api import llm_tool, logger, star from astrbot.api.event import AstrMessageEvent, MessageEventResult, filter @@ -62,13 +63,13 @@ class Main(star.Star): misfire_grace_time=60, ) elif "cron" in reminder: + trigger = CronTrigger(**self._parse_cron_expr(reminder["cron"])) self.scheduler.add_job( self._reminder_callback, - trigger="cron", + trigger=trigger, id=id_, args=[group, reminder], misfire_grace_time=60, - **self._parse_cron_expr(reminder["cron"]), ) def check_is_outdated(self, reminder: dict): @@ -101,10 +102,10 @@ class Main(star.Star): async def reminder_tool( self, event: AstrMessageEvent, - text: str = None, - datetime_str: str = None, - cron_expression: str = None, - human_readable_cron: str = None, + text: str | None = None, + datetime_str: str | None = None, + cron_expression: str | None = None, + human_readable_cron: str | None = None, ): """Call this function when user is asking for setting a reminder. @@ -139,17 +140,18 @@ class Main(star.Star): "id": str(uuid.uuid4()), } self.reminder_data[event.unified_msg_origin].append(d) + trigger = CronTrigger(**self._parse_cron_expr(cron_expression)) self.scheduler.add_job( self._reminder_callback, - "cron", + trigger, id=d["id"], misfire_grace_time=60, - **self._parse_cron_expr(cron_expression), args=[event.unified_msg_origin, d], ) if human_readable_cron: reminder_time = f"{human_readable_cron}(Cron: {cron_expression})" else: + assert datetime_str is not None d = {"text": text, "datetime": datetime_str, "id": str(uuid.uuid4())} self.reminder_data[event.unified_msg_origin].append(d) datetime_scheduled = datetime.datetime.strptime( diff --git a/packages/web_searcher/engines/__init__.py b/packages/web_searcher/engines/__init__.py index 706cfa87b..0e4b536fe 100644 --- a/packages/web_searcher/engines/__init__.py +++ b/packages/web_searcher/engines/__init__.py @@ -3,7 +3,7 @@ import urllib.parse from dataclasses import dataclass from aiohttp import ClientSession -from bs4 import BeautifulSoup +from bs4 import BeautifulSoup, Tag HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 6.1; rv:84.0) Gecko/20100101 Firefox/84.0", @@ -30,7 +30,7 @@ USER_AGENTS = [ @dataclass class SearchResult: title: str - url: str + url: str | Tag snippet: str def __str__(self) -> str: @@ -45,13 +45,13 @@ class SearchEngine: self.page = 1 self.headers = HEADERS - def _set_selector(self, selector: str) -> None: + def _set_selector(self, selector: str) -> str: raise NotImplementedError - def _get_next_page(self): + def _get_next_page(self, query: str): raise NotImplementedError - async def _get_html(self, url: str, data: dict = None) -> str: + async def _get_html(self, url: str, data: dict | None = None) -> str: headers = self.headers headers["Referer"] = url headers["User-Agent"] = random.choice(USER_AGENTS) @@ -92,9 +92,12 @@ class SearchEngine: links = soup.select(self._set_selector("links")) results = [] for link in links: - title = self.tidy_text( - link.select_one(self._set_selector("title")).text, - ) + # Safely get the title text (select_one may return None) + title_elem = link.select_one(self._set_selector("title")) + title = "" + if title_elem is not None: + title = self.tidy_text(title_elem.get_text()) + url = link.select_one(self._set_selector("url")) snippet = "" if title and url: diff --git a/packages/web_searcher/engines/sogo.py b/packages/web_searcher/engines/sogo.py index 382e7c937..e6367e66e 100644 --- a/packages/web_searcher/engines/sogo.py +++ b/packages/web_searcher/engines/sogo.py @@ -1,5 +1,6 @@ import random import re +from typing import cast from bs4 import BeautifulSoup @@ -29,7 +30,9 @@ class Sogo(SearchEngine): async def search(self, query: str, num_results: int) -> list[SearchResult]: results = await super().search(query, num_results) for result in results: - result.url = result.url.get("href") + if isinstance(result.url, str): + continue + result.url = cast(str, result.url.get("href")) if result.url.startswith("/link?"): result.url = self.base_url + result.url result.url = await self._parse_url(result.url) @@ -40,7 +43,10 @@ class Sogo(SearchEngine): soup = BeautifulSoup(html, "html.parser") script = soup.find("script") if script: - url = re.search(r'window.location.replace\("(.+?)"\)', script.string).group( - 1, + script_text = ( + script.string if script.string is not None else script.get_text() ) + match = re.search(r'window.location.replace\("(.+?)"\)', script_text) + if match: + url = match.group(1) return url