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

"""
BUFF 扫货服务（真实链路版）：
- Steam listings 分页拉取
- csgofloat 磨损校验
- buylisting 购买
- trade-confirmations/operate + buylisting(confirmation) 购买确认
- 多 Steam 账号余额切换与失败重试
"""
import json
import os
import random
import time
from dataclasses import dataclass
from decimal import Decimal, InvalidOperation
from typing import Any, Dict, List, Optional, Tuple
from urllib.parse import urlencode
from urllib.request import Request, build_opener, ProxyHandler
from urllib.error import HTTPError, URLError

from sqlalchemy.orm import Session

from app.database import SessionLocal
from app.models.buff_sweep_request import BuffSweepRequest
from app.models.buff_sweep_request_item import BuffSweepRequestItem
from app.models.buff_sweep_request_log import BuffSweepRequestLog
from app.models.buff_selection_request_item import BuffSelectionRequestItem
from app.models.steam_account import SteamAccount
from app.models.system_config import SystemConfig
from app.core.logger import logger


STEAM_API_BASE = "http://localhost:5000"
DEFAULT_PAGE_SIZE = 10
DEFAULT_MAX_SCAN_PER_ITEM = 60
MIN_BALANCE_CENTS = 50  # 0.5 USD


@dataclass
class LocalSteamAccount:
    steam_username: str
    mafile_path: str
    steam_id: int
    access_token: str
    refresh_token: str
    session_id: str
    steam_login_secure: str
    balance_cents: int = 0


@dataclass
class ProxyContext:
    mode: str  # none/account/api
    proxy_url: Optional[str] = None
    api_url: Optional[str] = None
    proxy_retry_times: int = 1
    proxy_retry_interval_ms: int = 800


def _get_system_config_int(db: Session, key: str, default: int) -> int:
    row = db.query(SystemConfig).filter(
        SystemConfig.config_key == key,
        SystemConfig.app_name == "backend-selection",
        SystemConfig.is_deleted == False,
    ).first()
    if not row or row.config_value in (None, ""):
        return default
    try:
        return int(row.config_value)
    except (ValueError, TypeError):
        return default


def _http_request(
    method: str,
    url: str,
    headers: Optional[Dict[str, str]] = None,
    body: Optional[bytes] = None,
    timeout: int = 25,
    proxy_url: Optional[str] = None,
) -> Tuple[int, str]:
    req = Request(url=url, method=method.upper(), data=body, headers=headers or {})
    opener = build_opener(ProxyHandler({"http": proxy_url, "https": proxy_url})) if proxy_url else build_opener()
    try:
        with opener.open(req, timeout=timeout) as resp:
            text = resp.read().decode("utf-8", errors="replace")
            return resp.getcode(), text
    except HTTPError as e:
        return e.code, e.read().decode("utf-8", errors="replace")
    except URLError as e:
        raise RuntimeError(f"请求失败: {e}") from e


def _post_json(
    url: str,
    payload: Dict[str, Any],
    timeout: int = 25,
) -> Tuple[int, Dict[str, Any], str]:
    body = json.dumps(payload).encode("utf-8")
    code, text = _http_request(
        method="POST",
        url=url,
        headers={"Content-Type": "application/json"},
        body=body,
        timeout=timeout,
        proxy_url=None,
    )
    try:
        parsed = json.loads(text) if text else {}
    except Exception:
        parsed = {}
    return code, parsed, text


def _parse_proxy_api_response(body: str) -> Optional[str]:
    txt = (body or "").strip()
    if not txt:
        return None
    try:
        obj = json.loads(txt)
    except Exception:
        obj = None
    if isinstance(obj, str):
        return obj if "://" in obj else f"http://{obj}"
    if isinstance(obj, dict):
        for k in ("proxyUrl", "proxy_url", "url", "proxy"):
            v = obj.get(k)
            if isinstance(v, str) and v.strip():
                return v if "://" in v else f"http://{v}"
        host = obj.get("ip") or obj.get("host") or obj.get("server")
        port = obj.get("port") or obj.get("serverPort")
        if host and port:
            return f"http://{host}:{port}"
    first = txt.splitlines()[0].split(",")[0].strip()
    if not first:
        return None
    return first if "://" in first else f"http://{first}"


def _is_proxy_related_failure(status_code: Optional[int], body: Optional[str], err_msg: Optional[str] = None) -> bool:
    if status_code in (407, 502, 503, 504):
        return True
    text = f"{body or ''} {err_msg or ''}".lower()
    tokens = [
        "proxy",
        "tunnel",
        "timed out",
        "timeout",
        "connection refused",
        "connection reset",
        "failed to connect",
        "unreachable",
        "network is unreachable",
        "name or service not known",
    ]
    return any(t in text for t in tokens)


