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

"""
认证服务
"""
import secrets
import hashlib
import json
import bcrypt
from datetime import datetime, timedelta
from typing import Optional, Tuple
from sqlalchemy.orm import Session
from app.models.user import User
from app.models.system_config import SystemConfig
from app.utils.timezone import beijing_now_naive
from app.models.session import Session as SessionModel
from app.models.card import Card
from app.models.software import Software
from app.models.user_software_subscription import UserSoftwareSubscription
from app.models.user_steam_account_limit import UserSteamAccountLimit
from app.models.login_attempt import LoginAttempt
from app.models.login_attempt_his import LoginAttemptHis
from app.models.password_reset_request import PasswordResetRequest
from app.utils.machine_code import raw_machine_info_to_machine_code
from app.core.logger import logger

# bcrypt 最大支持 72 字节，超过会报错，统一在哈希前截断
BCRYPT_MAX_PASSWORD_BYTES = 72

# 默认软件代码
DEFAULT_SOFTWARE_CODE = "buff-xp-steam-sh"

# 默认Session有效期(秒)
DEFAULT_SESSION_EXPIRY = 86400  # 24小时

# 登录失败配置
MAX_LOGIN_ATTEMPTS = 5          # 最大失败次数
LOGIN_LOCKOUT_MINUTES = 60      # 锁定时间(分钟)


def hash_password(password: str) -> str:
    """密码加密（bcrypt，传入前截断至 72 字节避免库报错）"""
    raw = password.encode("utf-8")
    if len(raw) > BCRYPT_MAX_PASSWORD_BYTES:
        raw = raw[:BCRYPT_MAX_PASSWORD_BYTES]
    return bcrypt.hashpw(raw, bcrypt.gensalt()).decode("utf-8")


def verify_password(plain_password: str, hashed_password: str) -> bool:
    """密码验证"""
    raw = plain_password.encode("utf-8")
    if len(raw) > BCRYPT_MAX_PASSWORD_BYTES:
        raw = raw[:BCRYPT_MAX_PASSWORD_BYTES]
    try:
        return bcrypt.checkpw(raw, hashed_password.encode("utf-8"))
    except Exception:
        return False


def generate_machine_code(disk_serial: str, mac_address: str) -> str:
    """生成机器码 (MD5)"""
    # 组合机器特征信息
    combined = f"{disk_serial}:{mac_address}"
    # MD5哈希
    return hashlib.md5(combined.encode('utf-8')).hexdigest()


def generate_session_id() -> str:
    """生成会话ID"""
    return secrets.token_hex(32)


