#!/usr/bin/env python3
"""
MySQL 数据库层

使用 MySQL 存储订阅和状态缓存数据。
特性:
- 连接池管理
- 请求日志记录
- 自动重连机制
"""

import os
import logging
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from contextlib import contextmanager

import pymysql
from pymysql.cursors import DictCursor
from dbutils.pooled_db import PooledDB
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

# ==================== 日志配置 ====================

logger = logging.getLogger('charger.db')
logger.setLevel(logging.INFO)

# 控制台输出
if not logger.handlers:
    handler = logging.StreamHandler()
    handler.setFormatter(logging.Formatter(
        '[%(asctime)s] [%(levelname)s] %(message)s',
        datefmt='%H:%M:%S'
    ))
    logger.addHandler(handler)


# ==================== 连接池管理 ====================

class Database:
    """MySQL 数据库连接池管理"""

    _pool: PooledDB = None
    _config = None

    @classmethod
    def _get_config(cls) -> dict:
        """获取数据库配置"""
        if cls._config is None:
            cls._config = {
                'host': os.getenv('DB_HOST', 'localhost'),
                'port': int(os.getenv('DB_PORT', 3306)),
                'user': os.getenv('DB_USER', 'root'),
                'password': os.getenv('DB_PASSWORD', ''),
                'database': os.getenv('DB_NAME', 'charger_monitor'),
                'charset': 'utf8mb4',
                'cursorclass': DictCursor,
                'autocommit': True,
            }
        return cls._config

    @classmethod
    def _get_pool(cls) -> PooledDB:
        """获取连接池（懒加载）"""
        if cls._pool is None:
            config = cls._get_config()
            try:
                cls._pool = PooledDB(
                    creator=pymysql,
                    maxconnections=10,      # 最大连接数
                    mincached=2,            # 初始空闲连接
                    maxcached=5,            # 最大空闲连接
                    maxshared=3,            # 最大共享连接
                    blocking=True,          # 连接池耗尽时阻塞等待
                    ping=1,                 # 检查连接是否可用
                    **config
                )
                logger.info(f"连接池初始化成功 [{config['host']}:{config['port']}/{config['database']}]")
            except Exception as e:
                logger.error(f"连接池初始化失败: {e}")
                raise
        return cls._pool

    @classmethod
    @contextmanager
    def get_connection(cls):
        """获取数据库连接（上下文管理器）"""
        pool = cls._get_pool()
        conn = pool.connection()
        try:
            yield conn
        finally:
            conn.close()

    @classmethod
    def execute(cls, sql: str, params: tuple = None, fetch_one: bool = False,
                fetch_all: bool = False, log_query: bool = True):
        """执行 SQL 并返回结果"""
        start_time = datetime.now()

        try:
            with cls.get_connection() as conn:
                with conn.cursor() as cursor:
                    cursor.execute(sql, params)

                    if fetch_one:
                        result = cursor.fetchone()
                    elif fetch_all:
                        result = cursor.fetchall()
                    else:
                        result = cursor.lastrowid

                    # 记录查询日志
                    if log_query:
                        elapsed = (datetime.now() - start_time).total_seconds() * 1000
                        logger.debug(f"SQL [{elapsed:.1f}ms] {sql[:100]}{'...' if len(sql) > 100 else ''}")

                    return result

        except pymysql.Error as e:
            elapsed = (datetime.now() - start_time).total_seconds() * 1000
            logger.error(f"SQL错误 [{elapsed:.1f}ms] {e}: {sql[:100]}")
            raise

    @classmethod
    def executemany(cls, sql: str, params_list: List[tuple], log_query: bool = True):
        """批量执行 SQL"""
        if not params_list:
            return 0

        start_time = datetime.now()

        try:
            with cls.get_connection() as conn:
                with conn.cursor() as cursor:
                    cursor.executemany(sql, params_list)
                    rowcount = cursor.rowcount

                    if log_query:
                        elapsed = (datetime.now() - start_time).total_seconds() * 1000
                        logger.debug(f"批量SQL [{elapsed:.1f}ms] {len(params_list)}行: {sql[:80]}")

                    return rowcount

        except pymysql.Error as e:
            elapsed = (datetime.now() - start_time).total_seconds() * 1000
            logger.error(f"批量SQL错误 [{elapsed:.1f}ms] {e}: {sql[:80]}")
            raise

    @classmethod
    def init_tables(cls):
        """初始化数据表"""
        with cls.get_connection() as conn:
            with conn.cursor() as cursor:
                # 订阅表
                cursor.execute("""
                    CREATE TABLE IF NOT EXISTS subscriptions (
                        id INT AUTO_INCREMENT PRIMARY KEY,
                        email VARCHAR(255) NOT NULL COMMENT '订阅邮箱',
                        feishu_name VARCHAR(64) DEFAULT '' COMMENT '飞书花名',
                        charger_type VARCHAR(10) DEFAULT 'all' COMMENT '充电桩类型: AC/DC/all',
                        vendor VARCHAR(16) DEFAULT 'all' COMMENT '订阅厂商: all/nio/gongniu',
                        week_start DATE COMMENT '订阅所在周（用于判断是否本周有效）',
                        created_at DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
                        notified_at DATETIME NULL COMMENT '通知发送时间',
                        is_priority TINYINT(1) DEFAULT 0 COMMENT 'VIP优先级',
                        expired TINYINT(1) DEFAULT 0 COMMENT '是否已过期',
                        INDEX idx_email (email),
                        INDEX idx_expired (expired),
                        INDEX idx_vendor (vendor),
                        INDEX idx_week_start (week_start),
                        INDEX idx_notified (notified_at),
                        INDEX idx_priority (is_priority, created_at)
                    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='订阅表'
                """)

                cursor.execute("SHOW COLUMNS FROM subscriptions LIKE 'vendor'")
                if not cursor.fetchone():
                    cursor.execute("""
                        ALTER TABLE subscriptions
                        ADD COLUMN vendor VARCHAR(16) DEFAULT 'all' COMMENT '订阅厂商: all/nio/gongniu'
                        AFTER charger_type
                    """)

                cursor.execute("SHOW INDEX FROM subscriptions WHERE Key_name = 'idx_vendor'")
                if not cursor.fetchone():
                    cursor.execute("ALTER TABLE subscriptions ADD INDEX idx_vendor (vendor)")

                # 充电桩状态缓存表
                cursor.execute("""
                    CREATE TABLE IF NOT EXISTS charger_state_cache (
                        id INT AUTO_INCREMENT PRIMARY KEY,
                        station_id VARCHAR(64) NOT NULL COMMENT '充电站ID',
                        connector_key VARCHAR(64) NOT NULL COMMENT '充电桩标识',
                        state_type VARCHAR(20) NOT NULL COMMENT '状态类型: free/charging/occupied',
                        started_at DATETIME NOT NULL COMMENT '状态开始时间',
                        updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
                        UNIQUE KEY uk_station_connector (station_id, connector_key),
                        INDEX idx_state_type (state_type)
                    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='充电桩状态缓存表'
                """)

                # 充电桩状态历史表（用于分析）
                cursor.execute("""
                    CREATE TABLE IF NOT EXISTS charger_status_history (
                        id INT AUTO_INCREMENT PRIMARY KEY,
                        station_id VARCHAR(64) NOT NULL COMMENT '充电站ID',
                        connector_id VARCHAR(64) NOT NULL COMMENT '充电桩ID',
                        connector_name VARCHAR(64) NOT NULL COMMENT '充电桩名称',
                        connector_type VARCHAR(10) NOT NULL COMMENT '类型: AC/DC',
        vendor VARCHAR(32) DEFAULT 'nio' COMMENT '厂商标识: nio/gongniu',
                        power DECIMAL(10,2) DEFAULT 0 COMMENT '功率(kW)',
                        state_type VARCHAR(20) NOT NULL COMMENT '状态: free/charging/occupied/offline',
                        state_desc VARCHAR(64) DEFAULT '' COMMENT '状态描述',
                        idle_seconds INT DEFAULT 0 COMMENT '空闲时长(秒)',
                        charging_seconds INT DEFAULT 0 COMMENT '充电时长(秒)',
                        occupied_seconds INT DEFAULT 0 COMMENT '占位时长(秒)',
                        remaining_minutes INT DEFAULT NULL COMMENT 'DC桩剩余充电时长(分钟)',
                        fetch_time DATETIME NOT NULL COMMENT '数据获取时间',
                        created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
                        UNIQUE KEY uk_station_name_time (station_id, connector_name, fetch_time),
                        INDEX idx_station_time (station_id, fetch_time),
                        INDEX idx_connector_time (connector_id, fetch_time),
                        INDEX idx_state_type (state_type)
                    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='充电桩状态历史表'
                """)

                # 最新状态快照表：实时接口只读这张小表，历史表只服务分析
                cursor.execute("""
                    CREATE TABLE IF NOT EXISTS charger_latest_status (
                        station_id VARCHAR(64) NOT NULL COMMENT '充电站ID',
                        connector_name VARCHAR(64) NOT NULL COMMENT '充电桩名称',
                        connector_id VARCHAR(64) NOT NULL COMMENT '充电桩ID',
                        connector_type VARCHAR(10) NOT NULL COMMENT '类型: AC/DC',
                        vendor VARCHAR(32) DEFAULT 'nio' COMMENT '厂商标识: nio/gongniu',
                        power DECIMAL(10,2) DEFAULT 0 COMMENT '功率(kW)',
                        state_type VARCHAR(20) NOT NULL COMMENT '状态: free/charging/occupied/offline',
                        state_desc VARCHAR(64) DEFAULT '' COMMENT '状态描述',
                        idle_seconds INT DEFAULT 0 COMMENT '空闲时长(秒)',
                        charging_seconds INT DEFAULT 0 COMMENT '充电时长(秒)',
                        occupied_seconds INT DEFAULT 0 COMMENT '占位时长(秒)',
                        remaining_minutes INT DEFAULT NULL COMMENT 'DC桩剩余充电时长(分钟)',
                        fetch_time DATETIME NOT NULL COMMENT '数据获取时间',
                        updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                        PRIMARY KEY (station_id, connector_name),
                        INDEX idx_station_fetch_time (station_id, fetch_time),
                        INDEX idx_latest_state_type (state_type),
                        INDEX idx_latest_vendor (vendor)
                    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='充电桩最新状态快照表'
                """)

                # 系统配置表
                cursor.execute("""
                    CREATE TABLE IF NOT EXISTS system_config (
                        `key` VARCHAR(64) PRIMARY KEY COMMENT '配置键',
                        `value` TEXT COMMENT '配置值',
                        updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间'
                    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='系统配置表'
                """)

                # 充电桩留言表
                cursor.execute("""
                    CREATE TABLE IF NOT EXISTS charger_comments (
                        id INT AUTO_INCREMENT PRIMARY KEY,
                        connector_id VARCHAR(64) NOT NULL COMMENT '充电桩ID',
                        connector_name VARCHAR(64) NOT NULL COMMENT '充电桩名称',
                        feishu_name VARCHAR(20) NOT NULL COMMENT '留言人飞书花名',
                        content VARCHAR(100) NOT NULL COMMENT '留言内容',
                        created_at DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '留言时间',
                        INDEX idx_connector (connector_id),
                        INDEX idx_created (created_at)
                    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='充电桩留言表'
                """)

                # 访问日志表
                cursor.execute("""
                    CREATE TABLE IF NOT EXISTS visit_logs (
                        id INT AUTO_INCREMENT PRIMARY KEY,
                        device_hash VARCHAR(16) NOT NULL COMMENT '设备指纹哈希(SHA256前16位)',
                        visited_at DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '访问时间',
                        INDEX idx_visited_date (visited_at),
                        INDEX idx_device_date (device_hash, visited_at)
                    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='访问日志表'
                """)

        logger.info("数据表初始化完成")

    @classmethod
    def close_pool(cls):
        """关闭连接池"""
        if cls._pool:
            # PooledDB 没有 close 方法，连接会在进程结束时自动释放
            cls._pool = None
            logger.info("连接池已关闭")


