添加安装版本统计上报

Author: jxxghp

app/api/endpoints/system.py

@@ -35,6 +35,7 @@
 from app.helper.rule import RuleHelper
 from app.helper.subscribe import SubscribeHelper
 from app.helper.system import SystemHelper
+from app.helper.usage import UsageHelper
 from app.log import logger
 from app.scheduler import Scheduler
 from app.schemas import ConfigChangeEventData
@@ -520,6 +521,14 @@ async def get_env_setting(_: User = Depends(get_current_active_user_async)):
     return schemas.Response(success=True, data=info)
 
 
+@router.get("/usage/statistic", summary="查询安装版本统计报表", response_model=schemas.Response)
+async def usage_statistic(_: User = Depends(get_current_active_user_async)):
+    """
+    查询安装版本统计报表
+    """
+    return schemas.Response(success=True, data=await UsageHelper().async_get_statistic())
+
+
 @router.post("/env", summary="更新系统配置", response_model=schemas.Response)
 async def set_env_setting(
     env: dict, _: User = Depends(get_current_active_superuser_async)

app/core/config.py

@@ -437,6 +437,8 @@ class ConfigModel(BaseModel):
     )
     # 插件安装数据共享
     PLUGIN_STATISTIC_SHARE: bool = True
+    # 安装版本统计上报
+    USAGE_STATISTIC_SHARE: bool = True
     # 是否开启插件热加载
     PLUGIN_AUTO_RELOAD: bool = False
     # 本地插件仓库目录，多个地址使用,分隔

app/helper/usage.py

@@ -0,0 +1,119 @@
+import platform
+from pathlib import Path
+from typing import Any, Dict
+
+from app.core.config import settings
+from app.log import logger
+from app.utils.http import AsyncRequestUtils, RequestUtils
+from app.utils.singleton import WeakSingleton
+from app.utils.system import SystemUtils
+from version import APP_VERSION, FRONTEND_VERSION
+
+
+class UsageHelper(metaclass=WeakSingleton):
+    """
+    安装版本统计上报
+    """
+
+    _usage_report = f"{settings.MP_SERVER_HOST}/usage/report"
+    _usage_statistic = f"{settings.MP_SERVER_HOST}/usage/statistic"
+
+    @staticmethod
+    def get_frontend_version() -> str:
+        """
+        获取当前前端版本。
+        """
+        if SystemUtils.is_frozen() and SystemUtils.is_windows():
+            version_file = settings.CONFIG_PATH.parent / "nginx" / "html" / "version.txt"
+        else:
+            version_file = Path(settings.FRONTEND_PATH) / "version.txt"
+        if version_file.exists():
+            try:
+                with open(version_file, "r") as file:
+                    version = str(file.read()).strip()
+                return version or FRONTEND_VERSION
+            except Exception as err:
+                logger.debug(f"加载版本文件 {version_file} 出错：{str(err)}")
+        return FRONTEND_VERSION
+
+    @staticmethod
+    def build_payload() -> Dict[str, Any]:
+        """
+        构建安装版本统计上报载荷。
+        """
+        return {
+            "user_uid": SystemUtils.generate_user_unique_id(),
+            "backend_version": APP_VERSION,
+            "frontend_version": UsageHelper.get_frontend_version(),
+            "version_flag": settings.VERSION_FLAG,
+            "platform": f"{platform.system()} {platform.release()}".strip(),
+            "arch": SystemUtils.cpu_arch(),
+        }
+
+    def report(self) -> bool:
+        """
+        上报当前安装实例的版本统计。
+        """
+        if not settings.USAGE_STATISTIC_SHARE:
+            return False
+        payload = self.build_payload()
+        if not payload.get("user_uid"):
+            return False
+        try:
+            res = RequestUtils(
+                proxies=settings.PROXY,
+                content_type="application/json",
+                timeout=5,
+            ).post(self._usage_report, json=payload)
+            return bool(res is not None and res.status_code == 200)
+        except Exception as err:
+            logger.debug(f"上报安装版本统计失败：{str(err)}")
+            return False
+
+    async def async_report(self) -> bool:
+        """
+        异步上报当前安装实例的版本统计。
+        """
+        if not settings.USAGE_STATISTIC_SHARE:
+            return False
+        payload = self.build_payload()
+        if not payload.get("user_uid"):
+            return False
+        try:
+            res = await AsyncRequestUtils(
+                proxies=settings.PROXY,
+                content_type="application/json",
+                timeout=5,
+            ).post(self._usage_report, json=payload)
+            return bool(res is not None and res.status_code == 200)
+        except Exception as err:
+            logger.debug(f"异步上报安装版本统计失败：{str(err)}")
+            return False
+
+    def get_statistic(self) -> Dict[str, Any]:
+        """
+        获取安装版本统计报表。
+        """
+        if not settings.USAGE_STATISTIC_SHARE:
+            return {}
+        try:
+            res = RequestUtils(proxies=settings.PROXY, timeout=10).get_res(self._usage_statistic)
+            if res is not None and res.status_code == 200:
+                return res.json()
+        except Exception as err:
+            logger.debug(f"获取安装版本统计报表失败：{str(err)}")
+        return {}
+
+    async def async_get_statistic(self) -> Dict[str, Any]:
+        """
+        异步获取安装版本统计报表。
+        """
+        if not settings.USAGE_STATISTIC_SHARE:
+            return {}
+        try:
+            res = await AsyncRequestUtils(proxies=settings.PROXY, timeout=10).get_res(self._usage_statistic)
+            if res is not None and res.status_code == 200:
+                return res.json()
+        except Exception as err:
+            logger.debug(f"异步获取安装版本统计报表失败：{str(err)}")
+        return {}