class AuthService:
    """认证服务"""

    def __init__(self, db: Session):
        self.db = db

    def _get_int_config(self, config_key: str, default: int) -> int:
        """
        从 system_config 表获取整型配置，app_name 包含 backend-client，读取失败时返回默认值。
        """
        try:
            cfg = (
                self.db.query(SystemConfig)
                .filter(
                    SystemConfig.config_key == config_key,
                    SystemConfig.app_name.contains("backend-client"),
                    SystemConfig.is_deleted == 0,
                )
                .first()
            )
            if not cfg or cfg.config_value is None:
                return default
            return int(str(cfg.config_value).strip())
        except Exception:
            return default

    def _get_software_id(self, software_code: str):
        """根据 software_code 解析 software_id，无效或未传则用默认软件"""
        code = (software_code or "").strip() or DEFAULT_SOFTWARE_CODE
        sw = self.db.query(Software).filter(
            Software.code == code,
            Software.is_deleted == 0
        ).first()
        return sw.id if sw else None

    def _get_subscription_expire(self, user_id: int, software_code: str) -> Tuple[Optional[datetime], Optional[int]]:
        """
        获取用户在某软件下的到期时间。优先读 user_software_subscription，兼容旧数据读 user.expire_time。
        返回 (expire_time, software_id)。
        """
        software_id = self._get_software_id(software_code)
        if software_id is None:
            return None, None
        sub = self.db.query(UserSoftwareSubscription).filter(
            UserSoftwareSubscription.user_id == user_id,
            UserSoftwareSubscription.software_id == software_id,
            UserSoftwareSubscription.is_deleted == 0
        ).first()
        if sub and sub.expire_time is not None:
            return sub.expire_time, software_id
        user = self.db.query(User).filter(User.id == user_id, User.is_deleted == 0).first()
        if user and user.expire_time is not None:
            return user.expire_time, software_id
        return None, software_id

    def _check_account_locked(self, username: str) -> Tuple[bool, str]:
        """
        检查账号是否被锁定

        Returns:
            (is_locked, message)
        """
        # 获取该用户名最近的登录失败记录
        lockout_time = beijing_now_naive() - timedelta(minutes=LOGIN_LOCKOUT_MINUTES)

        recent_failures = self.db.query(LoginAttempt).filter(
            LoginAttempt.username == username,
            LoginAttempt.attempted_at > lockout_time,
            LoginAttempt.success == 0
        ).count()

        if recent_failures >= MAX_LOGIN_ATTEMPTS:
            # 获取最早一次失败的时间
            earliest_failure = self.db.query(LoginAttempt).filter(
                LoginAttempt.username == username,
                LoginAttempt.attempted_at > lockout_time,
                LoginAttempt.success == 0
            ).order_by(LoginAttempt.attempted_at.asc()).first()

            if earliest_failure:
                # 计算解锁时间
                unlock_time = earliest_failure.attempted_at + timedelta(minutes=LOGIN_LOCKOUT_MINUTES)
                remaining_seconds = int((unlock_time - beijing_now_naive()).total_seconds())

                if remaining_seconds > 0:
                    minutes_left = remaining_seconds // 60
                    seconds_left = remaining_seconds % 60
                    return True, f"账号已锁定，请{minutes_left}分钟{seconds_left}秒后重试"

            return True, "账号已锁定，请稍后重试"

        return False, ""

    def _record_login_attempt(self, username: str, success: bool, machine_code: str = None,
                               ip_address: str = None, user_agent: str = None,
                               failure_reason: str = None):
        """记录登录尝试。主表 login_attempts 必写；历史表 login_attempts_his 可选（缺失时仅打日志）。"""
        try:
            attempt = LoginAttempt(
                username=username,
                success=success,
                machine_code=machine_code,
                ip_address=ip_address,
                user_agent=user_agent,
                failure_reason=failure_reason
            )
            self.db.add(attempt)
            self.db.commit()
        except Exception as e:
            self.db.rollback()
            logger.exception(f"记录登录尝试(主表)失败: {e}")
            return

        try:
            attempt_data = {
                "username": username,
                "success": success,
                "machine_code": machine_code,
                "ip_address": ip_address,
                "user_agent": user_agent,
                "failure_reason": failure_reason
            }
            attempt_his = LoginAttemptHis(
                username=username,
                success=success,
                machine_code=machine_code,
                ip_address=ip_address,
                user_agent=user_agent,
                failure_reason=failure_reason,
                action_type="INSERT",
                changed_fields=json.dumps(attempt_data, ensure_ascii=False),
                operator="system"
            )
            self.db.add(attempt_his)
            self.db.commit()
        except Exception as e:
            self.db.rollback()
            logger.warning(f"记录登录尝试(历史表)失败，已忽略: {e}")

        try:
            # 先写历史表，再改/删：将即将删除的登录尝试写入 login_attempts_his
            old_attempts_list = self.db.query(LoginAttempt).filter(
                LoginAttempt.attempted_at < beijing_now_naive() - timedelta(days=30)
            ).all()
            for a in old_attempts_list:
                self.db.add(LoginAttemptHis(
                    username=a.username,
                    success=a.success,
                    machine_code=a.machine_code,
                    ip_address=a.ip_address,
                    user_agent=a.user_agent,
                    failure_reason=a.failure_reason,
                    attempted_at=a.attempted_at,
                    action_type="DELETE",
                    operator="system",
                ))
            if old_attempts_list:
                self.db.query(LoginAttempt).filter(
                    LoginAttempt.attempted_at < beijing_now_naive() - timedelta(days=30)
                ).delete()
                self.db.commit()
        except Exception as e:
            self.db.rollback()
            logger.warning(f"清理过期登录记录失败: {e}")

        try:
            old_history = self.db.query(LoginAttemptHis).filter(
                LoginAttemptHis.created_at < beijing_now_naive() - timedelta(days=90)
            ).delete()
            if old_history > 0:
                self.db.commit()
        except Exception as e:
            self.db.rollback()
            logger.warning(f"清理登录历史记录失败: {e}")

    def _clear_login_attempts(self, username: str):
        """清除登录失败记录 (登录成功时调用)"""
        try:
            # 先写历史表，再改/删：将即将删除的失败尝试写入 login_attempts_his
            attempts_to_clear = self.db.query(LoginAttempt).filter(
                LoginAttempt.username == username,
                LoginAttempt.success == 0
            ).all()
            for a in attempts_to_clear:
                self.db.add(LoginAttemptHis(
                    username=a.username,
                    success=a.success,
                    machine_code=a.machine_code,
                    ip_address=a.ip_address,
                    user_agent=a.user_agent,
                    failure_reason=a.failure_reason,
                    attempted_at=a.attempted_at,
                    action_type="DELETE",
                    operator="system",
                ))
            self.db.query(LoginAttempt).filter(
                LoginAttempt.username == username,
                LoginAttempt.success == 0
            ).delete()
            self.db.commit()
        except Exception as e:
            self.db.rollback()
            logger.exception(f"清除登录记录失败: {e}")

    def register(self, username: str, password: str, email: str = None,
                 card_code: str = None, machine_code: str = None,
                 software_code: str = None) -> Tuple[bool, str, dict]:
        """
        用户注册。支持多软件：按 software_code 绑定到对应软件，卡密/到期写入 user_software_subscription。

        Args:
            username: 用户名
            password: 密码
            email: 邮箱(可选)
            card_code: 卡密代码(可选，注册时激活)
            machine_code: 机器码(可选)
            software_code: 软件特征码(可选，默认 buff-xp-steam-sh)

        Returns:
            (success, message, user_info)
        """
        try:
            # 检查用户名是否已存在
            existing_user = self.db.query(User).filter(
                User.username == username,
                User.is_deleted == 0
            ).first()
            if existing_user:
                return False, "用户名已存在", {}

            # 检查邮箱是否已存在(如果提供了邮箱)
            if email:
                existing_email = self.db.query(User).filter(
                    User.email == email,
                    User.is_deleted == 0
                ).first()
                if existing_email:
                    return False, "邮箱已被注册", {}

            sw_code = (software_code or "").strip() or DEFAULT_SOFTWARE_CODE
            software = self.db.query(Software).filter(
                Software.code == sw_code,
                Software.is_active == 1,
                Software.is_deleted == 0
            ).first()
            if not software:
                return False, "软件不存在或已禁用", {}

            expire_time = None
            card = None

            # 如果提供了卡密，检查并激活（卡密必须属于当前软件）
            if card_code:
                card = self.db.query(Card).filter(
                    Card.card_code == card_code,
                    Card.is_deleted == 0
                ).first()

                if not card:
                    return False, "卡密不存在", {}

                if card.status == 1:
                    return False, "卡密已被使用", {}

                if card.status == 2:
                    return False, "卡密已过期", {}

                if card.software_code != sw_code:
                    return False, "卡密类型不匹配", {}

                card.status = 1
                card.user_id = None
                card.activated_at = beijing_now_naive()
                card.expire_time = beijing_now_naive() + timedelta(days=card.days)
                expire_time = card.expire_time

            # 创建用户（用户表不再存 card_id/expire_time，由订阅表维护）
            hashed_password = hash_password(password)
            user = User(
                username=username,
                password=hashed_password,
                email=email,
                machine_code=machine_code,
                card_id=card.id if card else None,
                expire_time=expire_time,  # 兼容旧逻辑，同时写订阅表
                status=1
            )

            self.db.add(user)
            self.db.flush()

            if card:
                card.user_id = user.id

            # 用户-软件订阅：多软件共用用户表，按软件存卡密/到期
            sub = UserSoftwareSubscription(
                user_id=user.id,
                software_id=software.id,
                card_id=card.id if card else None,
                expire_time=expire_time,
                is_deleted=0
            )
            self.db.add(sub)

            # 新用户默认 Steam 账号绑定数量上限（默认 3，管理员可后续修改）
            limit_row = UserSteamAccountLimit(user_id=user.id, max_steam_accounts=3)
            self.db.add(limit_row)

            self.db.commit()

            logger.info(f"用户注册成功: {username}, software={sw_code}")

            return True, "注册成功", {
                "user_id": user.id,
                "username": user.username,
                "expire_time": user.expire_time.isoformat() if user.expire_time else None,
                "days_remaining": (user.expire_time - beijing_now_naive()).days if user.expire_time else 0
            }

        except Exception as e:
            self.db.rollback()
            logger.exception(f"用户注册失败: {e}")
            return False, f"注册失败: {str(e)}", {}

    def forgot_password(self, username: str, email: str) -> Tuple[bool, str]:
        """
        找回密码：校验用户名与邮箱匹配后写入工单表，由外部脚本扫描并发邮件。
        无论是否匹配均返回统一成功提示，避免泄露用户是否存在。
        """
        try:
            username = (username or "").strip()
            email = (email or "").strip()
            if not username or not email:
                return False, "请输入用户名和邮箱"

            user = self.db.query(User).filter(
                User.username == username,
                User.is_deleted == 0
            ).first()
            if not user:
                logger.info(f"找回密码：用户不存在 {username}")
                return True, "若该用户名与邮箱匹配，我们将把新密码发送至您的邮箱，请查收。"

            if not user.email or user.email.strip().lower() != email.strip().lower():
                logger.info(f"找回密码：邮箱不匹配 username={username}")
                return True, "若该用户名与邮箱匹配，我们将把新密码发送至您的邮箱，请查收。"

            # 写入工单表，状态 pending
            req = PasswordResetRequest(
                user_id=user.id,
                username=user.username,
                email=user.email,
                status="pending"
            )
            self.db.add(req)
            self.db.commit()
            logger.info(f"找回密码工单已创建: user_id={user.id}, username={username}")
            return True, "若该用户名与邮箱匹配，我们将把新密码发送至您的邮箱，请查收。"
        except Exception as e:
            self.db.rollback()
            logger.exception(f"找回密码申请失败: {e}")
            return False, "系统异常，请稍后重试；若持续发生请联系客服。"

    def login(self, username: str, password: str, machine_code: str,
              software_code: str = None) -> Tuple[bool, str, dict]:
        """
        用户登录。按 software_code 校验该软件下的订阅到期时间。

        Args:
            username: 用户名
            password: 密码
            machine_code: 机器码
            software_code: 软件特征码(可选，默认 buff-xp-steam-sh)

        Returns:
            (success, message, data)
        """
        try:
            # 1. 检查账号是否被锁定
            is_locked, lock_message = self._check_account_locked(username)
            if is_locked:
                logger.warning(f"登录被拒绝，账号已锁定: {username}")
                return False, lock_message, {}

            # 2. 查找用户
            user = self.db.query(User).filter(
                User.username == username,
                User.is_deleted == 0
            ).first()

            if not user:
                self._record_login_attempt(username, success=False, machine_code=machine_code)
                logger.warning(f"登录失败，用户不存在: {username}")
                return False, "用户名或密码错误", {}

            # 3. 验证密码
            if not verify_password(password, user.password):
                self._record_login_attempt(username, success=False, machine_code=machine_code)
                recent_failures = self._get_recent_failure_count(username)
                remaining_attempts = MAX_LOGIN_ATTEMPTS - recent_failures
                if remaining_attempts <= 0:
                    return False, "账号已锁定，请稍后重试", {}
                logger.warning(f"密码错误: {username}, 剩余尝试次数: {remaining_attempts}")
                return False, f"用户名或密码错误，还能尝试{remaining_attempts}次", {}

            # 4. 验证账户状态
            if user.status != 1:
                self._record_login_attempt(username, success=False, machine_code=machine_code)
                return False, "账号已被禁用", {}

            # 5. 机器码处理：首次登录绑定；已绑定时仅校验，不覆盖（换机必须先解绑）
            if user.machine_code is None:
                user.machine_code = machine_code
            elif user.machine_code != machine_code:
                self._record_login_attempt(username, success=False, machine_code=machine_code)
                return False, "该账号已绑定其他设备。如需换机请先在原设备解绑或联系客服", {}

            # 6. 按软件检查卡密有效期（优先 user_software_subscription，兼容 user.expire_time）
            expire_time, _ = self._get_subscription_expire(user.id, software_code or "")
            if expire_time and expire_time < beijing_now_naive():
                self._record_login_attempt(username, success=False, machine_code=machine_code)
                return False, "您的卡密已到期，请前往续费或联系客户支持", {}

            # 7. 创建会话
            session_id = generate_session_id()
            expires_at = beijing_now_naive() + timedelta(seconds=DEFAULT_SESSION_EXPIRY)
            session = SessionModel(
                session_id=session_id,
                user_id=user.id,
                machine_code=machine_code,
                is_active=True,
                expires_at=expires_at
            )
            self.db.add(session)
            self.db.commit()  # 先提交会话，确保会话落库；避免 _clear_login_attempts 失败时 rollback 导致会话丢失
            self._clear_login_attempts(username)  # 清除登录失败记录（失败不影响已提交的会话）

            days_remaining = 0
            if expire_time:
                days_remaining = (expire_time - beijing_now_naive()).days
            expire_str = expire_time.isoformat() if expire_time else None

            logger.info(f"用户登录成功: {username}")

            return True, "登录成功", {
                "user_id": user.id,
                "username": user.username,
                "session_id": session_id,
                "expire_time": expire_str,
                "days_remaining": days_remaining
            }

        except Exception as e:
            self.db.rollback()
            logger.exception(f"用户登录失败: {e}")
            return False, f"登录失败: {str(e)}", {}

    def _get_recent_failure_count(self, username: str) -> int:
        """获取最近失败次数"""
        lockout_time = beijing_now_naive() - timedelta(minutes=LOGIN_LOCKOUT_MINUTES)
        return self.db.query(LoginAttempt).filter(
            LoginAttempt.username == username,
            LoginAttempt.attempted_at > lockout_time,
            LoginAttempt.success == 0
        ).count()

    def logout(self, session_id: str) -> Tuple[bool, str]:
        """
        用户登出

        Args:
            session_id: 会话ID

        Returns:
            (success, message)
        """
        try:
            session = self.db.query(SessionModel).filter(
                SessionModel.session_id == session_id
            ).first()

            if session:
                session.is_active = False
                self.db.commit()
                logger.info(f"用户登出成功: {session_id}")

            return True, "登出成功"

        except Exception as e:
            self.db.rollback()
            logger.exception(f"用户登出失败: {e}")
            return False, f"登出失败: {str(e)}"

    def unbind(self, username: str, password: str, machine_code: str) -> Tuple[bool, str]:
        """
        机器解绑

        Args:
            username: 用户名
            password: 密码
            machine_code: 当前机器码

        Returns:
            (success, message)
        """
        try:
            # 查找用户
            user = self.db.query(User).filter(
                User.username == username,
                User.is_deleted == 0
            ).first()

            if not user:
                return False, "用户不存在"

            # 验证密码
            if not verify_password(password, user.password):
                return False, "密码错误"

            # 验证机器码
            if user.machine_code != machine_code:
                return False, "机器码不匹配"

            # 校验机器解绑冷却周期（单位：天）
            cooldown_days = self._get_int_config("machine_unbind_cooldown_days", 10)
            if cooldown_days > 0 and user.machine_code:
                bind_time = user.updated_at or user.created_at
                now = beijing_now_naive()
                allow_unbind_at = bind_time + timedelta(days=cooldown_days)
                if now < allow_unbind_at:
                    return False, f"当前设备绑定未满 {cooldown_days} 天，暂不支持解绑。请在 {allow_unbind_at.strftime('%Y-%m-%d %H:%M:%S')} 之后再试。"

            # 清除机器码
            user.machine_code = None

            # 先写历史表，再改/删：将即将删除的会话写入 sessions_his
            from app.utils.history import create_session_history
            sessions_to_remove = self.db.query(SessionModel).filter(
                SessionModel.user_id == user.id
            ).all()
            for s in sessions_to_remove:
                create_session_history(self.db, s, "DELETE")

            # 清除用户的所有会话(强制下线)
            self.db.query(SessionModel).filter(
                SessionModel.user_id == user.id
            ).delete()

            self.db.commit()

            logger.info(f"机器解绑成功: {username}")

            return True, "解绑成功，可在其他机器登录"

        except Exception as e:
            self.db.rollback()
            logger.exception(f"机器解绑失败: {e}")
            return False, f"解绑失败: {str(e)}"

    def change_password(self, username: str, old_password: str, new_password: str) -> Tuple[bool, str]:
        """
        修改密码（已登录用户通过当前密码修改为新密码）

        Args:
            username: 用户名
            old_password: 当前密码
            new_password: 新密码

        Returns:
            (success, message)
        """
        try:
            user = self.db.query(User).filter(
                User.username == username,
                User.is_deleted == 0
            ).first()

            if not user:
                return False, "用户不存在"

            if not verify_password(old_password, user.password):
                return False, "当前密码错误，请检查后重试"

            raw_new = new_password.encode("utf-8")
            if len(raw_new) < 6:
                return False, "新密码长度至少 6 位，请重新设置"

            if len(raw_new) > BCRYPT_MAX_PASSWORD_BYTES:
                return False, "新密码过长，请重新设置"

            user.password = hash_password(new_password)
            self.db.commit()
            logger.info(f"用户修改密码成功: {username}")
            return True, "密码修改成功，请使用新密码登录"

        except Exception as e:
            self.db.rollback()
            logger.exception(f"修改密码失败: {e}")
            return False, "系统异常，请稍后重试；若持续发生请联系客服"

    def verify_session(self, session_id: str) -> Optional[SessionModel]:
        """验证会话（仅校验会话存在且未过期，不校验机器码与用户卡密）"""
        session = self.db.query(SessionModel).filter(
            SessionModel.session_id == session_id,
            SessionModel.is_active == 1,
            SessionModel.expires_at > beijing_now_naive()
        ).first()
        return session

    def verify_session_with_machine_and_expiry(
        self,
        session_id: str,
        machine_code: str,
        software_code: Optional[str] = None,
    ) -> Tuple[Optional[SessionModel], Optional[int], Optional[str]]:
        """
        校验会话 + 机器码一致 + 用户未禁用 + 用户卡密未过期。
        与心跳逻辑一致，供所有需登录的接口统一使用。
        Returns:
            (session, None, None) 成功
            (None, error_code, error_message) 失败
            error_code 401: 未登录/会话过期；403: 机器码不匹配/账号禁用/卡密过期
        """
        session = self.verify_session(session_id)
        if not session:
            return None, 401, "会话已过期，请重新登录"
        mc = (machine_code or "").strip()
        if mc:
            # 与登录/心跳一致：session 存的是对 raw_machine_info 规范化后的 MD5；客户端可能传 raw（如 DISK=xxx）或已 MD5
            # 若已是 32 位小写 hex 则视为 MD5 直接比较，否则视为 raw 做规范化+MD5 再比较
            if len(mc) == 32 and all(c in "0123456789abcdef" for c in mc.lower()):
                mc_normalized = mc.lower()
            else:
                mc_normalized = raw_machine_info_to_machine_code(mc)
            if session.machine_code != mc_normalized:
                return None, 403, "账号已在其他设备登录，如非本人操作请联系客服"
        user = self.get_user_by_id(session.user_id)
        if not user:
            return None, 401, "用户不存在"
        if user.status != 1:
            return None, 403, "账号已被禁用"
        expire_time, _ = self._get_subscription_expire(user.id, software_code or "")
        if expire_time and expire_time < beijing_now_naive():
            return None, 403, "您的卡密已到期，请前往续费或联系客户支持"
        return session, None, None

    def get_user_by_id(self, user_id: int) -> Optional[User]:
        """根据ID获取用户"""
        return self.db.query(User).filter(
            User.id == user_id,
            User.is_deleted == 0
        ).first()

    def validate_session_or_token(self, username: str = None) -> Tuple[Optional[int], str, Optional[str]]:
        """
        验证会话或Token

        Returns:
            (user_id, message, session_id)
        """
        return None, "验证失败", None