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