perf: metrics

astrbot/core/platform/platform.py

@@ -1,4 +1,5 @@
 import abc
+import asyncio
 import uuid
 from asyncio import Queue
 from collections.abc import Coroutine
@@ -138,7 +139,9 @@ class Platform(abc.ABC):
 
         异步方法。
         """
-        await Metric.upload(msg_event_tick=1, adapter_name=self.meta().name)
+        asyncio.create_task(
+            Metric.upload(msg_event_tick=1, adapter_name=self.meta().name)
+        )
 
     def commit_event(self, event: AstrMessageEvent) -> None:
         """提交一个事件到事件队列。"""

astrbot/core/platform/sources/lark/lark_event.py

@@ -949,7 +949,9 @@ class LarkMessageEvent(AstrMessageEvent):
             buffer.squash_plain()
             await self.send(buffer)
 
-        await Metric.upload(msg_event_tick=1, adapter_name=self.platform_meta.name)
+        asyncio.create_task(
+            Metric.upload(msg_event_tick=1, adapter_name=self.platform_meta.name)
+        )
         self._has_send_oper = True
 
     async def send_streaming(self, generator, use_fallback: bool = False):
@@ -1000,7 +1002,9 @@ class LarkMessageEvent(AstrMessageEvent):
             if buffer:
                 buffer.squash_plain()
                 await self.send(buffer)
-            await Metric.upload(msg_event_tick=1, adapter_name=self.platform_meta.name)
+            asyncio.create_task(
+                Metric.upload(msg_event_tick=1, adapter_name=self.platform_meta.name)
+            )
             self._has_send_oper = True
 
         async def _flush_and_close_card() -> None:
@@ -1075,8 +1079,10 @@ class LarkMessageEvent(AstrMessageEvent):
         # If no text was produced at all, no card was created
         if card_id is None:
             if not fallback_used:
-                await Metric.upload(
-                    msg_event_tick=1, adapter_name=self.platform_meta.name
+                asyncio.create_task(
+                    Metric.upload(
+                        msg_event_tick=1, adapter_name=self.platform_meta.name
+                    )
                 )
                 self._has_send_oper = True
             return
@@ -1084,5 +1090,7 @@ class LarkMessageEvent(AstrMessageEvent):
         await _flush_and_close_card()
 
         # 内联父类 send_streaming 的副作用
-        await Metric.upload(msg_event_tick=1, adapter_name=self.platform_meta.name)
+        asyncio.create_task(
+            Metric.upload(msg_event_tick=1, adapter_name=self.platform_meta.name)
+        )
         self._has_send_oper = True

astrbot/core/utils/metrics.py

@@ -1,7 +1,10 @@
+import asyncio
 import os
 import socket
 import sys
 import uuid
+from contextlib import suppress
+from typing import Any
 
 import aiohttp
 
@@ -11,6 +14,14 @@ from astrbot.core.config import VERSION
 
 class Metric:
     _iid_cache = None
+    _has_uploaded_once = False
+    _upload_interval_seconds = 10 * 60
+    _max_pending_metric_groups = 64
+    _counter_fields = {"llm_tick", "msg_event_tick"}
+    _pending_metrics: dict[tuple[tuple[str, str], ...], dict[str, Any]] = {}
+    _flush_task: asyncio.Task | None = None
+    _lock: asyncio.Lock | None = None
+    _lock_loop: asyncio.AbstractEventLoop | None = None
 
     @staticmethod
     def get_installation_id():
@@ -40,25 +51,49 @@ class Metric:
             return "null"
 
     @staticmethod
-    async def upload(**kwargs) -> None:
-        """上传相关非敏感的指标以更好地了解 AstrBot 的使用情况。上传的指标不会包含任何有关消息文本、用户信息等敏感信息。
+    def _get_lock() -> asyncio.Lock:
+        loop = asyncio.get_running_loop()
+        if Metric._lock is None or Metric._lock_loop is not loop:
+            Metric._lock = asyncio.Lock()
+            Metric._lock_loop = loop
+        return Metric._lock
 