# ==================== 订阅管理 ====================

class SubscriptionDB:
    """订阅数据操作"""

    @classmethod
    def get_all(cls) -> Dict[str, dict]:
        """获取所有订阅"""
        rows = Database.execute(
            "SELECT * FROM subscriptions",
            fetch_all=True
        ) or []
        return {row['email']: dict(row) for row in rows}

    @classmethod
    def add(cls, email: str, feishu_name: str = '', charger_type: str = 'all',
            is_priority: int = 0, vendor: str = 'all') -> dict:
        """
        添加订阅

        Args:
            email: 订阅邮箱
            feishu_name: 飞书花名
            charger_type: 充电桩类型 AC/DC/all
            is_priority: VIP 优先级 (0=普通, 1=VIP)
            vendor: 订阅厂商 all/nio/gongniu

        Returns:
            {"success": bool, "message": str, "is_new": bool, "queue_position": int}
        """
        from datetime import date
        # 计算本周开始日期（周一）
        today = date.today()
        week_start = today - timedelta(days=today.weekday())

        # 先检查是否已存在有效订阅
        existing = Database.execute(
            "SELECT id, charger_type, vendor, expired FROM subscriptions WHERE email = %s",
            (email.lower(),),
            fetch_one=True
        )

        if existing and existing.get("expired") == 0:
            # 已存在有效订阅，不重复添加
            logger.info(f"订阅已存在: {email}")
            return {"success": False, "message": "该邮箱已订阅，无需重复订阅", "is_new": False, "queue_position": 0}

        # 计算当前排队位置
        queue_position = cls.get_queue_position(is_priority)

        if existing:
            # 存在已过期记录，更新而非插入（避免 UNIQUE 约束冲突）
            Database.execute(
                """UPDATE subscriptions
                   SET feishu_name = %s, charger_type = %s, vendor = %s, week_start = %s,
                       is_priority = %s, expired = 0, notified_at = NULL
                   WHERE email = %s""",
                (feishu_name, charger_type, vendor, week_start, is_priority, email.lower())
            )
        else:
            # 新订阅
            Database.execute(
                """INSERT INTO subscriptions (email, feishu_name, charger_type, vendor, week_start, is_priority)
                   VALUES (%s, %s, %s, %s, %s, %s)""",
                (email.lower(), feishu_name, charger_type, vendor, week_start, is_priority)
            )
        logger.info(f"新增订阅: {email} (花名: {feishu_name}, 类型: {charger_type}, 厂商: {vendor}, VIP: {is_priority})")
        return {"success": True, "message": "订阅成功", "is_new": True, "queue_position": queue_position}

    @classmethod
    def get_queue_position(cls, is_priority: int = 0) -> int:
        """
        计算新订阅者的排队位置

        Args:
            is_priority: 是否 VIP (VIP 排第 1 位)

        Returns:
            排队位置 (从 1 开始)
        """
        if is_priority:
            # VIP 直接排第 1 位
            return 1

        # 普通用户：统计当前未通知的订阅数 + VIP 数量
        count = Database.execute(
            """SELECT COUNT(*) as cnt FROM subscriptions
               WHERE expired = 0 AND notified_at IS NULL""",
            fetch_one=True
        )
        return (count.get("cnt", 0) if count else 0) + 1

    @classmethod
    def get_queue_count(cls) -> int:
        """获取当前排队人数（有效的未通知订阅数）"""
        count = Database.execute(
            """SELECT COUNT(*) as cnt FROM subscriptions
               WHERE expired = 0 AND notified_at IS NULL""",
            fetch_one=True
        )
        return count.get("cnt", 0) if count else 0

    @classmethod
    def get_active_count(cls) -> int:
        """获取有效订阅用户数（用于数据概览）"""
        count = Database.execute(
            """SELECT COUNT(*) as cnt FROM subscriptions
               WHERE expired = 0""",
            fetch_one=True
        )
        return count.get("cnt", 0) if count else 0

    @classmethod
    def get_unnotified_queue(cls) -> List[dict]:
        """
        获取未通知的订阅队列（VIP 在前，按创建时间排序）

        Returns:
            订阅列表，VIP 排在最前面
        """
        return Database.execute(
            """SELECT * FROM subscriptions
               WHERE expired = 0 AND notified_at IS NULL
               ORDER BY is_priority DESC, created_at ASC""",
            fetch_all=True
        ) or []

    @classmethod
    def get_is_priority(cls, email: str) -> int:
        """检查是否是 VIP 优先订阅（从 system_config 表读取白名单）"""
        try:
            row = Database.execute(
                "SELECT value FROM system_config WHERE `key` = %s",
                ("vip_email_list",),
                fetch_one=True
            )
            if row and row.get("value"):
                vip_emails = [e.strip().lower() for e in row["value"].split(",") if e.strip()]
                return 1 if email.lower() in vip_emails else 0
        except Exception:
            logger.warning(f"VIP白名单查询失败，降级为非VIP", exc_info=True)
        return 0

    @classmethod
    def remove(cls, email: str) -> bool:
        """移除订阅"""
        Database.execute(
            "DELETE FROM subscriptions WHERE email = %s",
            (email.lower(),)
        )
        logger.info(f"移除订阅: {email}")
        return True

    @classmethod
    def get_active_for_week(cls) -> List[dict]:
        """获取有效的订阅列表（未过期、未触发）"""
        return Database.execute(
            "SELECT * FROM subscriptions WHERE expired = 0",
            fetch_all=True
        ) or []

    @classmethod
    def get_active(cls) -> List[dict]:
        """获取有效订阅列表（兼容旧接口）"""
        return Database.execute(
            "SELECT * FROM subscriptions WHERE expired = 0",
            fetch_all=True
        ) or []

    @classmethod
    def mark_triggered(cls, email: str) -> bool:
        """标记订阅已触发（触发后删除订阅，生命周期结束）"""
        Database.execute(
            "DELETE FROM subscriptions WHERE email = %s",
            (email.lower(),)
        )
        logger.info(f"订阅触发并删除: {email}")
        return True

    @classmethod
    def mark_notified(cls, email: str) -> bool:
        """
        标记订阅已发送通知，并标记为过期（每人只通知一次）

        Args:
            email: 订阅邮箱

        Returns:
            是否成功
        """
        Database.execute(
            """UPDATE subscriptions
               SET notified_at = NOW(), expired = 1
               WHERE email = %s""",
            (email.lower(),)
        )
        logger.info(f"订阅已通知并过期: {email}")
        return True

    @classmethod
    def get_recent_notified(cls, minutes: int = 5) -> List[dict]:
        """
        获取最近 N 分钟内被通知但未过期的订阅者

        Args:
            minutes: 时间窗口（分钟）

        Returns:
            订阅列表
        """
        return Database.execute(
            """SELECT * FROM subscriptions
               WHERE expired = 0
                 AND notified_at IS NOT NULL
                 AND notified_at >= DATE_SUB(NOW(), INTERVAL %s MINUTE)""",
            (minutes,),
            fetch_all=True
        ) or []

    @classmethod
    def mark_expired(cls, email: str) -> bool:
        """标记订阅已过期"""
        Database.execute(
            "UPDATE subscriptions SET expired = 1 WHERE email = %s",
            (email.lower(),)
        )
        logger.info(f"订阅已过期: {email}")
        return True

    @classmethod
    def cleanup_weekly(cls):
        """每周五 00:00 清空所有订阅"""
        result = Database.execute("DELETE FROM subscriptions")
        logger.info(f"每周清理: 删除了 {result} 条订阅")
        return result

    @classmethod
    def cleanup_old_subscriptions(cls, weeks: int = 2):
        """清理N周前的旧订阅"""
        from datetime import date
        cutoff = date.today() - timedelta(weeks=weeks)
        result = Database.execute(
            "DELETE FROM subscriptions WHERE week_start < %s",
            (cutoff,)
        )
        logger.info(f"清理了 {result} 条旧订阅")
        return result


