buff3.0/backend-client-bak/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]) -> 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()