diff --git a/astrbot/core/platform/platform.py b/astrbot/core/platform/platform.py index a7c181217..b32891096 100644 --- a/astrbot/core/platform/platform.py +++ b/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: """提交一个事件到事件队列。""" diff --git a/astrbot/core/platform/sources/lark/lark_event.py b/astrbot/core/platform/sources/lark/lark_event.py index 1c7dd0b43..13b7ddec9 100644 --- a/astrbot/core/platform/sources/lark/lark_event.py +++ b/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 diff --git a/astrbot/core/utils/metrics.py b/astrbot/core/utils/metrics.py index 8fb146428..43955568e 100644 --- a/astrbot/core/utils/metrics.py +++ b/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)) diff --git a/dashboard/src/views/ExtensionPage.vue b/dashboard/src/views/ExtensionPage.vue index 7db1b8d1a..a7aaeaf72 100644 --- a/dashboard/src/views/ExtensionPage.vue +++ b/dashboard/src/views/ExtensionPage.vue @@ -636,15 +636,6 @@ const installDialogPluginLogo = computed(() => {
-
- -
- {{ tm("dialogs.install.title") }} @@ -839,7 +830,13 @@ const installDialogPluginLogo = computed(() => { {{ tm("buttons.cancel") }} - {{ + {{ tm("buttons.install") }}
diff --git a/dashboard/src/views/extension/useExtensionPage.js b/dashboard/src/views/extension/useExtensionPage.js index 74cd9d7ae..1fbaa0e25 100644 --- a/dashboard/src/views/extension/useExtensionPage.js +++ b/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(); } };