# ==================== 充电桩留言管理 ====================

class CommentDB:
    """充电桩留言数据操作"""

    @classmethod
    def add(cls, connector_id: str, connector_name: str, feishu_name: str, content: str) -> dict:
        """
        添加留言

        Args:
            connector_id: 充电桩ID
            connector_name: 充电桩名称
            feishu_name: 飞书花名
            content: 留言内容

        Returns:
            {"success": bool, "message": str}
        """
        # 参数验证
        if not feishu_name or len(feishu_name.strip()) == 0:
            return {"success": False, "message": "请填写飞书花名"}
        if len(feishu_name) > 10:
            return {"success": False, "message": "飞书花名不能超过10个字"}
        if not content or len(content.strip()) == 0:
            return {"success": False, "message": "留言内容不能为空"}
        if len(content) > 20:
            return {"success": False, "message": "留言内容不能超过20个字"}

        # 频率限制：同一花名同一桩5分钟内只能留1条
        recent = Database.execute(
            """SELECT id FROM charger_comments
               WHERE connector_id = %s AND feishu_name = %s
                 AND created_at >= DATE_SUB(NOW(), INTERVAL 5 MINUTE)
               LIMIT 1""",
            (connector_id, feishu_name.strip()),
            fetch_one=True
        )
        if recent:
            return {"success": False, "message": "5分钟内已留言，请稍后再试"}

        # 插入留言
        Database.execute(
            """INSERT INTO charger_comments (connector_id, connector_name, feishu_name, content)
               VALUES (%s, %s, %s, %s)""",
            (connector_id, connector_name, feishu_name.strip(), content.strip())
        )
        logger.info(f"新增留言: {connector_name} - {feishu_name}: {content}")
        return {"success": True, "message": "留言成功"}

    @classmethod
    def get_by_connector(cls, connector_id: str) -> List[dict]:
        """
        获取某充电桩的留言列表（24小时内有效）

        Args:
            connector_id: 充电桩ID

        Returns:
            留言列表，按时间倒序
        """
        rows = Database.execute(
            """SELECT id, feishu_name, content, created_at
               FROM charger_comments
               WHERE connector_id = %s
                 AND created_at >= DATE_SUB(NOW(), INTERVAL 24 HOUR)
               ORDER BY created_at DESC""",
            (connector_id,),
            fetch_all=True
        ) or []

        # 统一 created_at 为 ISO 格式字符串，供前端 Date() 解析
        for row in rows:
            if hasattr(row['created_at'], 'isoformat'):
                row['created_at'] = row['created_at'].isoformat()

        return rows

    @classmethod
    def get_counts(cls, connector_ids: List[str]) -> dict:
        """
        批量获取充电桩留言数量

        Args:
            connector_ids: 充电桩ID列表

        Returns:
            {connector_id: count} 字典
        """
        if not connector_ids:
            return {}

        placeholders = ', '.join(['%s'] * len(connector_ids))
        rows = Database.execute(
            f"""SELECT connector_id, COUNT(*) as cnt
                FROM charger_comments
                WHERE connector_id IN ({placeholders})
                  AND created_at >= DATE_SUB(NOW(), INTERVAL 24 HOUR)
                GROUP BY connector_id""",
            connector_ids,
            fetch_all=True
        ) or []

        return {row['connector_id']: row['cnt'] for row in rows}

    @classmethod
    def get_latest_comments(cls, connector_ids: List[str]) -> dict:
        """
        批量获取每个充电桩最新一条留言

        Returns:
            {connector_id: {content, feishu_name}} 字典
        """
        if not connector_ids:
            return {}

        placeholders = ', '.join(['%s'] * len(connector_ids))
        params = list(connector_ids) + list(connector_ids)
        rows = Database.execute(
            f"""SELECT connector_id, content, feishu_name, created_at
                FROM charger_comments
                WHERE connector_id IN ({placeholders})
                  AND created_at >= DATE_SUB(NOW(), INTERVAL 24 HOUR)
                  AND id IN (
                    SELECT MAX(id) FROM charger_comments
                    WHERE connector_id IN ({placeholders})
                      AND created_at >= DATE_SUB(NOW(), INTERVAL 24 HOUR)
                    GROUP BY connector_id
                  )""",
            params,
            fetch_all=True
        ) or []

        return {
            row['connector_id']: {
                'content': row['content'],
                'feishu_name': row['feishu_name'],
                'created_at': row['created_at'].isoformat() if hasattr(row['created_at'], 'isoformat') else str(row['created_at'])
            } for row in rows
        }

    @classmethod
    def cleanup_old(cls):
        """清理24小时前的旧留言"""
        result = Database.execute(
            "DELETE FROM charger_comments WHERE created_at < DATE_SUB(NOW(), INTERVAL 24 HOUR)"
        )
        if result:
            logger.info(f"清理了 {result} 条旧留言")
        return result


