Template

<type>([optional scope]): <subject>

[optional body]

[optional footer(s)]

/**

 * 文件描述：简要描述本文件的主要功能和作用。

 *

 * @author 你的名字 <email@example.com>

 * @version 1.0.0

 * @create 2025-12-30

 * @update 2025-12-30: 初始版本

 *             xxxx-xx-xx: 修改内容简述

 *

 * Copyright (c) 2025 Your Company. All rights reserved.

 */

def calculate_user_score(user_id: int, include_bonus: bool = True) -> float:

    """

    计算用户综合评分。

    该函数根据用户历史行为、活跃度和奖励规则计算最终评分。

    评分公式经过多次 A/B 测试优化，当前权重为：活跃度 40%、成就 35%、奖励 25%。

    参数

    ----------

    user_id : int

        用户唯一 ID，必须存在于数据库中。

    include_bonus : bool, optional

        是否包含临时奖励加分（默认 True）。

    返回值

    -------

    float

        用户综合评分，范围 0.0 ~ 100.0。

    异常

    -----

    ValueError

        当 user_id 不存在或为负数时抛出。

    NotFoundError

        当用户数据未加载时抛出。

    示例

    --------

    >>> calculate_user_score(12345)

    87.5

    >>> calculate_user_score(12345, include_bonus=False)

    82.3

    注意

    ----

    该函数依赖缓存，缓存失效时会自动重新计算。

    """

/**

 * 计算用户综合评分。

 *

 * 该函数根据用户历史行为、活跃度和奖励规则计算最终评分。

 * 评分公式经过多次 A/B 测试优化，当前权重为：活跃度 40%、成就 35%、奖励 25%。

 *

 * @param userId 用户唯一 ID，必须存在于数据库中

 * @param includeBonus 是否包含临时奖励加分，默认 true

 * @return 用户综合评分，范围 0.0 ~ 100.0

 * @throws ValueError 当 userId 不存在或为负数时抛出

 * @throws NotFoundError 当用户数据未加载时抛出

 * @example

 *   calculateUserScore(12345) // 返回 87.5

 * @since 1.2.0

 */

public double calculateUserScore(int userId, boolean includeBonus = true) {

    // ...

}

# ==============================================================================

# 关键算法块：用户行为衰减权重计算

# 原因：近期行为对评分影响更大，使用指数衰减模型模拟“遗忘曲线”

# 公式：weight = base ^ (days_elapsed / half_life)

# ==============================================================================

for record in user_records:

    # 计算行为发生天数间隔（基于当前日期）

    days_elapsed = (current_date - record.timestamp).days

    

    # 半衰期为 30 天，即 30 天后权重减半

    weight = decay_base ** (days_elapsed / HALF_LIFE_DAYS)  # NOTE: decay_base < 1

    

    total_score += record.value * weight

