""" BUFF 选品服务：第一步筛选候选 ID + 第二步利润判断与落库。 逻辑见 docs/buff_selection_filter_logic.md、docs/buff_selection_step2_logic.md。 """ import json import random from decimal import Decimal from typing import List, Dict, Any, Optional, Tuple from sqlalchemy import text from sqlalchemy.orm import Session from app.database import SessionLocal from app.models.buff_selection_request import BuffSelectionRequest from app.models.buff_selection_request_item import BuffSelectionRequestItem from app.models.buff_selection_request_log import BuffSelectionRequestLog from app.models.buff_selection_config import BuffSelectionConfig from app.models.system_config import SystemConfig from app.utils.timezone import beijing_now_naive from app.core.logger import logger def _get_system_config_value(db: Session, config_key: str, app_name: str = "backend-selection") -> Optional[str]: """读取 system_config 配置值。""" row = db.query(SystemConfig).filter( SystemConfig.config_key == config_key, SystemConfig.app_name == app_name, SystemConfig.is_deleted == False, ).first() return row.config_value if row else None def _get_system_config_int(db: Session, config_key: str, default: int = 0, app_name: str = "backend-selection") -> int: """读取 system_config 整型配置。""" val = _get_system_config_value(db, config_key, app_name) if val is None or val == "": return default try: return int(val) except (ValueError, TypeError): return default def _parse_ids_field(raw: Any) -> List[str]: """将 excludeIds/specifiedIds（字符串或数组）解析为 ID 列表。""" if raw is None: return [] if isinstance(raw, list): return [str(x).strip() for x in raw if str(x).strip()] s = str(raw).strip() if not s: return [] return [x.strip() for x in s.replace(",", "\n").splitlines() if x.strip()] def _collect_gun_codes(config: Dict[str, Any]) -> List[str]: """从配置中收集所有枪械分类编码（步枪/手枪/SMG/霰弹枪/机枪）。""" codes = [] for key in ("rifle", "pistol", "smg", "shotgun", "machinegun"): val = config.get(key) if isinstance(val, list): codes.extend([str(x).strip() for x in val if str(x).strip()]) return list(dict.fromkeys(codes)) def step1_candidate_ids(db: Session, config: Dict[str, Any]) -> Tuple[List[str], List[str]]: """ 第一步：根据基础过滤条件从 buff_goods_filtered_view 筛出候选 ID，排除排除 ID。 返回两个列表： - specified_ids: 指定饰品 ID 列表（不乱序，保持用户指定的顺序） - random_ids: 随机饰品 ID 列表（已乱序） """ conditions: List[str] = ["1=1"] params: Dict[str, Any] = {} # 枪械：都没选 -> 1=1；有选 -> 分类编码 IN (...) gun_codes = _collect_gun_codes(config) if gun_codes: placeholders = ", ".join([f":gun_{i}" for i in range(len(gun_codes))]) conditions.append(f"`分类编码` IN ({placeholders})") for i, c in enumerate(gun_codes): params[f"gun_{i}"] = c # 品质 quality = config.get("quality") if isinstance(quality, list) and quality: ph = ", ".join([f":quality_{i}" for i in range(len(quality))]) conditions.append(f"`品质` IN ({ph})") for i, v in enumerate(quality): params[f"quality_{i}"] = str(v).strip() # 类别 category = config.get("category") if isinstance(category, list) and category: ph = ", ".join([f":category_{i}" for i in range(len(category))]) conditions.append(f"`类别` IN ({ph})") for i, v in enumerate(category): params[f"category_{i}"] = str(v).strip() # 外观 exterior = config.get("exterior") if isinstance(exterior, list) and exterior: ph = ", ".join([f":exterior_{i}" for i in range(len(exterior))]) conditions.append(f"`外观` IN ({ph})") for i, v in enumerate(exterior): params[f"exterior_{i}"] = str(v).strip() # 收藏品 collection = config.get("collection") if isinstance(collection, list) and collection: ph = ", ".join([f":collection_{i}" for i in range(len(collection))]) conditions.append(f"`收藏品` IN ({ph})") for i, v in enumerate(collection): params[f"collection_{i}"] = str(v).strip() # BUFF 价格区间 pmin = config.get("buffPriceMin") pmax = config.get("buffPriceMax") if pmin is not None and pmax is not None: conditions.append("`当前在售最低价格` >= :buff_price_min AND `当前在售最低价格` <= :buff_price_max") params["buff_price_min"] = float(pmin) params["buff_price_max"] = float(pmax) elif pmin is not None: conditions.append("`当前在售最低价格` >= :buff_price_min") params["buff_price_min"] = float(pmin) elif pmax is not None: conditions.append("`当前在售最低价格` <= :buff_price_max") params["buff_price_max"] = float(pmax) # BUFF 在售数量 smin = config.get("buffSalesMin") smax = config.get("buffSalesMax") if smin is not None and smax is not None: conditions.append("`在售数量` >= :buff_sales_min AND `在售数量` <= :buff_sales_max") params["buff_sales_min"] = int(smin) params["buff_sales_max"] = int(smax) elif smin is not None: conditions.append("`在售数量` >= :buff_sales_min") params["buff_sales_min"] = int(smin) elif smax is not None: conditions.append("`在售数量` <= :buff_sales_max") params["buff_sales_max"] = int(smax) # 排除的饰品 ID（从随机候选中排除，不影响指定饰品） exclude_ids = _parse_ids_field(config.get("excludeIds")) if exclude_ids: ph = ", ".join([f":exclude_{i}" for i in range(len(exclude_ids))]) conditions.append(f"`BUFF饰品ID` NOT IN ({ph})") for i, v in enumerate(exclude_ids): params[f"exclude_{i}"] = v where_sql = " AND ".join(conditions) sql = f"SELECT DISTINCT `BUFF饰品ID` FROM buff_goods_filtered_view WHERE {where_sql}" try: rows = db.execute(text(sql), params).fetchall() candidate = [str(r[0]) for r in rows if r[0]] except Exception as e: logger.exception("step1 执行主查询异常: %s", e) candidate = [] # 指定饰品 ID（不乱序，保持用户指定的顺序） specified = _parse_ids_field(config.get("specifiedIds")) # 随机候选饰品：排除指定饰品后乱序 random_candidate = [gid for gid in candidate if gid not in specified] random.shuffle(random_candidate) return specified, random_candidate def _get_lowest_price(db: Session, goods_id: str) -> Optional[Decimal]: """获取饰品当前在售最低价格。""" sql = text( "SELECT MIN(`当前在售最低价格`) FROM buff_goods_filtered_view WHERE `BUFF饰品ID` = :gid" ) row = db.execute(sql, {"gid": goods_id}).fetchone() if row and row[0] is not None: return Decimal(str(row[0])) return None def _get_wear_ranges(db: Session, goods_id: str) -> List[Tuple[Decimal, Decimal, Decimal]]: """获取磨损区间列表：(最小磨损值, 最大磨损值, 区间底价)，按最小磨损值升序。""" sql = text( "SELECT `最小磨损值`, `最大磨损值`, `区间底价` FROM buff_goods_filtered_view " "WHERE `BUFF饰品ID` = :gid ORDER BY `最小磨损值` ASC" ) rows = db.execute(sql, {"gid": goods_id}).fetchall() out = [] for r in rows: if r[0] is not None and r[1] is not None and r[2] is not None: out.append((Decimal(str(r[0])), Decimal(str(r[1])), Decimal(str(r[2])))) return out def _get_ai_price_avgs(db: Session, goods_id: str) -> Tuple[Optional[Decimal], Optional[Decimal]]: """获取 7天/30天在售最低价格平均价。""" sql = text( "SELECT `7天在售最低价格平均价`, `30天在售最低价格平均价` " "FROM buff_goods_filtered_view WHERE `BUFF饰品ID` = :gid LIMIT 1" ) row = db.execute(sql, {"gid": goods_id}).fetchone() if not row: return None, None avg7 = Decimal(str(row[0])) if row[0] is not None else None avg30 = Decimal(str(row[1])) if row[1] is not None else None return avg7, avg30 def _assistant_check( db: Session, goods_id: str, assistant_level: str, current_lowest: Decimal ) -> Tuple[bool, str]: """ 第二步智能选品判断： - low: 不限制 - standard: 7天均价相对当前价需在配置涨跌幅范围内 - high: 30天均价相对当前价需在配置涨跌幅范围内 """ level = (assistant_level or "low").lower() if level == "low": return True, "通过" avg7, avg30 = _get_ai_price_avgs(db, goods_id) if level == "standard": if avg7 is None: return False, "7天均价缺失，智能选品标准模式不满足" inc = Decimal(str(_get_system_config_value(db, "buff_ai_standard_range_max_increase_ratio") or "0.3")) dec = Decimal(str(_get_system_config_value(db, "buff_ai_standard_range_max_decrease_ratio") or "0.5")) upper = current_lowest * (Decimal("1") + inc) lower = current_lowest * (Decimal("1") - dec) if avg7 > upper or avg7 < lower: return False, f"7天均价 {avg7} 不在智能选品标准范围 [{lower}, {upper}]" return True, "通过" if level == "high": if avg30 is None: return False, "30天均价缺失，智能选品高模式不满足" inc = Decimal(str(_get_system_config_value(db, "buff_ai_high_range_max_increase_ratio") or "0.4")) dec = Decimal(str(_get_system_config_value(db, "buff_ai_high_range_max_decrease_ratio") or "0.5")) upper = current_lowest * (Decimal("1") + inc) lower = current_lowest * (Decimal("1") - dec) if avg30 > upper or avg30 < lower: return False, f"30天均价 {avg30} 不在智能选品高范围 [{lower}, {upper}]" return True, "通过" # 兜底：未知值按 low 处理，避免阻断任务 return True, "通过" def _normal_ranges(ranges: List[Tuple[Decimal, Decimal, Decimal]]) -> List[Tuple[Decimal, Decimal, Decimal]]: """正常磨损区间：区间底价随最小磨损值递增而单调不升。""" if not ranges: return [] out = [ranges[0]] for i in range(1, len(ranges)): if ranges[i][2] <= out[-1][2]: out.append(ranges[i]) else: break return out def _qualified_ranges( normal: List[Tuple[Decimal, Decimal, Decimal]], compare_price: Decimal ) -> List[Tuple[Decimal, Decimal, Decimal]]: """满足利润的区间：区间底价 > 比较价，一旦不满足则截断。""" out = [] for r in normal: if r[2] > compare_price: out.append(r) else: break return out def _get_goods_display(db: Session, goods_id: str) -> Dict[str, Any]: """从视图取饰品名称、图片、Steam 链接（取一行即可）。""" sql = text( "SELECT `BUFF饰品名称`, `饰品图片URL地址`, `饰品STEAM链接` FROM buff_goods_filtered_view " "WHERE `BUFF饰品ID` = :gid LIMIT 1" ) row = db.execute(sql, {"gid": goods_id}).fetchone() if row: return { "goods_name": (row[0] or "").strip() or None, "icon_url": (row[1] or "").strip() or None, "steam_link": (row[2] or "").strip() or None, } return {"goods_name": None, "icon_url": None, "steam_link": None} def _sweep_factor(db: Session, level: str) -> Decimal: """扫货价格系数：low->min, standard->mid, high->max。""" key = { "low": "buff_sweep_price_factor_min", "standard": "buff_sweep_price_factor_mid", "high": "buff_sweep_price_factor_max", }.get((level or "low").lower(), "buff_sweep_price_factor_min") val = _get_system_config_value(db, key) or "0" return Decimal(str(val)) def _append_stage_log( db: Session, request_id: int, user_id: int, sequence_no: int, message: str, ) -> None: """追加选品阶段日志（检索/过滤/计算等），供前端日志弹窗展示。""" db.add(BuffSelectionRequestLog( request_id=request_id, user_id=user_id, goods_id="__stage__", goods_name=None, sequence_no=sequence_no, result="stage", message=message, )) db.commit() def _get_remaining_quota( db: Session, user_id: int, selection_quantity: int, items_limit_daily: int, today_start, selected_count: int, ) -> Tuple[int, str]: """ 计算剩余名额和是否应该结束选品。 返回 (remaining_quota, stop_reason) - remaining_quota: 剩余可入选的名额（0 表示已达上限） - stop_reason: 停止原因（空字符串表示可以继续） """ # 检查单次选品上限 if selection_quantity > 0 and selected_count >= selection_quantity: return 0, f"已达单次选品上限 {selection_quantity} 条" # 检查每日上限 today_items = db.query(BuffSelectionRequestItem).filter( BuffSelectionRequestItem.user_id == user_id, BuffSelectionRequestItem.created_at >= today_start, ).count() if today_items >= items_limit_daily: return 0, f"已达每日选品上限 {items_limit_daily} 条" # 计算剩余名额 remaining = selection_quantity - selected_count if selection_quantity > 0 else 999999 daily_remaining = items_limit_daily - today_items return min(remaining, daily_remaining), "" def _check_stop_request(db: Session, req: BuffSelectionRequest, request_id: int) -> bool: """检查是否收到用户停止请求。""" db.refresh(req) if getattr(req, "cancel_requested", False): logger.info("选品被用户请求停止: request_id=%s", request_id) req.status = "cancelled" db.commit() return True return False def _process_single_item( db: Session, request_id: int, user_id: int, goods_id: str, seq: int, profit_rate: float, buff_price_min: Optional[Decimal], buff_price_max: Optional[Decimal], assistant_level: str, steam_card_ratio: float, sweep_factor: Decimal, selected_count: int, ) -> Tuple[bool, int, str, Optional[Decimal], Optional[Decimal], Optional[str]]: """ 处理单个饰品，判断是否满足所有条件并入选。 返回: - selected: 是否入选 - new_selected_count: 更新后的已入选数量 - log_msg: 日志消息 - price_cny: 人民币价格（入选时有值） - price_usd: 美元价格（入选时有值） - wear_range_str: 磨损区间字符串（入选时有值） """ lowest = _get_lowest_price(db, goods_id) if lowest is None or lowest <= 0: return False, selected_count, "无在售最低价", None, None, None if buff_price_min is not None and lowest < buff_price_min: return False, selected_count, f"当前在售最低价 {lowest} 低于设置下限 {buff_price_min}", None, None, None if buff_price_max is not None and lowest > buff_price_max: return False, selected_count, f"当前在售最低价 {lowest} 高于设置上限 {buff_price_max}", None, None, None assistant_ok, assistant_msg = _assistant_check(db, goods_id, assistant_level, lowest) if not assistant_ok: return False, selected_count, assistant_msg, None, None, None compare_price = lowest * (Decimal("1") + Decimal(str(profit_rate))) ranges = _get_wear_ranges(db, goods_id) normal = _normal_ranges(ranges) qualified = _qualified_ranges(normal, compare_price) if not qualified: return False, selected_count, "无满足利润的磨损区间", None, None, None # 入选 display = _get_goods_display(db, goods_id) min_min = min(r[0] for r in qualified) max_max = max(r[1] for r in qualified) wear_range_str = f"{min_min}-{max_max}" price_cny = lowest * (Decimal("1") + sweep_factor) price_usd_raw = float(price_cny) / steam_card_ratio price_usd = Decimal(str(round(price_usd_raw, 2))) + Decimal("0.01") return True, selected_count + 1, "已入选", price_cny, price_usd, wear_range_str def run_selection_task(request_id: int, user_id: int, config_id: Optional[int]) -> None: """ 后台选品任务：更新状态为 running，执行 step1 + step2，最后更新为 completed/failed。 使用独立 db 会话，在请求返回后执行。 """ db = SessionLocal() try: req = db.query(BuffSelectionRequest).filter( BuffSelectionRequest.id == request_id, BuffSelectionRequest.user_id == user_id, ).first() if not req: logger.warning("选品请求单不存在或非本人: request_id=%s user_id=%s", request_id, user_id) return if req.status != "pending": logger.info("选品请求单已非 pending，跳过: request_id=%s status=%s", request_id, req.status) return req.status = "running" db.commit() # 处理竞态：如果在 pending->running 切换窗口已收到停止请求，立即结束任务 db.refresh(req) if getattr(req, "cancel_requested", False): logger.info("选品在启动阶段被用户请求停止: request_id=%s", request_id) req.status = "cancelled" db.commit() return config = {} if config_id: cfg_row = db.query(BuffSelectionConfig).filter( BuffSelectionConfig.id == config_id, BuffSelectionConfig.user_id == user_id, BuffSelectionConfig.is_deleted == False, ).first() if cfg_row and cfg_row.config_json: try: config = json.loads(cfg_row.config_json) except Exception: pass if not config: logger.warning("选品配置为空，请求单标记为失败: request_id=%s", request_id) req.status = "failed" db.commit() return profit_rate = config.get("profitRate") if profit_rate is None: profit_rate = 0 profit_rate = float(profit_rate) / 100.0 # 百分比转小数 steam_card_ratio = config.get("steamCardRatio") if steam_card_ratio is None or float(steam_card_ratio) <= 0: steam_card_ratio = 1.0 else: steam_card_ratio = float(steam_card_ratio) selection_quantity = config.get("selectionQuantity") if selection_quantity is None: # 用户未配置，使用系统配置的单次上限 selection_quantity = _get_system_config_int(db, "buff_selection_max_quantity", 300) else: selection_quantity = int(selection_quantity) assistant_level = (config.get("assistantLevel") or "low").lower() # BUFF 价格区间兜底校验：即使 step1 候选因数据异常混入，也在 step2 最终入选前拦截 raw_price_min = config.get("buffPriceMin") raw_price_max = config.get("buffPriceMax") try: buff_price_min = Decimal(str(raw_price_min)) if raw_price_min is not None else None except Exception: buff_price_min = None try: buff_price_max = Decimal(str(raw_price_max)) if raw_price_max is not None else None except Exception: buff_price_max = None # 每日选品条数上限（system_config），选品过程中达到则本单正常结束 today_start = beijing_now_naive().replace(hour=0, minute=0, second=0, microsecond=0) items_limit_daily = _get_system_config_int(db, "buff_selection_daily_items_limit_per_user", 900) sweep_level = (config.get("sweepAssistantLevel") or config.get("assistantLevel") or "low").lower() sweep_factor = _sweep_factor(db, sweep_level) # 第一步：检索候选 _append_stage_log(db, request_id, user_id, 0, "检索阶段：开始根据条件检索候选饰品") specified_ids, random_ids = step1_candidate_ids(db, config) _append_stage_log(db, request_id, user_id, 1, f"检索阶段：完成，指定饰品 {len(specified_ids)} 个，随机候选 {len(random_ids)} 个") # 第二步：分两阶段选品 # 阶段1：处理指定饰品（不乱序，优先入选） _append_stage_log(db, request_id, user_id, 2, "过滤阶段-指定饰品：开始处理指定饰品") specified_selected = 0 for seq, goods_id in enumerate(specified_ids, start=1): # 检查停止请求 if _check_stop_request(db, req, request_id): break # 检查名额 remaining, stop_reason = _get_remaining_quota( db, user_id, selection_quantity, items_limit_daily, today_start, specified_selected ) if remaining <= 0: logger.info("指定饰品选品结束: request_id=%s reason=%s", request_id, stop_reason) break selected, specified_selected, log_msg, price_cny, price_usd, wear_range = _process_single_item( db, request_id, user_id, goods_id, seq, profit_rate, buff_price_min, buff_price_max, assistant_level, steam_card_ratio, sweep_factor, specified_selected ) display = _get_goods_display(db, goods_id) db.add(BuffSelectionRequestLog( request_id=request_id, user_id=user_id, goods_id=goods_id, goods_name=display.get("goods_name"), sequence_no=seq, result="selected" if selected else "not_selected", message=log_msg, wear_range=wear_range if selected else None, price_cny=price_cny if selected else None, price_usd=price_usd if selected else None, )) db.commit() if selected: db.add(BuffSelectionRequestItem( request_id=request_id, user_id=user_id, goods_id=goods_id, goods_name=display.get("goods_name"), icon_url=display.get("icon_url"), steam_link=display.get("steam_link"), price_cny=price_cny, price_usd=price_usd, wear_range=wear_range, )) db.commit() _append_stage_log(db, request_id, user_id, 3, f"过滤阶段-指定饰品：结束，共入选 {specified_selected} 个指定饰品") # 阶段2：处理随机饰品（仅在指定饰品未占满名额时执行） total_selected = specified_selected if total_selected < selection_quantity and req.status != "cancelled": _append_stage_log(db, request_id, user_id, 4, "过滤阶段-随机饰品：开始处理随机候选饰品") random_selected = 0 base_seq = len(specified_ids) # 序号接续 for seq_offset, goods_id in enumerate(random_ids, start=1): # 检查停止请求 if _check_stop_request(db, req, request_id): break # 检查名额 remaining, stop_reason = _get_remaining_quota( db, user_id, selection_quantity, items_limit_daily, today_start, total_selected ) if remaining <= 0: logger.info("随机饰品选品结束: request_id=%s reason=%s", request_id, stop_reason) break seq = base_seq + seq_offset selected, _, log_msg, price_cny, price_usd, wear_range = _process_single_item( db, request_id, user_id, goods_id, seq, profit_rate, buff_price_min, buff_price_max, assistant_level, steam_card_ratio, sweep_factor, total_selected ) display = _get_goods_display(db, goods_id) db.add(BuffSelectionRequestLog( request_id=request_id, user_id=user_id, goods_id=goods_id, goods_name=display.get("goods_name"), sequence_no=seq, result="selected" if selected else "not_selected", message=log_msg, wear_range=wear_range if selected else None, price_cny=price_cny if selected else None, price_usd=price_usd if selected else None, )) db.commit() if selected: db.add(BuffSelectionRequestItem( request_id=request_id, user_id=user_id, goods_id=goods_id, goods_name=display.get("goods_name"), icon_url=display.get("icon_url"), steam_link=display.get("steam_link"), price_cny=price_cny, price_usd=price_usd, wear_range=wear_range, )) db.commit() total_selected += 1 random_selected += 1 _append_stage_log(db, request_id, user_id, 5, f"过滤阶段-随机饰品：结束，共入选 {random_selected} 个随机饰品") _append_stage_log(db, request_id, user_id, 9999998, f"计算阶段：结束，共入选 {total_selected} 个饰品") if req.status != "cancelled": req.status = "completed" db.commit() _append_stage_log(db, request_id, user_id, 9999999, f"选品任务结束，状态：{req.status}") logger.info("选品完成: request_id=%s specified=%s random=%s total=%s", request_id, specified_selected, total_selected - specified_selected, total_selected) except Exception as e: logger.exception("选品任务异常: request_id=%s %s", request_id, e) try: req = db.query(BuffSelectionRequest).filter( BuffSelectionRequest.id == request_id, BuffSelectionRequest.user_id == user_id, ).first() if req: req.status = "failed" db.commit() except Exception: db.rollback() finally: db.rollback() finally: db.close()