app/scheduler.py

@@ -36,6 +36,7 @@
 from app.helper.image import WallpaperHelper
 from app.helper.message import MessageHelper
 from app.helper.sites import SitesHelper  # noqa
+from app.helper.usage import UsageHelper
 from app.log import logger
 from app.schemas import Notification, NotificationType, Workflow
 from app.schemas.types import EventType, SystemConfigKey
@@ -264,6 +265,7 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
         "DATA_CLEANUP_DOWNLOAD_HISTORY_DAYS",
         "DATA_CLEANUP_SITE_USERDATA_DAYS",
         "DATA_CLEANUP_TRANSFER_HISTORY_DAYS",
+        "USAGE_STATISTIC_SHARE",
     }
 
     def __init__(self):
@@ -401,6 +403,11 @@ def init(self):
                     "func": self.agent_heartbeat,
                     "running": False,
                 },
+                "usage_report": {
+                    "name": "安装版本统计上报",
+                    "func": UsageHelper().report,
+                    "running": False,
+                },
             }
 
             # 创建定时服务
@@ -644,6 +651,17 @@ def init(self):
                     kwargs={"job_id": "agent_heartbeat"},
                 )
 
+            # 安装版本统计上报
+            if settings.USAGE_STATISTIC_SHARE:
+                self._scheduler.add_job(
+                    self.start,
+                    "interval",
+                    id="usage_report",
+                    name="安装版本统计上报",
+                    hours=24,
+                    kwargs={"job_id": "usage_report"},
+                )
+
             # 初始化工作流服务
             self.init_workflow_jobs()
 

app/startup/lifecycle.py

@@ -6,6 +6,7 @@
 from app.chain.system import SystemChain
 from app.core.config import global_vars
 from app.helper.system import SystemHelper
+from app.helper.usage import UsageHelper
 from app.startup.command_initializer import init_command, stop_command, restart_command
 from app.startup.modules_initializer import init_modules, stop_modules
 from app.startup.monitor_initializer import stop_monitor, init_monitor
@@ -29,6 +30,8 @@ async def init_extra():
     SystemHelper().set_system_modified()
     # 重启完成
     SystemChain().restart_finish()
+    # 上报当前安装版本
+    await UsageHelper().async_report()
 
 
 @asynccontextmanager