CREATE TABLE `user` (
  `id`              BIGINT UNSIGNED AUTO_INCREMENT COMMENT '用户ID'
                    PRIMARY KEY,
  `user_name`       VARCHAR(50) NOT NULL DEFAULT '' COMMENT '用户名（登录名）',
  `nick_name`       VARCHAR(50) NOT NULL DEFAULT '' COMMENT '昵称',
  `password`        VARCHAR(100) NOT NULL COMMENT '密码（加密存储）',
  `mobile`          VARCHAR(20) NOT NULL DEFAULT '' COMMENT '手机号',
  `email`           VARCHAR(100) NOT NULL DEFAULT '' COMMENT '邮箱',
  `status`          TINYINT NOT NULL DEFAULT 1 COMMENT '状态：0-禁用，1-正常，2-冻结',
  `is_vip`          TINYINT NOT NULL DEFAULT 0 COMMENT '是否VIP：0-否，1-是',
  `vip_expire_time` DATETIME NULL COMMENT 'VIP到期时间（NULL表示非VIP）',
  `created_at`      DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
  `updated_at`      DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
  `is_deleted`      TINYINT NOT NULL DEFAULT 0 COMMENT '逻辑删除：0-正常，1-已删除',
  `creator_id`      BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '创建人ID',
  `updater_id`      BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '更新人ID'
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='用户信息表';

"""全局日志配置模块

该模块提供项目统一的日志管理器，支持：
- 单例模式全局 logger
- 控制台 + 文件输出
- 日志文件按日期滚动（ RotatingFileHandler 或 TimedRotatingFileHandler ）
- JSON 结构化格式可选
- 配置从 settings 或环境变量加载
"""

import logging
import logging.handlers
import sys
from pathlib import Path
from typing import Optional

# 项目配置示例（实际项目中可替换为自己的 config 模块或 pydantic Settings）
class LogConfig:
    LOG_LEVEL: str = "INFO"                  # DEBUG/INFO/WARNING/ERROR/CRITICAL
    LOG_DIR: str = "logs"                     # 日志目录
    LOG_FILE_NAME: str = "app.log"            # 日志文件名
    LOG_MAX_BYTES: int = 50 * 1024 * 1024     # 单文件最大 50MB
    LOG_BACKUP_COUNT: int = 10                # 保留 10 个备份文件
    LOG_JSON_FORMAT: bool = False            # 是否启用 JSON 结构化日志
    LOG_CONSOLE: bool = True                 # 是否输出到控制台

# 可替换为你的实际配置加载方式
config = LogConfig()


class LoggerManager:
    """日志管理器（单例模式）"""

    _instance: Optional['LoggerManager'] = None
    _logger: Optional[logging.Logger] = None

    def __new__(cls) -> 'LoggerManager':
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

    def __init__(self) -> None:
        if self._logger is None:
            self._setup_logger()

    def _setup_logger(self) -> None:
        """初始化全局 logger"""
        logger_name = "app"  # 可改为项目名，如 "myproject" 或 __name__
        self._logger = logging.getLogger(logger_name)
        self._logger.setLevel(getattr(logging, config.LOG_LEVEL.upper()))

        # 避免重复添加 handler（多次导入模块时）
        if self._logger.handlers:
            return

        # 创建日志目录
        log_dir = Path(config.LOG_DIR)
        log_dir.mkdir(parents=True, exist_ok=True)
        log_file = log_dir / config.LOG_FILE_NAME

        # 格式化器
        if config.LOG_JSON_FORMAT:
            # 推荐第三方库：python-json-logger
            try:
                from python_json_logger.jsonlogger import JsonFormatter
                formatter = JsonFormatter(
                    '%(asctime)s %(name)s %(levelname)s %(filename)s %(lineno)d %(message)s'
                )
            except ImportError:  # fallback
                formatter = logging.Formatter(
                    '{"time":"%(asctime)s", "name":"%(name)s", "level":"%(levelname)s", '
                    '"file":"%(filename)s:%(lineno)d", "message":%(message)s}'
                )
        else:
            formatter = logging.Formatter(
                fmt='%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s',
                datefmt='%Y-%m-%d %H:%M:%S'
            )

        # 文件处理器 - 按大小滚动
        file_handler = logging.handlers.RotatingFileHandler(
            filename=log_file,
            maxBytes=config.LOG_MAX_BYTES,
            backupCount=config.LOG_BACKUP_COUNT,
            encoding='utf-8'
        )
        file_handler.setLevel(getattr(logging, config.LOG_LEVEL.upper()))
        file_handler.setFormatter(formatter)
        self._logger.addHandler(file_handler)

        # 可选：按天滚动（替代方案）
        # file_handler = logging.handlers.TimedRotatingFileHandler(
        #     filename=log_file,
        #     when='midnight',
        #     interval=1,
        #     backupCount=30,
        #     encoding='utf-8'
        # )

        # 控制台处理器
        if config.LOG_CONSOLE:
            console_handler = logging.StreamHandler(sys.stdout)
            console_handler.setLevel(getattr(logging, config.LOG_LEVEL.upper()))
            console_handler.setFormatter(formatter)
            self._logger.addHandler(console_handler)

    def get_logger(self) -> logging.Logger:
        """获取配置好的全局 logger"""
        return self._logger


# 全局 logger 实例（项目中统一导入使用）
logger = LoggerManager().get_logger()

from logger import logger  # 假设文件名为 logger.py

def process_task(task_id: str, url: str):
    logger.info("task.start", task_id=task_id, url=url)  # 推荐结构化键值对

    try:
        data = fetch_data(url)
        logger.info("fetch.success", task_id=task_id, data_size=len(data))
        
        result = process(data)
        logger.info("task.success", task_id=task_id)
        
    except Exception as e:
        logger.error(
            "task.failed",
            task_id=task_id,
            url=url,
            exc_info=e,  # 自动打印堆栈
            error=str(e)
        )
        raise