# ==================== 访问统计 ====================

class VisitDB:
    """访问日志操作"""

    @classmethod
    def record(cls, device_hash: str):
        """记录一次访问"""
        return Database.execute(
            "INSERT INTO visit_logs (device_hash) VALUES (%s)",
            (device_hash,)
        )

    @classmethod
    def get_today_stats(cls) -> dict:
        """获取今日访问统计（PV/UV）"""
        row = Database.execute(
            """SELECT COUNT(*) as pv, COUNT(DISTINCT device_hash) as uv
               FROM visit_logs
               WHERE DATE(visited_at) = CURDATE()""",
            fetch_one=True
        )
        return {'pv': row['pv'] or 0, 'uv': row['uv'] or 0} if row else {'pv': 0, 'uv': 0}

    @classmethod
    def get_visit_history(cls, days: int = 7) -> list:
        """获取近 N 天每天的 PV/UV 统计"""
        rows = Database.execute(
            """SELECT DATE(visited_at) as date,
                      COUNT(*) as pv,
                      COUNT(DISTINCT device_hash) as uv
               FROM visit_logs
               WHERE visited_at >= DATE_SUB(CURDATE(), INTERVAL %s DAY)
               GROUP BY DATE(visited_at)
               ORDER BY date ASC""",
            (days,),
            fetch_all=True
        )
        if not rows:
            return []
        return [
            {'date': row['date'].strftime('%Y-%m-%d'), 'pv': row['pv'], 'uv': row['uv']}
            for row in rows
        ]

    @classmethod
    def cleanup_old(cls, days: int = 90):
        """清理N天前的访问日志"""
        result = Database.execute(
            "DELETE FROM visit_logs WHERE visited_at < DATE_SUB(NOW(), INTERVAL %s DAY)",
            (days,)
        )
        if result:
            logger.info(f"清理了 {result} 条旧访问日志")
        return result


# ==================== 历史数据管理 ====================

