fix: harden backup import for duplicate platform stats

- 修复 replace 模式下主库清空失败仍继续导入的问题。
- 导入前对 platform_stats 重复键做聚合（count 累加），并统一时间戳判重格式。
- 非法 count 按 0 处理并告警（限流），补充对应测试。

Author: Clhikari

astrbot/core/backup/importer.py

@@ -12,7 +12,7 @@
 import shutil
 import zipfile
 from dataclasses import dataclass, field
-from datetime import datetime
+from datetime import datetime, timezone
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
@@ -452,7 +452,7 @@ async def _clear_main_db(self) -> None:
                         await session.execute(delete(model_class))
                         logger.debug(f"已清空表 {table_name}")
                     except Exception as e:
-                        logger.warning(f"清空表 {table_name} 失败: {e}")
+                        raise RuntimeError(f"清空表 {table_name} 失败: {e}") from e
 
     async def _clear_kb_data(self) -> None:
         """清空知识库数据"""
@@ -494,9 +494,18 @@ async def _import_main_database(
                     if not model_class:
                         logger.warning(f"未知的表: {table_name}")
                         continue
+                    normalized_rows = rows
+                    if table_name == "platform_stats":
+                        normalized_rows, duplicate_count = (
+                            self._merge_platform_stats_rows(rows)
+                        )
+                        if duplicate_count > 0:
+                            logger.warning(
+                                f"检测到 platform_stats 重复键 {duplicate_count} 条，已在导入前聚合"
+                            )
 
                     count = 0
-                    for row in rows:
+                    for row in normalized_rows:
                         try:
                             # 转换 datetime 字符串为 datetime 对象
                             row = self._convert_datetime_fields(row, model_class)
@@ -511,6 +520,86 @@ async def _import_main_database(
 
         return imported
 
+    def _merge_platform_stats_rows(
+        self, rows: list[dict[str, Any]]
+    ) -> tuple[list[dict[str, Any]], int]:
+        merged: dict[tuple[str, str, str], dict[str, Any]] = {}
+        timestamp_cache: dict[str, str] = {}
+        invalid_count_warned = 0
+        invalid_count_warn_limit = 5
+        duplicate_count = 0
+        for row in rows:
+            raw_timestamp = row.get("timestamp")
+            if isinstance(raw_timestamp, str):
+                normalized_timestamp = timestamp_cache.get(raw_timestamp)
+                if normalized_timestamp is None:
+                    normalized_timestamp = self._normalize_platform_stats_timestamp(
+                        raw_timestamp
+                    )
+                    timestamp_cache[raw_timestamp] = normalized_timestamp
+            else:
+                normalized_timestamp = self._normalize_platform_stats_timestamp(
+                    raw_timestamp
+                )
+            key = (
+                normalized_timestamp,
+                str(row.get("platform_id")),
+                str(row.get("platform_type")),
+            )
+            existing = merged.get(key)
+            if existing is None:
+                merged[key] = dict(row)
+                continue
+            duplicate_count += 1
+            existing_raw_count = existing.get("count", 0)
+            try:
+                existing_count = int(existing_raw_count)
+            except (TypeError, ValueError):
+                existing_count = 0
+                if invalid_count_warned < invalid_count_warn_limit:
+                    logger.warning(
+                        "platform_stats count 非法，已按 0 处理: "
+                        f"value={existing_raw_count!r}, key={key}"
+                    )
+                    invalid_count_warned += 1
+
+            incoming_raw_count = row.get("count", 0)
+            try:
+                incoming_count = int(incoming_raw_count)
+            except (TypeError, ValueError):
+                incoming_count = 0
+                if invalid_count_warned < invalid_count_warn_limit:
+                    logger.warning(
+                        "platform_stats count 非法，已按 0 处理: "
+                        f"value={incoming_raw_count!r}, key={key}"
+                    )
+                    invalid_count_warned += 1
+            existing["count"] = existing_count + incoming_count
+        return list(merged.values()), duplicate_count
+
+    def _normalize_platform_stats_timestamp(self, value: Any) -> str:
+        if isinstance(value, datetime):
+            dt = value
+            if dt.tzinfo is not None:
+                dt = dt.astimezone(timezone.utc)
+            return dt.isoformat()
+        if isinstance(value, str):
+            timestamp = value.strip()
+            if not timestamp:
+                return ""
+            if timestamp.endswith("Z"):
+                timestamp = f"{timestamp[:-1]}+00:00"
+            try:
+                dt = datetime.fromisoformat(timestamp)
+                if dt.tzinfo is not None:
+                    dt = dt.astimezone(timezone.utc)
+                return dt.isoformat()
+            except ValueError:
+                return value.strip()
+        if value is None:
+            return ""
+        return str(value)
+
     async def _import_knowledge_bases(
         self,
         zf: zipfile.ZipFile,

tests/test_backup.py

@@ -5,7 +5,7 @@
 import re
 import zipfile
 from datetime import datetime
-from unittest.mock import AsyncMock, MagicMock
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
@@ -308,6 +308,69 @@ def test_convert_datetime_fields(self):
         assert isinstance(result["created_at"], datetime)
         assert isinstance(result["updated_at"], datetime)
 
+    def test_merge_platform_stats_rows(self):
+        """测试 platform_stats 重复键会在导入前聚合"""
+        importer = AstrBotImporter(main_db=MagicMock())
+        rows = [
+            {
+                "id": 1,
+                "timestamp": "2025-12-13T20:00:00Z",
+                "platform_id": "webchat",
+                "platform_type": "unknown",
+                "count": 14,
+            },
+            {
+                "id": 80,
+                "timestamp": "2025-12-13T20:00:00+00:00",
+                "platform_id": "webchat",
+                "platform_type": "unknown",
+                "count": 3,
+            },
+            {
+                "id": 2,
+                "timestamp": "2025-12-13T21:00:00",
+                "platform_id": "aiocqhttp",
+                "platform_type": "unknown",
+                "count": 1,
+            },
+        ]
+
+        merged_rows, duplicate_count = importer._merge_platform_stats_rows(rows)
+
+        assert duplicate_count == 1
+        assert len(merged_rows) == 2
+        first = merged_rows[0]
+        assert first["timestamp"] == "2025-12-13T20:00:00Z"
+        assert first["platform_id"] == "webchat"
+        assert first["platform_type"] == "unknown"
+        assert first["count"] == 17
+
+    def test_merge_platform_stats_rows_warns_on_invalid_count(self):
+        """测试 platform_stats count 非法时会告警并按 0 处理"""
+        importer = AstrBotImporter(main_db=MagicMock())
+        rows = [
+            {
+                "timestamp": "2025-12-13T20:00:00+00:00",
+                "platform_id": "webchat",
+                "platform_type": "unknown",
+                "count": 5,
+            },
+            {
+                "timestamp": "2025-12-13T20:00:00Z",
+                "platform_id": "webchat",
+                "platform_type": "unknown",
+                "count": "bad-count",
+            },
+        ]
+
+        with patch("astrbot.core.backup.importer.logger.warning") as warning_mock:
+            merged_rows, duplicate_count = importer._merge_platform_stats_rows(rows)
+
+        assert duplicate_count == 1
+        assert len(merged_rows) == 1
+        assert merged_rows[0]["count"] == 5
+        assert warning_mock.called
+
     @pytest.mark.asyncio
     async def test_import_file_not_exists(self, mock_main_db, tmp_path):
         """测试导入不存在的文件"""
@@ -365,6 +428,32 @@ async def test_import_major_version_mismatch(self, mock_main_db, tmp_path):
         assert result.success is False
         assert any("主版本不兼容" in err for err in result.errors)
 
+    @pytest.mark.asyncio
+    async def test_import_replace_fails_when_clear_main_db_fails(
+        self, mock_main_db, tmp_path
+    ):
+        """测试 replace 模式下主库清空失败会直接终止导入"""
+        zip_path = tmp_path / "valid_backup.zip"
+        manifest = {
+            "version": "1.1",
+            "astrbot_version": VERSION,
+            "tables": {"platform_stats": 0},
+        }
+        main_data = {"platform_stats": []}
+        with zipfile.ZipFile(zip_path, "w") as zf:
+            zf.writestr("manifest.json", json.dumps(manifest))
+            zf.writestr("databases/main_db.json", json.dumps(main_data))
+
+        importer = AstrBotImporter(main_db=mock_main_db)
+        importer._clear_main_db = AsyncMock(
+            side_effect=RuntimeError("清空表 platform_stats 失败: db locked")
+        )
+
+        result = await importer.import_all(str(zip_path), mode="replace")
+
+        assert result.success is False
+        assert any("清空表 platform_stats 失败" in err for err in result.errors)
+
 
 class TestSecureFilename:
     """安全文件名函数测试"""