-        Powered by TickStats.
-        """
-        if os.environ.get("ASTRBOT_DISABLE_METRICS", "0") == "1":
-            return
-        base_url = "https://tickstats.soulter.top/api/metric/90a6c2a1"
-        kwargs["v"] = VERSION
-        kwargs["os"] = sys.platform
-        payload = {"metrics_data": kwargs}
+    @staticmethod
+    def _format_group_value(value: Any) -> str:
+        return repr(value)
+
+    @staticmethod
+    def _get_metric_group_key(kwargs: dict[str, Any]) -> tuple[tuple[str, str], ...]:
+        return tuple(
+            sorted(
+                (key, Metric._format_group_value(value))
+                for key, value in kwargs.items()
+                if key not in Metric._counter_fields
+            )
+        )
+
+    @staticmethod
+    def _get_metric_group_fields(kwargs: dict[str, Any]) -> dict[str, Any]:
+        return {
+            key: value
+            for key, value in kwargs.items()
+            if key not in Metric._counter_fields
+        }
+
+    @staticmethod
+    def _coerce_counter(value: Any) -> int:
         try:
-            kwargs["hn"] = socket.gethostname()
-        except Exception:
-            pass
-        try:
-            kwargs["iid"] = Metric.get_installation_id()
-        except Exception:
-            pass
+            return int(value)
+        except (TypeError, ValueError):
+            return 0
+
+    @staticmethod
+    def _ensure_flush_task_locked() -> None:
+        if Metric._flush_task is None or Metric._flush_task.done():
+            Metric._flush_task = asyncio.create_task(Metric._flush_periodically())
+
+    @staticmethod
+    async def _save_platform_stats(kwargs: dict[str, Any]) -> None:
         try:
             if "adapter_name" in kwargs:
                 await db_helper.insert_platform_stats(
@@ -68,6 +103,93 @@ class Metric:
         except Exception as e:
             logger.error(f"保存指标到数据库失败: {e}")
 
+    @staticmethod
+    async def _add_pending_metrics(kwargs: dict[str, Any]) -> None:
+        key = Metric._get_metric_group_key(kwargs)
+        immediate_metrics = None
+        should_flush = False
+        lock = Metric._get_lock()
+        async with lock:
+            if not Metric._has_uploaded_once:
+                Metric._has_uploaded_once = True
+                immediate_metrics = dict(kwargs)
+            else:
+                pending = Metric._pending_metrics.setdefault(
+                    key,
+                    Metric._get_metric_group_fields(kwargs),
+                )
+                for counter_field in Metric._counter_fields:
+                    if counter_field in kwargs:
+                        pending[counter_field] = pending.get(
+                            counter_field,
+                            0,
+                        ) + Metric._coerce_counter(kwargs[counter_field])
+                Metric._ensure_flush_task_locked()
+                should_flush = (
+                    len(Metric._pending_metrics) > Metric._max_pending_metric_groups
+                )
+
+        if immediate_metrics is not None:
+            await Metric._post_metrics(immediate_metrics)
+            return
+        if should_flush:
+            await Metric.flush()
+
+    @staticmethod
+    async def _flush_periodically() -> None:
+        try:
+            while True:
+                await asyncio.sleep(Metric._upload_interval_seconds)
+                await Metric.flush()
+
+                lock = Metric._get_lock()
+                async with lock:
+                    if not Metric._pending_metrics:
+                        Metric._flush_task = None
+                        return
+        except asyncio.CancelledError:
+            raise
+        except Exception:
+            pass
+        finally:
+            current_task = asyncio.current_task()
+            with suppress(RuntimeError):
+                lock = Metric._get_lock()
+                async with lock:
+                    if Metric._flush_task is current_task:
+                        Metric._flush_task = None
+
+    @staticmethod
+    async def flush() -> None:
+        """Flush pending metrics immediately."""
+        lock = Metric._get_lock()
+        async with lock:
+            pending_metrics = list(Metric._pending_metrics.values())
+            Metric._pending_metrics = {}
+
+        for metrics_data in pending_metrics:
+            await Metric._post_metrics(metrics_data)
+            await asyncio.sleep(0.25)
+
+    @staticmethod
+    async def _post_metrics(metrics_data: dict[str, Any]) -> None:
+        if os.environ.get("ASTRBOT_DISABLE_METRICS", "0") == "1":
+            return
+
+        base_url = "https://tickstats.soulter.top/api/metric/90a6c2a1"
+        payload_metrics = dict(metrics_data)
+        payload_metrics["v"] = VERSION
+        payload_metrics["os"] = sys.platform
+        try:
+            payload_metrics["hn"] = socket.gethostname()
+        except Exception:
+            pass
+        try:
+            payload_metrics["iid"] = Metric.get_installation_id()
+        except Exception:
+            pass
+        payload = {"metrics_data": payload_metrics}
+
         try:
             async with aiohttp.ClientSession(trust_env=True) as session:
                 async with session.post(base_url, json=payload, timeout=3) as response:
@@ -75,3 +197,15 @@ class Metric:
                         pass
         except Exception:
             pass
+
+    @staticmethod
+    async def upload(**kwargs) -> None:
+        """上传相关非敏感的指标以更好地了解 AstrBot 的使用情况。上传的指标不会包含任何有关消息文本、用户信息等敏感信息。
+
+        Powered by TickStats.
+        """
+        if os.environ.get("ASTRBOT_DISABLE_METRICS", "0") == "1":
+            return
+
+        await Metric._save_platform_stats(kwargs)
+        await Metric._add_pending_metrics(dict(kwargs))

dashboard/src/views/ExtensionPage.vue

@@ -636,15 +636,6 @@ const installDialogPluginLogo = computed(() => {
     <div
       class="v-card v-card--density-default rounded-lg v-card--variant-elevated"
     >
-      <div class="v-card__loader">
-        <v-progress-linear
-          :indeterminate="loading_"
-          color="primary"
-          height="2"
-          :active="loading_"
-        ></v-progress-linear>
-      </div>
-
       <v-card-title class="text-h3 pa-4 pb-0 pl-6">
         {{ tm("dialogs.install.title") }}
       </v-card-title>
@@ -839,7 +830,13 @@ const installDialogPluginLogo = computed(() => {
         <v-btn color="grey" variant="text" @click="closeInstallDialog">{{
           tm("buttons.cancel")
         }}</v-btn>
-        <v-btn color="primary" variant="text" @click="newExtension">{{
+        <v-btn
+          color="primary"
+          variant="text"
+          :loading="loading_"
+          :disabled="loading_"
+          @click="newExtension"
+        >{{
           tm("buttons.install")
         }}</v-btn>
       </div>

dashboard/src/views/extension/useExtensionPage.js

@@ -1189,23 +1189,19 @@ export const useExtensionPage = () => {
     versionCompatibilityDialog.show = false;
   };
 
-  const handleInstallResponse = async (resData, { toastStatus = false } = {}) => {
+  const handleInstallResponse = async (resData) => {
     if (
       resData.status === "warning" &&
       resData.data?.warning_type === "astrbot_version_incompatible"
     ) {
-      onLoadingDialogResult(2, resData.message, -1);
+      toast(resData.message, "warning");
       showVersionCompatibilityWarning(resData.message);
       await refreshExtensionsAfterInstallFailure();
       return false;
     }
 
-    if (toastStatus) {
-      toast(resData.message, resData.status === "ok" ? "success" : "error");
-    }
-
     if (resData.status === "error") {
-      onLoadingDialogResult(2, resData.message, -1);
+      toast(resData.message, "error");
       await refreshExtensionsAfterInstallFailure();
       return false;
     }
@@ -1239,7 +1235,7 @@ export const useExtensionPage = () => {
       extension_url.value = "";
     }
 
-    onLoadingDialogResult(1, resData.message);
+    toast(resData.message, "success");
     dialog.value = false;
     selectedMarketInstallPlugin.value = null;
     await getExtensions();
@@ -1263,35 +1259,21 @@ export const useExtensionPage = () => {
       toast(tm("messages.dontFillBoth"), "error");
       return;
     }
-    loading_.value = true;
-    loadingDialog.title = tm("status.loading");
-    loadingDialog.show = true;
-
     const source = upload_file.value !== null ? "file" : "url";
-    toast(
-      source === "file"
-        ? tm("messages.installing")
-        : tm("messages.installingFromUrl") + " " + extension_url.value,
-      "primary",
-    );
+    loading_.value = true;
 
     try {
       const res = await performInstallRequest({ source, ignoreVersionCheck });
       loading_.value = false;
 
-      const canContinue = await handleInstallResponse(res.data, {
-        toastStatus: source === "url",
-      });
+      const canContinue = await handleInstallResponse(res.data);
       if (!canContinue) return;
 
       await finalizeSuccessfulInstall(res.data, source);
     } catch (err) {
       loading_.value = false;
       const message = resolveErrorMessage(err, tm("messages.installFailed"));
-      if (source === "url") {
-        toast(message, "error");
-      }
-      onLoadingDialogResult(2, message, -1);
+      toast(message, "error");
       await refreshExtensionsAfterInstallFailure();
     }
   };