def _resolve_steam_msg_dir() -> Optional[str]:
    candidates = []
    env_dir = (os.environ.get("STEAM_MSG_DIR") or "").strip()
    if env_dir:
        candidates.append(env_dir)

    home = os.path.expanduser("~")
    candidates.extend([
        os.path.join(home, "Library", "Application Support", "steam-msg"),
        os.path.join(home, "AppData", "Roaming", "steam-msg"),
        os.path.join(os.getcwd(), "steam-msg"),
    ])

    # backend-client/app/services -> project root
    project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", ".."))
    candidates.append(os.path.join(project_root, "steam-msg"))

    for c in candidates:
        if c and os.path.isdir(c):
            return c
    return None


def _resolve_proxy_context() -> ProxyContext:
    steam_msg_dir = _resolve_steam_msg_dir()
    if not steam_msg_dir:
        return ProxyContext(mode="none")
    cfg_path = os.path.join(steam_msg_dir, "proxy_config.json")
    if not os.path.exists(cfg_path):
        return ProxyContext(mode="none")
    try:
        with open(cfg_path, "r", encoding="utf-8") as f:
            cfg = json.load(f)
    except Exception:
        return ProxyContext(mode="none")
    mode = (cfg.get("effectiveMode") or "").strip().lower()
    if mode not in ("api", "account"):
        return ProxyContext(mode="none")
    brand_id = cfg.get("effectiveBrandId")
    brand = (cfg.get("configs") or {}).get(brand_id) if brand_id else None
    if not isinstance(brand, dict):
        return ProxyContext(mode="none")
    if mode == "account":
        host = (brand.get("serverHost") or "").strip()
        port = int(brand.get("serverPort") or 0)
        if not host or port <= 0:
            return ProxyContext(mode="none")
        user = (brand.get("serverUsername") or "").strip()
        pwd = brand.get("serverPassword") or ""
        if user:
            return ProxyContext(mode="account", proxy_url=f"http://{user}:{pwd}@{host}:{port}")
        return ProxyContext(mode="account", proxy_url=f"http://{host}:{port}")

    api_url = (brand.get("apiUrl") or "").strip()
    if not api_url:
        return ProxyContext(mode="none")
    try:
        code, text = _http_request("GET", api_url, timeout=12)
        if code < 200 or code >= 300:
            return ProxyContext(mode="api", api_url=api_url, proxy_url=None)
        return ProxyContext(mode="api", api_url=api_url, proxy_url=_parse_proxy_api_response(text))
    except Exception:
        return ProxyContext(mode="api", api_url=api_url, proxy_url=None)


def _refresh_api_proxy(ctx: ProxyContext) -> bool:
    if ctx.mode != "api" or not ctx.api_url:
        return False
    try:
        # 加时间戳避免代理平台缓存同一条
        sep = "&" if "?" in ctx.api_url else "?"
        refresh_url = f"{ctx.api_url}{sep}_ts={random.randint(100000, 999999)}"
        code, text = _http_request("GET", refresh_url, timeout=12)
        if code < 200 or code >= 300:
            return False
        parsed = _parse_proxy_api_response(text)
        if not parsed:
            return False
        ctx.proxy_url = parsed
        return True
    except Exception:
        return False


def _ensure_proxy_ready(proxy_ctx: Optional[ProxyContext], refresh_if_needed: bool = True) -> Tuple[bool, str]:
    """扫货链路强制代理：没有可用代理时必须失败，禁止直连 Steam。"""
    if not proxy_ctx or proxy_ctx.mode == "none":
        return False, "未配置可用代理（扫货链路禁止直连 Steam）"
    if proxy_ctx.mode == "api" and not proxy_ctx.proxy_url and refresh_if_needed:
        _refresh_api_proxy(proxy_ctx)
    if not proxy_ctx.proxy_url:
        return False, f"代理模式={proxy_ctx.mode} 但未获取到可用代理地址（扫货链路禁止直连 Steam）"
    return True, ""


