refactor: 后端也加入 umami 统计

xiaomusic/analytics.py

@@ -1,16 +1,19 @@
 import asyncio
+import platform
 from datetime import datetime
 
+import aiohttp
 from ga4mp import GtagMP
 
 from xiaomusic import __version__
 
 
 class Analytics:
-    def __init__(self, log):
+    def __init__(self, log, config):
         self.gtag = None
         self.current_date = None
         self.log = log
+        self.config = config
         self.init()
 
     def init(self):
@@ -27,6 +30,35 @@ class Analytics:
         self.gtag = gtag
         self.log.info("analytics init ok")
 
+    async def send_startup_event(self):
+        event = self.gtag.create_new_event(name="startup")
+        event.set_event_param(name="version", value=__version__)
+        await self._send(event)
+
+    async def send_daily_event(self):
+        current_date = datetime.now().strftime("%Y-%m-%d")
+        if self.current_date == current_date:
+            return
+
+        event = self.gtag.create_new_event(name="daily_active_user")
+        event.set_event_param(name="version", value=__version__)
+        event.set_event_param(name="date", value=current_date)
+        await self._send(event)
+        self.current_date = current_date
+
+    async def send_play_event(self, name, sec, hardware):
+        event = self.gtag.create_new_event(name="play")
+        event.set_event_param(name="version", value=__version__)
+        event.set_event_param(name="music", value=name)
+        event.set_event_param(name="sec", value=sec)
+        event.set_event_param(name="hardware", value=hardware)
+        await self._send(event)
+
+    async def _send(self, event):
+        await self.post_to_umami(event)
+        events = [event]
+        await self.run_with_cancel(self.gtag.send, events)
+
     async def run_with_cancel(self, func, *args, **kwargs):
         try:
             asyncio.ensure_future(asyncio.to_thread(func, *args, **kwargs))
@@ -35,50 +67,49 @@ class Analytics:
             self.log.warning(f"analytics run_with_cancel failed {e}")
             return None
 
-    async def send_startup_event(self):
+    async def post_to_umami(self, event):
-        try:
+        url = "https://umami.hanxi.cc/api/send"
-            await self.run_with_cancel(self._send_startup_event)
+        user_agent = self._get_user_agent()
-        except Exception as e:
+        event.set_event_param(name="useragent", value=user_agent)
-            self.log.warning(f"analytics send_startup_event failed {e}")
+        data = {
-            self.init()
+            "payload": {
+                "hostname": self.config.hostname,
+                "language": "zh-CN",
+                "referrer": "",
+                "screen": "430x932",
+                "title": "后端统计",
+                "url": "/backend",
+                "website": "7bfb0890-4115-4260-8892-b391513e7e99",
+                "name": event.get_event_name(),
+                "data": event.get_event_params(),
+            },
+            "type": "event",
+        }
 
-    def _send_startup_event(self):
+        async with aiohttp.ClientSession() as session:
-        event = self.gtag.create_new_event(name="startup")
+            headers = {
-        event.set_event_param(name="version", value=__version__)
+                "User-Agent": user_agent,
-        event_list = [event]
+            }
-        self.gtag.send(events=event_list)
+            # self.log.info(f"headers {headers}, {data}")
+            async with session.post(url, json=data, headers=headers) as response:
+                self.log.info(f"umami Status: {response.status}")
+                await response.text()
 
-    async def send_daily_event(self):
+    def _get_user_agent(self):
-        try:
+        # 获取系统信息
-            await self.run_with_cancel(self._send_daily_event)
+        os_name = platform.system()  # 操作系统名称，如 'Windows', 'Linux', 'Darwin'
-        except Exception as e:
+        os_version = platform.version()  # 操作系统版本号
-            self.log.warning(f"analytics send_daily_event failed {e}")
+        architecture = platform.architecture()[0]  # '32bit' or '64bit'
-            self.init()
+        machine = platform.machine()  # 机器类型，如 'x86_64', 'arm64'
 
-    def _send_daily_event(self):
+        # 获取 Python 版本信息
-        current_date = datetime.now().strftime("%Y-%m-%d")
+        python_version = platform.python_version()  # Python 版本
-        if self.current_date == current_date:
-            return
 
-        event = self.gtag.create_new_event(name="daily_active_user")
+        # 组合 User-Agent 字符串
-        event.set_event_param(name="version", value=__version__)
+        user_agent = (
-        event.set_event_param(name="date", value=current_date)
+            f"XiaoMusic/{__version__} "
-        event_list = [event]
+            f"({os_name} {os_version}; {architecture}; {machine}) "
-        self.gtag.send(events=event_list)
+            f"Python/{python_version}"
-        self.current_date = current_date
+        )
 
-    async def send_play_event(self, name, sec, hardware):
+        return user_agent
-        try:
-            await self.run_with_cancel(self._send_play_event, name, sec, hardware)
-        except Exception as e:
-            self.log.warning(f"analytics send_play_event failed {e}")
-            self.init()
-
-    def _send_play_event(self, name, sec, hardware):
-        event = self.gtag.create_new_event(name="play")
-        event.set_event_param(name="version", value=__version__)
-        event.set_event_param(name="music", value=name)
-        event.set_event_param(name="sec", value=sec)
-        event.set_event_param(name="hardware", value=hardware)
-        event_list = [event]
-        self.gtag.send(events=event_list)

xiaomusic/xiaomusic.py

@@ -103,7 +103,7 @@ class XiaoMusic:
         self.update_devices()
 
         # 启动统计
-        self.analytics = Analytics(self.log)
+        self.analytics = Analytics(self.log, self.config)
 
         debug_config = deepcopy_data_no_sensitive_info(self.config)
         self.log.info(f"Startup OK. {debug_config}")