class HistoryDB:
    """充电桩状态历史数据操作"""

    @classmethod
    def _build_status_params(cls, station_id: str, connectors: List[dict], fetch_time: datetime) -> List[tuple]:
        """构造历史表和最新快照表共用的状态参数。"""
        params_list = []
        for c in connectors:
            state = c.get("state", {})
            state_desc = state.get("desc", "")
            if c.get("type") == "DC" and c.get("state", {}).get("type") == "charging" and c.get("charging_progress") is not None:
                state_desc = str(c.get("charging_progress"))
            # 解析剩余时长（DC桩专属）
            remaining_minutes = None
            if c.get("estimated_charging_ready_time"):
                try:
                    remaining_minutes = int(c.get("estimated_charging_ready_time"))
                except (ValueError, TypeError):
                    pass

            # 获取厂商标识，默认 nio
            vendor = c.get("vendor", "nio")

            params_list.append((
                station_id,
                vendor,
                c.get("connector_id", ""),
                c.get("connector_name", ""),
                c.get("type", "AC"),
                c.get("power", 0),
                state.get("type", "unknown"),
                state_desc,
                c.get("idle_seconds", 0),
                c.get("charging_seconds", 0),
                c.get("occupied_seconds", 0),
                remaining_minutes,
                fetch_time
            ))
        return params_list

    @classmethod
    def save_status(cls, station_id: str, connectors: List[dict], fetch_time: datetime = None):
        """
        保存充电桩状态快照

        Args:
            station_id: 充电站ID
            connectors: 充电桩状态列表
            fetch_time: 数据获取时间，默认当前时间
        """
        if not connectors:
            return 0

        if fetch_time is None:
            fetch_time = datetime.now()

        params_list = cls._build_status_params(station_id, connectors, fetch_time)

        saved = Database.executemany(
            """INSERT IGNORE INTO charger_status_history
               (station_id, vendor, connector_id, connector_name, connector_type, power,
                state_type, state_desc, idle_seconds, charging_seconds, occupied_seconds, remaining_minutes, fetch_time)
               VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)""",
            params_list
        )
        cls.save_latest_status(params_list)
        return saved

    @classmethod
    def save_latest_status(cls, params_list: List[tuple]) -> int:
        """更新最新状态快照表。"""
        return Database.executemany(
            """INSERT INTO charger_latest_status
               (station_id, vendor, connector_id, connector_name, connector_type, power,
                state_type, state_desc, idle_seconds, charging_seconds, occupied_seconds, remaining_minutes, fetch_time)
               VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
               ON DUPLICATE KEY UPDATE
                 vendor = VALUES(vendor),
                 connector_id = VALUES(connector_id),
                 connector_type = VALUES(connector_type),
                 power = VALUES(power),
                 state_type = VALUES(state_type),
                 state_desc = VALUES(state_desc),
                 idle_seconds = VALUES(idle_seconds),
                 charging_seconds = VALUES(charging_seconds),
                 occupied_seconds = VALUES(occupied_seconds),
                 remaining_minutes = VALUES(remaining_minutes),
                 fetch_time = VALUES(fetch_time)""",
            params_list
        )

    @classmethod
    def get_latest_snapshot(cls, station_id: str) -> List[dict]:
        """读取最新状态快照。实时监控接口优先使用这张小表。"""
        return Database.execute(
            """SELECT connector_id, connector_name, connector_type, vendor, power,
                      state_type, state_desc, idle_seconds, charging_seconds,
                      occupied_seconds, remaining_minutes, fetch_time
               FROM charger_latest_status
               WHERE station_id = %s
               ORDER BY
                 connector_type DESC,
                 CAST(connector_name AS UNSIGNED) DESC""",
            (station_id,),
            fetch_all=True
        ) or []

    @classmethod
    def get_history(cls, station_id: str, connector_id: str = None,
                    start_time: datetime = None, end_time: datetime = None,
                    limit: int = 1000) -> List[dict]:
        """
        查询历史数据

        Args:
            station_id: 充电站ID
            connector_id: 充电桩ID（可选，不传则查询所有）
            start_time: 开始时间
            end_time: 结束时间
            limit: 最大返回条数
        """
        sql = "SELECT * FROM charger_status_history WHERE station_id = %s"
        params = [station_id]

        if connector_id:
            sql += " AND connector_id = %s"
            params.append(connector_id)

        if start_time:
            sql += " AND fetch_time >= %s"
            params.append(start_time)

        if end_time:
            sql += " AND fetch_time <= %s"
            params.append(end_time)

        sql += " ORDER BY fetch_time DESC LIMIT %s"
        params.append(limit)

        return Database.execute(sql, tuple(params), fetch_all=True) or []

    @classmethod
    def get_statistics(cls, station_id: str, days: int = 7) -> dict:
        """
        获取统计数据

        Args:
            station_id: 充电站ID
            days: 统计最近N天的数据
        """
        sql = """
            SELECT
                connector_id,
                connector_name,
                connector_type,
                COUNT(*) as total_records,
                SUM(CASE WHEN state_type = 'free' THEN 1 ELSE 0 END) as free_count,
                SUM(CASE WHEN state_type = 'charging' THEN 1 ELSE 0 END) as charging_count,
                SUM(CASE WHEN state_type = 'occupied' THEN 1 ELSE 0 END) as occupied_count,
                ROUND(SUM(CASE WHEN state_type = 'free' THEN 1 ELSE 0 END) * 100.0 / COUNT(*), 1) as free_rate
            FROM charger_status_history
            WHERE station_id = %s AND fetch_time >= DATE_SUB(NOW(), INTERVAL %s DAY)
            GROUP BY connector_id, connector_name, connector_type
            ORDER BY connector_id
        """
        rows = Database.execute(sql, (station_id, days), fetch_all=True) or []
        return {row['connector_id']: dict(row) for row in rows}

    @classmethod
    def cleanup_old_data(cls, days: int = 90):
        """清理N天前的历史数据"""
        result = Database.execute(
            "DELETE FROM charger_status_history WHERE fetch_time < DATE_SUB(NOW(), INTERVAL %s DAY)",
            (days,)
        )
        logger.info(f"清理了 {result} 条历史数据（{days}天前）")
        return result

    # ==================== 数据分析查询 ====================

    @classmethod
    def get_summary(cls, station_id: str, days: int = 7) -> dict:
        """
        获取汇总统计数据

        Returns:
            {
                "total_sessions": int,        # 充电次数（基于状态变化）
                "avg_charging_minutes": float, # 平均充电时长（分钟）
                "utilization_rate": float,     # 整体使用率（%）
                "peak_hour": int,              # 高峰时段（小时 0-23）
                "ac_utilization": float,       # AC 使用率
                "dc_utilization": float,       # DC 使用率
                "ac_avg_charging_minutes": float,  # AC 平均充电时长
                "dc_avg_charging_minutes": float,  # DC 平均充电时长
            }
        """
        # 整体统计
        overall = Database.execute("""
            SELECT
                COUNT(*) as total_records,
                SUM(CASE WHEN state_type = 'charging' THEN 1 ELSE 0 END) as charging_count,
                ROUND(AVG(CASE WHEN state_type = 'charging' THEN charging_seconds END) / 60, 1) as avg_charging_minutes,
                ROUND(SUM(CASE WHEN state_type = 'charging' THEN 1 ELSE 0 END) * 100.0 / COUNT(*), 1) as utilization_rate
            FROM charger_status_history
            WHERE station_id = %s AND fetch_time >= DATE_SUB(NOW(), INTERVAL %s DAY)
        """, (station_id, days), fetch_one=True) or {}

        # 高峰时段
        peak = Database.execute("""
            SELECT HOUR(fetch_time) as hour, COUNT(*) as cnt
            FROM charger_status_history
            WHERE station_id = %s AND fetch_time >= DATE_SUB(NOW(), INTERVAL %s DAY)
              AND state_type = 'charging'
            GROUP BY HOUR(fetch_time)
            ORDER BY cnt DESC
            LIMIT 1
        """, (station_id, days), fetch_one=True)

        # AC 统计（使用率和平均充电时长）
        ac_stats = Database.execute("""
            SELECT
                ROUND(SUM(CASE WHEN state_type = 'charging' THEN 1 ELSE 0 END) * 100.0 / COUNT(*), 1) as utilization,
                ROUND(AVG(CASE WHEN state_type = 'charging' THEN charging_seconds END) / 60, 1) as avg_charging_minutes
            FROM charger_status_history
            WHERE station_id = %s AND fetch_time >= DATE_SUB(NOW(), INTERVAL %s DAY)
              AND connector_type = 'AC'
        """, (station_id, days), fetch_one=True) or {}

        # DC 统计（使用率和平均充电时长）
        dc_stats = Database.execute("""
            SELECT
                ROUND(SUM(CASE WHEN state_type = 'charging' THEN 1 ELSE 0 END) * 100.0 / COUNT(*), 1) as utilization,
                ROUND(AVG(CASE WHEN state_type = 'charging' THEN charging_seconds END) / 60, 1) as avg_charging_minutes
            FROM charger_status_history
            WHERE station_id = %s AND fetch_time >= DATE_SUB(NOW(), INTERVAL %s DAY)
              AND connector_type = 'DC'
        """, (station_id, days), fetch_one=True) or {}

        return {
            "total_sessions": overall.get("charging_count") or 0,
            "avg_charging_minutes": overall.get("avg_charging_minutes") or 0,
            "utilization_rate": overall.get("utilization_rate") or 0,
            "peak_hour": peak.get("hour") if peak else 0,
            "ac_utilization": ac_stats.get("utilization") or 0,
            "dc_utilization": dc_stats.get("utilization") or 0,
            "ac_avg_charging_minutes": ac_stats.get("avg_charging_minutes") or 0,
            "dc_avg_charging_minutes": dc_stats.get("avg_charging_minutes") or 0,
        }

    @classmethod
    def get_hourly(cls, station_id: str, days: int = 7) -> List[dict]:
        """
        获取每小时使用率矩阵（用于热力图）

        Returns:
            [{"day": "周一", "hour": 0, "utilization": 45.2}, ...]
        """
        rows = Database.execute("""
            SELECT
                DAYNAME(fetch_time) as day_name,
                DAYOFWEEK(fetch_time) as day_num,
                HOUR(fetch_time) as hour,
                COUNT(*) as total,
                SUM(CASE WHEN state_type = 'charging' THEN 1 ELSE 0 END) as charging_count,
                ROUND(SUM(CASE WHEN state_type = 'charging' THEN 1 ELSE 0 END) * 100.0 / COUNT(*), 1) as utilization
            FROM charger_status_history
            WHERE station_id = %s AND fetch_time >= DATE_SUB(NOW(), INTERVAL %s DAY)
            GROUP BY DAYOFWEEK(fetch_time), DAYNAME(fetch_time), HOUR(fetch_time)
            ORDER BY day_num, hour
        """, (station_id, days), fetch_all=True) or []

        # 映射英文星期为中文
        day_map = {
            "Sunday": "周日", "Monday": "周一", "Tuesday": "周二",
            "Wednesday": "周三", "Thursday": "周四", "Friday": "周五", "Saturday": "周六"
        }

        return [{
            "day": day_map.get(r["day_name"], r["day_name"]),
            "day_num": r["day_num"],
            "hour": r["hour"],
            "utilization": r["utilization"]
        } for r in rows]

    @classmethod
    def get_trend(cls, station_id: str, days: int = 7) -> List[dict]:
        """
        获取每日使用率趋势（AC/DC 分开）

        Returns:
            [{"date": "2024-01-15", "ac_utilization": 45.2, "dc_utilization": 62.1}, ...]
        """
        rows = Database.execute("""
            SELECT
                DATE(fetch_time) as date,
                connector_type,
                ROUND(SUM(CASE WHEN state_type = 'charging' THEN 1 ELSE 0 END) * 100.0 / COUNT(*), 1) as utilization
            FROM charger_status_history
            WHERE station_id = %s AND fetch_time >= DATE_SUB(NOW(), INTERVAL %s DAY)
            GROUP BY DATE(fetch_time), connector_type
            ORDER BY date
        """, (station_id, days), fetch_all=True) or []

        # 按日期聚合 AC/DC
        trend_map = {}
        for r in rows:
            date_str = r["date"].strftime("%m-%d") if hasattr(r["date"], "strftime") else str(r["date"])
            if date_str not in trend_map:
                trend_map[date_str] = {"date": date_str, "ac_utilization": 0, "dc_utilization": 0}
            if r["connector_type"] == "AC":
                trend_map[date_str]["ac_utilization"] = r["utilization"]
            elif r["connector_type"] == "DC":
                trend_map[date_str]["dc_utilization"] = r["utilization"]

        return list(trend_map.values())

    @classmethod
    def get_connector_rank(cls, station_id: str, days: int = 7) -> List[dict]:
        """
        获取各充电桩使用率排名

        Returns:
            [{"connector_id": "AC-01", "connector_name": "1号枪", "connector_type": "AC", "utilization": 45.2}, ...]
        """
        rows = Database.execute("""
            SELECT
                connector_name,
                connector_type,
                ROUND(SUM(CASE WHEN state_type = 'charging' THEN 1 ELSE 0 END) * 100.0 / COUNT(*), 1) as utilization
            FROM charger_status_history
            WHERE station_id = %s AND fetch_time >= DATE_SUB(NOW(), INTERVAL %s DAY)
            GROUP BY connector_name, connector_type
            ORDER BY utilization DESC
        """, (station_id, days), fetch_all=True) or []

        return [dict(r) for r in rows]

    @classmethod
    def get_overview(cls, station_id: str, days: int = 7) -> dict:
        """
        获取数据概览

        Returns:
            {
                "subscription_count": int,           # 当前订阅邮箱数
                "ac_avg_charging_minutes": float,    # AC 平均充电时长（分钟）
                "dc_avg_charging_minutes": float,    # DC 平均充电时长（分钟）
                "grab_hour": str,                    # 工作日抢桩时段（如 "09:15"）
                "charging_sessions": int,            # 充电次数
                "estimated_kwh": float,              # 预估充电电量（kWh）
                "ac_kwh": float,                     # AC 预估电量
                "dc_kwh": float,                     # DC 预估电量
            }
        """
        # 1. 订阅数
        sub_count = Database.execute(
            "SELECT COUNT(*) as cnt FROM subscriptions WHERE expired = 0",
            fetch_one=True
        )
        subscription_count = sub_count.get("cnt", 0) if sub_count else 0

        # 2. AC/DC 平均充电时长：每次充电事件的平均时长
        # 每次充电事件的时长 = 该次充电期间最大的 charging_seconds
        # 按充电桩分组，找出每次充电事件（连续充电记录的最大值）
        ac_session_stats = Database.execute("""
            SELECT
                COUNT(*) as session_count,
                AVG(max_seconds) as avg_seconds
            FROM (
                SELECT connector_name, MAX(charging_seconds) as max_seconds
                FROM charger_status_history
                WHERE station_id = %s AND fetch_time >= DATE_SUB(NOW(), INTERVAL %s DAY)
                  AND connector_type = 'AC' AND state_type = 'charging' AND charging_seconds > 0
                GROUP BY connector_name, DATE(fetch_time)
            ) as sessions
        """, (station_id, days), fetch_one=True) or {}

        dc_session_stats = Database.execute("""
            SELECT
                COUNT(*) as session_count,
                AVG(max_seconds) as avg_seconds
            FROM (
                SELECT connector_name, MAX(charging_seconds) as max_seconds
                FROM charger_status_history
                WHERE station_id = %s AND fetch_time >= DATE_SUB(NOW(), INTERVAL %s DAY)
                  AND connector_type = 'DC' AND state_type = 'charging' AND charging_seconds > 0
                GROUP BY connector_name, DATE(fetch_time)
            ) as sessions
        """, (station_id, days), fetch_one=True) or {}

        ac_avg_minutes = round(float(ac_session_stats.get("avg_seconds") or 0) / 60, 1)
        dc_avg_minutes = round(float(dc_session_stats.get("avg_seconds") or 0) / 60, 1)
        ac_session_count = ac_session_stats.get("session_count") or 0
        dc_session_count = dc_session_stats.get("session_count") or 0

        # 3. 总充电时长和预估电量（AC 7kW, DC 20kW）
        # 正确计算：统计充电记录数 × 采集间隔（假设平均 120 秒）
        # 不再用 SUM(charging_seconds)，因为那是快照数据会严重重复计算
        avg_collect_interval = 120  # 平均采集间隔（秒）

        ac_stats = Database.execute("""
            SELECT COUNT(*) as charging_records
            FROM charger_status_history
            WHERE station_id = %s AND fetch_time >= DATE_SUB(NOW(), INTERVAL %s DAY)
              AND connector_type = 'AC' AND state_type = 'charging'
        """, (station_id, days), fetch_one=True) or {}

        dc_stats = Database.execute("""
            SELECT COUNT(*) as charging_records
            FROM charger_status_history
            WHERE station_id = %s AND fetch_time >= DATE_SUB(NOW(), INTERVAL %s DAY)
              AND connector_type = 'DC' AND state_type = 'charging'
        """, (station_id, days), fetch_one=True) or {}

        ac_charging_records = ac_stats.get("charging_records") or 0
        dc_charging_records = dc_stats.get("charging_records") or 0

        # 电量 = 功率 × 时间（小时）
        # 时间 = 充电记录数 × 采集间隔
        ac_kwh = round(ac_charging_records * avg_collect_interval / 3600 * 7, 2)
        dc_kwh = round(dc_charging_records * avg_collect_interval / 3600 * 20, 2)
        estimated_kwh = round(ac_kwh + dc_kwh, 2)

        # 4. 工作日抢桩截止时间：9:00-10:00 第一次所有充电桩都非空闲的时刻
        grab_hour = "--"
        grab_data = Database.execute("""
            SELECT
                DATE(fetch_time) as date,
                HOUR(fetch_time) as hour,
                MINUTE(fetch_time) as minute,
                COUNT(*) as total,
                SUM(CASE WHEN state_type = 'free' THEN 1 ELSE 0 END) as free_count
            FROM charger_status_history
            WHERE station_id = %s
              AND fetch_time >= DATE_SUB(NOW(), INTERVAL %s DAY)
              AND DAYOFWEEK(fetch_time) BETWEEN 2 AND 6
              AND HOUR(fetch_time) = 9
            GROUP BY DATE(fetch_time), HOUR(fetch_time), MINUTE(fetch_time)
            HAVING free_count = 0
            ORDER BY date ASC, hour ASC, minute ASC
        """, (station_id, days), fetch_all=True) or []

        # 取第一个满载时刻（所有充电桩都在使用）
        if grab_data:
            first_full = grab_data[0]
            grab_hour = f"{first_full['hour']:02d}:{first_full['minute']:02d}"

        # 5. 充电次数
        charging_sessions = ac_session_count + dc_session_count

        # 6. 获取工作日 9:00-10:00 每天的满载时刻
        grab_times = cls.get_grab_times(station_id, days)

        return {
            "subscription_count": subscription_count,
            "ac_avg_charging_minutes": ac_avg_minutes,
            "dc_avg_charging_minutes": dc_avg_minutes,
            "grab_hour": grab_hour,
            "grab_times": grab_times,
            "charging_sessions": charging_sessions,
            "estimated_kwh": estimated_kwh,
            "ac_kwh": ac_kwh,
            "dc_kwh": dc_kwh,
        }

    @classmethod
    def get_grab_times(cls, station_id: str, days: int = 7) -> List[dict]:
        """
        获取工作日 9:00-10:00 每天的满载时刻

        Returns:
            [{"day": "周一", "minute": 15, "time": "9:15", "full": True}, ...]
        """
        # 查询每天 9:00-10:00 第一个满载时刻
        rows = Database.execute("""
            SELECT
                DAYNAME(fetch_time) as day_name,
                DAYOFWEEK(fetch_time) as day_num,
                MINUTE(fetch_time) as minute,
                COUNT(*) as total,
                SUM(CASE WHEN state_type = 'free' THEN 1 ELSE 0 END) as free_count
            FROM charger_status_history
            WHERE station_id = %s
              AND fetch_time >= DATE_SUB(NOW(), INTERVAL %s DAY)
              AND DAYOFWEEK(fetch_time) BETWEEN 2 AND 6
              AND HOUR(fetch_time) = 9
            GROUP BY DAYNAME(fetch_time), DAYOFWEEK(fetch_time), MINUTE(fetch_time)
            HAVING free_count = 0
            ORDER BY day_num ASC, minute ASC
        """, (station_id, days), fetch_all=True) or []

        # 映射英文星期为中文
        day_map = {
            "Monday": "周一", "Tuesday": "周二", "Wednesday": "周三",
            "Thursday": "周四", "Friday": "周五"
        }

        # 按天取第一个满载时刻
        seen_days = set()
        result = []
        for r in rows:
            day = day_map.get(r["day_name"])
            if day and day not in seen_days:
                seen_days.add(day)
                result.append({
                    "day": day,
                    "minute": r["minute"],
                    "time": f"9:{r['minute']:02d}",
                    "full": True
                })

        return result


