refactor: inline platform stats merge normalization

Author: zouyonghe

astrbot/core/backup/importer.py

@@ -549,34 +549,72 @@ def _merge_platform_stats_rows(
         non_mergeable: list[dict[str, Any]] = []
         invalid_count_warned = 0
 
-        for row in rows:
-            normalized_row, normalized_timestamp, is_timestamp_valid = (
-                self._normalize_platform_stats_row(row)
-            )
+        def normalize_timestamp(value: Any) -> str | None:
+            if isinstance(value, datetime):
+                return self._to_utc_iso(value)
+            if isinstance(value, str):
+                timestamp = value.strip()
+                if not timestamp:
+                    return None
+                if timestamp.endswith("Z"):
+                    timestamp = f"{timestamp[:-1]}+00:00"
+                try:
+                    return self._to_utc_iso(datetime.fromisoformat(timestamp))
+                except ValueError:
+                    return None
+            return None
+
+        def build_key(row: dict[str, Any]) -> tuple[str, str, str] | None:
+            normalized_timestamp = normalize_timestamp(row.get("timestamp"))
+            if normalized_timestamp is not None:
+                row["timestamp"] = normalized_timestamp
+            platform_id = row.get("platform_id")
+            platform_type = row.get("platform_type")
+            if (
+                normalized_timestamp is None
+                or not isinstance(platform_id, str)
+                or not isinstance(platform_type, str)
+            ):
+                return None
+            return (normalized_timestamp, platform_id, platform_type)
+
+        def parse_count(
+            raw_count: Any,
+            key_for_log: tuple[Any, Any, Any],
+            warned_count: int,
+        ) -> tuple[int, int]:
+            if warned_count >= PLATFORM_STATS_INVALID_COUNT_WARN_LIMIT:
+                try:
+                    return int(raw_count), warned_count
+                except (TypeError, ValueError):
+                    return 0, warned_count
+            try:
+                return int(raw_count), warned_count
+            except (TypeError, ValueError):
+                logger.warning(
+                    "platform_stats count 非法，已按 0 处理: value=%r, key=%s",
+                    raw_count,
+                    key_for_log,
+                )
+                return 0, warned_count + 1
 
-            platform_id = normalized_row.get("platform_id")
-            platform_type = normalized_row.get("platform_type")
+        for row in rows:
+            normalized_row = dict(row)
+            key = build_key(normalized_row)
             key_for_log = (
-                normalized_timestamp if is_timestamp_valid else "<invalid_timestamp>",
-                repr(platform_id),
-                repr(platform_type),
+                normalized_row.get("timestamp"),
+                repr(normalized_row.get("platform_id")),
+                repr(normalized_row.get("platform_type")),
             )
-            count, invalid_count_warned = self._parse_platform_stats_count(
-                normalized_row.get("count", 0),
-                key_for_log,
-                invalid_count_warned,
+            count, invalid_count_warned = parse_count(
+                normalized_row.get("count", 0), key_for_log, invalid_count_warned
             )
             normalized_row["count"] = count
 
-            if not is_timestamp_valid:
-                non_mergeable.append(normalized_row)
-                continue
-
-            if not isinstance(platform_id, str) or not isinstance(platform_type, str):
+            if key is None:
                 non_mergeable.append(normalized_row)
                 continue
 
-            key = (normalized_timestamp, platform_id, platform_type)
             existing = merged.get(key)
             if existing is None:
                 merged[key] = normalized_row
@@ -585,71 +623,13 @@ def _merge_platform_stats_rows(
 
         return [*non_mergeable, *merged.values()]
 
-    def _parse_platform_stats_count(
-        self,
-        raw_count: Any,
-        key_for_log: tuple[str, str, str],
-        warned_count: int,
-    ) -> tuple[int, int]:
-        if warned_count >= PLATFORM_STATS_INVALID_COUNT_WARN_LIMIT:
-            try:
-                return int(raw_count), warned_count
-            except (TypeError, ValueError):
-                return 0, warned_count
-        try:
-            return int(raw_count), warned_count
-        except (TypeError, ValueError):
-            logger.warning(
-                "platform_stats count 非法，已按 0 处理: value=%r, key=%s",
-                raw_count,
-                key_for_log,
-            )
-            return 0, warned_count + 1
-
-    def _normalize_platform_stats_row(
-        self, row: dict[str, Any]
-    ) -> tuple[dict[str, Any], str, bool]:
-        normalized_row = dict(row)
-        raw_timestamp = normalized_row.get("timestamp")
-        normalized_timestamp = self._normalize_platform_stats_timestamp(raw_timestamp)
-        if normalized_timestamp is None:
-            if isinstance(raw_timestamp, str):
-                normalized_row["timestamp"] = raw_timestamp.strip()
-            elif raw_timestamp is None:
-                normalized_row["timestamp"] = ""
-            else:
-                normalized_row["timestamp"] = str(raw_timestamp)
-            return normalized_row, normalized_row["timestamp"], False
-        normalized_row["timestamp"] = normalized_timestamp
-        return normalized_row, normalized_timestamp, True
-
     def _to_utc_iso(self, dt: datetime) -> str:
         if dt.tzinfo is None:
             dt = dt.replace(tzinfo=timezone.utc)
         else:
             dt = dt.astimezone(timezone.utc)
         return dt.isoformat()
 
-    def _normalize_platform_stats_timestamp(self, value: Any) -> str | None:
-        if isinstance(value, datetime):
-            return self._to_utc_iso(value)
-
-        if isinstance(value, str):
-            timestamp = value.strip()
-            if not timestamp:
-                return None
-            if timestamp.endswith("Z"):
-                timestamp = f"{timestamp[:-1]}+00:00"
-            try:
-                return self._to_utc_iso(datetime.fromisoformat(timestamp))
-            except ValueError:
-                return None
-
-        if value is None:
-            return None
-
-        return None
-
     async def _import_knowledge_bases(
         self,
         zf: zipfile.ZipFile,

tests/test_backup.py

@@ -377,17 +377,30 @@ def test_merge_platform_stats_rows(self):
         assert aiocq_row is not None
         assert aiocq_row["timestamp"] == "2025-12-13T21:00:00+00:00"
 
-    def test_normalize_platform_stats_timestamp_treats_naive_as_utc(self):
-        """测试 naive timestamp 会统一转为显式 UTC 偏移"""
+    def test_merge_platform_stats_rows_normalizes_naive_timestamp_to_utc(self):
+        """测试 platform_stats 合并前会将 naive timestamp 标准化为 UTC 偏移"""
         importer = AstrBotImporter(main_db=MagicMock())
 
-        normalized = importer._normalize_platform_stats_timestamp("2025-12-13T21:00:00")
-        assert normalized == "2025-12-13T21:00:00+00:00"
+        rows = [
+            {
+                "timestamp": "2025-12-13T21:00:00",
+                "platform_id": "webchat",
+                "platform_type": "unknown",
+                "count": 1,
+            },
+            {
+                "timestamp": datetime(2025, 12, 13, 22, 0, 0),
+                "platform_id": "telegram",
+                "platform_type": "unknown",
+                "count": 1,
+            },
+        ]
 
-        normalized_dt = importer._normalize_platform_stats_timestamp(
-            datetime(2025, 12, 13, 21, 0, 0)
-        )
-        assert normalized_dt == "2025-12-13T21:00:00+00:00"
+        merged_rows = importer._merge_platform_stats_rows(rows)
+        assert len(merged_rows) == 2
+        by_platform = {row["platform_id"]: row for row in merged_rows}
+        assert by_platform["webchat"]["timestamp"] == "2025-12-13T21:00:00+00:00"
+        assert by_platform["telegram"]["timestamp"] == "2025-12-13T22:00:00+00:00"
 
     def test_merge_platform_stats_rows_warns_on_invalid_count(self):
         """测试 platform_stats count 非法时会告警并按 0 处理（含上限）"""