def _http_request_with_proxy_retry(
    method: str,
    url: str,
    headers: Optional[Dict[str, str]] = None,
    body: Optional[bytes] = None,
    timeout: int = 25,
    proxy_ctx: Optional[ProxyContext] = None,
    debug_meta: Optional[Dict[str, Any]] = None,
) -> Tuple[int, str]:
    ok, err = _ensure_proxy_ready(proxy_ctx, refresh_if_needed=True)
    if not ok:
        raise RuntimeError(err)
    proxy_url = proxy_ctx.proxy_url if proxy_ctx else None
    retry_times = max(0, int(proxy_ctx.proxy_retry_times if proxy_ctx else 0))
    retry_interval = max(0, int(proxy_ctx.proxy_retry_interval_ms if proxy_ctx else 0))

    last_err: Optional[Exception] = None
    code: Optional[int] = None
    text: Optional[str] = None
    for attempt in range(retry_times + 1):
        try:
            code, text = _http_request(
                method,
                url,
                headers=headers,
                body=body,
                timeout=timeout,
                proxy_url=proxy_ctx.proxy_url if proxy_ctx else proxy_url,
            )
            if not (
                proxy_ctx
                and proxy_ctx.mode == "api"
                and _is_proxy_related_failure(code, text)
                and attempt < retry_times
            ):
                return code, text
            if debug_meta is not None:
                debug_meta["retry_attempts"] = int(debug_meta.get("retry_attempts", 0)) + 1
            if retry_interval > 0:
                time.sleep(retry_interval / 1000.0)
            refreshed = _refresh_api_proxy(proxy_ctx)
            if debug_meta is not None and refreshed:
                debug_meta["proxy_refreshed"] = True
        except Exception as e:
            last_err = e
            if not (
                proxy_ctx
                and proxy_ctx.mode == "api"
                and _is_proxy_related_failure(None, None, str(e))
                and attempt < retry_times
            ):
                raise
            if debug_meta is not None:
                debug_meta["retry_attempts"] = int(debug_meta.get("retry_attempts", 0)) + 1
            if retry_interval > 0:
                time.sleep(retry_interval / 1000.0)
            refreshed = _refresh_api_proxy(proxy_ctx)
            if debug_meta is not None and refreshed:
                debug_meta["proxy_refreshed"] = True

    if code is not None and text is not None:
        return code, text
    if last_err:
        raise last_err
    raise RuntimeError("代理重试后请求失败")


def _post_steam_api_with_proxy_retry(
    path: str,
    payload: Dict[str, Any],
    proxy_ctx: Optional[ProxyContext],
    timeout: int = 25,
    debug_meta: Optional[Dict[str, Any]] = None,
) -> Tuple[int, Dict[str, Any], str]:
    url = f"{STEAM_API_BASE.rstrip('/')}/{path.lstrip('/')}"
    ok, err = _ensure_proxy_ready(proxy_ctx, refresh_if_needed=True)
    if not ok:
        raise RuntimeError(err)
    retry_times = max(0, int(proxy_ctx.proxy_retry_times if proxy_ctx else 0))
    retry_interval = max(0, int(proxy_ctx.proxy_retry_interval_ms if proxy_ctx else 0))

    last_code = 0
    last_text = ""
    last_parsed: Dict[str, Any] = {}
    for attempt in range(retry_times + 1):
        req_payload = dict(payload)
        if proxy_ctx and proxy_ctx.proxy_url:
            req_payload["ProxyUrl"] = proxy_ctx.proxy_url
        code, parsed, text = _post_json(url, req_payload, timeout=timeout)
        last_code, last_text, last_parsed = code, text, parsed
        if not (
            proxy_ctx
            and proxy_ctx.mode == "api"
            and _is_proxy_related_failure(code, text)
            and attempt < retry_times
        ):
            return code, parsed, text
        if debug_meta is not None:
            debug_meta["retry_attempts"] = int(debug_meta.get("retry_attempts", 0)) + 1
        if retry_interval > 0:
            time.sleep(retry_interval / 1000.0)
        refreshed = _refresh_api_proxy(proxy_ctx)
        if debug_meta is not None and refreshed:
            debug_meta["proxy_refreshed"] = True

    return last_code, last_parsed, last_text


def _load_accounts_json() -> Dict[str, Any]:
    steam_msg_dir = _resolve_steam_msg_dir()
    if not steam_msg_dir:
        return {}
    path = os.path.join(steam_msg_dir, "accounts.json")
    if not os.path.exists(path):
        return {}
    try:
        with open(path, "r", encoding="utf-8") as f:
            root = json.load(f)
        return root.get("accounts") if isinstance(root.get("accounts"), dict) else {}
    except Exception:
        return {}


def _build_local_account(steam_username: str, raw: Dict[str, Any]) -> Optional[LocalSteamAccount]:
    try:
        steam_id = int(raw.get("steamId") or 0)
    except Exception:
        steam_id = 0
    mafile = (raw.get("steam_guard_path") or "").strip()
    if not mafile or not os.path.exists(mafile):
        return None
    access_token = (raw.get("accessToken") or "").strip()
    refresh_token = (raw.get("refreshToken") or "").strip()
    steam_login_secure = (raw.get("steamLoginSecure") or "").strip()
    if steam_id <= 0 or not access_token or not refresh_token or not steam_login_secure:
        return None
    return LocalSteamAccount(
        steam_username=steam_username,
        mafile_path=mafile,
        steam_id=steam_id,
        access_token=access_token,
        refresh_token=refresh_token,
        session_id=(raw.get("sessionId") or "").strip(),
        steam_login_secure=steam_login_secure,
    )