# ==================== 状态缓存管理 ====================

class StateCacheDB:
    """充电桩状态缓存操作"""

    STATION_ID = os.getenv('STATION_ID', '')

    @classmethod
    def get_free_cache(cls) -> Dict[str, datetime]:
        """获取空闲状态缓存"""
        rows = Database.execute(
            "SELECT connector_key, started_at FROM charger_state_cache WHERE station_id = %s AND state_type = 'free'",
            (cls.STATION_ID,),
            fetch_all=True
        ) or []
        return {row['connector_key']: row['started_at'] for row in rows}

    @classmethod
    def get_charging_cache(cls) -> Dict[str, datetime]:
        """获取充电状态缓存"""
        rows = Database.execute(
            "SELECT connector_key, started_at FROM charger_state_cache WHERE station_id = %s AND state_type = 'charging'",
            (cls.STATION_ID,),
            fetch_all=True
        ) or []
        return {row['connector_key']: row['started_at'] for row in rows}

    @classmethod
    def get_occupied_cache(cls) -> Dict[str, datetime]:
        """获取占位状态缓存"""
        rows = Database.execute(
            "SELECT connector_key, started_at FROM charger_state_cache WHERE station_id = %s AND state_type = 'occupied'",
            (cls.STATION_ID,),
            fetch_all=True
        ) or []
        return {row['connector_key']: row['started_at'] for row in rows}

    @classmethod
    def update_states(cls, connectors: List[dict]) -> tuple:
        """
        更新充电桩状态缓存（使用事务保证原子性）
        支持 free/charging/occupied 三种状态

        Returns:
            (free_cache, charging_cache, occupied_cache) 三个缓存字典
        """
        now = datetime.now()
        current_free = set()
        current_charging = set()
        current_occupied = set()

        for c in connectors:
            state_type = c.get("state", {}).get("type", "")
            # 使用 connector_name 作为唯一标识，因为 NIO API 返回的 connector_id 都是 '1'
            key = c.get("connector_name") or c.get("connector_id")

            if state_type == "free":
                current_free.add(key)
            elif state_type == "charging":
                current_charging.add(key)
            elif state_type == "occupied":
                current_occupied.add(key)

        # 使用事务保证原子性
        with Database.get_connection() as conn:
            # DBUtils 连接需要通过底层连接设置 autocommit
            underlying_conn = conn._con if hasattr(conn, '_con') else conn
            try:
                underlying_conn.autocommit(False)
            except AttributeError:
                # pymysql 连接使用 begin() 开启事务
                pass

            try:
                with conn.cursor() as cursor:
                    # 获取现有缓存
                    cursor.execute(
                        "SELECT connector_key, state_type FROM charger_state_cache WHERE station_id = %s",
                        (cls.STATION_ID,)
                    )
                    existing_rows = cursor.fetchall()
                    existing = {row['connector_key']: row['state_type'] for row in existing_rows}

                    # 删除不再存在的状态
                    to_delete = set(existing.keys()) - current_free - current_charging - current_occupied
                    if to_delete:
                        cursor.executemany(
                            "DELETE FROM charger_state_cache WHERE station_id = %s AND connector_key = %s",
                            [(cls.STATION_ID, key) for key in to_delete]
                        )

                    # 插入或更新空闲状态
                    for key in current_free:
                        if existing.get(key) != 'free':
                            cursor.execute(
                                """INSERT INTO charger_state_cache (station_id, connector_key, state_type, started_at)
                                   VALUES (%s, %s, 'free', %s)
                                   ON DUPLICATE KEY UPDATE state_type = 'free', started_at = %s""",
                                (cls.STATION_ID, key, now, now)
                            )

                    # 插入或更新充电状态
                    for key in current_charging:
                        if existing.get(key) != 'charging':
                            cursor.execute(
                                """INSERT INTO charger_state_cache (station_id, connector_key, state_type, started_at)
                                   VALUES (%s, %s, 'charging', %s)
                                   ON DUPLICATE KEY UPDATE state_type = 'charging', started_at = %s""",
                                (cls.STATION_ID, key, now, now)
                            )

                    # 插入或更新占位状态
                    for key in current_occupied:
                        if existing.get(key) != 'occupied':
                            cursor.execute(
                                """INSERT INTO charger_state_cache (station_id, connector_key, state_type, started_at)
                                   VALUES (%s, %s, 'occupied', %s)
                                   ON DUPLICATE KEY UPDATE state_type = 'occupied', started_at = %s""",
                                (cls.STATION_ID, key, now, now)
                            )

                conn.commit()
            except Exception as e:
                conn.rollback()
                logger.error(f"状态缓存更新失败: {e}")
                raise

        # 返回缓存（三种状态）
        return cls.get_free_cache(), cls.get_charging_cache(), cls.get_occupied_cache()

    @classmethod
    def clear(cls):
        """清空状态缓存"""
        Database.execute(
            "DELETE FROM charger_state_cache WHERE station_id = %s",
            (cls.STATION_ID,)
        )


