""" 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]) -> List[str]: """ 第一步：根据基础过滤条件从 buff_goods_filtered_view 筛出候选 ID，合并指定 ID，排除排除 ID，乱序。 返回初始化要筛选的饰品 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")) all_ids = list(dict.fromkeys(candidate + specified)) random.shuffle(all_ids) return all_ids 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 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 = 0 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, "检索阶段：开始根据条件检索候选饰品") candidate_ids = step1_candidate_ids(db, config) _append_stage_log(db, request_id, user_id, 1, f"检索阶段：完成，共检索到 {len(candidate_ids)} 个候选饰品") _append_stage_log(db, request_id, user_id, 2, "过滤阶段：开始执行价格、智能选品与利润过滤") _append_stage_log(db, request_id, user_id, 3, "计算阶段：开始逐个计算区间与目标价格") selected_count = 0 for seq, goods_id in enumerate(candidate_ids, start=1): # 用户请求停止：重新加载请求单并检查 cancel_requested db.refresh(req) if getattr(req, "cancel_requested", False): logger.info("选品被用户请求停止: request_id=%s", request_id) req.status = "cancelled" db.commit() break # 本单选品数量达标则正常结束 if selection_quantity > 0 and selected_count >= selection_quantity: break # 当日选品总条数已达 system_config 上限则正常结束（不再为本单继续选） today_items = db.query(BuffSelectionRequestItem).filter( BuffSelectionRequestItem.user_id == user_id, BuffSelectionRequestItem.created_at >= today_start, ).count() if today_items >= items_limit_daily: logger.info("选品过程达到每日条数上限，正常结束: request_id=%s items_limit=%s", request_id, items_limit_daily) break lowest = _get_lowest_price(db, goods_id) if lowest is None or lowest <= 0: log_msg = "无在售最低价" db.add(BuffSelectionRequestLog( request_id=request_id, user_id=user_id, goods_id=goods_id, goods_name=None, sequence_no=seq, result="not_selected", message=log_msg, )) db.commit() continue if buff_price_min is not None and lowest < buff_price_min: display = _get_goods_display(db, goods_id) log_msg = f"当前在售最低价 {lowest} 低于设置下限 {buff_price_min}" 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="not_selected", message=log_msg, )) db.commit() continue if buff_price_max is not None and lowest > buff_price_max: display = _get_goods_display(db, goods_id) log_msg = f"当前在售最低价 {lowest} 高于设置上限 {buff_price_max}" 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="not_selected", message=log_msg, )) db.commit() continue assistant_ok, assistant_msg = _assistant_check(db, goods_id, assistant_level, lowest) if not assistant_ok: 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="not_selected", message=assistant_msg, )) db.commit() continue 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: log_msg = "无满足利润的磨损区间" 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="not_selected", message=log_msg, )) db.commit() continue # 入选 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}" # 扫货价 = 商品最低价 × (1 + 扫货系数) 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") 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_str, )) 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", message="已入选", wear_range=wear_range_str, price_cny=price_cny, price_usd=price_usd, )) db.commit() selected_count += 1 _append_stage_log(db, request_id, user_id, 9999998, f"计算阶段：结束，共入选 {selected_count} 个饰品") 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 selected=%s", request_id, selected_count) 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()