buff_selection_service.py 26 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656
  1. """
  2. BUFF 选品服务:第一步筛选候选 ID + 第二步利润判断与落库。
  3. 逻辑见 docs/buff_selection_filter_logic.md、docs/buff_selection_step2_logic.md。
  4. """
  5. import json
  6. import random
  7. from decimal import Decimal
  8. from typing import List, Dict, Any, Optional, Tuple
  9. from sqlalchemy import text
  10. from sqlalchemy.orm import Session
  11. from app.database import SessionLocal
  12. from app.models.buff_selection_request import BuffSelectionRequest
  13. from app.models.buff_selection_request_item import BuffSelectionRequestItem
  14. from app.models.buff_selection_request_log import BuffSelectionRequestLog
  15. from app.models.buff_selection_config import BuffSelectionConfig
  16. from app.models.system_config import SystemConfig
  17. from app.utils.timezone import beijing_now_naive
  18. from app.core.logger import logger
  19. def _get_system_config_value(db: Session, config_key: str, app_name: str = "backend-selection") -> Optional[str]:
  20. """读取 system_config 配置值。"""
  21. row = db.query(SystemConfig).filter(
  22. SystemConfig.config_key == config_key,
  23. SystemConfig.app_name == app_name,
  24. SystemConfig.is_deleted == False,
  25. ).first()
  26. return row.config_value if row else None
  27. def _get_system_config_int(db: Session, config_key: str, default: int = 0, app_name: str = "backend-selection") -> int:
  28. """读取 system_config 整型配置。"""
  29. val = _get_system_config_value(db, config_key, app_name)
  30. if val is None or val == "":
  31. return default
  32. try:
  33. return int(val)
  34. except (ValueError, TypeError):
  35. return default
  36. def _parse_ids_field(raw: Any) -> List[str]:
  37. """将 excludeIds/specifiedIds(字符串或数组)解析为 ID 列表。"""
  38. if raw is None:
  39. return []
  40. if isinstance(raw, list):
  41. return [str(x).strip() for x in raw if str(x).strip()]
  42. s = str(raw).strip()
  43. if not s:
  44. return []
  45. return [x.strip() for x in s.replace(",", "\n").splitlines() if x.strip()]
  46. def _collect_gun_codes(config: Dict[str, Any]) -> List[str]:
  47. """从配置中收集所有枪械分类编码(步枪/手枪/SMG/霰弹枪/机枪)。"""
  48. codes = []
  49. for key in ("rifle", "pistol", "smg", "shotgun", "machinegun"):
  50. val = config.get(key)
  51. if isinstance(val, list):
  52. codes.extend([str(x).strip() for x in val if str(x).strip()])
  53. return list(dict.fromkeys(codes))
  54. def step1_candidate_ids(db: Session, config: Dict[str, Any]) -> Tuple[List[str], List[str]]:
  55. """
  56. 第一步:根据基础过滤条件从 buff_goods_filtered_view 筛出候选 ID,排除排除 ID。
  57. 返回两个列表:
  58. - specified_ids: 指定饰品 ID 列表(不乱序,保持用户指定的顺序)
  59. - random_ids: 随机饰品 ID 列表(已乱序)
  60. """
  61. conditions: List[str] = ["1=1"]
  62. params: Dict[str, Any] = {}
  63. # 枪械:都没选 -> 1=1;有选 -> 分类编码 IN (...)
  64. gun_codes = _collect_gun_codes(config)
  65. if gun_codes:
  66. placeholders = ", ".join([f":gun_{i}" for i in range(len(gun_codes))])
  67. conditions.append(f"`分类编码` IN ({placeholders})")
  68. for i, c in enumerate(gun_codes):
  69. params[f"gun_{i}"] = c
  70. # 品质
  71. quality = config.get("quality")
  72. if isinstance(quality, list) and quality:
  73. ph = ", ".join([f":quality_{i}" for i in range(len(quality))])
  74. conditions.append(f"`品质` IN ({ph})")
  75. for i, v in enumerate(quality):
  76. params[f"quality_{i}"] = str(v).strip()
  77. # 类别
  78. category = config.get("category")
  79. if isinstance(category, list) and category:
  80. ph = ", ".join([f":category_{i}" for i in range(len(category))])
  81. conditions.append(f"`类别` IN ({ph})")
  82. for i, v in enumerate(category):
  83. params[f"category_{i}"] = str(v).strip()
  84. # 外观
  85. exterior = config.get("exterior")
  86. if isinstance(exterior, list) and exterior:
  87. ph = ", ".join([f":exterior_{i}" for i in range(len(exterior))])
  88. conditions.append(f"`外观` IN ({ph})")
  89. for i, v in enumerate(exterior):
  90. params[f"exterior_{i}"] = str(v).strip()
  91. # 收藏品
  92. collection = config.get("collection")
  93. if isinstance(collection, list) and collection:
  94. ph = ", ".join([f":collection_{i}" for i in range(len(collection))])
  95. conditions.append(f"`收藏品` IN ({ph})")
  96. for i, v in enumerate(collection):
  97. params[f"collection_{i}"] = str(v).strip()
  98. # BUFF 价格区间
  99. pmin = config.get("buffPriceMin")
  100. pmax = config.get("buffPriceMax")
  101. if pmin is not None and pmax is not None:
  102. conditions.append("`当前在售最低价格` >= :buff_price_min AND `当前在售最低价格` <= :buff_price_max")
  103. params["buff_price_min"] = float(pmin)
  104. params["buff_price_max"] = float(pmax)
  105. elif pmin is not None:
  106. conditions.append("`当前在售最低价格` >= :buff_price_min")
  107. params["buff_price_min"] = float(pmin)
  108. elif pmax is not None:
  109. conditions.append("`当前在售最低价格` <= :buff_price_max")
  110. params["buff_price_max"] = float(pmax)
  111. # BUFF 在售数量
  112. smin = config.get("buffSalesMin")
  113. smax = config.get("buffSalesMax")
  114. if smin is not None and smax is not None:
  115. conditions.append("`在售数量` >= :buff_sales_min AND `在售数量` <= :buff_sales_max")
  116. params["buff_sales_min"] = int(smin)
  117. params["buff_sales_max"] = int(smax)
  118. elif smin is not None:
  119. conditions.append("`在售数量` >= :buff_sales_min")
  120. params["buff_sales_min"] = int(smin)
  121. elif smax is not None:
  122. conditions.append("`在售数量` <= :buff_sales_max")
  123. params["buff_sales_max"] = int(smax)
  124. # 排除的饰品 ID(从随机候选中排除,不影响指定饰品)
  125. exclude_ids = _parse_ids_field(config.get("excludeIds"))
  126. if exclude_ids:
  127. ph = ", ".join([f":exclude_{i}" for i in range(len(exclude_ids))])
  128. conditions.append(f"`BUFF饰品ID` NOT IN ({ph})")
  129. for i, v in enumerate(exclude_ids):
  130. params[f"exclude_{i}"] = v
  131. where_sql = " AND ".join(conditions)
  132. sql = f"SELECT DISTINCT `BUFF饰品ID` FROM buff_goods_filtered_view WHERE {where_sql}"
  133. try:
  134. rows = db.execute(text(sql), params).fetchall()
  135. candidate = [str(r[0]) for r in rows if r[0]]
  136. except Exception as e:
  137. logger.exception("step1 执行主查询异常: %s", e)
  138. candidate = []
  139. # 指定饰品 ID(不乱序,保持用户指定的顺序)
  140. specified = _parse_ids_field(config.get("specifiedIds"))
  141. # 随机候选饰品:排除指定饰品后乱序
  142. random_candidate = [gid for gid in candidate if gid not in specified]
  143. random.shuffle(random_candidate)
  144. return specified, random_candidate
  145. def _get_lowest_price(db: Session, goods_id: str) -> Optional[Decimal]:
  146. """获取饰品当前在售最低价格。"""
  147. sql = text(
  148. "SELECT MIN(`当前在售最低价格`) FROM buff_goods_filtered_view WHERE `BUFF饰品ID` = :gid"
  149. )
  150. row = db.execute(sql, {"gid": goods_id}).fetchone()
  151. if row and row[0] is not None:
  152. return Decimal(str(row[0]))
  153. return None
  154. def _get_wear_ranges(db: Session, goods_id: str) -> List[Tuple[Decimal, Decimal, Decimal]]:
  155. """获取磨损区间列表:(最小磨损值, 最大磨损值, 区间底价),按最小磨损值升序。"""
  156. sql = text(
  157. "SELECT `最小磨损值`, `最大磨损值`, `区间底价` FROM buff_goods_filtered_view "
  158. "WHERE `BUFF饰品ID` = :gid ORDER BY `最小磨损值` ASC"
  159. )
  160. rows = db.execute(sql, {"gid": goods_id}).fetchall()
  161. out = []
  162. for r in rows:
  163. if r[0] is not None and r[1] is not None and r[2] is not None:
  164. out.append((Decimal(str(r[0])), Decimal(str(r[1])), Decimal(str(r[2]))))
  165. return out
  166. def _get_ai_price_avgs(db: Session, goods_id: str) -> Tuple[Optional[Decimal], Optional[Decimal]]:
  167. """获取 7天/30天在售最低价格平均价。"""
  168. sql = text(
  169. "SELECT `7天在售最低价格平均价`, `30天在售最低价格平均价` "
  170. "FROM buff_goods_filtered_view WHERE `BUFF饰品ID` = :gid LIMIT 1"
  171. )
  172. row = db.execute(sql, {"gid": goods_id}).fetchone()
  173. if not row:
  174. return None, None
  175. avg7 = Decimal(str(row[0])) if row[0] is not None else None
  176. avg30 = Decimal(str(row[1])) if row[1] is not None else None
  177. return avg7, avg30
  178. def _assistant_check(
  179. db: Session, goods_id: str, assistant_level: str, current_lowest: Decimal
  180. ) -> Tuple[bool, str]:
  181. """
  182. 第二步智能选品判断:
  183. - low: 不限制
  184. - standard: 7天均价相对当前价需在配置涨跌幅范围内
  185. - high: 30天均价相对当前价需在配置涨跌幅范围内
  186. """
  187. level = (assistant_level or "low").lower()
  188. if level == "low":
  189. return True, "通过"
  190. avg7, avg30 = _get_ai_price_avgs(db, goods_id)
  191. if level == "standard":
  192. if avg7 is None:
  193. return False, "7天均价缺失,智能选品标准模式不满足"
  194. inc = Decimal(str(_get_system_config_value(db, "buff_ai_standard_range_max_increase_ratio") or "0.3"))
  195. dec = Decimal(str(_get_system_config_value(db, "buff_ai_standard_range_max_decrease_ratio") or "0.5"))
  196. upper = current_lowest * (Decimal("1") + inc)
  197. lower = current_lowest * (Decimal("1") - dec)
  198. if avg7 > upper or avg7 < lower:
  199. return False, f"7天均价 {avg7} 不在智能选品标准范围 [{lower}, {upper}]"
  200. return True, "通过"
  201. if level == "high":
  202. if avg30 is None:
  203. return False, "30天均价缺失,智能选品高模式不满足"
  204. inc = Decimal(str(_get_system_config_value(db, "buff_ai_high_range_max_increase_ratio") or "0.4"))
  205. dec = Decimal(str(_get_system_config_value(db, "buff_ai_high_range_max_decrease_ratio") or "0.5"))
  206. upper = current_lowest * (Decimal("1") + inc)
  207. lower = current_lowest * (Decimal("1") - dec)
  208. if avg30 > upper or avg30 < lower:
  209. return False, f"30天均价 {avg30} 不在智能选品高范围 [{lower}, {upper}]"
  210. return True, "通过"
  211. # 兜底:未知值按 low 处理,避免阻断任务
  212. return True, "通过"
  213. def _normal_ranges(ranges: List[Tuple[Decimal, Decimal, Decimal]]) -> List[Tuple[Decimal, Decimal, Decimal]]:
  214. """正常磨损区间:区间底价随最小磨损值递增而单调不升。"""
  215. if not ranges:
  216. return []
  217. out = [ranges[0]]
  218. for i in range(1, len(ranges)):
  219. if ranges[i][2] <= out[-1][2]:
  220. out.append(ranges[i])
  221. else:
  222. break
  223. return out
  224. def _qualified_ranges(
  225. normal: List[Tuple[Decimal, Decimal, Decimal]], compare_price: Decimal
  226. ) -> List[Tuple[Decimal, Decimal, Decimal]]:
  227. """满足利润的区间:区间底价 > 比较价,一旦不满足则截断。"""
  228. out = []
  229. for r in normal:
  230. if r[2] > compare_price:
  231. out.append(r)
  232. else:
  233. break
  234. return out
  235. def _get_goods_display(db: Session, goods_id: str) -> Dict[str, Any]:
  236. """从视图取饰品名称、图片、Steam 链接(取一行即可)。"""
  237. sql = text(
  238. "SELECT `BUFF饰品名称`, `饰品图片URL地址`, `饰品STEAM链接` FROM buff_goods_filtered_view "
  239. "WHERE `BUFF饰品ID` = :gid LIMIT 1"
  240. )
  241. row = db.execute(sql, {"gid": goods_id}).fetchone()
  242. if row:
  243. return {
  244. "goods_name": (row[0] or "").strip() or None,
  245. "icon_url": (row[1] or "").strip() or None,
  246. "steam_link": (row[2] or "").strip() or None,
  247. }
  248. return {"goods_name": None, "icon_url": None, "steam_link": None}
  249. def _sweep_factor(db: Session, level: str) -> Decimal:
  250. """扫货价格系数:low->min, standard->mid, high->max。"""
  251. key = {
  252. "low": "buff_sweep_price_factor_min",
  253. "standard": "buff_sweep_price_factor_mid",
  254. "high": "buff_sweep_price_factor_max",
  255. }.get((level or "low").lower(), "buff_sweep_price_factor_min")
  256. val = _get_system_config_value(db, key) or "0"
  257. return Decimal(str(val))
  258. def _append_stage_log(
  259. db: Session,
  260. request_id: int,
  261. user_id: int,
  262. sequence_no: int,
  263. message: str,
  264. ) -> None:
  265. """追加选品阶段日志(检索/过滤/计算等),供前端日志弹窗展示。"""
  266. db.add(BuffSelectionRequestLog(
  267. request_id=request_id,
  268. user_id=user_id,
  269. goods_id="__stage__",
  270. goods_name=None,
  271. sequence_no=sequence_no,
  272. result="stage",
  273. message=message,
  274. ))
  275. db.commit()
  276. def _get_remaining_quota(
  277. db: Session,
  278. user_id: int,
  279. selection_quantity: int,
  280. items_limit_daily: int,
  281. today_start,
  282. selected_count: int,
  283. ) -> Tuple[int, str]:
  284. """
  285. 计算剩余名额和是否应该结束选品。
  286. 返回 (remaining_quota, stop_reason)
  287. - remaining_quota: 剩余可入选的名额(0 表示已达上限)
  288. - stop_reason: 停止原因(空字符串表示可以继续)
  289. """
  290. # 检查单次选品上限
  291. if selection_quantity > 0 and selected_count >= selection_quantity:
  292. return 0, f"已达单次选品上限 {selection_quantity} 条"
  293. # 检查每日上限
  294. today_items = db.query(BuffSelectionRequestItem).filter(
  295. BuffSelectionRequestItem.user_id == user_id,
  296. BuffSelectionRequestItem.created_at >= today_start,
  297. ).count()
  298. if today_items >= items_limit_daily:
  299. return 0, f"已达每日选品上限 {items_limit_daily} 条"
  300. # 计算剩余名额
  301. remaining = selection_quantity - selected_count if selection_quantity > 0 else 999999
  302. daily_remaining = items_limit_daily - today_items
  303. return min(remaining, daily_remaining), ""
  304. def _check_stop_request(db: Session, req: BuffSelectionRequest, request_id: int) -> bool:
  305. """检查是否收到用户停止请求。"""
  306. db.refresh(req)
  307. if getattr(req, "cancel_requested", False):
  308. logger.info("选品被用户请求停止: request_id=%s", request_id)
  309. req.status = "cancelled"
  310. db.commit()
  311. return True
  312. return False
  313. def _process_single_item(
  314. db: Session,
  315. request_id: int,
  316. user_id: int,
  317. goods_id: str,
  318. seq: int,
  319. profit_rate: float,
  320. buff_price_min: Optional[Decimal],
  321. buff_price_max: Optional[Decimal],
  322. assistant_level: str,
  323. steam_card_ratio: float,
  324. sweep_factor: Decimal,
  325. selected_count: int,
  326. ) -> Tuple[bool, int, str, Optional[Decimal], Optional[Decimal], Optional[str]]:
  327. """
  328. 处理单个饰品,判断是否满足所有条件并入选。
  329. 返回:
  330. - selected: 是否入选
  331. - new_selected_count: 更新后的已入选数量
  332. - log_msg: 日志消息
  333. - price_cny: 人民币价格(入选时有值)
  334. - price_usd: 美元价格(入选时有值)
  335. - wear_range_str: 磨损区间字符串(入选时有值)
  336. """
  337. lowest = _get_lowest_price(db, goods_id)
  338. if lowest is None or lowest <= 0:
  339. return False, selected_count, "无在售最低价", None, None, None
  340. if buff_price_min is not None and lowest < buff_price_min:
  341. return False, selected_count, f"当前在售最低价 {lowest} 低于设置下限 {buff_price_min}", None, None, None
  342. if buff_price_max is not None and lowest > buff_price_max:
  343. return False, selected_count, f"当前在售最低价 {lowest} 高于设置上限 {buff_price_max}", None, None, None
  344. assistant_ok, assistant_msg = _assistant_check(db, goods_id, assistant_level, lowest)
  345. if not assistant_ok:
  346. return False, selected_count, assistant_msg, None, None, None
  347. compare_price = lowest * (Decimal("1") + Decimal(str(profit_rate)))
  348. ranges = _get_wear_ranges(db, goods_id)
  349. normal = _normal_ranges(ranges)
  350. qualified = _qualified_ranges(normal, compare_price)
  351. if not qualified:
  352. return False, selected_count, "无满足利润的磨损区间", None, None, None
  353. # 入选
  354. display = _get_goods_display(db, goods_id)
  355. min_min = min(r[0] for r in qualified)
  356. max_max = max(r[1] for r in qualified)
  357. wear_range_str = f"{min_min}-{max_max}"
  358. price_cny = lowest * (Decimal("1") + sweep_factor)
  359. price_usd_raw = float(price_cny) / steam_card_ratio
  360. price_usd = Decimal(str(round(price_usd_raw, 2))) + Decimal("0.01")
  361. return True, selected_count + 1, "已入选", price_cny, price_usd, wear_range_str
  362. def run_selection_task(request_id: int, user_id: int, config_id: Optional[int]) -> None:
  363. """
  364. 后台选品任务:更新状态为 running,执行 step1 + step2,最后更新为 completed/failed。
  365. 使用独立 db 会话,在请求返回后执行。
  366. """
  367. db = SessionLocal()
  368. try:
  369. req = db.query(BuffSelectionRequest).filter(
  370. BuffSelectionRequest.id == request_id,
  371. BuffSelectionRequest.user_id == user_id,
  372. ).first()
  373. if not req:
  374. logger.warning("选品请求单不存在或非本人: request_id=%s user_id=%s", request_id, user_id)
  375. return
  376. if req.status != "pending":
  377. logger.info("选品请求单已非 pending,跳过: request_id=%s status=%s", request_id, req.status)
  378. return
  379. req.status = "running"
  380. db.commit()
  381. # 处理竞态:如果在 pending->running 切换窗口已收到停止请求,立即结束任务
  382. db.refresh(req)
  383. if getattr(req, "cancel_requested", False):
  384. logger.info("选品在启动阶段被用户请求停止: request_id=%s", request_id)
  385. req.status = "cancelled"
  386. db.commit()
  387. return
  388. config = {}
  389. if config_id:
  390. cfg_row = db.query(BuffSelectionConfig).filter(
  391. BuffSelectionConfig.id == config_id,
  392. BuffSelectionConfig.user_id == user_id,
  393. BuffSelectionConfig.is_deleted == False,
  394. ).first()
  395. if cfg_row and cfg_row.config_json:
  396. try:
  397. config = json.loads(cfg_row.config_json)
  398. except Exception:
  399. pass
  400. if not config:
  401. logger.warning("选品配置为空,请求单标记为失败: request_id=%s", request_id)
  402. req.status = "failed"
  403. db.commit()
  404. return
  405. profit_rate = config.get("profitRate")
  406. if profit_rate is None:
  407. profit_rate = 0
  408. profit_rate = float(profit_rate) / 100.0 # 百分比转小数
  409. steam_card_ratio = config.get("steamCardRatio")
  410. if steam_card_ratio is None or float(steam_card_ratio) <= 0:
  411. steam_card_ratio = 1.0
  412. else:
  413. steam_card_ratio = float(steam_card_ratio)
  414. selection_quantity = config.get("selectionQuantity")
  415. if selection_quantity is None:
  416. # 用户未配置,使用系统配置的单次上限
  417. selection_quantity = _get_system_config_int(db, "buff_selection_max_quantity", 300)
  418. else:
  419. selection_quantity = int(selection_quantity)
  420. assistant_level = (config.get("assistantLevel") or "low").lower()
  421. # BUFF 价格区间兜底校验:即使 step1 候选因数据异常混入,也在 step2 最终入选前拦截
  422. raw_price_min = config.get("buffPriceMin")
  423. raw_price_max = config.get("buffPriceMax")
  424. try:
  425. buff_price_min = Decimal(str(raw_price_min)) if raw_price_min is not None else None
  426. except Exception:
  427. buff_price_min = None
  428. try:
  429. buff_price_max = Decimal(str(raw_price_max)) if raw_price_max is not None else None
  430. except Exception:
  431. buff_price_max = None
  432. # 每日选品条数上限(system_config),选品过程中达到则本单正常结束
  433. today_start = beijing_now_naive().replace(hour=0, minute=0, second=0, microsecond=0)
  434. items_limit_daily = _get_system_config_int(db, "buff_selection_daily_items_limit_per_user", 900)
  435. sweep_level = (config.get("sweepAssistantLevel") or config.get("assistantLevel") or "low").lower()
  436. sweep_factor = _sweep_factor(db, sweep_level)
  437. # 第一步:检索候选
  438. _append_stage_log(db, request_id, user_id, 0, "检索阶段:开始根据条件检索候选饰品")
  439. specified_ids, random_ids = step1_candidate_ids(db, config)
  440. _append_stage_log(db, request_id, user_id, 1,
  441. f"检索阶段:完成,指定饰品 {len(specified_ids)} 个,随机候选 {len(random_ids)} 个")
  442. # 第二步:分两阶段选品
  443. # 阶段1:处理指定饰品(不乱序,优先入选)
  444. _append_stage_log(db, request_id, user_id, 2, "过滤阶段-指定饰品:开始处理指定饰品")
  445. specified_selected = 0
  446. for seq, goods_id in enumerate(specified_ids, start=1):
  447. # 检查停止请求
  448. if _check_stop_request(db, req, request_id):
  449. break
  450. # 检查名额
  451. remaining, stop_reason = _get_remaining_quota(
  452. db, user_id, selection_quantity, items_limit_daily, today_start, specified_selected
  453. )
  454. if remaining <= 0:
  455. logger.info("指定饰品选品结束: request_id=%s reason=%s", request_id, stop_reason)
  456. break
  457. selected, specified_selected, log_msg, price_cny, price_usd, wear_range = _process_single_item(
  458. db, request_id, user_id, goods_id, seq, profit_rate,
  459. buff_price_min, buff_price_max, assistant_level,
  460. steam_card_ratio, sweep_factor, specified_selected
  461. )
  462. display = _get_goods_display(db, goods_id)
  463. db.add(BuffSelectionRequestLog(
  464. request_id=request_id,
  465. user_id=user_id,
  466. goods_id=goods_id,
  467. goods_name=display.get("goods_name"),
  468. sequence_no=seq,
  469. result="selected" if selected else "not_selected",
  470. message=log_msg,
  471. wear_range=wear_range if selected else None,
  472. price_cny=price_cny if selected else None,
  473. price_usd=price_usd if selected else None,
  474. ))
  475. db.commit()
  476. if selected:
  477. db.add(BuffSelectionRequestItem(
  478. request_id=request_id,
  479. user_id=user_id,
  480. goods_id=goods_id,
  481. goods_name=display.get("goods_name"),
  482. icon_url=display.get("icon_url"),
  483. steam_link=display.get("steam_link"),
  484. price_cny=price_cny,
  485. price_usd=price_usd,
  486. wear_range=wear_range,
  487. ))
  488. db.commit()
  489. _append_stage_log(db, request_id, user_id, 3,
  490. f"过滤阶段-指定饰品:结束,共入选 {specified_selected} 个指定饰品")
  491. # 阶段2:处理随机饰品(仅在指定饰品未占满名额时执行)
  492. total_selected = specified_selected
  493. if total_selected < selection_quantity and req.status != "cancelled":
  494. _append_stage_log(db, request_id, user_id, 4, "过滤阶段-随机饰品:开始处理随机候选饰品")
  495. random_selected = 0
  496. base_seq = len(specified_ids) # 序号接续
  497. for seq_offset, goods_id in enumerate(random_ids, start=1):
  498. # 检查停止请求
  499. if _check_stop_request(db, req, request_id):
  500. break
  501. # 检查名额
  502. remaining, stop_reason = _get_remaining_quota(
  503. db, user_id, selection_quantity, items_limit_daily, today_start, total_selected
  504. )
  505. if remaining <= 0:
  506. logger.info("随机饰品选品结束: request_id=%s reason=%s", request_id, stop_reason)
  507. break
  508. seq = base_seq + seq_offset
  509. selected, _, log_msg, price_cny, price_usd, wear_range = _process_single_item(
  510. db, request_id, user_id, goods_id, seq, profit_rate,
  511. buff_price_min, buff_price_max, assistant_level,
  512. steam_card_ratio, sweep_factor, total_selected
  513. )
  514. display = _get_goods_display(db, goods_id)
  515. db.add(BuffSelectionRequestLog(
  516. request_id=request_id,
  517. user_id=user_id,
  518. goods_id=goods_id,
  519. goods_name=display.get("goods_name"),
  520. sequence_no=seq,
  521. result="selected" if selected else "not_selected",
  522. message=log_msg,
  523. wear_range=wear_range if selected else None,
  524. price_cny=price_cny if selected else None,
  525. price_usd=price_usd if selected else None,
  526. ))
  527. db.commit()
  528. if selected:
  529. db.add(BuffSelectionRequestItem(
  530. request_id=request_id,
  531. user_id=user_id,
  532. goods_id=goods_id,
  533. goods_name=display.get("goods_name"),
  534. icon_url=display.get("icon_url"),
  535. steam_link=display.get("steam_link"),
  536. price_cny=price_cny,
  537. price_usd=price_usd,
  538. wear_range=wear_range,
  539. ))
  540. db.commit()
  541. total_selected += 1
  542. random_selected += 1
  543. _append_stage_log(db, request_id, user_id, 5,
  544. f"过滤阶段-随机饰品:结束,共入选 {random_selected} 个随机饰品")
  545. _append_stage_log(db, request_id, user_id, 9999998, f"计算阶段:结束,共入选 {total_selected} 个饰品")
  546. if req.status != "cancelled":
  547. req.status = "completed"
  548. db.commit()
  549. _append_stage_log(db, request_id, user_id, 9999999, f"选品任务结束,状态:{req.status}")
  550. logger.info("选品完成: request_id=%s specified=%s random=%s total=%s",
  551. request_id, specified_selected, total_selected - specified_selected, total_selected)
  552. except Exception as e:
  553. logger.exception("选品任务异常: request_id=%s %s", request_id, e)
  554. try:
  555. req = db.query(BuffSelectionRequest).filter(
  556. BuffSelectionRequest.id == request_id,
  557. BuffSelectionRequest.user_id == user_id,
  558. ).first()
  559. if req:
  560. req.status = "failed"
  561. db.commit()
  562. except Exception:
  563. db.rollback()
  564. finally:
  565. db.rollback()
  566. finally:
  567. db.close()