# ==================== 兼容旧 MockDB 接口 ====================

class MockDB:
    """兼容层：模拟 MockDB 接口，实际调用 MySQL"""

    # 订阅管理
    get_subscriptions = SubscriptionDB.get_all
    add_subscription = SubscriptionDB.add
    remove_subscription = SubscriptionDB.remove
    get_active_subscriptions = SubscriptionDB.get_active
    mark_triggered = SubscriptionDB.mark_triggered
    save_subscriptions = lambda subs: True

    # 状态缓存
    get_free_state_cache = StateCacheDB.get_free_cache

    @classmethod
    def save_free_state_cache(cls, cache):
        return True

    @classmethod
    def update_free_state(cls, connectors):
        free_cache, _ = StateCacheDB.update_states(connectors)
        import time
        return {k: v.timestamp() if isinstance(v, datetime) else v for k, v in free_cache.items()}

    @classmethod
    def clear_all(cls):
        """清空所有数据"""
        with Database.get_connection() as conn:
            with conn.cursor() as cursor:
                cursor.execute("DELETE FROM subscriptions")
                cursor.execute("DELETE FROM charger_state_cache")

    @classmethod
    def init_sample_data(cls):
        """初始化示例数据"""
        cls.add_subscription("demo@example.com")


# ==================== 初始化 ====================

