fix: harden attachment export ids and tune io chunking

Author: zouyonghe

astrbot/core/backup/exporter.py

@@ -369,10 +369,15 @@ def _export_attachments_sync(
         """在单个线程中批量导出附件，减少高频线程切换。"""
         for attachment in attachments:
             file_path = attachment.get("path", "")
-            attachment_id = attachment.get("attachment_id", "")
+            attachment_id = attachment.get("attachment_id")
             try:
                 if not file_path:
                     continue
+                if not attachment_id:
+                    logger.warning(
+                        f"跳过附件导出：attachment_id 为空 (path={file_path})"
+                    )
+                    continue
                 # 使用 attachment_id 作为文件名
                 ext = os.path.splitext(file_path)[1]
                 archive_path = f"files/attachments/{attachment_id}{ext}"

astrbot/core/utils/io.py

@@ -19,6 +19,7 @@
 from .astrbot_path import get_astrbot_data_path, get_astrbot_path, get_astrbot_temp_path
 
 logger = logging.getLogger("astrbot")
+_DOWNLOAD_READ_CHUNK_SIZE = 64 * 1024
 
 
 def on_error(func, path, exc_info) -> None:
@@ -197,7 +198,7 @@ async def _stream_to_file(
     known_total = total_size if total_size > 0 else None
 
     while True:
-        chunk = await stream.read(8192)
+        chunk = await stream.read(_DOWNLOAD_READ_CHUNK_SIZE)
         if not chunk:
             break
         await asyncio.to_thread(file_obj.write, chunk)

tests/performance/test_benchmarks.py

@@ -176,7 +176,7 @@ async def bench_file_component_get_file() -> None:
         await file_component.get_file()
 
     async def bench_to_thread_exists() -> None:
-        await asyncio.to_thread(Path.exists, exists_path)
+        await asyncio.to_thread(exists_path.exists)
 
     async def bench_export_attachments_existing() -> None:
         if zip_path.exists():

tests/test_backup.py

@@ -275,6 +275,34 @@ async def test_export_attachments_exports_existing_and_skips_missing(
         assert "files/attachments/att_ok.txt" in namelist
         assert "files/attachments/att_missing.txt" not in namelist
 
+    @pytest.mark.asyncio
+    async def test_export_attachments_skips_empty_attachment_id(
+        self, mock_main_db, tmp_path
+    ):
+        """测试附件导出：attachment_id 为空时跳过，避免覆盖冲突。"""
+        exporter = AstrBotExporter(main_db=mock_main_db, kb_manager=None)
+
+        file_a = tmp_path / "a.txt"
+        file_b = tmp_path / "b.txt"
+        file_a.write_text("a", encoding="utf-8")
+        file_b.write_text("b", encoding="utf-8")
+        zip_path = tmp_path / "attachments_empty_id.zip"
+
+        attachments = [
+            {"attachment_id": "", "path": str(file_a)},
+            {"path": str(file_b)},
+            {"attachment_id": "att_ok", "path": str(file_a)},
+        ]
+
+        with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
+            await exporter._export_attachments(zf, attachments)
+
+        with zipfile.ZipFile(zip_path, "r") as zf:
+            namelist = zf.namelist()
+
+        assert "files/attachments/att_ok.txt" in namelist
+        assert "files/attachments/.txt" not in namelist
+
 
 class TestAstrBotImporter:
     """AstrBotImporter 类测试"""