buff3.0/cleanup-selection-logs/cleanup_selection_logs.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
清理选品日志表 buff_selection_request_log 的历史数据
- 每1小时执行一次
- 对每个用户，保留最新一次选品的所有日志，删除旧选品的日志
- 如果用户只有1次选品记录，则不删除
- 记录清理结果到日志
"""
import os
import sys
import time
import configparser
from datetime import datetime, timezone, timedelta
from pathlib import Path

import pymysql

# 北京时间 UTC+8，与 backend-client 一致
BEIJING_TZ = timezone(timedelta(hours=8))


def beijing_now() -> datetime:
    """当前北京时间（naive）"""
    return datetime.now(BEIJING_TZ).replace(tzinfo=None)


def load_config():
    """从 config.ini 加载配置，返回 config 对象"""
    conf_path = Path(__file__).resolve().parent / "config.ini"
    if not conf_path.exists():
        print(f"错误: 未找到配置文件 {conf_path}，请复制 config.ini.example 为 config.ini 并填写", file=sys.stderr)
        sys.exit(1)
    config = configparser.ConfigParser()
    config.read(conf_path, encoding="utf-8")
    return config


def get_db_connection(config):
    """创建数据库连接"""
    return pymysql.connect(
        host=config.get("database", "host"),
        port=config.getint("database", "port"),
        user=config.get("database", "user"),
        password=config.get("database", "password"),
        database=config.get("database", "database"),
        charset=config.get("database", "charset"),
        cursorclass=pymysql.cursors.DictCursor,
    )


def run_once(config) -> int:
    """
    执行一次清理：删除每个用户的旧选品日志
    每个用户保留最新一次选品的所有日志
    返回: 删除的记录数量
    """
    now_dt = beijing_now()
    now_str = now_dt.strftime("%Y-%m-%d %H:%M:%S")

    conn = get_db_connection(config)
    try:
        with conn.cursor() as cursor:
            # 查出每个用户最新一次选品的 request_id
            cursor.execute("""
                SELECT user_id, MAX(request_id) as max_request_id
                FROM buff_selection_request_log
                GROUP BY user_id
            """)
            latest_requests = cursor.fetchall()

            if not latest_requests:
                print(f"[{now_str}] 无选品日志需要清理")
                return 0

            # 删除不在保留列表中的日志（保留每个用户最新一次选品的日志）
            cursor.execute("""
                DELETE FROM buff_selection_request_log
                WHERE (user_id, request_id) NOT IN (
                    SELECT user_id, max_request_id FROM (
                        SELECT user_id, MAX(request_id) as max_request_id
                        FROM buff_selection_request_log
                        GROUP BY user_id
                    ) AS latest_requests
                )
            """)
            deleted_count = cursor.rowcount

            conn.commit()

            # 查询剩余记录数
            cursor.execute("SELECT COUNT(*) as remaining_count FROM buff_selection_request_log")
            remaining_count = cursor.fetchone()["remaining_count"]

            if deleted_count == 0:
                print(f"[{now_str}] 无历史选品日志需要清理（每个用户仅保留最新一次选品日志）")
                return 0

            print(f"[{now_str}] 清理完成:")
            print(f"  - 删除 {deleted_count} 条选品日志")
            print(f"  - 剩余 {remaining_count} 条选品日志")
            return deleted_count
    finally:
        conn.close()


def main():
    config = load_config()
    interval_seconds = config.getint("script", "interval_seconds", fallback=3600)

    print(f"=" * 50)
    print(f"选品日志清理脚本已启动")
    print(f"清理周期: 每 {interval_seconds} 秒执行一次 ({interval_seconds // 60} 分钟)")
    print(f"清理规则: 每个用户保留最新一次选品的所有日志")
    print(f"启动时间: {beijing_now().strftime('%Y-%m-%d %H:%M:%S')}")
    print(f"=" * 50)

    # 启动时先执行一次清理
    try:
        run_once(config)
    except Exception as e:
        print(f"[{beijing_now()}] 启动时清理失败: {e}", file=sys.stderr)

    print(f"\n开始定时清理，每 {interval_seconds} 秒执行一次...\n")

    while True:
        time.sleep(interval_seconds)
        try:
            run_once(config)
        except Exception as e:
            print(f"[{beijing_now()}] 清理异常: {e}", file=sys.stderr)


if __name__ == "__main__":
    main()