def _load_active_steam_accounts(db: Session, user_id: int) -> List[LocalSteamAccount]:
    local_accounts = _load_accounts_json()
    rows = (
        db.query(SteamAccount)
        .filter(
            SteamAccount.user_id == user_id,
            SteamAccount.is_deleted == False,
            SteamAccount.is_active == True,
        )
        .order_by(SteamAccount.id.asc())
        .all()
    )
    out: List[LocalSteamAccount] = []
    for r in rows:
        raw = local_accounts.get(r.steam_username)
        if not isinstance(raw, dict):
            continue
        acc = _build_local_account(r.steam_username, raw)
        if acc:
            out.append(acc)
    return out


def _refresh_wallet_balance(
    account: LocalSteamAccount,
    proxy_ctx: Optional[ProxyContext],
    debug_meta: Optional[Dict[str, Any]] = None,
) -> int:
    payload: Dict[str, Any] = {"SteamLoginSecure": account.steam_login_secure}
    code, resp, body = _post_steam_api_with_proxy_retry(
        "wallet-balance",
        payload,
        proxy_ctx=proxy_ctx,
        timeout=25,
        debug_meta=debug_meta,
    )
    if code != 200:
        raise RuntimeError(f"获取余额失败: http={code} body={body[:200]}")
    amount_raw = str(resp.get("amount") or "0").strip()
    try:
        cents = int(float(amount_raw))
    except Exception as e:
        raise RuntimeError(f"余额格式错误: {amount_raw}") from e
    account.balance_cents = max(0, cents)
    return account.balance_cents


def _parse_wear_range(wear_range: str) -> Optional[Tuple[Decimal, Decimal]]:
    if not wear_range:
        return None
    parts = [p.strip() for p in str(wear_range).split("-")]
    if len(parts) != 2:
        return None
    try:
        lo = Decimal(parts[0])
        hi = Decimal(parts[1])
    except (InvalidOperation, ValueError):
        return None
    if lo > hi:
        lo, hi = hi, lo
    return lo, hi


def _format_item_exception_message(account_name: str, reason: str) -> str:
    """统一扫货异常展示文案：用户友好提示 + 内部简要原因。"""
    clean_reason = " ".join((reason or "").split())[:180] or "未知异常"
    return f"扫货异常，已跳过当前饰品并继续。原因：账号={account_name}；{clean_reason}"


def _build_render_url(steam_url: str, start: int, count: int) -> str:
    base = steam_url.rstrip("/")
    if not base.endswith("/render"):
        base = f"{base}/render"
    query = urlencode({
        "query": "",
        "start": start,
        "count": count,
        "country": "US",
        "language": "schinese",
        "currency": 1,
    })
    return f"{base}/?{query}"


def _extract_inspect_value(page_json: Dict[str, Any], listing: Dict[str, Any]) -> Optional[str]:
    asset = listing.get("asset") or {}
    appid = str(asset.get("appid") or "")
    contextid = str(asset.get("contextid") or "")
    asset_id = str(asset.get("id") or "")
    if not appid or not contextid or not asset_id:
        return None
    assets_root = page_json.get("assets") or {}
    app_obj = assets_root.get(appid) or {}
    ctx_obj = app_obj.get(contextid) or {}
    asset_obj = ctx_obj.get(asset_id) or {}
    props = asset_obj.get("asset_properties") or []
    if not isinstance(props, list):
        return None
    for p in props:
        if str(p.get("propertyid")) == "6":
            v = p.get("string_value")
            if v:
                return str(v).strip()
    return None


def _fetch_float_value(
    inspect_value: str,
    proxy_ctx: Optional[ProxyContext],
    debug_meta: Optional[Dict[str, Any]] = None,
) -> Optional[Decimal]:
    if not inspect_value:
        return None
    csgofloat_url = "https://inspect.csgofloat.xyz/?" + urlencode({
        "url": f"steam://run/730//+csgo_econ_action_preview%20{inspect_value}"
    })
    headers = {
        "Referer": "https://steamcommunity.com/",
        "Origin": "https://steamcommunity.com",
        "User-Agent": "Mozilla/5.0",
    }
    code, text = _http_request_with_proxy_retry(
        "GET",
        csgofloat_url,
        headers=headers,
        timeout=20,
        proxy_ctx=proxy_ctx,
        debug_meta=debug_meta,
    )
    if code != 200:
        return None
    try:
        data = json.loads(text)
        fv = data.get("iteminfo", {}).get("floatvalue")
        if fv is None:
            return None
        return Decimal(str(fv))
    except Exception:
        return None


