buff2.0/backend-client/app/services/heartbeat_service.py

"""
心跳服务
"""
from datetime import datetime, timedelta
from typing import Dict, Any, Tuple, Optional
from sqlalchemy.orm import Session
from app.models.user import User
from app.utils.timezone import beijing_now_naive
from app.models.session import Session as SessionModel
from app.models.user_software_subscription import UserSoftwareSubscription
from app.models.software import Software
from app.models.system_config import SystemConfig
from app.utils.history import create_session_history
from app.core.logger import logger

DEFAULT_SOFTWARE_CODE = "buff-xp-steam-sh"


def _get_subscription_expire(db: Session, user_id: int, software_code: str) -> Optional[datetime]:
    """获取用户在某软件下的到期时间，优先订阅表，兼容 user.expire_time"""
    code = (software_code or "").strip() or DEFAULT_SOFTWARE_CODE
    sw = db.query(Software).filter(Software.code == code, Software.is_deleted == 0).first()
    if not sw:
        return None
    sub = db.query(UserSoftwareSubscription).filter(
        UserSoftwareSubscription.user_id == user_id,
        UserSoftwareSubscription.software_id == sw.id,
        UserSoftwareSubscription.is_deleted == 0
    ).first()
    if sub and sub.expire_time is not None:
        return sub.expire_time
    user = db.query(User).filter(User.id == user_id, User.is_deleted == 0).first()
    return user.expire_time if user else None


class HeartbeatService:
    """心跳服务"""
    
    def __init__(self, db: Session):
        self.db = db
    
    def heartbeat(self, session_id: str, machine_code: str,
                  software_code: str = None) -> Tuple[bool, str, dict]:
        """
        心跳检测。按 software_code 返回该软件下的到期/剩余天数。

        Args:
            session_id: 会话ID
            machine_code: 机器码
            software_code: 软件特征码(可选，默认 buff-xp-steam-sh)

        Returns:
            (success, message, data)
        """
        try:
            session = self.db.query(SessionModel).filter(
                SessionModel.session_id == session_id,
                SessionModel.is_active == 1,
                SessionModel.expires_at > beijing_now_naive()
            ).first()
            if not session:
                return False, "会话已过期，请重新登录", {}
            if session.machine_code != machine_code:
                return False, "账号已在其他设备登录，如非本人操作请联系客服", {}

            user = self.db.query(User).filter(
                User.id == session.user_id,
                User.is_deleted == 0
            ).first()
            if not user:
                return False, "用户不存在", {}
            if user.status != 1:
                return False, "账号已被禁用", {}

            expire_time = _get_subscription_expire(self.db, user.id, software_code or "")
            if expire_time and expire_time < beijing_now_naive():
                return False, "您的卡密已到期，请前往续费或联系客户支持", {}

            config = self.get_client_config()
            renewal_threshold = config.get('session_renewal_threshold', 1800)
            extension_duration = config.get('session_extension_duration', 3600)
            remaining_seconds = int((session.expires_at - beijing_now_naive()).total_seconds())
            renewed = False
            if remaining_seconds < renewal_threshold:
                expires_at_before = session.expires_at
                create_session_history(self.db, session, 'RENEW')
                session.expires_at = beijing_now_naive() + timedelta(seconds=extension_duration)
                self.db.flush()
                renewed = True
                logger.info(f"Session自动续期: session_id={session_id}, "
                           f"原过期时间={expires_at_before.isoformat()}, "
                           f"新过期时间={session.expires_at.isoformat()}")
                remaining_seconds = int((session.expires_at - beijing_now_naive()).total_seconds())

            days_remaining = 0
            if expire_time:
                days_remaining = (expire_time - beijing_now_naive()).days

            logger.debug(f"心跳检测成功: session_id={session_id}, renewed={renewed}")
            return True, "成功", {
                "config": config,
                "user_info": {
                    "expire_time": expire_time.isoformat() if expire_time else None,
                    "days_remaining": days_remaining
                },
                "session_info": {
                    "expires_at": session.expires_at.isoformat(),
                    "remaining_seconds": remaining_seconds,
                    "renewed": renewed
                }
            }
        except Exception as e:
            logger.exception(f"心跳检测失败: {e}")
            return False, f"心跳检测失败: {str(e)}", {}
    
    def get_client_config(self) -> Dict[str, Any]:
        """获取客户端配置"""
        config_map = {
            'heartbeat_interval_seconds': ('int', 10),
            'client_heartbeat_interval_seconds': ('int', 20),
            'max_heartbeat_failures': ('int', 20),
            'timestamp_tolerance_seconds': ('int', 300),
            'prevent_duplicate_login': ('bool', True),
            'session_renewal_threshold': ('int', 1800),
            'session_extension_duration': ('int', 3600),
            'session_default_expiry_seconds': ('int', 86400),
            'max_login_attempts': ('int', 5),
            'login_lockout_minutes': ('int', 60),
            'announcement_cache_seconds': ('int', 300),
            'key_sync_interval_seconds': ('int', 60),
            'key_rotation_check_interval_minutes': ('int', 30),
            'max_key_rotation_failures': ('int', 20),
        }
        
        result = {}
        
        for key, (config_type, default) in config_map.items():
            value = self.db.query(SystemConfig).filter(
                SystemConfig.config_key == key,
                SystemConfig.app_name.contains('backend-client'),
                SystemConfig.is_deleted == 0
            ).first()
            
            if value:
                try:
                    if config_type == 'int':
                        result[key] = int(value.config_value)
                    elif config_type == 'bool':
                        result[key] = value.config_value.lower() == 'true'
                    else:
                        result[key] = value.config_value
                except (ValueError, AttributeError):
                    result[key] = default
            else:
                result[key] = default
        
        # 客户端心跳间隔优先使用 client_heartbeat_interval_seconds，兼容旧 key
        result['heartbeat_interval_seconds'] = result.get(
            'client_heartbeat_interval_seconds'
        ) or result.get('heartbeat_interval_seconds', 20)
        
        return result
    
    def get_config_value(self, config_key: str, app_name: str = 'backend-client', 
                         config_type: str = 'string', default: Any = None) -> Any:
        """获取单个配置值"""
        config = self.db.query(SystemConfig).filter(
            SystemConfig.config_key == config_key,
            SystemConfig.app_name.contains(app_name),
            SystemConfig.is_deleted == 0
        ).first()
        
        if not config:
            return default
        
        try:
            if config_type == 'int':
                return int(config.config_value)
            elif config_type == 'bool':
                return config.config_value.lower() == 'true'
            else:
                return config.config_value
        except (ValueError, AttributeError):
            return default