kela-logger 1.1.3

A lightweight logging utility for Python applications

kela-logger

轻量级匿名使用统计收集库，用于收集应用使用数据（不收集任何个人信息）。

安装

pip install kela-logger

API

log_install() -> bool

上报安装事件。应用首次安装时调用一次。

from kela_logger import log_install

log_install()

log_data(data: dict) -> bool

上报使用数据。传入完整的统计数据，会覆盖服务端之前的记录。

from kela_logger import log_data

log_data({
    "workspace_count": 10,      # 工作空间数量
    "chat_count": 200,          # 对话次数
    "excel_count": 20,          # Excel 文件数量
    "field_count": 500,         # 字段数量
    "tool_usage": {             # 工具使用统计
        "sql_query": 100,
        "chart": 50,
        "export": 30
    }
})

get_installation_id() -> str

获取设备唯一标识（基于硬件信息生成的哈希值）。

set_debug(enabled: bool)

开启/关闭调试模式，开启后会打印请求详情。

业务层使用建议

1. 安装时上报

在应用首次安装完成后调用一次：

import os
from kela_logger import log_install

# 使用标记文件判断是否首次安装
install_flag = os.path.expanduser("~/.myapp_installed")

if not os.path.exists(install_flag):
    log_install()
    # 创建标记文件
    with open(install_flag, "w") as f:
        f.write("1")

2. 数据上报策略

建议采用 间隔时间 + 数据变化 双重判断，避免频繁请求：

import time
import json
import os
from kela_logger import log_data

class StatsReporter:
    def __init__(self, interval_minutes=60):
        self.interval = interval_minutes * 60  # 转换为秒
        self.state_file = os.path.expanduser("~/.myapp_stats_state")
        self.last_report_time = 0
        self.last_data_hash = ""
        self._load_state()

    def _load_state(self):
        """加载上次上报状态"""
        if os.path.exists(self.state_file):
            try:
                with open(self.state_file, "r") as f:
                    state = json.load(f)
                    self.last_report_time = state.get("time", 0)
                    self.last_data_hash = state.get("hash", "")
            except:
                pass

    def _save_state(self, data_hash):
        """保存上报状态"""
        try:
            with open(self.state_file, "w") as f:
                json.dump({
                    "time": time.time(),
                    "hash": data_hash
                }, f)
        except:
            pass

    def _get_data_hash(self, data):
        """计算数据哈希，用于判断是否有变化"""
        return hash(json.dumps(data, sort_keys=True))

    def report_if_needed(self, data):
        """
        判断是否需要上报
        条件：距离上次上报超过 interval 分钟 且 数据有变化
        """
        now = time.time()
        data_hash = self._get_data_hash(data)

        # 检查时间间隔
        time_passed = now - self.last_report_time >= self.interval

        # 检查数据是否变化
        data_changed = data_hash != self.last_data_hash

        if time_passed and data_changed:
            success = log_data(data)
            if success:
                self.last_report_time = now
                self.last_data_hash = data_hash
                self._save_state(data_hash)
            return success

        return False

    def force_report(self, data):
        """强制上报（如应用关闭时）"""
        success = log_data(data)
        if success:
            data_hash = self._get_data_hash(data)
            self._save_state(data_hash)
        return success


# 使用示例
reporter = StatsReporter(interval_minutes=60)

# 在关闭工作空间时调用
def on_workspace_close():
    stats = get_current_stats()  # 你的业务方法，获取当前统计数据
    reporter.report_if_needed(stats)

# 在应用退出时强制上报
def on_app_exit():
    stats = get_current_stats()
    reporter.force_report(stats)

3. 简化版（仅时间间隔判断）

如果不需要判断数据变化，可以用更简单的方式：

import time
from kela_logger import log_data

_last_report = 0
REPORT_INTERVAL = 60 * 60  # 60分钟

def report_stats(data):
    global _last_report
    now = time.time()

    if now - _last_report >= REPORT_INTERVAL:
        if log_data(data):
            _last_report = now
            return True
    return False

数据说明

• Installation ID: 基于硬件信息（主板+硬盘序列号）生成的哈希值，用于识别设备
• 所有数据均为聚合统计数据，不包含任何个人信息
• 数据用于了解产品使用情况，改进产品体验

隐私

本库仅收集匿名统计数据：

• 设备标识（硬件哈希，无法反推原始信息）
• 功能使用次数统计

不收集：

• 用户个人信息
• 文件内容
• 具体操作记录

License

MIT License