""" API路由 - v1版本 完整实现认证、充值、心跳、公告、密钥管理等核心接口 符合统一用户友好错误提示规范 """ from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query, Request from pydantic import BaseModel, Field from typing import Optional, Dict, Any, List from sqlalchemy.orm import Session from datetime import datetime import time import hashlib import json from app.database import get_db from app.core.encrypt import encryptor from app.middleware.security import create_encrypted_response, secure_endpoint, decrypt_request_data from app.services.auth_service import AuthService, verify_password from app.services.recharge_service import RechargeService from app.services.heartbeat_service import HeartbeatService from app.services.announcement_service import AnnouncementService from app.services.client_version_service import ClientVersionService from app.services.key_service import KeyService from app.services.steam_account_service import SteamAccountService from app.utils.response import create_client_response, UserFriendlyError from app.models.buff_filter_option import BuffFilterOption from app.models.buff_selection_config import BuffSelectionConfig from app.models.buff_selection_request import BuffSelectionRequest from app.models.buff_goods_attr import BuffGoodsAttr from app.services.buff_selection_service import run_selection_task from fastapi.responses import JSONResponse from app.utils.timezone import beijing_now_naive from app.models.system_config import SystemConfig from app.models.buff_selection_request_item import BuffSelectionRequestItem from app.models.buff_selection_request_log import BuffSelectionRequestLog from app.models.buff_sweep_request import BuffSweepRequest from app.models.buff_sweep_request_item import BuffSweepRequestItem from app.models.buff_sweep_request_log import BuffSweepRequestLog from app.models.steam_account import SteamAccount from app.services.buff_sweep_service import run_sweep_task from app.utils.machine_code import raw_machine_info_to_machine_code as _raw_to_machine_code from app.core.logger import logger router = APIRouter() # ============ 请求/响应模型 ============ class LoginRequest(BaseModel): """登录请求""" username: str = Field(..., min_length=3, max_length=100, description="用户名") password: str = Field(..., min_length=6, max_length=100, description="密码") machine_code: str = Field(..., description="机器码") class RegisterRequest(BaseModel): """注册请求""" username: str = Field(..., min_length=3, max_length=100, description="用户名") password: str = Field(..., min_length=6, max_length=100, description="密码") email: Optional[str] = Field(None, description="邮箱") card_code: Optional[str] = Field(None, description="卡密代码") class RechargeRequest(BaseModel): """充值请求""" username: str = Field(..., description="用户名") password: str = Field(..., description="密码") card_code: str = Field(..., description="卡密代码") class UnbindRequest(BaseModel): """解绑请求""" username: str = Field(..., description="用户名") password: str = Field(..., description="密码") class ForgotPasswordRequest(BaseModel): """找回密码请求""" username: str = Field(..., min_length=1, description="用户名") email: str = Field(..., min_length=1, description="邮箱") class HeartbeatRequest(BaseModel): """心跳请求""" session_id: str = Field(..., description="会话ID") machine_code: str = Field(..., description="机器码") class AnnouncementsRequest(BaseModel): """公告请求""" page: int = Field(1, ge=1, description="页码") page_size: int = Field(10, ge=1, le=100, description="每页数量") class LogoutRequest(BaseModel): """登出请求""" session_id: str = Field(..., description="会话ID") class ClientConfigRequest(BaseModel): """客户端配置请求""" session_id: str = Field(..., description="会话ID") class ClientVersionCheckRequest(BaseModel): """客户端版本检查请求""" client_version: str = Field(..., description="当前客户端版本") platform: Optional[str] = Field("windows", description="平台: windows/macos/linux") channel: Optional[str] = Field("stable", description="通道: stable/beta") # ============ 响应模型 ============ class UserInfoResponse(BaseModel): """用户信息响应""" user_id: int username: str email: Optional[str] expire_time: Optional[str] days_remaining: int status: int class AnnouncementsResponse(BaseModel): """公告响应""" data: List[Dict[str, Any]] total: int page: int page_size: int class HeartbeatResponse(BaseModel): """心跳响应""" status: str session_id: str expire_time: str config: Dict[str, Any] # ============ 辅助函数 ============ def handle_service_result(success: bool, message: str, data: Any = None, error_mapping: Optional[Dict[str, int]] = None) -> Dict[str, Any]: """ 处理服务层返回结果，转换为统一错误响应格式 Args: success: 是否成功 message: 消息 data: 响应数据 error_mapping: 错误消息到状态码的映射 Returns: 统一格式的响应字典 """ if success: return create_client_response(200, message, data) # 默认错误映射 if error_mapping is None: error_mapping = { "密码错误": 401, "用户名": 401, "不存在": 404, "禁用": 403, "到期": 403, "已使用": 403, "已过期": 403 } # 根据消息内容确定错误码 status_code = 400 # 默认 for key, code in error_mapping.items(): if key in message: status_code = code break # 返回错误响应（使用统一的错误响应格式） return create_client_response(status_code, message, None) # ============ 认证接口 ============ def _raw_machine_info_to_machine_code(decrypted_params: dict) -> str: """从解密参数中取 raw_machine_info，规范化后由后端统一计算 MD5 作为 machine_code（不信任客户端传入的 machine_code）""" raw = (decrypted_params.get("raw_machine_info") or decrypted_params.get("machine_code") or "").strip() return _raw_to_machine_code(raw) @router.post("/auth/login") @secure_endpoint(require_login=False, require_machine_code=False) async def login(decrypted_params: dict, request: Request, db: Session, session=None, user=None): """ 用户登录（安全端点：请求体为加密 params，响应为加密 data） - 解密后取 username、password、raw_machine_info - 后端对 raw_machine_info 做 MD5 得到 machine_code，不信任客户端传入的 machine_code 统一错误提示：401 用户名或密码错误；403 卡密已过期/机器码不匹配/账号被禁用 """ try: username = (decrypted_params.get("username") or "").strip() password = decrypted_params.get("password") or "" machine_code = _raw_machine_info_to_machine_code(decrypted_params) if not username or not password: return handle_service_result(False, "请求参数错误，请检查后重试", None, {"参数": 400}) client_version = (decrypted_params.get("client_version") or "").strip() platform = (decrypted_params.get("platform") or "windows").strip() channel = (decrypted_params.get("channel") or "stable").strip() if client_version: version_service = ClientVersionService(db) should_block, result = version_service.should_block_client( client_version=client_version, platform=platform, channel=channel, ) if should_block: return create_client_response(403, "客户端版本过低，请更新到最新版本后继续使用", { "version_check": result }) software_code = (decrypted_params.get("software_code") or "").strip() auth_service = AuthService(db) success, message, data = auth_service.login( username=username, password=password, machine_code=machine_code, software_code=software_code or None ) if success: logger.info(f"登录成功: {username}") return create_client_response(200, message, data) error_mapping = {"密码错误": 401, "用户名": 401, "禁用": 403, "到期": 403, "设备": 403, "绑定": 403} return handle_service_result(success, message, None, error_mapping) except Exception as e: logger.exception(f"登录接口异常: {e}") return create_client_response(500, "系统异常，请稍后重试；若持续发生请联系客服", None) @router.post("/auth/register") @secure_endpoint(require_login=False, require_machine_code=False) async def register(decrypted_params: dict, request: Request, db: Session, session=None, user=None): """ 用户注册（安全端点：请求体为加密 params） 统一错误提示：400 用户名已存在 """ try: username = (decrypted_params.get("username") or "").strip() password = decrypted_params.get("password") or "" email = decrypted_params.get("email") or "" card_code = decrypted_params.get("card_code") or "" software_code = (decrypted_params.get("software_code") or "").strip() if not username or not password: return handle_service_result(False, "请求参数错误，请检查后重试", None, {"参数": 400}) auth_service = AuthService(db) success, message, data = auth_service.register( username=username, password=password, email=email or None, card_code=card_code or None, software_code=software_code or None ) if success: return create_client_response(200, message, data) return handle_service_result(success, message, None, {"已存在": 400}) except Exception as e: logger.exception(f"注册接口异常: {e}") return create_client_response(500, "系统异常，请稍后重试；若持续发生请联系客服", None) @router.post("/auth/logout") async def logout(request: Request, db: Session = Depends(get_db)): """ 用户登出 - 支持两种加密格式：① encrypted_params + aes_key（与 secure_endpoint 一致）② 单 blob：encrypted + key_version - 也支持明文：{"session_id": "..."}（兼容） - 登出成功后数据库 sessions 表对应记录的 is_active 置为 0 """ start_time = time.time() session_id = "" try: body = await request.json() if not isinstance(body, dict): logger.warning("[logout] 请求体非 dict") return { **create_client_response(400, "请求参数错误，请检查后重试", None), "_execution_time": int((time.time() - start_time) * 1000), } body_keys = list(body.keys()) if body else [] logger.info(f"[logout] 收到登出请求 body_keys={body_keys}") # 格式1：encrypted_params + aes_key（与登录/心跳一致，部分客户端走此格式） encrypted_params = (body.get("encrypted_params") or "").strip() aes_key_b64 = (body.get("aes_key") or "").strip() if encrypted_params and aes_key_b64: try: key_version = int(body.get("key_version") or 1) decrypted = decrypt_request_data( encrypted_params, key_version, db, encrypted_aes_key=aes_key_b64 ) session_id = (decrypted.get("session_id") or "").strip() logger.info(f"[logout] 解密(encrypted_params)得到 session_id 长度={len(session_id)}") except Exception as e: logger.warning(f"[logout] encrypted_params 解密失败: {e}") return { **create_client_response(400, "请求参数错误，请检查后重试", None), "_execution_time": int((time.time() - start_time) * 1000), } # 格式2：单 blob encrypted + key_version（api_client.make_request） elif body.get("encrypted") and body.get("key_version") is not None: try: key_version = int(body.get("key_version", 1)) decrypted = encryptor.decrypt_from_client( (body.get("encrypted") or "").strip(), key_version, db ) session_id = (decrypted.get("session_id") or "").strip() logger.info(f"[logout] 解密(encrypted blob)得到 session_id 长度={len(session_id)}") except Exception as e: logger.warning(f"[logout] encrypted blob 解密失败: {e}") return { **create_client_response(400, "请求参数错误，请检查后重试", None), "_execution_time": int((time.time() - start_time) * 1000), } else: # 明文 session_id（兼容） session_id = (body.get("session_id") or "").strip() if session_id: logger.info(f"[logout] 明文 session_id 长度={len(session_id)}") if not session_id: logger.warning("[logout] 未解析到 session_id") return { **create_client_response(400, "请求参数错误，请检查后重试", None), "_execution_time": int((time.time() - start_time) * 1000), } auth_service = AuthService(db) success, message = auth_service.logout(session_id=session_id) logger.info(f"[logout] 登出结果 success={success} session_id_len={len(session_id)}") execution_time = int((time.time() - start_time) * 1000) return { **create_client_response(200 if success else 400, message), "_execution_time": execution_time, } except HTTPException: raise except Exception as e: logger.exception(f"登出接口异常: {e}") raise HTTPException( status_code=500, detail="系统异常，请稍后重试；若持续发生请联系客服", ) @router.post("/auth/forgot-password") async def forgot_password(request_data: ForgotPasswordRequest, db: Session = Depends(get_db)): """ 找回密码：提交用户名和邮箱，匹配则写入工单表，由后台脚本扫描并发送新密码邮件。 为防泄露，无论是否匹配均返回相同提示。 """ try: auth_service = AuthService(db) success, message = auth_service.forgot_password( username=request_data.username.strip(), email=request_data.email.strip() ) return create_client_response(200 if success else 500, message, None) except Exception as e: logger.exception(f"找回密码接口异常: {e}") return create_client_response(500, "系统异常，请稍后重试；若持续发生请联系客服", None) @router.post("/auth/recharge") @secure_endpoint(require_login=False, require_machine_code=False) async def recharge(decrypted_params: dict, request: Request, db: Session, session=None, user=None): """ 卡密充值（安全端点：请求体为加密 params） 统一错误提示：404 卡密不存在；403 卡密已使用/已过期 """ try: username = (decrypted_params.get("username") or "").strip() password = decrypted_params.get("password") or "" card_code = (decrypted_params.get("card_code") or "").strip() software_code = (decrypted_params.get("software_code") or "").strip() if not username or not password or not card_code: return handle_service_result(False, "请求参数错误，请检查后重试", None, {"参数": 400}) recharge_service = RechargeService(db) success, message, data = recharge_service.recharge( username=username, password=password, card_code=card_code, software_code=software_code or None ) if success: return create_client_response(200, message, data) return handle_service_result(success, message, None, {"不存在": 404, "已使用": 403, "已过期": 403}) except Exception as e: logger.exception(f"充值接口异常: {e}") return create_client_response(500, "系统异常，请稍后重试；若持续发生请联系客服", None) @router.post("/auth/unbind") @secure_endpoint(require_login=False, require_machine_code=False) async def unbind(decrypted_params: dict, request: Request, db: Session, session=None, user=None): """ 机器解绑（安全端点：请求体为加密 params） 统一错误提示：404 用户不存在；401 密码错误 """ try: username = (decrypted_params.get("username") or "").strip() password = decrypted_params.get("password") or "" if not username or not password: return handle_service_result(False, "请求参数错误，请检查后重试", None, {"参数": 400}) machine_code = _raw_machine_info_to_machine_code(decrypted_params) auth_service = AuthService(db) success, message = auth_service.unbind( username=username, password=password, machine_code=machine_code ) if success: return create_client_response(200, message) return handle_service_result(success, message, None, {"不存在": 404, "密码错误": 401}) except Exception as e: logger.exception(f"解绑接口异常: {e}") return create_client_response(500, "系统异常，请稍后重试；若持续发生请联系客服", None) @router.post("/auth/change-password") @secure_endpoint(require_login=False, require_machine_code=False) async def change_password(decrypted_params: dict, request: Request, db: Session, session=None, user=None): """ 修改密码（安全端点：请求体为加密 params，含 username、old_password、new_password） 统一错误提示：用户不存在、当前密码错误、新密码格式不符合要求 """ try: username = (decrypted_params.get("username") or "").strip() old_password = (decrypted_params.get("old_password") or "").strip() new_password = (decrypted_params.get("new_password") or "").strip() if not username or not old_password or not new_password: return handle_service_result(False, "请求参数错误，请检查后重试", None, {"参数": 400}) auth_service = AuthService(db) success, message = auth_service.change_password( username=username, old_password=old_password, new_password=new_password ) if success: return create_client_response(200, message) return handle_service_result( success, message, None, {"用户不存在": 404, "当前密码错误": 401, "新密码": 400} ) except Exception as e: logger.exception(f"修改密码接口异常: {e}") return create_client_response(500, "系统异常，请稍后重试；若持续发生请联系客服", None) # ============ 心跳接口 ============ @router.post("/heartbeat") @secure_endpoint(require_login=True, require_machine_code=True) async def heartbeat(decrypted_params: dict, request: Request, db: Session, session=None, user=None): """ 心跳检测（安全端点：请求体为加密 params，含 session_id、machine_code/raw_machine_info） - 与登录一致：对 raw_machine_info 或 machine_code 做 MD5 后与 session 绑定值比较 - AES+RSA 加密请求与响应 - 需要登录验证 - 返回系统配置（带签名响应，防止中间人攻击） 统一错误提示： - 401: 会话已过期 - 403: 卡密已到期 """ try: session_id = (decrypted_params.get("session_id") or "").strip() # 与登录接口一致：后端统一用 MD5 计算 machine_code，再与 session.machine_code 比较 machine_code = _raw_machine_info_to_machine_code(decrypted_params) software_code = (decrypted_params.get("software_code") or "").strip() client_version = (decrypted_params.get("client_version") or "").strip() platform = (decrypted_params.get("platform") or "").strip() channel = (decrypted_params.get("channel") or "").strip() if not session_id or not machine_code: return {"code": 400, "msg": "请求参数错误，请检查后重试"} heartbeat_service = HeartbeatService(db) success, message, data = heartbeat_service.heartbeat( session_id=session_id, machine_code=machine_code, software_code=software_code or None, client_version=client_version, platform=platform, channel=channel, ) if success: logger.debug(f"心跳检测成功: session_id={session_id}, renewed={data.get('session_info', {}).get('renewed', False)}") return {"code": 200, "msg": message, **data} else: error_code = 400 if "过期" in message or "登录" in message: error_code = 401 elif "到期" in message or "禁用" in message or "版本" in message: error_code = 403 return {"code": error_code, "msg": message} except Exception as e: logger.exception(f"心跳接口异常: {e}") return {"code": 500, "msg": "系统异常，请稍后重试；若持续发生请联系客服"} @router.post("/client/version/check") @secure_endpoint(require_login=False, require_machine_code=False) async def post_client_version_check(decrypted_params: dict, request: Request, db: Session, session=None, user=None): """ 客户端版本检查（安全端点） - 登录前/登录后均可调用 - 返回最新版本、强更策略、下载地址、富文本更新日志 """ try: client_version = (decrypted_params.get("client_version") or "").strip() platform = (decrypted_params.get("platform") or "windows").strip() channel = (decrypted_params.get("channel") or "stable").strip() if not client_version: return create_client_response(400, "请求参数错误，请检查后重试", None) version_service = ClientVersionService(db) result = version_service.build_check_result( client_version=client_version, platform=platform, channel=channel, ) return create_client_response(200, "成功", result) except Exception as e: logger.exception(f"客户端版本检查接口异常: {e}") return create_client_response(500, "系统异常，请稍后重试；若持续发生请联系客服", None) # ============ 公告接口 ============ @router.get("/announcements") async def get_announcements( page: int = Query(1, ge=1, description="页码"), page_size: int = Query(10, ge=1, le=100, description="每页数量"), db: Session = Depends(get_db) ): """ 获取公告列表 - AES+RSA加密 - 无需登录 """ start_time = time.time() try: announcement_service = AnnouncementService(db) success, message, data = announcement_service.get_announcements( page=page, page_size=page_size ) execution_time = int((time.time() - start_time) * 1000) return { **create_client_response(200 if success else 400, message, data), "_execution_time": execution_time } except Exception as e: logger.exception(f"获取公告接口异常: {e}") raise HTTPException( status_code=500, detail="系统异常，请稍后重试；若持续发生请联系客服" ) @router.post("/announcements") @secure_endpoint(require_login=False, require_machine_code=False) async def post_announcements(decrypted_params: dict, request: Request, db: Session, session=None, user=None): """ 获取公告列表（安全端点：请求体为加密 params，含 page、page_size、keyword） """ try: page = int(decrypted_params.get("page") or 1) page_size = int(decrypted_params.get("page_size") or 10) if page < 1: page = 1 if page_size < 1 or page_size > 100: page_size = 10 announcement_service = AnnouncementService(db) success, message, data = announcement_service.get_announcements( page=page, page_size=page_size ) if success: return create_client_response(200, message, data) return create_client_response(400, message, data) except Exception as e: logger.exception(f"POST 公告接口异常: {e}") return create_client_response(500, "系统异常，请稍后重试", None) # ============ 用户信息接口 ============ @router.post("/user/profile") @secure_endpoint(require_login=True, require_machine_code=True) async def get_profile(decrypted_params: dict, request: Request, db: Session, session=None, user=None): """ 获取用户信息（安全端点：请求体为加密 params，含 session_id、machine_code） - AES+RSA 加密请求与响应 - 需要登录验证 - 返回带签名的安全响应 统一错误提示： - 401: 会话已过期 - 403: 账号在其他设备登录 """ try: if not user: return {"code": 404, "msg": "用户不存在"} days_remaining = 0 if user.expire_time: days_remaining = (user.expire_time - beijing_now_naive()).days user_data = { "user_id": user.id, "username": user.username, "email": user.email, "expire_time": user.expire_time.isoformat() if user.expire_time else None, "days_remaining": days_remaining, "status": user.status } logger.debug(f"获取用户信息成功: user_id={user.id}") return {"code": 200, "msg": "成功", **user_data} except Exception as e: logger.exception(f"获取用户信息异常: {e}") return {"code": 500, "msg": "系统异常，请稍后重试；若持续发生请联系客服"} # ============ 密钥管理接口 ============ def _post_keys_public_encrypt_response( public_key_pem: str, key_version: int, private_key_pem: str, client_aes_key_b64: str, timestamp: int, nonce: str, ) -> Dict[str, Any]: """ 构造 POST /keys/public 的加密响应（与客户端约定一致）： data.encrypted_key = AES-CBC( response_key+response_iv, client_aes_key, iv=client_aes_key[16:32] ) data.encrypted_data = AES-CBC( json(public_key, key_version, signing_public_key), response_key, response_iv ) data.signature = RSA-PSS( timestamp:nonce:encrypted_data_b64 ) """ import base64 import json import secrets from cryptography.hazmat.primitives import serialization, hashes from cryptography.hazmat.primitives.asymmetric import padding as rsa_padding from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes from cryptography.hazmat.primitives.padding import PKCS7 from cryptography.hazmat.backends import default_backend client_aes_key_b64 = (client_aes_key_b64 or "").strip() client_aes = base64.b64decode(client_aes_key_b64) if len(client_aes) < 32: raise ValueError("aes_key 长度不足") key_c = client_aes[:32] iv_c = client_aes[16:32] response_key = secrets.token_bytes(32) response_iv = secrets.token_bytes(16) payload = json.dumps({ "public_key": public_key_pem, "key_version": key_version, "signing_public_key": public_key_pem, }).encode() padder = PKCS7(128).padder() padded = padder.update(payload) + padder.finalize() cipher_r = Cipher(algorithms.AES(response_key), modes.CBC(response_iv), backend=default_backend()) enc_r = cipher_r.encryptor() encrypted_data_bytes = enc_r.update(padded) + enc_r.finalize() encrypted_data_b64 = base64.b64encode(encrypted_data_bytes).decode() key_iv = response_key + response_iv # 48 字节 # CBC 需 PKCS7 填充，否则客户端 OpenSSL 解密时会按填充校验导致 bad decrypt padder = PKCS7(128).padder() key_iv_padded = padder.update(key_iv) + padder.finalize() cipher_c = Cipher(algorithms.AES(key_c), modes.CBC(iv_c), backend=default_backend()) enc_c = cipher_c.encryptor() encrypted_key_bytes = enc_c.update(key_iv_padded) + enc_c.finalize() encrypted_key_b64 = base64.b64encode(encrypted_key_bytes).decode() logger.info("[keys/public 加密响应] key_iv_len=48 padded_len=%s encrypted_key_len=%s", len(key_iv_padded), len(encrypted_key_bytes)) sign_string = f"{timestamp}:{nonce}:{encrypted_data_b64}" private_key = serialization.load_pem_private_key( private_key_pem.encode(), password=None, backend=default_backend() ) signature = private_key.sign( sign_string.encode(), rsa_padding.PSS(mgf=rsa_padding.MGF1(hashes.SHA256()), salt_length=rsa_padding.PSS.MAX_LENGTH), hashes.SHA256(), ) signature_b64 = base64.b64encode(signature).decode() return { "encrypted_key": encrypted_key_b64, "encrypted_data": encrypted_data_b64, "signature": signature_b64, "signing_public_key": public_key_pem, "timestamp": timestamp, "nonce": nonce, } @router.post("/keys/public") async def post_public_key(request: Request, db: Session = Depends(get_db)): """ POST 获取RSA公钥（与客户端握手约定一致） 请求体: encrypted_params (AES加密的机器特征), aes_key (base64), timestamp, nonce 响应 data: encrypted_key, encrypted_data, signature, signing_public_key, timestamp, nonce """ import base64 from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes from cryptography.hazmat.backends import default_backend try: body = await request.json() encrypted_params_b64 = (body.get("encrypted_params") or "").strip() aes_key_b64 = (body.get("aes_key") or "").strip() ts = body.get("timestamp", 0) nonce = (body.get("nonce") or "").strip() logger.info( "[POST /keys/public] encrypted_params_b64_len=%s aes_key_b64_len=%s ts=%s", len(encrypted_params_b64), len(aes_key_b64), ts, ) if not aes_key_b64 or not encrypted_params_b64: return create_client_response(400, "请求参数错误", None) aes_bytes = base64.b64decode(aes_key_b64) logger.info("[POST /keys/public] aes_bytes decoded len=%s (需>=32)", len(aes_bytes)) if len(aes_bytes) < 32: return create_client_response(400, "aes_key 无效", None) key_a = aes_bytes[:32] iv_a = aes_bytes[16:32] try: enc_bytes = base64.b64decode(encrypted_params_b64) except Exception: return create_client_response(400, "encrypted_params 无效", None) cipher = Cipher(algorithms.AES(key_a), modes.CBC(iv_a), backend=default_backend()) dec = cipher.decryptor() decrypted_padded = dec.update(enc_bytes) + dec.finalize() try: from cryptography.hazmat.primitives.padding import PKCS7 unpadder = PKCS7(128).unpadder() decrypted_bytes = unpadder.update(decrypted_padded) + unpadder.finalize() except Exception: decrypted_bytes = decrypted_padded try: raw_machine_info = decrypted_bytes.decode("utf-8", errors="replace") except Exception: raw_machine_info = "" logger.debug("POST /keys/public raw_machine_info length=%s", len(raw_machine_info)) key_service = KeyService(db) success, message, data = key_service.get_active_key_pair() if not success or not data: return create_client_response(500, message or "系统初始化中，请稍后再试", None) public_key_pem = data["public_key"] private_key_pem = data["private_key"] key_version = data["key_version"] response_data = _post_keys_public_encrypt_response( public_key_pem, key_version, private_key_pem, aes_key_b64, ts, nonce ) return create_client_response(200, "获取成功", response_data) except Exception as e: logger.exception("POST /keys/public 异常: %s", e) return create_client_response(500, "系统异常，请稍后重试", None) @router.get("/keys/public") async def get_public_key(db: Session = Depends(get_db)): """ 获取RSA公钥 - 无需登录 - 返回当前活跃版本的RSA公钥 统一错误提示： - 500: 系统初始化中 """ start_time = time.time() try: key_service = KeyService(db) success, message, data = key_service.get_active_public_key() execution_time = int((time.time() - start_time) * 1000) if success: return { **create_client_response(200, message, data), "_execution_time": execution_time } else: # 公钥不存在时，返回友好的系统初始化提示 return { **create_client_response(500, "系统初始化中，请稍后再试"), "_execution_time": execution_time } except Exception as e: logger.exception(f"获取公钥接口异常: {e}") raise HTTPException( status_code=500, detail="系统异常，请稍后重试；若持续发生请联系客服" ) @router.post("/keys/sync") @secure_endpoint(require_login=True, require_machine_code=True) async def sync_public_key(decrypted_params: dict, request: Request, db: Session, session=None, user=None): """ 同步RSA公钥（已登录状态，符合技术方案 4.3.2） - 请求体为加密 params，含 session_id、machine_code - 返回当前激活的 public_key 与 key_version（可能比客户端更新） """ try: key_service = KeyService(db) success, message, data = key_service.get_active_public_key() if not success or not data: return {"code": 500, "msg": message or "系统初始化中，请稍后再试"} return { "code": 200, "msg": "成功", "public_key": data.get("public_key"), "key_version": data.get("key_version"), } except Exception as e: logger.exception(f"同步公钥异常: {e}") return {"code": 500, "msg": "系统异常，请稍后重试"} @router.get("/keys/version") async def get_key_version(db: Session = Depends(get_db)): """ 获取当前RSA密钥版本 - 无需登录 """ start_time = time.time() try: key_service = KeyService(db) success, message, data = key_service.get_current_key_version() execution_time = int((time.time() - start_time) * 1000) if success: return { **create_client_response(200, message, data), "_execution_time": execution_time } else: return { **create_client_response(500, message), "_execution_time": execution_time } except Exception as e: logger.exception(f"获取密钥版本接口异常: {e}") raise HTTPException( status_code=500, detail="系统异常，请稍后重试；若持续发生请联系客服" ) # ============ 配置管理接口 ============ @router.get("/config/client") async def get_client_config(db: Session = Depends(get_db)): """ 获取客户端配置 - 无需登录 - 返回客户端相关的系统配置 """ start_time = time.time() try: key_service = KeyService(db) success, message, data = key_service.get_client_configs() execution_time = int((time.time() - start_time) * 1000) if success: return { **create_client_response(200, message, data), "_execution_time": execution_time } else: return { **create_client_response(500, message), "_execution_time": execution_time } except Exception as e: logger.exception(f"获取客户端配置接口异常: {e}") raise HTTPException( status_code=500, detail="系统异常，请稍后重试；若持续发生请联系客服" ) # ============ HMAC密钥管理接口 ============ class HmacKeyRequest(BaseModel): """HMAC密钥请求""" username: str = Field(..., description="用户名") machine_code: str = Field(..., description="机器码") @router.post("/keys/hmac") async def sync_hmac_key(request: HmacKeyRequest, db: Session = Depends(get_db)): """ 同步/获取动态HMAC密钥 - 无需登录（但需要验证用户名和机器码） - 用于客户端获取或刷新HMAC签名密钥 - 密钥有效期24小时，支持自动轮换 统一错误提示： - 401: 用户名不存在/密码错误 - 403: 账号被禁用/机器码不匹配 - 500: 系统异常 """ start_time = time.time() try: # 导入HMAC密钥服务 from app.services.hmac_key_service import HmacKeyService hmac_service = HmacKeyService(db) # 先验证用户 is_valid, msg, user_id = hmac_service.validate_user_for_hmac( request.username, request.machine_code ) if not is_valid: # 根据错误类型返回不同的状态码 error_mapping = { "不存在": 401, "禁用": 403, "不匹配": 403, } status_code = 400 for key, code in error_mapping.items(): if key in msg: status_code = code break return { **create_client_response(status_code, msg, None), "_execution_time": int((time.time() - start_time) * 1000) } # 创建HMAC密钥 success, message, data = hmac_service.create_hmac_key( user_id=user_id, username=request.username, machine_code=request.machine_code ) execution_time = int((time.time() - start_time) * 1000) if success: logger.info(f"HMAC密钥同步成功: {request.username}") return { **create_client_response(200, message, data), "_execution_time": execution_time } else: # 根据错误类型返回不同的状态码 error_mapping = { "不存在": 401, "禁用": 403, "不匹配": 403, "已过期": 403 } status_code = 400 for key, code in error_mapping.items(): if key in message: status_code = code break return { **create_client_response(status_code, message, None), "_execution_time": execution_time } except Exception as e: logger.exception(f"HMAC密钥同步接口异常: {e}") raise HTTPException( status_code=500, detail="系统异常，请稍后重试；若持续发生请联系客服" ) @router.get("/keys/hmac/status") async def get_hmac_key_status( username: str = Query(..., description="用户名"), machine_code: str = Query(..., description="机器码"), db: Session = Depends(get_db) ): """ 检查HMAC密钥状态 - 无需登录 - 返回当前密钥是否有效及过期时间 """ start_time = time.time() try: from app.services.hmac_key_service import HmacKeyService hmac_service = HmacKeyService(db) # 先验证用户 is_valid, msg, user_id = hmac_service.validate_user_for_hmac(username, machine_code) if not is_valid: return create_client_response(401, msg, None) # 获取当前密钥状态 success, message, data = hmac_service.get_hmac_key(user_id, machine_code) execution_time = int((time.time() - start_time) * 1000) if success: return { **create_client_response(200, message, data), "_execution_time": execution_time } else: return { **create_client_response(404, message), "_execution_time": execution_time } except Exception as e: logger.exception(f"获取HMAC密钥状态接口异常: {e}") raise HTTPException( status_code=500, detail="系统异常，请稍后重试；若持续发生请联系客服" ) @router.delete("/keys/hmac") async def revoke_hmac_key( username: str = Query(..., description="用户名"), machine_code: str = Query(..., description="机器码"), db: Session = Depends(get_db) ): """ 撤销HMAC密钥 - 需要登录 - 用于登出时清除HMAC密钥 """ start_time = time.time() try: from app.services.hmac_key_service import HmacKeyService from app.services.auth_service import AuthService # 验证用户 auth_service = AuthService(db) user_id, msg, _ = auth_service.validate_session_or_token(username) if user_id is None: return create_client_response(401, msg or "会话已过期", None) # 撤销HMAC密钥 hmac_service = HmacKeyService(db) success, message = hmac_service.revoke_hmac_key(user_id, machine_code) execution_time = int((time.time() - start_time) * 1000) return { **create_client_response(200 if success else 500, message, None), "_execution_time": execution_time } except Exception as e: logger.exception(f"撤销HMAC密钥接口异常: {e}") raise HTTPException( status_code=500, detail="系统异常，请稍后重试；若持续发生请联系客服" ) # ============ Steam账号管理接口 ============ class CreateSteamAccountRequest(BaseModel): """创建/绑定 Steam 账号请求（敏感信息存客户端本地，仅传账号标识）""" user_id: int = Field(..., description="用户ID") steam_username: str = Field(..., description="Steam用户名") steam_id: Optional[str] = Field(None, description="Steam ID（来自客户端登录验证）") class UpdateSteamAccountRequest(BaseModel): """更新Steam账号请求""" session_id: str = Field(..., description="会话ID") machine_code: str = Field(..., description="机器码（raw 或 md5）") account_id: int = Field(..., description="账号ID") is_active: Optional[bool] = Field(None, description="是否启用") class DeleteSteamAccountRequest(BaseModel): """删除Steam账号请求""" user_id: int = Field(..., description="用户ID") account_id: int = Field(..., description="账号ID") class BindSteamAccountRequest(BaseModel): """验证绑定Steam账号请求""" user_id: int = Field(..., description="用户ID") steam_username: str = Field(..., description="Steam用户名") steam_id: str = Field(..., description="Steam ID（来自客户端登录验证）") class SteamAccountLimitRequest(BaseModel): """获取Steam账号绑定限制请求""" user_id: int = Field(..., description="用户ID") @router.post("/steam/account-limit") async def get_steam_account_limit(request: Request, db: Session = Depends(get_db)): """ 获取用户Steam账号绑定限制（最大可绑定数、当前已验证绑定数） - 请求体为加密格式：encrypted + key_version（与 client-tauri make_request 一致） - 响应为明文 JSON（code, msg, data），便于 client-tauri 直接解析并展示最大可绑定数 """ try: body = await request.json() if not isinstance(body, dict) or not body.get("encrypted") or body.get("key_version") is None: return JSONResponse( status_code=400, content=create_client_response(400, "请求参数错误，请检查后重试", None) ) key_version = int(body.get("key_version", 1)) decrypted = encryptor.decrypt_from_client( (body.get("encrypted") or "").strip(), key_version, db ) user_id = decrypted.get("user_id") if user_id is None: return JSONResponse( status_code=400, content=create_client_response(400, "请求参数错误，请检查后重试", None) ) user_id = int(user_id) steam_service = SteamAccountService(db) success, message, data = steam_service.get_steam_account_limit(user_id) code = 200 if success else 400 payload = create_client_response(code, message, data if data else {}) return JSONResponse(status_code=200, content=payload) except Exception as e: logger.exception(f"获取Steam账号限制接口异常: {e}") return JSONResponse( status_code=200, content=create_client_response(500, "系统异常，请稍后重试；若持续发生请联系客服", None) ) @router.post("/steam/accounts") async def create_steam_account(request: Request, db: Session = Depends(get_db)): """ 创建/绑定Steam账号（与 client-tauri make_request 一致：请求体为 encrypted + key_version） - 解密后需包含 user_id, steam_username, steam_id（可选） """ try: body = await request.json() if not isinstance(body, dict) or not body.get("encrypted") or body.get("key_version") is None: return create_encrypted_response(None, db, 400, "请求参数错误，请检查后重试") key_version = int(body.get("key_version", 1)) decrypted = encryptor.decrypt_from_client( (body.get("encrypted") or "").strip(), key_version, db ) user_id = decrypted.get("user_id") steam_username = (decrypted.get("steam_username") or "").strip() if user_id is None or not steam_username: return create_encrypted_response(None, db, 400, "请求参数错误，请检查后重试") user_id = int(user_id) steam_id = decrypted.get("steam_id") if steam_id is not None and isinstance(steam_id, str) and not steam_id.strip(): steam_id = None steam_service = SteamAccountService(db) success, message, data = steam_service.create_steam_account( user_id=user_id, steam_username=steam_username, steam_id=steam_id, ) if success: logger.info(f"Steam账号绑定成功: user_id={user_id}, steam_username={steam_username}") return create_encrypted_response(data, db, 200, message) error_code = 400 if "已绑定" in message: error_code = 403 if "最大绑定数量" in message: error_code = 403 return create_encrypted_response(None, db, error_code, message) except Exception as e: logger.exception(f"创建Steam账号接口异常: {e}") return create_encrypted_response(None, db, 500, "系统异常，请稍后重试；若持续发生请联系客服") @router.get("/steam/accounts") async def get_steam_accounts( user_id: int = Query(..., description="用户ID"), db: Session = Depends(get_db) ): """ 获取用户绑定的Steam账号列表（明文 Query，供管理端或调试） - 需登录验证 """ start_time = time.time() try: steam_service = SteamAccountService(db) success, message, data = steam_service.get_accounts_by_user(user_id) logger.debug(f"获取Steam账号列表: user_id={user_id}, count={len(data) if data else 0}") return create_encrypted_response(data if data else [], db, 200 if success else 400, message) except Exception as e: logger.exception(f"获取Steam账号列表接口异常: {e}") return create_encrypted_response(None, db, 500, "系统异常，请稍后重试；若持续发生请联系客服") @router.post("/steam/accounts/list") async def post_steam_accounts_list(request: Request, db: Session = Depends(get_db)): """ 获取用户绑定的Steam账号列表（加密请求体；响应为明文 code/msg/data，便于 client-tauri 直接展示列表） """ try: body = await request.json() if not isinstance(body, dict) or not body.get("encrypted") or body.get("key_version") is None: return JSONResponse( status_code=200, content=create_client_response(400, "请求参数错误，请检查后重试", None) ) key_version = int(body.get("key_version", 1)) decrypted = encryptor.decrypt_from_client( (body.get("encrypted") or "").strip(), key_version, db ) user_id = decrypted.get("user_id") if user_id is None: return JSONResponse( status_code=200, content=create_client_response(400, "请求参数错误，请检查后重试", None) ) user_id = int(user_id) steam_service = SteamAccountService(db) success, message, data = steam_service.get_accounts_by_user(user_id) logger.debug(f"获取Steam账号列表(list): user_id={user_id}, count={len(data) if data else 0}") code = 200 if success else 400 payload = create_client_response(code, message, data if data else []) return JSONResponse(status_code=200, content=payload) except Exception as e: logger.exception(f"获取Steam账号列表(list)接口异常: {e}") return JSONResponse( status_code=200, content=create_client_response(500, "系统异常，请稍后重试；若持续发生请联系客服", None) ) @router.post("/steam/accounts/bind") async def bind_steam_account(request: Request, db: Session = Depends(get_db)): """ 对已新增的Steam账号进行验证绑定（写入steam_id） - 请求体：encrypted + key_version，解密后需包含 user_id, steam_username, steam_id - 需先新增账号再调用本接口，已验证绑定数不得超过配置上限 """ try: body = await request.json() if not isinstance(body, dict) or not body.get("encrypted") or body.get("key_version") is None: return JSONResponse( status_code=200, content=create_client_response(400, "请求参数错误，请检查后重试", None) ) key_version = int(body.get("key_version", 1)) decrypted = encryptor.decrypt_from_client( (body.get("encrypted") or "").strip(), key_version, db ) user_id = decrypted.get("user_id") steam_username = (decrypted.get("steam_username") or "").strip() steam_id = (decrypted.get("steam_id") or "").strip() if user_id is None or not steam_username: return JSONResponse( status_code=200, content=create_client_response(400, "请求参数错误，请检查后重试", None) ) user_id = int(user_id) if not steam_id: return JSONResponse( status_code=200, content=create_client_response(400, "steam_id 不能为空", None) ) steam_service = SteamAccountService(db) success, message, data = steam_service.bind_steam_account( user_id=user_id, steam_username=steam_username, steam_id=steam_id, ) if success: logger.info(f"Steam账号验证绑定成功: user_id={user_id}, steam_username={steam_username}, steam_id={steam_id}") return JSONResponse(status_code=200, content=create_client_response(200, message, data)) error_code = 400 if "已绑定" in message or "最大绑定数量" in message: error_code = 403 if "请先新增" in message: error_code = 400 return JSONResponse(status_code=200, content=create_client_response(error_code, message, None)) except Exception as e: logger.exception(f"Steam账号验证绑定接口异常: {e}") return JSONResponse( status_code=200, content=create_client_response(500, "系统异常，请稍后重试；若持续发生请联系客服", None) ) @router.put("/steam/accounts") async def update_steam_account( request: UpdateSteamAccountRequest, db: Session = Depends(get_db) ): """ 更新Steam账号状态 - AES+RSA加密 - 需要登录验证 """ try: # 统一校验：会话有效 + 机器码一致 + 账号可用 + 订阅未过期 auth_service = AuthService(db) session, err_code, err_msg = auth_service.verify_session_with_machine_and_expiry( request.session_id, request.machine_code ) if not session: return create_encrypted_response(None, db, err_code or 401, err_msg or "会话已过期，请重新登录") steam_service = SteamAccountService(db) success, message = steam_service.update_account( account_id=request.account_id, user_id=session.user_id, is_active=request.is_active ) if success: return create_encrypted_response(None, db, 200, message) error_code = 400 if "不存在" in message: error_code = 404 return create_encrypted_response(None, db, error_code, message) except Exception as e: logger.exception(f"更新Steam账号接口异常: {e}") return create_encrypted_response(None, db, 500, "系统异常，请稍后重试；若持续发生请联系客服") @router.delete("/steam/accounts") async def delete_steam_account(request: Request, db: Session = Depends(get_db)): """ 删除Steam账号（仅未绑定可删） - 请求体：encrypted + key_version，解密后需包含 user_id、account_id """ try: body = await request.json() if not isinstance(body, dict) or not body.get("encrypted") or body.get("key_version") is None: return JSONResponse( status_code=200, content=create_client_response(400, "请求参数错误，请检查后重试", None) ) key_version = int(body.get("key_version", 1)) decrypted = encryptor.decrypt_from_client( (body.get("encrypted") or "").strip(), key_version, db ) user_id = decrypted.get("user_id") account_id = decrypted.get("account_id") if user_id is None or account_id is None: return JSONResponse( status_code=200, content=create_client_response(400, "请求参数错误，请检查后重试", None) ) user_id = int(user_id) account_id = int(account_id) steam_service = SteamAccountService(db) success, message = steam_service.delete_account( account_id=account_id, user_id=user_id ) if success: logger.info(f"Steam账号删除成功: account_id={account_id}") return JSONResponse(status_code=200, content=create_client_response(200, message, None)) error_code = 400 if "不存在" in message: error_code = 404 if "已绑定" in message: error_code = 403 return JSONResponse(status_code=200, content=create_client_response(error_code, message, None)) except Exception as e: logger.exception(f"删除Steam账号接口异常: {e}") return JSONResponse( status_code=200, content=create_client_response(500, "系统异常，请稍后重试；若持续发生请联系客服", None) ) # ============ BUFF 选品筛选项（从 buff_filter_option 表加载） ============ def _build_filter_options_data(db: Session, game: str) -> Dict[str, Any]: """构建选品筛选项数据（供 GET 明文与 POST 安全端点共用）。""" query = db.query(BuffFilterOption).filter( BuffFilterOption.game == game, BuffFilterOption.is_enabled == True ).order_by(BuffFilterOption.filter_type, BuffFilterOption.parent_value, BuffFilterOption.sort_order, BuffFilterOption.id) rows = query.all() rifle, pistol, smg, shotgun, machinegun = [], [], [], [], [] quality, category, exterior, collection = [], [], [], [] itemset_roots, itemset_children = [], [] for r in rows: opt = {"label": r.label, "value": r.value} if r.filter_type == "category" and r.parent_value == "rifle": rifle.append(opt) elif r.filter_type == "category" and r.parent_value == "pistol": pistol.append(opt) elif r.filter_type == "category" and r.parent_value == "smg": smg.append(opt) elif r.filter_type == "category" and r.parent_value == "shotgun": shotgun.append(opt) elif r.filter_type == "category" and r.parent_value == "machinegun": machinegun.append(opt) elif r.filter_type == "rarity": quality.append(opt) elif r.filter_type == "quality": category.append(opt) elif r.filter_type == "exterior": exterior.append(opt) elif r.filter_type == "itemset": if r.parent_value is None: itemset_roots.append(r) else: itemset_children.append(r) collection.append(opt) collection_tree = [] for root in sorted(itemset_roots, key=lambda x: (x.sort_order, x.id)): children = [c for c in itemset_children if c.parent_value == root.value] children.sort(key=lambda x: (x.sort_order, x.id)) collection_tree.append({ "title": root.label, "value": root.value, "key": root.value, "children": [{"title": c.label, "value": c.value, "key": c.value} for c in children], }) return { "rifle": rifle, "pistol": pistol, "smg": smg, "shotgun": shotgun, "machinegun": machinegun, "quality": quality, "category": category, "exterior": exterior, "collection": collection, "collectionTree": collection_tree, } @router.post("/buff/filter-options") @secure_endpoint(require_login=True, require_machine_code=True) async def post_buff_filter_options_secure( decrypted_params: dict, request: Request, db: Session, session=None, user=None ): """ 获取 BUFF 选品筛选项（安全端点：POST，入参出参密文，时间戳与签名校验）。 解密参数：game（可选，默认 csgo）。会话与机器码由装饰器校验。 日志：入参密文/明文、出参密文/明文由 SecureEndpoint 与 APILoggingASGIMiddleware 统一记录。 """ try: game = (decrypted_params.get("game") or "csgo").strip() or "csgo" data = _build_filter_options_data(db, game) return {"code": 200, "msg": "获取成功", "data": data} except Exception as e: logger.exception(f"POST /buff/filter-options 异常: {e}") return {"code": 500, "msg": "系统异常，请稍后重试", "data": None} @router.get("/buff/filter-options") async def get_buff_filter_options( game: str = Query("csgo", description="游戏标识"), session_id: str = Query(..., description="会话ID"), machine_code: str = Query(..., description="机器码"), db: Session = Depends(get_db) ): """ 获取 BUFF 选品检索栏下拉选项。需登录，校验会话+机器码+用户过期。 """ try: if not (session_id or "").strip() or not (machine_code or "").strip(): return JSONResponse(status_code=200, content=create_client_response(401, "请先登录", None)) auth_service = AuthService(db) session, err_code, err_msg = auth_service.verify_session_with_machine_and_expiry( session_id.strip(), (machine_code or "").strip() ) if session is None: return JSONResponse(status_code=200, content=create_client_response(err_code or 401, err_msg or "会话已过期，请重新登录", None)) data = _build_filter_options_data(db, game) return JSONResponse(status_code=200, content=create_client_response(200, "获取成功", data)) except Exception as e: logger.exception(f"获取BUFF筛选项异常: {e}") return JSONResponse( status_code=200, content=create_client_response(500, "系统异常，请稍后重试", None) ) @router.get("/buff/goods-attr-list") async def get_buff_goods_attr_list( goods_ids: str = Query(..., description="逗号分隔的 BUFF 饰品 goods_id 列表"), session_id: str = Query(..., description="会话ID"), machine_code: str = Query(..., description="机器码"), db: Session = Depends(get_db) ): """ 根据 goods_id 列表查询 buff_goods_attr 表。需登录，校验会话+机器码+用户过期。 """ try: if not (session_id or "").strip() or not (machine_code or "").strip(): return JSONResponse(status_code=200, content=create_client_response(401, "请先登录", None)) auth_service = AuthService(db) session, err_code, err_msg = auth_service.verify_session_with_machine_and_expiry( session_id.strip(), (machine_code or "").strip() ) if session is None: return JSONResponse(status_code=200, content=create_client_response(err_code or 401, err_msg or "会话已过期，请重新登录", None)) id_list = [x.strip() for x in (goods_ids or "").split(",") if (x and x.strip())] if not id_list: return JSONResponse(status_code=200, content=create_client_response(200, "获取成功", [])) # 限制单次查询数量，避免过长 id_list = id_list[:500] rows = db.query(BuffGoodsAttr).filter( BuffGoodsAttr.goods_id.in_(id_list) ).all() data = [ {"goods_id": r.goods_id, "name": r.name or "", "icon_url": r.icon_url or ""} for r in rows ] return JSONResponse(status_code=200, content=create_client_response(200, "获取成功", data)) except Exception as e: logger.exception(f"获取BUFF商品属性列表异常: {e}") return JSONResponse( status_code=200, content=create_client_response(500, "系统异常，请稍后重试", None) ) @router.post("/buff/goods-attr-list") @secure_endpoint(require_login=True, require_machine_code=True) async def post_buff_goods_attr_list_secure( decrypted_params: dict, request: Request, db: Session, session=None, user=None ): """ 根据 goods_id 列表获取商品属性（安全端点：POST，入参出参密文，时间戳与签名校验）。 解密参数：goods_ids（逗号分隔）。日志由 SecureEndpoint 与 APILoggingASGIMiddleware 统一记录。 """ try: goods_ids_raw = (decrypted_params.get("goods_ids") or "").strip() id_list = [x.strip() for x in goods_ids_raw.split(",") if (x and x.strip())] if not id_list: return {"code": 200, "msg": "获取成功", "data": []} id_list = id_list[:500] rows = db.query(BuffGoodsAttr).filter(BuffGoodsAttr.goods_id.in_(id_list)).all() data = [{"goods_id": r.goods_id, "name": r.name or "", "icon_url": r.icon_url or ""} for r in rows] return {"code": 200, "msg": "获取成功", "data": data} except Exception as e: logger.exception(f"POST /buff/goods-attr-list 异常: {e}") return {"code": 500, "msg": "系统异常，请稍后重试", "data": []} # ============ BUFF 选品配置（保存/加载用户检索条件+选品条件） ============ @router.post("/buff/selection-config") async def save_buff_selection_config(request: Request, db: Session = Depends(get_db)): """ 保存 BUFF 选品配置（检索条件+选品条件）。 请求体 JSON：session_id（必填）, machine_code（必填）, name（可选）, config_json（必填）, is_default（可选）。 校验会话+机器码+用户过期后按 user_id + name 写入或更新，同用户至多一个 is_default=1。 """ try: body = await request.json() if not isinstance(body, dict): return JSONResponse(status_code=200, content=create_client_response(400, "请求参数错误，请检查后重试", None)) session_id = (body.get("session_id") or "").strip() machine_code = (body.get("machine_code") or "").strip() if not session_id: return JSONResponse(status_code=200, content=create_client_response(401, "请先登录", None)) if not machine_code: return JSONResponse(status_code=200, content=create_client_response(400, "请求参数错误：缺少 machine_code", None)) auth_service = AuthService(db) session, err_code, err_msg = auth_service.verify_session_with_machine_and_expiry(session_id, machine_code) if session is None: return JSONResponse(status_code=200, content=create_client_response(err_code or 401, err_msg or "会话已过期，请重新登录", None)) user_id = session.user_id name = (body.get("name") or "默认").strip() or "默认" config_json = body.get("config_json") if config_json is None: return JSONResponse(status_code=200, content=create_client_response(400, "缺少配置内容 config_json", None)) if not isinstance(config_json, dict): try: config_json = json.loads(config_json) if isinstance(config_json, str) else None except Exception: config_json = None if not isinstance(config_json, dict): return JSONResponse(status_code=200, content=create_client_response(400, "config_json 格式错误", None)) is_default = body.get("is_default", True) if not isinstance(is_default, bool): is_default = is_default in (1, "1", True, "true") existing = db.query(BuffSelectionConfig).filter( BuffSelectionConfig.user_id == user_id, BuffSelectionConfig.name == name, BuffSelectionConfig.is_deleted == False ).first() config_str = json.dumps(config_json, ensure_ascii=False) if existing: existing.config_json = config_str existing.is_default = bool(is_default) existing.updated_at = beijing_now_naive() db.commit() row_id = existing.id else: row = BuffSelectionConfig( user_id=user_id, name=name, config_json=config_str, is_default=bool(is_default) ) db.add(row) db.commit() db.refresh(row) row_id = row.id if is_default: db.query(BuffSelectionConfig).filter( BuffSelectionConfig.user_id == user_id, BuffSelectionConfig.id != row_id, BuffSelectionConfig.is_deleted == False ).update({BuffSelectionConfig.is_default: False}, synchronize_session=False) db.commit() return JSONResponse(status_code=200, content=create_client_response(200, "配置信息已保存", {"id": row_id})) except Exception as e: logger.exception(f"保存BUFF选品配置异常: {e}") db.rollback() return JSONResponse(status_code=200, content=create_client_response(500, "系统异常，请稍后重试", None)) @router.get("/buff/selection-config") async def get_buff_selection_config( session_id: str = Query(..., description="会话ID"), machine_code: str = Query(..., description="机器码"), db: Session = Depends(get_db) ): """ 获取当前用户的 BUFF 选品配置列表及默认配置。 校验会话+机器码+用户过期后返回 data.list 与 data.defaultConfig。 """ try: if not (session_id or "").strip(): return JSONResponse(status_code=200, content=create_client_response(401, "请先登录", None)) if not (machine_code or "").strip(): return JSONResponse(status_code=200, content=create_client_response(400, "请求参数错误：缺少 machine_code", None)) auth_service = AuthService(db) session, err_code, err_msg = auth_service.verify_session_with_machine_and_expiry( session_id.strip(), (machine_code or "").strip() ) if session is None: return JSONResponse(status_code=200, content=create_client_response(err_code or 401, err_msg or "会话已过期，请重新登录", None)) user_id = session.user_id rows = db.query(BuffSelectionConfig).filter( BuffSelectionConfig.user_id == user_id, BuffSelectionConfig.is_deleted == False ).order_by(BuffSelectionConfig.is_default.desc(), BuffSelectionConfig.updated_at.desc()).all() list_data = [{"id": r.id, "name": r.name, "is_default": r.is_default, "updated_at": r.updated_at.isoformat() if r.updated_at else None} for r in rows] default_config = None for r in rows: if r.is_default and r.config_json: try: default_config = json.loads(r.config_json) except Exception: pass break return JSONResponse(status_code=200, content=create_client_response(200, "获取成功", { "list": list_data, "defaultConfig": default_config })) except Exception as e: logger.exception(f"获取BUFF选品配置异常: {e}") return JSONResponse(status_code=200, content=create_client_response(500, "系统异常，请稍后重试", None)) @router.post("/buff/secure/selection-config") @secure_endpoint(require_login=True, require_machine_code=True) async def post_buff_selection_config_secure( decrypted_params: dict, request: Request, db: Session, session=None, user=None ): """ 选品配置获取或保存（安全端点：POST，入参出参密文，时间戳与签名校验）。 解密参数：action（get|save）；save 时另需 name、config_json、is_default。日志统一记录入参/出参密文与明文。 """ try: user_id = session.user_id action = (decrypted_params.get("action") or "get").strip().lower() if action == "get": rows = db.query(BuffSelectionConfig).filter( BuffSelectionConfig.user_id == user_id, BuffSelectionConfig.is_deleted == False ).order_by(BuffSelectionConfig.is_default.desc(), BuffSelectionConfig.updated_at.desc()).all() list_data = [{"id": r.id, "name": r.name, "is_default": r.is_default, "updated_at": r.updated_at.isoformat() if r.updated_at else None} for r in rows] default_config = None for r in rows: if r.is_default and r.config_json: try: default_config = json.loads(r.config_json) except Exception: pass break return {"code": 200, "msg": "获取成功", "data": {"list": list_data, "defaultConfig": default_config}} elif action == "save": name = (decrypted_params.get("name") or "默认").strip() or "默认" config_json = decrypted_params.get("config_json") if config_json is None: return {"code": 400, "msg": "缺少配置内容 config_json", "data": None} if not isinstance(config_json, dict): try: config_json = json.loads(config_json) if isinstance(config_json, str) else None except Exception: config_json = None if not isinstance(config_json, dict): return {"code": 400, "msg": "config_json 格式错误", "data": None} is_default = decrypted_params.get("is_default", True) if not isinstance(is_default, bool): is_default = is_default in (1, "1", True, "true") config_str = json.dumps(config_json, ensure_ascii=False) existing = db.query(BuffSelectionConfig).filter( BuffSelectionConfig.user_id == user_id, BuffSelectionConfig.name == name, BuffSelectionConfig.is_deleted == False ).first() if existing: existing.config_json = config_str existing.is_default = bool(is_default) existing.updated_at = beijing_now_naive() db.commit() row_id = existing.id else: row = BuffSelectionConfig( user_id=user_id, name=name, config_json=config_str, is_default=bool(is_default) ) db.add(row) db.commit() db.refresh(row) row_id = row.id if is_default: db.query(BuffSelectionConfig).filter( BuffSelectionConfig.user_id == user_id, BuffSelectionConfig.id != row_id, BuffSelectionConfig.is_deleted == False ).update({BuffSelectionConfig.is_default: False}, synchronize_session=False) db.commit() return {"code": 200, "msg": "配置信息已保存", "data": {"id": row_id}} else: return {"code": 400, "msg": "action 必须为 get 或 save", "data": None} except Exception as e: logger.exception(f"POST /buff/secure/selection-config 异常: {e}") try: db.rollback() except Exception: pass return {"code": 500, "msg": "系统异常，请稍后重试", "data": None} def _get_system_config_int(db: Session, key: str, default: int = 0) -> int: """读取 system_config 整型配置（backend-selection）。""" row = db.query(SystemConfig).filter( SystemConfig.config_key == key, SystemConfig.app_name == "backend-selection", SystemConfig.is_deleted == False, ).first() if not row or not row.config_value: return default try: return int(row.config_value) except (ValueError, TypeError): return default @router.post("/buff/selection-start") async def start_buff_selection( request: Request, background_tasks: BackgroundTasks, db: Session = Depends(get_db), ): """ 开始选品（异步）。 请求体：session_id（必填）, machine_code（必填）, config_id（可选）。 校验会话+机器码+用户过期与每日次数/条数限制后，创建选品请求单并立即返回；后台异步执行选品逻辑。 """ try: body = await request.json() if not isinstance(body, dict): return JSONResponse(status_code=200, content=create_client_response(400, "请求参数错误，请检查后重试", None)) session_id = (body.get("session_id") or "").strip() machine_code = (body.get("machine_code") or "").strip() if not session_id: return JSONResponse(status_code=200, content=create_client_response(401, "请先登录", None)) if not machine_code: return JSONResponse(status_code=200, content=create_client_response(400, "请求参数错误：缺少 machine_code", None)) auth_service = AuthService(db) session, err_code, err_msg = auth_service.verify_session_with_machine_and_expiry(session_id, machine_code) if session is None: return JSONResponse(status_code=200, content=create_client_response(err_code or 401, err_msg or "会话已过期，请重新登录", None)) user_id = session.user_id # 今日 0 点（北京时间） today_start = beijing_now_naive().replace(hour=0, minute=0, second=0, microsecond=0) # 每日选品次数限制 times_limit = _get_system_config_int(db, "buff_selection_daily_times_limit_per_user", 3) today_requests = db.query(BuffSelectionRequest).filter( BuffSelectionRequest.user_id == user_id, BuffSelectionRequest.created_at >= today_start, ).count() if today_requests >= times_limit: return JSONResponse( status_code=200, content=create_client_response(403, f"今日选品次数已达上限（{times_limit} 次），请明日再试", None), ) # 每日选品条数限制（今日已入选的饰品条数） items_limit = _get_system_config_int(db, "buff_selection_daily_items_limit_per_user", 900) today_items = db.query(BuffSelectionRequestItem).filter( BuffSelectionRequestItem.user_id == user_id, BuffSelectionRequestItem.created_at >= today_start, ).count() if today_items >= items_limit: return JSONResponse( status_code=200, content=create_client_response(403, f"今日选品条数已达上限（{items_limit} 条），请明日再试", None), ) config_id = body.get("config_id") if config_id is not None: try: config_id = int(config_id) except (TypeError, ValueError): config_id = None if config_id is None: default_cfg = db.query(BuffSelectionConfig).filter( BuffSelectionConfig.user_id == user_id, BuffSelectionConfig.is_default == True, BuffSelectionConfig.is_deleted == False, ).first() if default_cfg: config_id = default_cfg.id req = BuffSelectionRequest( user_id=user_id, config_id=config_id, status="pending", ) db.add(req) db.commit() db.refresh(req) request_id = req.id background_tasks.add_task(run_selection_task, request_id, user_id, config_id) return JSONResponse( status_code=200, content=create_client_response(200, "选品已开始，正在后台执行", {"request_id": request_id}), ) except Exception as e: logger.exception("开始选品异常: %s", e) try: db.rollback() except Exception: pass return JSONResponse(status_code=200, content=create_client_response(500, "系统异常，请稍后重试", None)) @router.post("/buff/selection-log") async def post_buff_selection_log( request: Request, db: Session = Depends(get_db), ): """ 获取最新一次选品请求的选品日志（非安全端点）。 请求体：session_id（必填）, machine_code（必填）。 返回结构与安全端点保持一致，便于联调。 """ try: body = await request.json() if not isinstance(body, dict): return JSONResponse(status_code=200, content=create_client_response(400, "请求参数错误，请检查后重试", None)) session_id = (body.get("session_id") or "").strip() machine_code = (body.get("machine_code") or "").strip() if not session_id: return JSONResponse(status_code=200, content=create_client_response(401, "请先登录", None)) if not machine_code: return JSONResponse(status_code=200, content=create_client_response(400, "请求参数错误：缺少 machine_code", None)) auth_service = AuthService(db) session, err_code, err_msg = auth_service.verify_session_with_machine_and_expiry(session_id, machine_code) if session is None: return JSONResponse(status_code=200, content=create_client_response(err_code or 401, err_msg or "会话已过期，请重新登录", None)) user_id = session.user_id req = ( db.query(BuffSelectionRequest) .filter(BuffSelectionRequest.user_id == user_id) .order_by(BuffSelectionRequest.id.desc()) .first() ) if not req: return JSONResponse( status_code=200, content=create_client_response( 200, "ok", { "request_id": None, "status": None, "started_at": None, "ended_at": None, "logs": [], }, ), ) started_at = req.created_at.strftime("%Y-%m-%d %H:%M:%S") if req.created_at else None ended_at = None if req.status in ("completed", "failed", "cancelled") and req.updated_at: ended_at = req.updated_at.strftime("%Y-%m-%d %H:%M:%S") logs = ( db.query(BuffSelectionRequestLog) .filter( BuffSelectionRequestLog.request_id == req.id, BuffSelectionRequestLog.user_id == user_id, ) .order_by(BuffSelectionRequestLog.sequence_no.asc(), BuffSelectionRequestLog.created_at.asc()) .all() ) item_map = {} if logs: selected_goods = [l.goods_id for l in logs if l.result == "selected"] if selected_goods: items = ( db.query(BuffSelectionRequestItem) .filter( BuffSelectionRequestItem.request_id == req.id, BuffSelectionRequestItem.user_id == user_id, BuffSelectionRequestItem.goods_id.in_(selected_goods), ) .all() ) for it in items: item_map[it.goods_id] = { "wear_range": it.wear_range, "price_cny": float(it.price_cny) if it.price_cny is not None else None, "price_usd": float(it.price_usd) if it.price_usd is not None else None, "steam_link": it.steam_link or "", } log_list = [] for l in logs: row = { "created_at": l.created_at.strftime("%Y-%m-%d %H:%M:%S") if l.created_at else None, "goods_id": l.goods_id, "goods_name": l.goods_name or "", "result": l.result, "message": l.message or "", } if l.result == "selected": log_wear_range = getattr(l, "wear_range", None) log_price_cny = float(l.price_cny) if getattr(l, "price_cny", None) is not None else None log_price_usd = float(l.price_usd) if getattr(l, "price_usd", None) is not None else None if log_wear_range is not None or log_price_cny is not None or log_price_usd is not None: row["wear_range"] = log_wear_range row["price_cny"] = log_price_cny row["price_usd"] = log_price_usd row["steam_link"] = "" elif l.goods_id in item_map: row["wear_range"] = item_map[l.goods_id]["wear_range"] row["price_cny"] = item_map[l.goods_id]["price_cny"] row["price_usd"] = item_map[l.goods_id]["price_usd"] row["steam_link"] = item_map[l.goods_id]["steam_link"] else: row["wear_range"] = None row["price_cny"] = None row["price_usd"] = None row["steam_link"] = "" else: row["wear_range"] = None row["price_cny"] = None row["price_usd"] = None row["steam_link"] = "" log_list.append(row) return JSONResponse( status_code=200, content=create_client_response( 200, "ok", { "request_id": req.id, "status": req.status, "started_at": started_at, "ended_at": ended_at, "logs": log_list, }, ), ) except Exception as e: logger.exception("POST /buff/selection-log 异常: %s", e) try: db.rollback() except Exception: pass return JSONResponse(status_code=200, content=create_client_response(500, "系统异常，请稍后重试", None)) @router.post("/buff/selection-result") async def post_buff_selection_result( request: Request, db: Session = Depends(get_db), ): """ 获取最新一次选品请求的状态及选中饰品列表（非安全端点）。 请求体：session_id（必填）, machine_code（必填）。 返回结构与安全端点保持一致（含 steam_link）。 """ try: body = await request.json() if not isinstance(body, dict): return JSONResponse(status_code=200, content=create_client_response(400, "请求参数错误，请检查后重试", None)) session_id = (body.get("session_id") or "").strip() machine_code = (body.get("machine_code") or "").strip() if not session_id: return JSONResponse(status_code=200, content=create_client_response(401, "请先登录", None)) if not machine_code: return JSONResponse(status_code=200, content=create_client_response(400, "请求参数错误：缺少 machine_code", None)) auth_service = AuthService(db) session, err_code, err_msg = auth_service.verify_session_with_machine_and_expiry(session_id, machine_code) if session is None: return JSONResponse(status_code=200, content=create_client_response(err_code or 401, err_msg or "会话已过期，请重新登录", None)) user_id = session.user_id req = ( db.query(BuffSelectionRequest) .filter(BuffSelectionRequest.user_id == user_id) .order_by(BuffSelectionRequest.id.desc()) .first() ) if not req: return JSONResponse( status_code=200, content=create_client_response( 200, "ok", { "request_id": None, "status": None, "started_at": None, "ended_at": None, "items": [], }, ), ) started_at = req.created_at.strftime("%Y-%m-%d %H:%M:%S") if req.created_at else None ended_at = None if req.status in ("completed", "failed", "cancelled") and req.updated_at: ended_at = req.updated_at.strftime("%Y-%m-%d %H:%M:%S") item_rows = ( db.query(BuffSelectionRequestItem) .filter( BuffSelectionRequestItem.request_id == req.id, BuffSelectionRequestItem.user_id == user_id, ) .order_by(BuffSelectionRequestItem.id.asc()) .all() ) items = [] for it in item_rows: items.append({ "id": it.id, "goods_id": it.goods_id, "goods_name": it.goods_name or "", "icon_url": it.icon_url or "", "steam_link": it.steam_link or "", "wear_range": it.wear_range or "", "price_cny": float(it.price_cny) if it.price_cny is not None else None, "price_usd": float(it.price_usd) if it.price_usd is not None else None, "created_at": it.created_at.strftime("%Y-%m-%d %H:%M:%S") if it.created_at else None, }) return JSONResponse( status_code=200, content=create_client_response( 200, "ok", { "request_id": req.id, "status": req.status, "started_at": started_at, "ended_at": ended_at, "items": items, }, ), ) except Exception as e: logger.exception("POST /buff/selection-result 异常: %s", e) try: db.rollback() except Exception: pass return JSONResponse(status_code=200, content=create_client_response(500, "系统异常，请稍后重试", None)) @router.post("/buff/secure/selection-start") @secure_endpoint(require_login=True, require_machine_code=True) async def post_buff_selection_start_secure( decrypted_params: dict, request: Request, db: Session, session=None, user=None ): """ 开始选品（安全端点：POST，入参出参密文，时间戳与签名校验）。 解密参数：config_id（可选）。日志由 SecureEndpoint 与 APILoggingASGIMiddleware 统一记录。 """ import asyncio try: user_id = session.user_id unfinished_reqs = ( db.query(BuffSelectionRequest) .filter( BuffSelectionRequest.user_id == user_id, BuffSelectionRequest.status.in_(("pending", "running")), ) .order_by(BuffSelectionRequest.id.asc()) .all() ) cleaned_request_ids = [] if unfinished_reqs: for old_req in unfinished_reqs: old_req.cancel_requested = True old_req.status = "cancelled" cleaned_request_ids.append(old_req.id) db.flush() today_start = beijing_now_naive().replace(hour=0, minute=0, second=0, microsecond=0) times_limit = _get_system_config_int(db, "buff_selection_daily_times_limit_per_user", 3) today_requests = db.query(BuffSelectionRequest).filter( BuffSelectionRequest.user_id == user_id, BuffSelectionRequest.created_at >= today_start, ).count() if today_requests >= times_limit: return {"code": 403, "msg": f"今日选品次数已达上限（{times_limit} 次），请明日再试", "data": None} items_limit = _get_system_config_int(db, "buff_selection_daily_items_limit_per_user", 900) today_items = db.query(BuffSelectionRequestItem).filter( BuffSelectionRequestItem.user_id == user_id, BuffSelectionRequestItem.created_at >= today_start, ).count() if today_items >= items_limit: return {"code": 403, "msg": f"今日选品条数已达上限（{items_limit} 条），请明日再试", "data": None} config_id = decrypted_params.get("config_id") if config_id is not None: try: config_id = int(config_id) except (TypeError, ValueError): config_id = None if config_id is None: default_cfg = db.query(BuffSelectionConfig).filter( BuffSelectionConfig.user_id == user_id, BuffSelectionConfig.is_default == True, BuffSelectionConfig.is_deleted == False, ).first() if default_cfg: config_id = default_cfg.id req = BuffSelectionRequest(user_id=user_id, config_id=config_id, status="pending") db.add(req) db.commit() db.refresh(req) request_id = req.id loop = asyncio.get_event_loop() loop.run_in_executor(None, lambda: run_selection_task(request_id, user_id, config_id)) msg = "选品已开始，正在后台执行" if cleaned_request_ids: msg = f"已停止{len(cleaned_request_ids)}个历史未结束选品任务，{msg}" return { "code": 200, "msg": msg, "data": { "request_id": request_id, "cleaned_request_ids": cleaned_request_ids, }, } except Exception as e: logger.exception("POST /buff/secure/selection-start 异常: %s", e) try: db.rollback() except Exception: pass return {"code": 500, "msg": "系统异常，请稍后重试", "data": None} @router.post("/buff/secure/selection-log") @secure_endpoint(require_login=True, require_machine_code=True) async def post_buff_selection_log_secure( decrypted_params: dict, request: Request, db: Session, session=None, user=None ): """ 获取最新一次选品请求的选品日志（安全端点）。 返回：request_id, status, started_at, ended_at, logs（含每条是否选中、选中时的区间/价格，未选中时的原因）。 若请求单未结束（pending/running），前端可定时轮询本接口直至结束。 """ try: user_id = session.user_id req = ( db.query(BuffSelectionRequest) .filter(BuffSelectionRequest.user_id == user_id) .order_by(BuffSelectionRequest.id.desc()) .first() ) if not req: return { "code": 200, "msg": "ok", "data": { "request_id": None, "status": None, "started_at": None, "ended_at": None, "max_sequence_no": 0, "logs": [], }, } started_at = req.created_at.strftime("%Y-%m-%d %H:%M:%S") if req.created_at else None ended_at = None if req.status in ("completed", "failed", "cancelled") and req.updated_at: ended_at = req.updated_at.strftime("%Y-%m-%d %H:%M:%S") last_sequence_no_raw = decrypted_params.get("last_sequence_no") limit_raw = decrypted_params.get("limit") try: last_sequence_no = int(last_sequence_no_raw) if last_sequence_no_raw is not None else None except (TypeError, ValueError): last_sequence_no = None try: limit = int(limit_raw) if limit_raw is not None else 200 except (TypeError, ValueError): limit = 200 if limit <= 0: limit = 200 if limit > 500: limit = 500 logs_q = ( db.query(BuffSelectionRequestLog) .filter( BuffSelectionRequestLog.request_id == req.id, BuffSelectionRequestLog.user_id == user_id, ) ) if last_sequence_no is not None and last_sequence_no > 0: logs_q = logs_q.filter(BuffSelectionRequestLog.sequence_no > last_sequence_no) logs = ( logs_q .order_by(BuffSelectionRequestLog.sequence_no.desc(), BuffSelectionRequestLog.created_at.desc()) .limit(limit) .all() ) item_map = {} if logs: selected_goods = [l.goods_id for l in logs if l.result == "selected"] if selected_goods: items = ( db.query(BuffSelectionRequestItem) .filter( BuffSelectionRequestItem.request_id == req.id, BuffSelectionRequestItem.user_id == user_id, BuffSelectionRequestItem.goods_id.in_(selected_goods), ) .all() ) for it in items: item_map[it.goods_id] = { "wear_range": it.wear_range, "price_cny": float(it.price_cny) if it.price_cny is not None else None, "price_usd": float(it.price_usd) if it.price_usd is not None else None, } log_list = [] for l in logs: row = { "created_at": l.created_at.strftime("%Y-%m-%d %H:%M:%S") if l.created_at else None, "goods_id": l.goods_id, "goods_name": l.goods_name or "", "result": l.result, "message": l.message or "", } if l.result == "selected": log_wear_range = getattr(l, "wear_range", None) log_price_cny = float(l.price_cny) if getattr(l, "price_cny", None) is not None else None log_price_usd = float(l.price_usd) if getattr(l, "price_usd", None) is not None else None if log_wear_range is not None or log_price_cny is not None or log_price_usd is not None: row["wear_range"] = log_wear_range row["price_cny"] = log_price_cny row["price_usd"] = log_price_usd elif l.goods_id in item_map: row["wear_range"] = item_map[l.goods_id]["wear_range"] row["price_cny"] = item_map[l.goods_id]["price_cny"] row["price_usd"] = item_map[l.goods_id]["price_usd"] else: row["wear_range"] = None row["price_cny"] = None row["price_usd"] = None else: row["wear_range"] = None row["price_cny"] = None row["price_usd"] = None log_list.append(row) max_sequence_no = 0 if logs: max_sequence_no = max(int(l.sequence_no or 0) for l in logs) elif last_sequence_no is not None and last_sequence_no > 0: max_sequence_no = last_sequence_no return { "code": 200, "msg": "ok", "data": { "request_id": req.id, "status": req.status, "started_at": started_at, "ended_at": ended_at, "max_sequence_no": max_sequence_no, "logs": log_list, }, } except Exception as e: logger.exception("POST /buff/secure/selection-log 异常: %s", e) try: db.rollback() except Exception: pass return {"code": 500, "msg": "系统异常，请稍后重试", "data": None} @router.post("/buff/secure/selection-stop") @secure_endpoint(require_login=True, require_machine_code=True) async def post_buff_selection_stop_secure( decrypted_params: dict, request: Request, db: Session, session=None, user=None ): """ 停止当前用户所有进行中选品任务（安全端点）。 将 status=pending/running 的请求单均标记 cancel_requested； 后台选品任务在循环中检测到后即退出并将状态置为 cancelled。 """ try: user_id = session.user_id unfinished = ( db.query(BuffSelectionRequest) .filter( BuffSelectionRequest.user_id == user_id, BuffSelectionRequest.status.in_(["pending", "running"]), ) .order_by(BuffSelectionRequest.id.asc()) .all() ) if not unfinished: return {"code": 200, "msg": "当前无进行中的选品任务", "data": None} for r in unfinished: r.cancel_requested = True db.commit() ids = [r.id for r in unfinished] msg = ( "已发送停止请求，选品任务将尽快结束" if len(ids) == 1 else f"已停止 {len(ids)} 个进行中的选品任务" ) return {"code": 200, "msg": msg, "data": {"request_ids": ids, "request_id": ids[-1]}} except Exception as e: logger.exception("POST /buff/secure/selection-stop 异常: %s", e) try: db.rollback() except Exception: pass return {"code": 500, "msg": "系统异常，请稍后重试", "data": None} @router.post("/buff/secure/selection-stop-logout") @secure_endpoint(require_login=True, require_machine_code=True, require_subscription_valid=False) async def post_buff_selection_stop_logout_secure( decrypted_params: dict, request: Request, db: Session, session=None, user=None ): """ 退登专用停止选品任务（安全端点）。 仅校验签名、会话和机器码，不校验卡密有效期；用于被动退登场景强一致停止后台选品线程。 """ try: user_id = session.user_id unfinished = ( db.query(BuffSelectionRequest) .filter( BuffSelectionRequest.user_id == user_id, BuffSelectionRequest.status.in_(["pending", "running"]), ) .order_by(BuffSelectionRequest.id.asc()) .all() ) if not unfinished: return {"code": 200, "msg": "当前无进行中的选品任务", "data": None} for r in unfinished: r.cancel_requested = True db.commit() ids = [r.id for r in unfinished] return { "code": 200, "msg": "已发送停止请求，选品任务将尽快结束", "data": {"request_ids": ids, "request_id": ids[-1]}, } except Exception as e: logger.exception("POST /buff/secure/selection-stop-logout 异常: %s", e) try: db.rollback() except Exception: pass return {"code": 500, "msg": "系统异常，请稍后重试", "data": None} @router.post("/buff/secure/selection-result") @secure_endpoint(require_login=True, require_machine_code=True) async def post_buff_selection_result_secure( decrypted_params: dict, request: Request, db: Session, session=None, user=None ): """ 获取最新一次选品请求的状态及选中饰品列表（安全端点）。 用于前台「选品数据」表格定时刷新：返回 request_id, status, items； 当 status 为 completed、failed 或 cancelled 时，前台可停止定时刷新。 若传入 selection_request_id，则返回该次选品请求（用于本地扫货按指定批次拉取饰品列表）。 """ try: user_id = session.user_id selection_request_id_raw = decrypted_params.get("selection_request_id") req = None if selection_request_id_raw is not None: try: sid = int(selection_request_id_raw) except (TypeError, ValueError): sid = 0 if sid > 0: req = ( db.query(BuffSelectionRequest) .filter( BuffSelectionRequest.user_id == user_id, BuffSelectionRequest.id == sid, ) .first() ) if req is None: req = ( db.query(BuffSelectionRequest) .filter(BuffSelectionRequest.user_id == user_id) .order_by(BuffSelectionRequest.id.desc()) .first() ) if not req: return { "code": 200, "msg": "ok", "data": { "request_id": None, "status": None, "started_at": None, "ended_at": None, "items": [], }, } started_at = req.created_at.strftime("%Y-%m-%d %H:%M:%S") if req.created_at else None ended_at = None if req.status in ("completed", "failed", "cancelled") and req.updated_at: ended_at = req.updated_at.strftime("%Y-%m-%d %H:%M:%S") item_rows = ( db.query(BuffSelectionRequestItem) .filter( BuffSelectionRequestItem.request_id == req.id, BuffSelectionRequestItem.user_id == user_id, ) .order_by(BuffSelectionRequestItem.created_at.desc(), BuffSelectionRequestItem.id.desc()) .all() ) items = [] for it in item_rows: items.append({ "id": it.id, "goods_id": it.goods_id, "goods_name": it.goods_name or "", "icon_url": it.icon_url or "", "steam_link": it.steam_link or "", "wear_range": it.wear_range or "", "price_cny": float(it.price_cny) if it.price_cny is not None else None, "price_usd": float(it.price_usd) if it.price_usd is not None else None, "created_at": it.created_at.strftime("%Y-%m-%d %H:%M:%S") if it.created_at else None, }) return { "code": 200, "msg": "ok", "data": { "request_id": req.id, "status": req.status, "started_at": started_at, "ended_at": ended_at, "items": items, }, } except Exception as e: logger.exception("POST /buff/secure/selection-result 异常: %s", e) try: db.rollback() except Exception: pass return {"code": 500, "msg": "系统异常，请稍后重试", "data": None} @router.post("/buff/secure/selection-item-delete") @secure_endpoint(require_login=True, require_machine_code=True) async def post_buff_selection_item_delete_secure( decrypted_params: dict, request: Request, db: Session, session=None, user=None ): """ 删除选品明细（安全端点）。 - 仅允许删除当前用户最新一次选品请求的数据 - 删除后扫货不再扫描该饰品 """ try: user_id = session.user_id item_id_raw = decrypted_params.get("item_id") if item_id_raw is None: return {"code": 400, "msg": "缺少 item_id", "data": None} try: item_id = int(item_id_raw) except (TypeError, ValueError): return {"code": 400, "msg": "item_id 格式错误", "data": None} latest_req = ( db.query(BuffSelectionRequest) .filter(BuffSelectionRequest.user_id == user_id) .order_by(BuffSelectionRequest.id.desc()) .first() ) if not latest_req: return {"code": 404, "msg": "暂无选品数据", "data": None} row = ( db.query(BuffSelectionRequestItem) .filter( BuffSelectionRequestItem.id == item_id, BuffSelectionRequestItem.user_id == user_id, BuffSelectionRequestItem.request_id == latest_req.id, ) .first() ) if not row: return {"code": 404, "msg": "未找到可删除的选品数据", "data": None} db.delete(row) db.commit() return {"code": 200, "msg": "删除成功", "data": {"item_id": item_id}} except Exception as e: logger.exception("POST /buff/secure/selection-item-delete 异常: %s", e) try: db.rollback() except Exception: pass return {"code": 500, "msg": "系统异常，请稍后重试", "data": None} @router.post("/buff/secure/sweep-bound-steam-usernames") @secure_endpoint(require_login=True, require_machine_code=True) async def post_buff_sweep_bound_steam_usernames_secure( decrypted_params: dict, request: Request, db: Session, session=None, user=None ): """ 扫货专用：返回当前用户在后端已绑定且启用的 Steam 账号用户名列表。 客户端仅允许使用这些用户名与本地 accounts.json 交集进行扫货。 """ try: user_id = session.user_id rows = ( db.query(SteamAccount) .filter( SteamAccount.user_id == user_id, SteamAccount.is_deleted == False, SteamAccount.is_active == True, ) .order_by(SteamAccount.id.asc()) .all() ) names = [str(r.steam_username).strip() for r in rows if r.steam_username] return {"code": 200, "msg": "ok", "data": {"steam_usernames": names}} except Exception as e: logger.exception("POST /buff/secure/sweep-bound-steam-usernames 异常: %s", e) try: db.rollback() except Exception: pass return {"code": 500, "msg": "系统异常，请稍后重试", "data": None} @router.post("/buff/secure/sweep-start") @secure_endpoint(require_login=True, require_machine_code=True) async def post_buff_sweep_start_secure( decrypted_params: dict, request: Request, db: Session, session=None, user=None ): """ 开始扫货（安全端点）。 - 默认基于最新一次选品请求的数据 - 若已有 pending/running 的扫货请求，则禁止重复开始 """ import asyncio try: user_id = session.user_id unfinished_reqs = ( db.query(BuffSweepRequest) .filter( BuffSweepRequest.user_id == user_id, BuffSweepRequest.status.in_(("pending", "running")), ) .order_by(BuffSweepRequest.id.asc()) .all() ) cleaned_request_ids = [] if unfinished_reqs: for old_req in unfinished_reqs: old_req.cancel_requested = True old_req.status = "cancelled" cleaned_request_ids.append(old_req.id) db.flush() latest_selection = ( db.query(BuffSelectionRequest) .filter(BuffSelectionRequest.user_id == user_id) .order_by(BuffSelectionRequest.id.desc()) .first() ) if not latest_selection: return {"code": 404, "msg": "暂无选品数据，请先完成选品", "data": None} item_count = ( db.query(BuffSelectionRequestItem) .filter( BuffSelectionRequestItem.user_id == user_id, BuffSelectionRequestItem.request_id == latest_selection.id, ) .count() ) if item_count <= 0: return {"code": 400, "msg": "最新选品数据为空，请先选品或检查是否已删除全部饰品", "data": None} req = BuffSweepRequest( user_id=user_id, selection_request_id=latest_selection.id, status="pending", ) db.add(req) db.commit() db.refresh(req) client_local = bool(decrypted_params.get("client_local")) if not client_local: loop = asyncio.get_event_loop() loop.run_in_executor(None, lambda: run_sweep_task(req.id, user_id, latest_selection.id)) msg = "扫货已开始，正在后台执行" else: msg = "扫货请求单已创建，等待客户端本地执行" if cleaned_request_ids: msg = f"已停止{len(cleaned_request_ids)}个历史未结束扫货任务，{msg}" return { "code": 200, "msg": msg, "data": { "request_id": req.id, "selection_request_id": latest_selection.id, "cleaned_request_ids": cleaned_request_ids, }, } except Exception as e: logger.exception("POST /buff/secure/sweep-start 异常: %s", e) try: db.rollback() except Exception: pass return {"code": 500, "msg": "系统异常，请稍后重试", "data": None} @router.post("/buff/secure/sweep-stop") @secure_endpoint(require_login=True, require_machine_code=True) async def post_buff_sweep_stop_secure( decrypted_params: dict, request: Request, db: Session, session=None, user=None ): """停止当前用户所有进行中的扫货任务（安全端点，与 sweep-start 清理逻辑一致）。""" try: user_id = session.user_id unfinished = ( db.query(BuffSweepRequest) .filter( BuffSweepRequest.user_id == user_id, BuffSweepRequest.status.in_(("pending", "running")), ) .order_by(BuffSweepRequest.id.asc()) .all() ) if not unfinished: return {"code": 200, "msg": "当前无进行中的扫货任务", "data": None} for r in unfinished: r.cancel_requested = True if r.status in ("pending", "running"): r.status = "cancelled" db.commit() ids = [r.id for r in unfinished] msg = "扫货已停止" if len(ids) == 1 else f"已停止 {len(ids)} 个进行中的扫货任务" return {"code": 200, "msg": msg, "data": {"request_ids": ids, "request_id": ids[-1]}} except Exception as e: logger.exception("POST /buff/secure/sweep-stop 异常: %s", e) try: db.rollback() except Exception: pass return {"code": 500, "msg": "系统异常，请稍后重试", "data": None} @router.post("/buff/secure/sweep-log") @secure_endpoint(require_login=True, require_machine_code=True) async def post_buff_sweep_log_secure( decrypted_params: dict, request: Request, db: Session, session=None, user=None ): """获取最新一次扫货日志（安全端点）。""" try: user_id = session.user_id req = ( db.query(BuffSweepRequest) .filter(BuffSweepRequest.user_id == user_id) .order_by(BuffSweepRequest.id.desc()) .first() ) if not req: return { "code": 200, "msg": "ok", "data": { "request_id": None, "status": None, "started_at": None, "ended_at": None, "max_sequence_no": 0, "logs": [], }, } last_sequence_no_raw = decrypted_params.get("last_sequence_no") limit_raw = decrypted_params.get("limit") try: last_sequence_no = int(last_sequence_no_raw) if last_sequence_no_raw is not None else None except (TypeError, ValueError): last_sequence_no = None try: limit = int(limit_raw) if limit_raw is not None else 200 except (TypeError, ValueError): limit = 200 if limit <= 0: limit = 200 if limit > 500: limit = 500 logs_q = ( db.query(BuffSweepRequestLog) .filter( BuffSweepRequestLog.user_id == user_id, BuffSweepRequestLog.sweep_request_id == req.id, ) ) if last_sequence_no is not None and last_sequence_no > 0: logs_q = logs_q.filter(BuffSweepRequestLog.sequence_no > last_sequence_no) logs = ( logs_q .order_by(BuffSweepRequestLog.sequence_no.desc(), BuffSweepRequestLog.created_at.desc()) .limit(limit) .all() ) max_sequence_no = 0 if logs: max_sequence_no = max(int(l.sequence_no or 0) for l in logs) elif last_sequence_no is not None and last_sequence_no > 0: max_sequence_no = last_sequence_no return { "code": 200, "msg": "ok", "data": { "request_id": req.id, "status": req.status, "started_at": req.created_at.strftime("%Y-%m-%d %H:%M:%S") if req.created_at else None, "ended_at": req.updated_at.strftime("%Y-%m-%d %H:%M:%S") if req.status in ("completed", "failed", "cancelled") and req.updated_at else None, "max_sequence_no": max_sequence_no, "logs": [ { "created_at": l.created_at.strftime("%Y-%m-%d %H:%M:%S") if l.created_at else None, "goods_id": l.goods_id, "goods_name": l.goods_name or "", "result": l.result, "message": l.message or "", "retry_attempts": int(l.retry_attempts or 0), "proxy_refreshed": bool(l.proxy_refreshed), "target_price_usd": float(l.target_price_usd) if l.target_price_usd is not None else None, "purchase_price_usd": float(l.purchase_price_usd) if l.purchase_price_usd is not None else None, } for l in logs ], }, } except Exception as e: logger.exception("POST /buff/secure/sweep-log 异常: %s", e) try: db.rollback() except Exception: pass return {"code": 500, "msg": "系统异常，请稍后重试", "data": None} @router.post("/buff/secure/sweep-log-client-sync") @secure_endpoint(require_login=True, require_machine_code=True) async def post_buff_sweep_log_client_sync_secure( decrypted_params: dict, request: Request, db: Session, session=None, user=None ): """ 客户端本地扫货日志同步到后端数据库（安全端点）。 说明： - 客户端以本地 steam-msg/{request_id}.txt 为准 - 后端按 user_id + sweep_request_id 做覆盖写入（先删后插），保留最新快照 """ try: user_id = session.user_id sweep_request_id_raw = decrypted_params.get("sweep_request_id") if sweep_request_id_raw is None: return {"code": 400, "msg": "缺少 sweep_request_id", "data": None} try: sweep_request_id = int(sweep_request_id_raw) except (TypeError, ValueError): return {"code": 400, "msg": "sweep_request_id 格式错误", "data": None} if sweep_request_id <= 0: return {"code": 400, "msg": "sweep_request_id 必须大于0", "data": None} logs = decrypted_params.get("logs") if not isinstance(logs, list): return {"code": 400, "msg": "logs 格式错误", "data": None} selection_request_id_raw = decrypted_params.get("selection_request_id") status_raw = str(decrypted_params.get("status") or "").strip().lower() allowed_status = {"pending", "running", "completed", "failed", "cancelled"} normalized_status = status_raw if status_raw in allowed_status else None req = ( db.query(BuffSweepRequest) .filter( BuffSweepRequest.id == sweep_request_id, BuffSweepRequest.user_id == user_id, ) .first() ) if req is None: selection_request_id = None try: if selection_request_id_raw is not None: selection_request_id = int(selection_request_id_raw) except (TypeError, ValueError): selection_request_id = None if not selection_request_id or selection_request_id <= 0: latest_selection = ( db.query(BuffSelectionRequest) .filter(BuffSelectionRequest.user_id == user_id) .order_by(BuffSelectionRequest.id.desc()) .first() ) if latest_selection: selection_request_id = latest_selection.id if not selection_request_id or selection_request_id <= 0: return {"code": 400, "msg": "缺少有效 selection_request_id，无法创建扫货请求单", "data": None} req = BuffSweepRequest( id=sweep_request_id, user_id=user_id, selection_request_id=selection_request_id, status=normalized_status or "pending", ) db.add(req) elif normalized_status: req.status = normalized_status # 覆盖式写入：删除同请求历史日志，再插入本次快照 db.query(BuffSweepRequestLog).filter( BuffSweepRequestLog.user_id == user_id, BuffSweepRequestLog.sweep_request_id == sweep_request_id, ).delete(synchronize_session=False) insert_count = 0 for idx, row in enumerate(logs, start=1): if not isinstance(row, dict): continue goods_id = str(row.get("goods_id") or "").strip() goods_name = str(row.get("goods_name") or "").strip() result = str(row.get("result") or "").strip() or "skipped" message = str(row.get("message") or "").strip() if not goods_id: goods_id = "-" target_price = row.get("target_price_usd") purchase_price = row.get("purchase_price_usd") try: target_price = float(target_price) if target_price is not None else None except Exception: target_price = None try: purchase_price = float(purchase_price) if purchase_price is not None else None except Exception: purchase_price = None db.add(BuffSweepRequestLog( sweep_request_id=sweep_request_id, user_id=user_id, goods_id=goods_id, goods_name=goods_name, sequence_no=idx, result=result, message=message[:512] if message else "", retry_attempts=0, proxy_refreshed=False, target_price_usd=target_price, purchase_price_usd=purchase_price, )) insert_count += 1 db.commit() return {"code": 200, "msg": "ok", "data": {"sweep_request_id": sweep_request_id, "inserted": insert_count}} except Exception as e: logger.exception("POST /buff/secure/sweep-log-client-sync 异常: %s", e) try: db.rollback() except Exception: pass return {"code": 500, "msg": "系统异常，请稍后重试", "data": None} @router.post("/buff/secure/sweep-item-client-sync") @secure_endpoint(require_login=True, require_machine_code=True) async def post_buff_sweep_item_client_sync_secure( decrypted_params: dict, request: Request, db: Session, session=None, user=None ): """ 客户端本地扫货成功明细同步到后端数据库（安全端点）。 说明： - 客户端提交成功购买 items 快照（含 steam_account） - 默认：按 user_id + sweep_request_id 覆盖写入（先删后插） - append_only=true：不删除历史行，每笔成功购买追加一行（同一选品明细可有多笔成交，勿按 selection_item_id 覆盖） """ try: user_id = session.user_id append_only = bool(decrypted_params.get("append_only")) sweep_request_id_raw = decrypted_params.get("sweep_request_id") if sweep_request_id_raw is None: return {"code": 400, "msg": "缺少 sweep_request_id", "data": None} try: sweep_request_id = int(sweep_request_id_raw) except (TypeError, ValueError): return {"code": 400, "msg": "sweep_request_id 格式错误", "data": None} if sweep_request_id <= 0: return {"code": 400, "msg": "sweep_request_id 必须大于0", "data": None} items = decrypted_params.get("items") if not isinstance(items, list): return {"code": 400, "msg": "items 格式错误", "data": None} selection_request_id_raw = decrypted_params.get("selection_request_id") req = ( db.query(BuffSweepRequest) .filter( BuffSweepRequest.id == sweep_request_id, BuffSweepRequest.user_id == user_id, ) .first() ) if req is None: selection_request_id = None try: if selection_request_id_raw is not None: selection_request_id = int(selection_request_id_raw) except (TypeError, ValueError): selection_request_id = None if not selection_request_id or selection_request_id <= 0: latest_selection = ( db.query(BuffSelectionRequest) .filter(BuffSelectionRequest.user_id == user_id) .order_by(BuffSelectionRequest.id.desc()) .first() ) if latest_selection: selection_request_id = latest_selection.id if not selection_request_id or selection_request_id <= 0: return {"code": 400, "msg": "缺少有效 selection_request_id，无法创建扫货请求单", "data": None} req = BuffSweepRequest( id=sweep_request_id, user_id=user_id, selection_request_id=selection_request_id, status="pending", ) db.add(req) if not append_only: # 覆盖式写入：删除同请求历史成功明细，再插入本次快照 db.query(BuffSweepRequestItem).filter( BuffSweepRequestItem.user_id == user_id, BuffSweepRequestItem.sweep_request_id == sweep_request_id, ).delete(synchronize_session=False) insert_count = 0 for row in items: if not isinstance(row, dict): continue try: selection_item_id = int(row.get("selection_item_id") or 0) except (TypeError, ValueError): selection_item_id = 0 if selection_item_id <= 0: continue goods_id = str(row.get("goods_id") or "").strip() if not goods_id: continue try: item_selection_request_id = int( row.get("selection_request_id") or req.selection_request_id ) except (TypeError, ValueError): item_selection_request_id = req.selection_request_id target_price = row.get("target_price_usd") purchase_price = row.get("purchase_price_usd") purchase_wear = row.get("purchase_wear") try: target_price = float(target_price) if target_price is not None else None except Exception: target_price = None try: purchase_price = float(purchase_price) if purchase_price is not None else None except Exception: purchase_price = None try: purchase_wear = float(purchase_wear) if purchase_wear is not None else None except Exception: purchase_wear = None steam_account = str(row.get("steam_account") or "").strip() or None db.add(BuffSweepRequestItem( sweep_request_id=sweep_request_id, user_id=user_id, selection_request_id=item_selection_request_id, selection_item_id=selection_item_id, goods_id=goods_id, goods_name=str(row.get("goods_name") or "").strip() or None, icon_url=str(row.get("icon_url") or "").strip() or None, wear_range=str(row.get("wear_range") or "").strip() or None, purchase_wear=purchase_wear, target_price_usd=target_price, purchase_price_usd=purchase_price, steam_account=steam_account, )) insert_count += 1 db.commit() return {"code": 200, "msg": "ok", "data": {"sweep_request_id": sweep_request_id, "inserted": insert_count}} except Exception as e: logger.exception("POST /buff/secure/sweep-item-client-sync 异常: %s", e) try: db.rollback() except Exception: pass return {"code": 500, "msg": "系统异常，请稍后重试", "data": None} @router.post("/buff/secure/sweep-result") @secure_endpoint(require_login=True, require_machine_code=True) async def post_buff_sweep_result_secure( decrypted_params: dict, request: Request, db: Session, session=None, user=None ): """ 获取扫货成功列表（安全端点）。 - 结果始终返回该用户全部扫货成功记录 - 支持分页 page/page_size """ try: user_id = session.user_id page = int(decrypted_params.get("page") or 1) page_size = int(decrypted_params.get("page_size") or 10) if page < 1: page = 1 if page_size < 1 or page_size > 100: page_size = 10 latest_req = ( db.query(BuffSweepRequest) .filter(BuffSweepRequest.user_id == user_id) .order_by(BuffSweepRequest.id.desc()) .first() ) q = db.query(BuffSweepRequestItem).filter(BuffSweepRequestItem.user_id == user_id) total = q.count() rows = ( q.order_by(BuffSweepRequestItem.created_at.desc(), BuffSweepRequestItem.id.desc()) .offset((page - 1) * page_size) .limit(page_size) .all() ) return { "code": 200, "msg": "ok", "data": { "request_id": latest_req.id if latest_req else None, "status": latest_req.status if latest_req else None, "started_at": latest_req.created_at.strftime("%Y-%m-%d %H:%M:%S") if latest_req and latest_req.created_at else None, "ended_at": latest_req.updated_at.strftime("%Y-%m-%d %H:%M:%S") if latest_req and latest_req.status in ("completed", "failed", "cancelled") and latest_req.updated_at else None, "page": page, "page_size": page_size, "total": total, "items": [ { "id": r.id, "sweep_request_id": r.sweep_request_id, "selection_request_id": r.selection_request_id, "selection_item_id": r.selection_item_id, "goods_id": r.goods_id, "goods_name": r.goods_name or "", "icon_url": r.icon_url or "", "wear_range": r.wear_range or "", "purchase_wear": str(r.purchase_wear) if r.purchase_wear is not None else None, "target_price_usd": float(r.target_price_usd) if r.target_price_usd is not None else None, "purchase_price_usd": float(r.purchase_price_usd) if r.purchase_price_usd is not None else None, "steam_account": r.steam_account, "created_at": r.created_at.strftime("%Y-%m-%d %H:%M:%S") if r.created_at else None, } for r in rows ], }, } except Exception as e: logger.exception("POST /buff/secure/sweep-result 异常: %s", e) try: db.rollback() except Exception: pass return {"code": 500, "msg": "系统异常，请稍后重试", "data": None} # ============ HMAC 密钥（安全端点，登录后密钥交换） ============ @router.post("/keys/hmac/secure") @secure_endpoint(require_login=True, require_machine_code=True) async def post_keys_hmac_secure( decrypted_params: dict, request: Request, db: Session, session=None, user=None ): """ 同步/获取动态HMAC密钥（安全端点：POST，入参出参密文，时间戳与签名校验）。 会话与机器码由装饰器校验，无需解密参数。日志统一记录入参/出参密文与明文。 """ try: from app.services.hmac_key_service import HmacKeyService hmac_service = HmacKeyService(db) success, message, data = hmac_service.create_hmac_key( user_id=user.id, username=user.username, machine_code=session.machine_code, ) if success and data: return {"code": 200, "msg": message or "成功", "data": data} return {"code": 400, "msg": message or "HMAC密钥获取失败", "data": None} except Exception as e: logger.exception("POST /keys/hmac/secure 异常: %s", e) return {"code": 500, "msg": "系统异常，请稍后重试", "data": None}