Merge branch 'AstrBotDevs:master' into master

Author: elecvoid243

astrbot/core/pipeline/result_decorate/stage.py

@@ -289,7 +289,9 @@ async def process(
                         ),
                     )
                 else:
-                    result.chain.insert(0, Plain(f"🤔 思考: {reasoning_content}\n"))
+                    result.chain.insert(
+                        0, Plain(f"🤔 思考: {reasoning_content}\n\n────\n")
+                    )
 
             if should_tts and tts_provider:
                 new_chain = []

astrbot/core/star/updator.py

@@ -1,10 +1,9 @@
 import os
-import shutil
 import zipfile
 
 from astrbot.core import logger
 from astrbot.core.utils.astrbot_path import get_astrbot_plugin_path
-from astrbot.core.utils.io import ensure_dir, on_error, remove_dir
+from astrbot.core.utils.io import ensure_dir, remove_dir
 
 from ..star.star import StarMetadata
 from ..updator import RepoZipUpdator
@@ -72,28 +71,9 @@ async def update(
 
     def unzip_file(self, zip_path: str, target_dir: str) -> None:
         ensure_dir(target_dir)
-        update_dir = ""
         logger.info(f"Extracting archive: {zip_path}")
         with zipfile.ZipFile(zip_path, "r") as z:
-            update_dir = z.namelist()[0]
+            update_dir = self._resolve_archive_root_dir(z.namelist())
             z.extractall(target_dir)
 
-        files = os.listdir(os.path.join(target_dir, update_dir))
-        for f in files:
-            if os.path.isdir(os.path.join(target_dir, update_dir, f)):
-                if os.path.exists(os.path.join(target_dir, f)):
-                    shutil.rmtree(os.path.join(target_dir, f), onerror=on_error)
-            elif os.path.exists(os.path.join(target_dir, f)):
-                os.remove(os.path.join(target_dir, f))
-            shutil.move(os.path.join(target_dir, update_dir, f), target_dir)
-
-        try:
-            logger.info(
-                f"Removing temporary files: {zip_path} and {os.path.join(target_dir, update_dir)}",
-            )
-            shutil.rmtree(os.path.join(target_dir, update_dir), onerror=on_error)
-            os.remove(zip_path)
-        except BaseException:
-            logger.warning(
-                f"Failed to remove update files; you can manually delete {zip_path} and {os.path.join(target_dir, update_dir)}",
-            )
+        self._finalize_extracted_archive(zip_path, target_dir, update_dir)

astrbot/core/utils/io.py

@@ -289,9 +289,23 @@ async def download_dashboard(
             )
         except BaseException as _:
             if latest:
-                dashboard_release_url = "https://github.com/AstrBotDevs/AstrBot/releases/latest/download/dist.zip"
+                # Resolve latest release tag from GitHub API to construct correct asset URL
+                ssl_context = ssl.create_default_context(cafile=certifi.where())
+                async with aiohttp.ClientSession(
+                    connector=aiohttp.TCPConnector(ssl=ssl_context),
+                    trust_env=True,
+                ) as session:
+                    async with session.get(
+                        "https://api.github.com/repos/AstrBotDevs/AstrBot/releases/latest",
+                        timeout=30,
+                        headers={"Accept": "application/vnd.github+json"},
+                    ) as api_resp:
+                        api_resp.raise_for_status()
+                        release_data = await api_resp.json()
+                        tag = release_data["tag_name"]
             else:
-                dashboard_release_url = f"https://github.com/AstrBotDevs/AstrBot/releases/download/{version}/dist.zip"
+                tag = version
+            dashboard_release_url = f"https://github.com/AstrBotDevs/AstrBot/releases/download/{tag}/AstrBot-{tag}-dashboard.zip"
             if proxy:
                 dashboard_release_url = f"{proxy}/{dashboard_release_url}"
             await download_file(

astrbot/core/zip_updator.py

@@ -234,30 +234,87 @@ def parse_github_url(self, url: str):
     def unzip_file(self, zip_path: str, target_dir: str) -> None:
         """解压缩文件, 并将压缩包内**第一个**文件夹内的文件移动到 target_dir"""
         ensure_dir(target_dir)
-        update_dir = ""
         with zipfile.ZipFile(zip_path, "r") as z:
-            update_dir = z.namelist()[0]
+            update_dir = self._resolve_archive_root_dir(z.namelist())
             z.extractall(target_dir)
         logger.debug(f"解压文件完成: {zip_path}")
 
-        files = os.listdir(os.path.join(target_dir, update_dir))
+        self._finalize_extracted_archive(zip_path, target_dir, update_dir)
+
+    @staticmethod
+    def _resolve_archive_root_dir(entries: list[str]) -> str:
+        normalized_entries = [os.path.normpath(entry) for entry in entries]
+        portable_entries = [entry.replace("\\", "/") for entry in normalized_entries]
+        root_candidates: list[str] = []
+
+        for raw_entry, normalized_entry, portable_entry in zip(
+            entries, normalized_entries, portable_entries
+        ):
+            if normalized_entry == ".":
+                continue
+
+            has_children = any(
+                other_entry != portable_entry
+                and other_entry.startswith(f"{portable_entry}/")
+                for other_entry in portable_entries
+            )
+            if raw_entry.endswith(("/", "\\")) or has_children:
+                root_candidates.append(normalized_entry)
+                continue
+
+            parent_portable, _, _ = portable_entry.rpartition("/")
+            if not parent_portable:
+                return ""
+            root_candidates.append(parent_portable.replace("/", os.sep))
+
+        if not root_candidates:
+            return ""
+        return os.path.commonpath(root_candidates)
+
+    def _finalize_extracted_archive(
+        self,
+        zip_path: str,
+        target_dir: str,
+        update_dir: str,
+    ) -> None:
+        target_root_path = os.path.normpath(target_dir)
+
+        def _join_under_root(root: str, *parts: str) -> str:
+            path = os.path.normpath(os.path.join(root, *parts))
+            try:
+                if os.path.commonpath([root, path]) != root:
+                    raise ValueError("path escapes root directory")
+            except ValueError as exc:
+                raise ValueError("path escapes root directory") from exc
+            return path
+
+        if not update_dir:
+            try:
+                os.remove(zip_path)
+            except Exception:
+                logger.warning(f"删除更新文件失败，可以手动删除 {zip_path}")
+            return
+
+        update_root_path = _join_under_root(target_root_path, update_dir)
+
+        files = os.listdir(update_root_path)
         for f in files:
-            if os.path.isdir(os.path.join(target_dir, update_dir, f)):
-                if os.path.exists(os.path.join(target_dir, f)):
-                    shutil.rmtree(os.path.join(target_dir, f), onerror=on_error)
-            elif os.path.exists(os.path.join(target_dir, f)):
-                os.remove(os.path.join(target_dir, f))
-            shutil.move(os.path.join(target_dir, update_dir, f), target_dir)
+            update_item_path = _join_under_root(update_root_path, f)
+            target_item_path = _join_under_root(target_root_path, f)
+            if os.path.isdir(update_item_path):
+                if os.path.exists(target_item_path):
+                    shutil.rmtree(target_item_path, onerror=on_error)
+            elif os.path.exists(target_item_path):
+                os.remove(target_item_path)
+            shutil.move(update_item_path, target_root_path)
 
         try:
-            logger.debug(
-                f"删除临时更新文件: {zip_path} 和 {os.path.join(target_dir, update_dir)}",
-            )
-            shutil.rmtree(os.path.join(target_dir, update_dir), onerror=on_error)
+            logger.debug(f"删除临时更新文件: {zip_path} 和 {update_root_path}")
+            shutil.rmtree(update_root_path, onerror=on_error)
             os.remove(zip_path)
-        except BaseException:
+        except Exception:
             logger.warning(
-                f"删除更新文件失败，可以手动删除 {zip_path} 和 {os.path.join(target_dir, update_dir)}",
+                f"删除更新文件失败，可以手动删除 {zip_path} 和 {update_root_path}"
             )
 
     def format_name(self, name: str) -> str: