buff3.0/backend-client/app/services/buff_selection_service.py

"""
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()