def _random_addr_fields() -> Dict[str, str]:
    return {
        "first_name": f"user{random.randint(1000, 9999)}",
        "last_name": f"ln{random.randint(1000, 9999)}",
        "billing_address": f"{random.randint(10, 9999)} Main St",
        "billing_address_two": f"Apt {random.randint(1, 999)}",
        "billing_country": "US",
        "billing_city": f"City{random.randint(10, 999)}",
        "billing_state": "AK",
        "billing_postal_code": str(random.randint(10000, 99999)),
    }


def _post_buylisting(
    listing_id: str,
    steam_url: str,
    account: LocalSteamAccount,
    subtotal: int,
    fee: int,
    total: int,
    confirmation: str,
    proxy_ctx: Optional[ProxyContext],
    debug_meta: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
    url = f"https://steamcommunity.com/market/buylisting/{listing_id}"
    fields = _random_addr_fields()
    form_data = {
        "sessionid": account.session_id or "",
        "currency": "1",
        "subtotal": str(subtotal),
        "fee": str(fee),
        "total": str(total),
        "quantity": "1",
        "first_name": fields["first_name"],
        "last_name": fields["last_name"],
        "billing_address": fields["billing_address"],
        "billing_address_two": fields["billing_address_two"],
        "billing_country": fields["billing_country"],
        "billing_city": fields["billing_city"],
        "billing_state": fields["billing_state"],
        "billing_postal_code": fields["billing_postal_code"],
        "save_my_address": "0",
        "tradefee_tax": "0",
        "confirmation": confirmation,
    }
    body = urlencode(form_data).encode("utf-8")
    cookie = (
        "timezoneOffset=28800,0; "
        f"sessionid={account.session_id}; "
        f"steamLoginSecure={account.steam_login_secure};"
    )
    headers = {
        "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
        "Origin": "https://steamcommunity.com",
        "Referer": steam_url,
        "Cookie": cookie,
        "User-Agent": "Mozilla/5.0",
        "Accept": "*/*",
    }
    code, text = _http_request_with_proxy_retry(
        method="POST",
        url=url,
        headers=headers,
        body=body,
        timeout=25,
        proxy_ctx=proxy_ctx,
        debug_meta=debug_meta,
    )
    try:
        parsed = json.loads(text) if text else {}
    except Exception:
        parsed = {}
    if code != 200 and not parsed:
        raise RuntimeError(f"buylisting 请求失败: http={code} body={text[:200]}")
    return parsed


def _operate_confirmation(
    account: LocalSteamAccount,
    confirmation_id: str,
    proxy_ctx: Optional[ProxyContext],
    debug_meta: Optional[Dict[str, Any]] = None,
) -> bool:
    payload: Dict[str, Any] = {
        "MaFilePath": account.mafile_path,
        "Session": {
            "SteamId": account.steam_id,
            "AccessToken": account.access_token,
            "RefreshToken": account.refresh_token,
            "SessionId": account.session_id or "",
        },
        "ConfirmationId": int(str(confirmation_id)),
        "Operation": "accept",
    }
    code, resp, body = _post_steam_api_with_proxy_retry(
        "trade-confirmations/operate",
        payload,
        proxy_ctx=proxy_ctx,
        timeout=35,
        debug_meta=debug_meta,
    )
    if code != 200:
        return False
    if isinstance(resp.get("session"), dict):
        session = resp.get("session") or {}
        account.session_id = str(session.get("sessionId") or account.session_id or "")
        account.access_token = str(session.get("accessToken") or account.access_token)
        account.refresh_token = str(session.get("refreshToken") or account.refresh_token)
        try:
            account.steam_id = int(session.get("steamId") or account.steam_id)
        except Exception:
            pass
    return bool(resp.get("success") is True or resp.get("success") == 1)


def _purchase_one_listing(
    account: LocalSteamAccount,
    steam_url: str,
    listing_id: str,
    subtotal: int,
    converted_steam_fee: int,
    converted_publisher_fee: int,
    proxy_ctx: Optional[ProxyContext],
    debug_meta: Optional[Dict[str, Any]] = None,
) -> Tuple[bool, Optional[int], str]:
    fee = int(converted_steam_fee) + int(converted_publisher_fee)
    total = int(subtotal) + int(fee)
    if account.balance_cents < total:
        return False, None, f"账号余额不足（余额={account.balance_cents}，需={total}）"

    first = _post_buylisting(
        listing_id=listing_id,
        steam_url=steam_url,
        account=account,
        subtotal=subtotal,
        fee=fee,
        total=total,
        confirmation="0",
        proxy_ctx=proxy_ctx,
        debug_meta=debug_meta,
    )
    wallet = first.get("wallet_info") if isinstance(first, dict) else None
    if isinstance(wallet, dict) and int(wallet.get("success") or 0) == 1:
        try:
            account.balance_cents = int(float(str(wallet.get("wallet_balance") or "0")))
        except Exception:
            account.balance_cents = max(0, account.balance_cents - total)
        return True, total, "购买成功（无需确认）"

    if first.get("need_confirmation") is True:
        confirmation_id = str((first.get("confirmation") or {}).get("confirmation_id") or "").strip()
        if not confirmation_id:
            return False, None, "需要确认但未返回 confirmation_id"

        ok = _operate_confirmation(account, confirmation_id, proxy_ctx, debug_meta=debug_meta)
        if not ok:
            return False, None, "确认订单失败（trade-confirmations/operate）"

        second = _post_buylisting(
            listing_id=listing_id,
            steam_url=steam_url,
            account=account,
            subtotal=subtotal,
            fee=fee,
            total=total,
            confirmation=confirmation_id,
            proxy_ctx=proxy_ctx,
            debug_meta=debug_meta,
        )
        wallet2 = second.get("wallet_info") if isinstance(second, dict) else None
        if isinstance(wallet2, dict) and int(wallet2.get("success") or 0) == 1:
            try:
                account.balance_cents = int(float(str(wallet2.get("wallet_balance") or "0")))
            except Exception:
                account.balance_cents = max(0, account.balance_cents - total)
            return True, total, "购买成功（确认后完成）"
        return False, None, f"确认后购买失败: {json.dumps(second, ensure_ascii=False)[:220]}"

    return False, None, f"购买失败: {json.dumps(first, ensure_ascii=False)[:220]}"


def _scan_item_and_try_buy(
    item: BuffSelectionRequestItem,
    target_price_usd: Decimal,
    wear_min: Decimal,
    wear_max: Decimal,
    account: LocalSteamAccount,
    proxy_ctx: Optional[ProxyContext],
    page_size: int,
    max_scan: int,
    debug_meta: Optional[Dict[str, Any]] = None,
) -> Tuple[bool, Optional[Decimal], Optional[Decimal], str]:
    if not item.steam_link:
        return False, None, None, "缺少 steam_link，无法扫货"

    # 仅第一页：每个饰品只请求一次 render（不翻页；单页内最多处理 max_scan 条）
    scanned = 0
    render_url = _build_render_url(item.steam_link, 0, page_size)
    headers = {"User-Agent": "Mozilla/5.0", "Accept": "application/json"}
    code, text = _http_request_with_proxy_retry(
        "GET",
        render_url,
        headers=headers,
        timeout=20,
        proxy_ctx=proxy_ctx,
        debug_meta=debug_meta,
    )
    if code != 200:
        return False, None, None, f"拉取 listings 失败: http={code}"
    try:
        page_json = json.loads(text)
    except Exception:
        return False, None, None, "拉取 listings 返回非 JSON"

    listinginfo = page_json.get("listinginfo") or {}
    if not isinstance(listinginfo, dict) or not listinginfo:
        return False, None, None, "无可用 listing 数据"

    for listing_id, listing in listinginfo.items():
        if scanned >= max_scan:
            break
        scanned += 1

        try:
            subtotal = int(listing.get("converted_price") or 0)
            steam_fee = int(listing.get("converted_steam_fee") or 0)
            publisher_fee = int(listing.get("converted_publisher_fee") or 0)
        except Exception:
            continue
        total_cents = subtotal + steam_fee + publisher_fee
        total_usd = (Decimal(total_cents) / Decimal("100")).quantize(Decimal("0.01"))
        if total_usd > target_price_usd:
            continue

        inspect_value = _extract_inspect_value(page_json, listing)
        if not inspect_value:
            continue
        fv = _fetch_float_value(inspect_value, proxy_ctx, debug_meta=debug_meta)
        if fv is None or fv < wear_min or fv > wear_max:
            continue

        ok, spent_cents, msg = _purchase_one_listing(
            account=account,
            steam_url=item.steam_link,
            listing_id=str(listing_id),
            subtotal=subtotal,
            converted_steam_fee=steam_fee,
            converted_publisher_fee=publisher_fee,
            proxy_ctx=proxy_ctx,
            debug_meta=debug_meta,
        )
        if ok:
            purchase_price = (Decimal(spent_cents or total_cents) / Decimal("100")).quantize(Decimal("0.01"))
            return True, purchase_price, fv, msg
        # 购买失败（包括确认失败）按流程跳过当前 listing，继续下一个

    return False, None, None, "未命中可购买 listing 或购买失败"


def run_sweep_task(sweep_request_id: int, user_id: int, selection_request_id: int) -> None:
    db = SessionLocal()
    try:
        req = db.query(BuffSweepRequest).filter(
            BuffSweepRequest.id == sweep_request_id,
            BuffSweepRequest.user_id == user_id,
        ).first()
        if not req:
            logger.warning("扫货请求不存在: sweep_request_id=%s user_id=%s", sweep_request_id, user_id)
            return
        if req.status != "pending":
            logger.info("扫货请求状态非 pending，跳过: %s", req.status)
            return
        if req.cancel_requested:
            req.status = "cancelled"
            db.commit()
            logger.info("扫货请求已取消，任务未启动: sweep_request_id=%s", sweep_request_id)
            return

        req.status = "running"
        db.commit()

        proxy_ctx = _resolve_proxy_context()
        proxy_ctx.proxy_retry_times = max(0, _get_system_config_int(db, "buff_sweep_proxy_retry_times", 1))
        proxy_ctx.proxy_retry_interval_ms = max(0, _get_system_config_int(db, "buff_sweep_proxy_retry_interval_ms", 800))
        proxy_ok, proxy_err = _ensure_proxy_ready(proxy_ctx, refresh_if_needed=True)
        if not proxy_ok:
            req.status = "failed"
            db.add(BuffSweepRequestLog(
                sweep_request_id=sweep_request_id,
                user_id=user_id,
                goods_id="-",
                goods_name="",
                sequence_no=0,
                result="failed",
                message=proxy_err,
                retry_attempts=0,
                proxy_refreshed=False,
            ))
            db.commit()
            return
        accounts = _load_active_steam_accounts(db, user_id)
        if not accounts:
            req.status = "failed"
            db.add(BuffSweepRequestLog(
                sweep_request_id=sweep_request_id,
                user_id=user_id,
                goods_id="-",
                goods_name="",
                sequence_no=0,
                result="failed",
                message="未找到可用 Steam 账号（请先在本地验证并绑定）",
                retry_attempts=0,
                proxy_refreshed=False,
            ))
            db.commit()
            return

        # 初始化余额；余额不足阈值的账号后续自动跳过
        for acc in accounts:
            try:
                _refresh_wallet_balance(acc, proxy_ctx)
            except Exception as e:
                acc.balance_cents = 0
                logger.warning("刷新余额失败，账号将跳过: %s err=%s", acc.steam_username, e)

        rows = (
            db.query(BuffSelectionRequestItem)
            .filter(
                BuffSelectionRequestItem.user_id == user_id,
                BuffSelectionRequestItem.request_id == selection_request_id,
            )
            .order_by(BuffSelectionRequestItem.id.asc())
            .all()
        )

        page_size = max(1, _get_system_config_int(db, "buff_sweep_steam_listings_page_size", DEFAULT_PAGE_SIZE))
        max_scan = max(1, _get_system_config_int(db, "buff_sweep_steam_listings_max_scan_per_item", DEFAULT_MAX_SCAN_PER_ITEM))
        success_count = 0
        account_idx = 0

        for seq, row in enumerate(rows, start=1):
            db.refresh(req)
            if req.cancel_requested:
                req.status = "cancelled"
                db.add(BuffSweepRequestLog(
                    sweep_request_id=sweep_request_id,
                    user_id=user_id,
                    goods_id=row.goods_id,
                    goods_name=row.goods_name,
                    sequence_no=seq,
                    result="stopped",
                    message="用户已停止扫货",
                    retry_attempts=0,
                    proxy_refreshed=False,
                    target_price_usd=row.price_usd,
                    purchase_price_usd=None,
                ))
                db.commit()
                return

            target_price = row.price_usd
            wear = _parse_wear_range(row.wear_range or "")
            if target_price is None or target_price <= 0 or wear is None:
                db.add(BuffSweepRequestLog(
                    sweep_request_id=sweep_request_id,
                    user_id=user_id,
                    goods_id=row.goods_id,
                    goods_name=row.goods_name,
                    sequence_no=seq,
                    result="skipped",
                    message="选品数据不完整（价格/磨损区间）",
                    retry_attempts=0,
                    proxy_refreshed=False,
                    target_price_usd=row.price_usd,
                    purchase_price_usd=None,
                ))
                db.commit()
                continue
            wear_min, wear_max = wear

            if all(a.balance_cents < MIN_BALANCE_CENTS for a in accounts):
                db.add(BuffSweepRequestLog(
                    sweep_request_id=sweep_request_id,
                    user_id=user_id,
                    goods_id=row.goods_id,
                    goods_name=row.goods_name,
                    sequence_no=seq,
                    result="failed",
                    message="所有 Steam 账号余额不足（<0.5），停止扫货",
                    retry_attempts=0,
                    proxy_refreshed=False,
                    target_price_usd=target_price,
                    purchase_price_usd=None,
                ))
                req.status = "failed"
                db.commit()
                return

            bought = False
            final_msg = "未命中可购买 listing"
            final_purchase_price: Optional[Decimal] = None
            final_purchase_wear: Optional[Decimal] = None
            used_account = None
            item_debug = {"retry_attempts": 0, "proxy_refreshed": False}

            # 单饰品单账号轮询策略：每个饰品只分配一个账号扫描，账号按 A->B->C->A 轮转。
            selected_account: Optional[LocalSteamAccount] = None
            for _ in range(len(accounts)):
                acc = accounts[account_idx % len(accounts)]
                account_idx += 1
                if acc.balance_cents >= MIN_BALANCE_CENTS:
                    selected_account = acc
                    break

            if not selected_account:
                final_msg = "无可用账号（余额不足）"
            else:
                used_account = selected_account.steam_username
                try:
                    ok, purchase_price_usd, purchase_wear, msg = _scan_item_and_try_buy(
                        item=row,
                        target_price_usd=Decimal(str(target_price)),
                        wear_min=wear_min,
                        wear_max=wear_max,
                        account=selected_account,
                        proxy_ctx=proxy_ctx,
                        page_size=page_size,
                        max_scan=max_scan,
                        debug_meta=item_debug,
                    )
                    final_msg = msg
                    if ok:
                        bought = True
                        final_purchase_price = purchase_price_usd
                        final_purchase_wear = purchase_wear
                except Exception as e:
                    # 单饰品仅由单一账号扫描；异常时只记录并继续下一个饰品。
                    logger.exception(
                        "扫货单品异常（单饰品单账号策略）: sweep_request_id=%s goods_id=%s steam_account=%s error=%s",
                        sweep_request_id,
                        row.goods_id,
                        selected_account.steam_username,
                        e,
                    )
                    final_msg = _format_item_exception_message(selected_account.steam_username, str(e))

            if bought:
                db.add(BuffSweepRequestItem(
                    sweep_request_id=sweep_request_id,
                    user_id=user_id,
                    selection_request_id=selection_request_id,
                    selection_item_id=row.id,
                    goods_id=row.goods_id,
                    goods_name=row.goods_name,
                    icon_url=row.icon_url,
                    wear_range=row.wear_range,
                    purchase_wear=final_purchase_wear,
                    target_price_usd=row.price_usd,
                    purchase_price_usd=final_purchase_price,
                    steam_account=used_account,
                ))
                db.add(BuffSweepRequestLog(
                    sweep_request_id=sweep_request_id,
                    user_id=user_id,
                    goods_id=row.goods_id,
                    goods_name=row.goods_name,
                    sequence_no=seq,
                    result="success",
                    message=f"扫货成功，账号={used_account}",
                    retry_attempts=int(item_debug.get("retry_attempts", 0)),
                    proxy_refreshed=bool(item_debug.get("proxy_refreshed", False)),
                    target_price_usd=row.price_usd,
                    purchase_price_usd=final_purchase_price,
                ))
                success_count += 1
            else:
                db.add(BuffSweepRequestLog(
                    sweep_request_id=sweep_request_id,
                    user_id=user_id,
                    goods_id=row.goods_id,
                    goods_name=row.goods_name,
                    sequence_no=seq,
                    result="skipped",
                    message=(f"账号={used_account}；{final_msg}" if used_account else final_msg),
                    retry_attempts=int(item_debug.get("retry_attempts", 0)),
                    proxy_refreshed=bool(item_debug.get("proxy_refreshed", False)),
                    target_price_usd=row.price_usd,
                    purchase_price_usd=None,
                ))
            db.commit()

        db.refresh(req)
        if req.cancel_requested or req.status == "cancelled":
            req.status = "cancelled"
            db.commit()
            logger.info(
                "扫货任务在收尾阶段检测到取消请求: sweep_request_id=%s selection_request_id=%s",
                sweep_request_id,
                selection_request_id,
            )
            return

        req.status = "completed"
        db.commit()
        logger.info(
            "扫货完成: sweep_request_id=%s selection_request_id=%s success=%s",
            sweep_request_id,
            selection_request_id,
            success_count,
        )
    except Exception as e:
        logger.exception("扫货任务异常: sweep_request_id=%s error=%s", sweep_request_id, e)
        try:
            req = db.query(BuffSweepRequest).filter(
                BuffSweepRequest.id == sweep_request_id,
                BuffSweepRequest.user_id == user_id,
            ).first()
            if req:
                req.status = "failed"
                db.commit()
        except Exception:
            db.rollback()
    finally:
        db.close()