def init_db():
    """初始化数据库"""
    try:
        # 初始化连接池（懒加载，这里只是测试连接）
        Database._get_pool()
        Database.init_tables()
        logger.info("MySQL 连接成功")
    except Exception as e:
        logger.error(f"MySQL 连接失败: {e}")
        raise


# 命令行工具
if __name__ == "__main__":
    import sys

    # 设置详细日志
    logging.basicConfig(level=logging.DEBUG)

    print("MySQL 数据库管理工具")
    print("=" * 40)

    # 初始化数据库
    init_db()

    if len(sys.argv) > 1:
        cmd = sys.argv[1]

        if cmd == "subscriptions":
            subs = MockDB.get_subscriptions()
            print(f"订阅列表 ({len(subs)} 条):")
            for email, data in subs.items():
                print(f"  - {email}: {data}")

        elif cmd == "cache":
            cache = MockDB.get_free_state_cache()
            print(f"空闲缓存 ({len(cache)} 条):")
            for key, ts in cache.items():
                if isinstance(ts, datetime):
                    print(f"  - {key}: {ts.isoformat()}")
                else:
                    print(f"  - {key}: {datetime.fromtimestamp(ts).isoformat()}")

        elif cmd == "clear":
            MockDB.clear_all()
            print("已清空所有数据")

        elif cmd == "init":
            MockDB.init_sample_data()
            print("已初始化示例数据")

        elif cmd == "create-tables":
            Database.init_tables()
            print("已创建数据表")

        else:
            print(f"未知命令: {cmd}")
            print("可用命令: subscriptions, cache, clear, init, create-tables")
    else:
        print("用法: python db.py <command>")
        print("命令:")
        print("  subscriptions  - 查看订阅列表")
        print("  cache          - 查看空闲缓存")
        print("  clear          - 清空所有数据")
        print("  init           - 初始化示例数据")
        print("  create-tables  - 创建数据表")
