""" 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 # 智能选品助手（低=1=1；标准=7天；高=30天） assistant = (config.get("assistantLevel") or "low").lower() if assistant == "standard": inc = _get_system_config_value(db, "buff_ai_standard_range_max_increase_ratio") or "0.3" dec = _get_system_config_value(db, "buff_ai_standard_range_max_decrease_ratio") or "0.5" conditions.append( "`7天在售最低价格平均价` <= `当前在售最低价格` * (1 + :ai_std_inc) " "AND `7天在售最低价格平均价` >= `当前在售最低价格` * (1 - :ai_std_dec)" ) params["ai_std_inc"] = float(inc) params["ai_std_dec"] = float(dec) elif assistant == "high": inc = _get_system_config_value(db, "buff_ai_high_range_max_increase_ratio") or "0.4" dec = _get_system_config_value(db, "buff_ai_high_range_max_decrease_ratio") or "0.5" conditions.append( "`30天在售最低价格平均价` <= `当前在售最低价格` * (1 + :ai_high_inc) " "AND `30天在售最低价格平均价` >= `当前在售最低价格` * (1 - :ai_high_dec)" ) params["ai_high_inc"] = float(inc) params["ai_high_dec"] = float(dec) 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 DISTINCT `当前在售最低价格` 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 _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 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() 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) # 每日选品条数上限（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) # 第一步 candidate_ids = step1_candidate_ids(db, config) 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 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}" price_cny = compare_price * (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="已入选", )) db.commit() selected_count += 1 req.status = "completed" db.commit() 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()