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

"""
心跳服务
"""
import json
from datetime import datetime, timedelta
from typing import Dict, Any, Tuple, Optional
from sqlalchemy.orm import Session
from sqlalchemy import or_
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.services.client_version_service import ClientVersionService
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,
                  client_version: str = "",
                  platform: str = "",
                  channel: str = "") -> 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, "您的卡密已到期，请前往续费或联系客户支持", {}

            version_service = ClientVersionService(self.db)
            should_block, version_check = version_service.should_block_client(
                client_version=client_version,
                platform=platform,
                channel=channel,
            )
            if should_block:
                return False, "客户端版本过低，请更新到最新版本后继续使用", {
                    "version_check": version_check
                }

            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,
                "version_check": version_check,
                "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', 864000),
            '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),
            # BUFF 扫货相关配置（来自 backend-selection）
            'buff_sweep_proxy_retry_times': ('int', 1),
            'buff_sweep_proxy_retry_interval_ms': ('int', 800),
            'buff_sweep_steam_listings_page_size': ('int', 10),
            'buff_sweep_steam_listings_max_scan_per_item': ('int', 60),
            'buff_sweep_pause_every_items': ('int', 0),
            'buff_sweep_pause_seconds': ('int', 0),
            # 客户端：每个选品饰品扫描+并发阶段结束后，再处理下一饰品前暂停秒数（非「每页 render」）
            'buff_sweep_pause_after_render_seconds': ('int', 0),
            'buff_sweep_proxy_rotate_every_items': ('int', 0),
            # 新增：扫货线程数最大数量
            'buff_sweep_max_worker_count': ('int', 32),
            # 新增：Steam卡价比例最小值
            'buff_steam_card_ratio_min': ('int', 4),
            # 新增：选品数量最大值
            'buff_selection_max_quantity': ('int', 300),
        }

        def parse_config_value(raw_value: Any, config_type: str) -> Any:
            """按配置类型解析字符串值，解析失败时返回原始值。"""
            try:
                if raw_value is None:
                    return None
                text = str(raw_value).strip()
                if config_type == 'int':
                    return int(text)
                if config_type == 'bool':
                    return text.lower() in ('true', '1', 'yes', 'on')
                if config_type == 'json':
                    return json.loads(text) if text else None
                return raw_value
            except Exception:
                return raw_value

        result = {key: default for key, (_, default) in config_map.items()}

        # 同步 backend-client / backend-selection 两类配置到客户端
        # 冲突优先级：backend-selection > backend-client（更贴近选品/扫货业务域）
        rows = self.db.query(SystemConfig).filter(
            SystemConfig.is_deleted == 0,
            or_(
                SystemConfig.app_name == 'backend-client',
                SystemConfig.app_name == 'backend-selection'
            )
        ).all()

        priority = {'backend-client': 1, 'backend-selection': 2}
        # 同 app 下多行同 key 时按 id 稳定覆盖（后者 id 更大优先）；表上 (config_key,is_deleted) 唯一，但历史数据可能重复
        rows.sort(key=lambda row: (priority.get(row.app_name or '', 0), row.id or 0))
        for row in rows:
            row_type = (row.config_type or 'string').lower()
            parsed = parse_config_value(row.config_value, row_type)
            result[row.config_key] = parsed

        # 客户端心跳间隔优先使用 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