fix: eventlet换成simple-websocket,避免与Playwright asyncio冲突

eventlet的monkey-patch会破坏Playwright的Node.js子进程通信(EPIPE)
simple-websocket是Flask-SocketIO推荐的轻量驱动,无侵入性

Dockerfile

@@ -18,5 +18,7 @@ EXPOSE 9000
 VOLUME ["/app/data"]
 
 ENV FLASK_DEBUG=0
+ENV TZ=Asia/Shanghai
+RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone
 
 CMD ["python", "run.py"]

main.py

@@ -1,18 +1,27 @@
+"""
+独立抢购脚本（命令行模式）— 优化版
+用法: python main.py
+"""
 import asyncio
 import yaml
-import time
 from playwright.async_api import async_playwright
 from utils.stealth import stealth_async
 from utils.auth import Authenticator
 from utils.timer import PrecisionTimer
 
+BUY_TEXTS = ["立即抢购", "立即购买", "马上抢", "立即秒杀"]
+MAX_RETRIES = 5
+CONCURRENT_TABS = 2
+
+
 async def snatch(config):
     auth = Authenticator(config.get("auth_file", "auth_state.json"))
     timer = PrecisionTimer()
     timer.sync_time()
 
     async with async_playwright() as p:
-        browser, context = await auth.get_context(p, headless=config.get("headless", False))
+        browser, context = await auth.get_context(
+            p, headless=config.get("headless", False))
         page = await context.new_page()
         await stealth_async(page)
 
@@ -23,7 +32,7 @@ async def snatch(config):
 
         # 1. 预热：先打开页面
         print(f"正在打开商品页面: {target_url}")
-        await page.goto(target_url)
+        await page.goto(target_url, wait_until='networkidle', timeout=20000)
 
         # 如果未登录，处理登录
         if not auth.has_auth():
@@ -31,83 +40,111 @@ async def snatch(config):
             await auth.login(target_url)
             await context.close()
             await browser.close()
-            browser, context = await auth.get_context(p, headless=config.get("headless", False))
+            browser, context = await auth.get_context(
+                p, headless=config.get("headless", False))
             page = await context.new_page()
             await stealth_async(page)
             print("已重新加载登录状态，正在打开商品页面...")
-            await page.goto(target_url)
+            await page.goto(target_url, wait_until='networkidle',
+                            timeout=20000)
 
-        # 2. 等待抢购时间
+        # 2. 等待抢购时间（提前 500ms 触发）
         snatch_time = config.get("snatch_time")
         if snatch_time:
-            await timer.wait_until(snatch_time)
+            await timer.wait_until_early(snatch_time, early_ms=500)
 
-        # 3. 抢购核心逻辑
-        max_retries = 3
-        for attempt in range(max_retries):
+        # 3. 并发抢购
+        pages = [page]
+        for _ in range(CONCURRENT_TABS - 1):
             try:
-                # 刷新页面，让预售按钮变为可点击
-                if attempt > 0:
-                    await asyncio.sleep(0.3)
-                await page.reload(wait_until='domcontentloaded', timeout=10000)
-                await asyncio.sleep(0.5)
+                p2 = await context.new_page()
+                await stealth_async(p2)
+                await p2.goto(target_url, wait_until='commit', timeout=10000)
+                pages.append(p2)
+            except Exception:
+                pass
 
-                # 点击购买按钮（兼容多种文案）
+        tasks = [_do_purchase(pg, i) for i, pg in enumerate(pages)]
+        results = await asyncio.gather(*tasks, return_exceptions=True)
+
+        for r in results:
+            print(f"结果: {r}")
+            if isinstance(r, str) and ('已提交' in r or '已发送' in r):
+                print("✅ 抢购成功！")
+                break
+
+        await asyncio.sleep(10)
+        await browser.close()
+
+
+async def _do_purchase(page, tab_index=0):
+    """极速购买流程"""
+    for attempt in range(MAX_RETRIES):
+        try:
+            await page.reload(wait_until='commit', timeout=8000)
+            await asyncio.sleep(0.3)
+
+            # 点击购买按钮
             buy_btn = None
-                for text in ["立即抢购", "立即购买", "马上抢", "立即秒杀"]:
+            for text in BUY_TEXTS:
                 loc = page.get_by_text(text, exact=False)
-                    if await loc.count() > 0:
+                try:
+                    await loc.first.wait_for(state="visible", timeout=1500)
                     buy_btn = loc.first
                     break
+                except Exception:
+                    continue
 
             if not buy_btn:
-                    if attempt < max_retries - 1:
-                        print(f"第{attempt+1}次未找到购买按钮，重试...")
+                if attempt < MAX_RETRIES - 1:
+                    print(f"tab{tab_index} 第{attempt+1}次未找到按钮，重试...")
+                    await asyncio.sleep(0.05)
                     continue
-                    print("错误: 未找到购买按钮")
-                    break
+                return f"tab{tab_index}: 未找到购买按钮"
 
-                await buy_btn.click(timeout=3000)
-                print("点击购买按钮")
+            await buy_btn.click(timeout=2000)
+            print(f"tab{tab_index}: 点击购买按钮")
 
-                # 处理 SKU 选择（如果弹出规格选择框）
-                await asyncio.sleep(0.5)
+            # 处理 SKU 弹窗
             try:
                 confirm_btn = page.get_by_text("确定", exact=True)
-                    if await confirm_btn.count() > 0 and await confirm_btn.first.is_visible():
-                        # 自动选择第一个可用的 SKU 选项
-                        sku_items = page.locator('.sku-item:not(.disabled), .sku_item:not(.disabled), [class*="sku"] [class*="item"]:not([class*="disabled"])')
+                await confirm_btn.first.wait_for(state="visible",
+                                                 timeout=1500)
+                sku_sel = ('.sku-item:not(.disabled), '
+                           '.sku_item:not(.disabled), '
+                           '[class*="sku"] [class*="item"]'
+                           ':not([class*="disabled"])')
+                sku_items = page.locator(sku_sel)
                 if await sku_items.count() > 0:
                     await sku_items.first.click()
-                            print("自动选择 SKU")
-                            await asyncio.sleep(0.3)
-                        await confirm_btn.first.click(timeout=3000)
-                        print("点击确定（SKU）")
+                    print(f"tab{tab_index}: 自动选择 SKU")
+                    await asyncio.sleep(0.1)
+                await confirm_btn.first.click(timeout=2000)
+                print(f"tab{tab_index}: 点击确定")
             except Exception:
                 pass
 
             # 提交订单
             submit_btn = page.get_by_text("提交订单")
-                await submit_btn.wait_for(state="visible", timeout=8000)
+            await submit_btn.wait_for(state="visible", timeout=6000)
             await submit_btn.click()
-                print("点击提交订单！抢购请求已发送！")
-                break
+            return f"tab{tab_index}: 抢购请求已提交"
 
         except Exception as e:
-                if attempt < max_retries - 1:
-                    print(f"第{attempt+1}次尝试失败: {e}，重试...")
+            if attempt < MAX_RETRIES - 1:
+                print(f"tab{tab_index} 第{attempt+1}次失败: {e}，重试...")
+                await asyncio.sleep(0.05)
             else:
-                    print(f"抢购失败: {e}")
+                return f"tab{tab_index}: 抢购失败: {e}"
 
-        # 保持浏览器打开一段时间查看结果
-        await asyncio.sleep(10)
-        await browser.close()
+    return f"tab{tab_index}: 重试次数用尽"
 
 
 def load_config():
     with open("config.yaml", "r", encoding="utf-8") as f:
         return yaml.safe_load(f)
 
+
 if __name__ == "__main__":
     config = load_config()
     asyncio.run(snatch(config))

requirements.txt

@@ -1,5 +1,6 @@
 flask>=3.0
 flask-socketio>=5.3
+simple-websocket>=1.0
 playwright==1.52.0
 playwright-stealth>=1.0
 pyyaml>=6.0

server/app.py

@@ -1,26 +1,214 @@
 import os
 import sys
+import hashlib
+import time
 
 sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
 
-from flask import Flask, redirect, url_for
+from flask import (Flask, redirect, url_for, request, session,
+                   render_template, jsonify)
 from flask_socketio import SocketIO
-from server.database import init_db
-from server.routers import accounts, tasks, orders
+from server.database import init_db, get_db
 
 socketio = SocketIO(cors_allowed_origins="*")
 
+# 防爆破：最大失败次数 & 锁定时间（秒）
+MAX_FAIL = 5
+LOCK_SECONDS = 300  # 5分钟
+
+
+def _hash_pwd(pwd):
+    return hashlib.sha256(pwd.encode()).hexdigest()
+
+
+def _get_admin():
+    db = get_db()
+    row = db.execute('SELECT * FROM admin WHERE id = 1').fetchone()
+    db.close()
+    return row
+
+
+def _is_locked():
+    admin = _get_admin()
+    if not admin:
+        return False
+    if admin['fail_count'] >= MAX_FAIL and admin['locked_until']:
+        try:
+            from datetime import datetime
+            locked = datetime.strptime(admin['locked_until'],
+                                       '%Y-%m-%d %H:%M:%S')
+            if datetime.now() < locked:
+                return True
+            # 锁定已过期，重置
+            db = get_db()
+            db.execute("UPDATE admin SET fail_count=0, locked_until='' "
+                       "WHERE id=1")
+            db.commit()
+            db.close()
+        except Exception:
+            pass
+    return False
+
+
+def _record_fail():
+    from datetime import datetime, timedelta
+    db = get_db()
+    admin = db.execute('SELECT fail_count FROM admin WHERE id=1').fetchone()
+    if admin:
+        new_count = (admin['fail_count'] or 0) + 1
+        locked_until = ''
+        if new_count >= MAX_FAIL:
+            locked_until = (datetime.now() + timedelta(
+                seconds=LOCK_SECONDS)).strftime('%Y-%m-%d %H:%M:%S')
+        db.execute("UPDATE admin SET fail_count=?, locked_until=?, "
+                   "updated_at=datetime('now','localtime') WHERE id=1",
+                   (new_count, locked_until))
+        db.commit()
+    db.close()
+
+
+def _reset_fail():
+    db = get_db()
+    db.execute("UPDATE admin SET fail_count=0, locked_until='', "
+               "updated_at=datetime('now','localtime') WHERE id=1")
+    db.commit()
+    db.close()
+
 
 def create_app():
     app = Flask(
         __name__,
-        template_folder=os.path.join(os.path.dirname(__file__), '..', 'templates'),
-        static_folder=os.path.join(os.path.dirname(__file__), '..', 'static'),
+        template_folder=os.path.join(os.path.dirname(__file__), '..',
+                                     'templates'),
+        static_folder=os.path.join(os.path.dirname(__file__), '..',
+                                   'static'),
     )
-    app.secret_key = 'snatcher-secret-key-change-me'
+    app.secret_key = os.environ.get(
+        'SECRET_KEY', 'snatcher-' + _hash_pwd('change-me')[:16])
 
     init_db()
 
+    # ── 登录 & 初始设置密码 ──
+    @app.route('/login', methods=['GET', 'POST'])
+    def login():
+        admin = _get_admin()
+        # 首次使用：跳转设置密码
+        if not admin:
+            return redirect(url_for('setup'))
+
+        if request.method == 'POST':
+            if _is_locked():
+                remain = ''
+                try:
+                    from datetime import datetime
+                    locked = datetime.strptime(
+                        _get_admin()['locked_until'], '%Y-%m-%d %H:%M:%S')
+                    remain = f'{int((locked - datetime.now()).total_seconds())}秒'
+                except Exception:
+                    remain = f'{LOCK_SECONDS}秒'
+                return render_template(
+                    'login.html',
+                    error=f'登录失败次数过多，请{remain}后再试')
+
+            pwd = request.form.get('password', '')
+            if _hash_pwd(pwd) == admin['password_hash']:
+                _reset_fail()
+                session['authenticated'] = True
+                session.permanent = True
+                app.permanent_session_lifetime = __import__(
+                    'datetime').timedelta(days=7)
+                return redirect(request.args.get('next', '/'))
+            else:
+                _record_fail()
+                admin = _get_admin()
+                remaining = MAX_FAIL - (admin['fail_count'] or 0)
+                if remaining <= 0:
+                    return render_template(
+                        'login.html',
+                        error=f'账号已锁定{LOCK_SECONDS // 60}分钟')
+                return render_template(
+                    'login.html',
+                    error=f'密码错误，还剩{remaining}次机会')
+
+        return render_template('login.html')
+
+    @app.route('/setup', methods=['GET', 'POST'])
+    def setup():
+        admin = _get_admin()
+        if admin:
+            return redirect(url_for('login'))
+
+        if request.method == 'POST':
+            pwd = request.form.get('password', '')
+            pwd2 = request.form.get('password2', '')
+            if len(pwd) < 6:
+                return render_template('setup.html',
+                                       error='密码至少6位')
+            if pwd != pwd2:
+                return render_template('setup.html',
+                                       error='两次密码不一致')
+            db = get_db()
+            db.execute(
+                "INSERT INTO admin (id, password_hash) VALUES (1, ?)",
+                (_hash_pwd(pwd),))
+            db.commit()
+            db.close()
+            session['authenticated'] = True
+            session.permanent = True
+            app.permanent_session_lifetime = __import__(
+                'datetime').timedelta(days=7)
+            return redirect('/')
+
+        return render_template('setup.html')
+
+    @app.route('/logout')
+    def logout():
+        session.clear()
+        return redirect('/login')
+
+    @app.route('/change_password', methods=['GET', 'POST'])
+    def change_password():
+        if not session.get('authenticated'):
+            return redirect(url_for('login'))
+        if request.method == 'POST':
+            old = request.form.get('old_password', '')
+            new = request.form.get('new_password', '')
+            new2 = request.form.get('new_password2', '')
+            admin = _get_admin()
+            if _hash_pwd(old) != admin['password_hash']:
+                return render_template('change_password.html',
+                                       error='原密码错误')
+            if len(new) < 6:
+                return render_template('change_password.html',
+                                       error='新密码至少6位')
+            if new != new2:
+                return render_template('change_password.html',
+                                       error='两次密码不一致')
+            db = get_db()
+            db.execute(
+                "UPDATE admin SET password_hash=?, "
+                "updated_at=datetime('now','localtime') WHERE id=1",
+                (_hash_pwd(new),))
+            db.commit()
+            db.close()
+            return render_template('change_password.html',
+                                   success='密码修改成功')
+        return render_template('change_password.html')
+
+    @app.before_request
+    def require_auth():
+        allowed = ('/login', '/setup')
+        if request.path in allowed or request.path.startswith('/static'):
+            return
+        if request.path.startswith('/socket.io'):
+            return
+        admin = _get_admin()
+        if not admin:
+            return redirect(url_for('setup'))
+        if not session.get('authenticated'):
+            return redirect(url_for('login', next=request.path))
+
+    from server.routers import accounts, tasks, orders
     app.register_blueprint(accounts.bp)
     app.register_blueprint(tasks.bp)
     app.register_blueprint(orders.bp)
@@ -53,19 +241,15 @@ def _register_socketio_events():
     def handle_keyboard(data):
         s = get_session(data.get('session_id', ''))
         if s and s.loop and s.loop.is_running():
-            text = data.get('text', '')
-            if text:
             asyncio.run_coroutine_threadsafe(
-                    s.handle_keyboard(text), s.loop)
+                s.handle_keyboard(data.get('text', '')), s.loop)
 
     @socketio.on('rb_key', namespace='/rb')
     def handle_key(data):
         s = get_session(data.get('session_id', ''))
         if s and s.loop and s.loop.is_running():
-            key = data.get('key', '')
-            if key:
             asyncio.run_coroutine_threadsafe(
-                    s.handle_key(key), s.loop)
+                s.handle_key(data.get('key', '')), s.loop)
 
     @socketio.on('rb_close', namespace='/rb')
     def handle_close(data):

server/database.py

@@ -54,6 +54,15 @@ def init_db():
             FOREIGN KEY (task_id) REFERENCES tasks(id),
             FOREIGN KEY (account_id) REFERENCES accounts(id)
         );
+
+        CREATE TABLE IF NOT EXISTS admin (
+            id INTEGER PRIMARY KEY CHECK (id = 1),
+            password_hash TEXT NOT NULL,
+            fail_count INTEGER DEFAULT 0,
+            locked_until TEXT DEFAULT '',
+            created_at TEXT DEFAULT (datetime('now', 'localtime')),
+            updated_at TEXT DEFAULT (datetime('now', 'localtime'))
+        );
     ''')
     conn.commit()
     conn.close()

server/services/auth_service.py

@@ -146,7 +146,7 @@ async def login_with_password(account_id, phone, password):
             # 保存 storage_state + 精简 cookies
             auth_path = get_auth_path(account_id)
             await context.storage_state(path=auth_path)
-            _save_cookies_file(account_id, context)
+            await _save_cookies_file_async(context, account_id)
 
             return True, "登录成功"
 

server/services/cart_service.py

@@ -1,16 +1,17 @@
 """
 购物车预售商品抓取服务
-通过 Playwright 打开购物车页面，拦截 API + DOM 提取商品信息
+通过 Playwright 打开购物车页面，从 DOM 提取商品信息，
+并通过点击商品图片获取跳转 URL 来提取真实 itemID。
 """
 import asyncio
-import json
+import re
 from playwright.async_api import async_playwright
 from utils.stealth import stealth_async
 from server.services.auth_service import get_browser_context, has_auth
 
 CART_URL = "https://weidian.com/new-cart/index.php"
 
-# 从 DOM 提取购物车商品（含尝试从 Vue 组件获取 itemID）
+# 从 DOM 提取购物车商品基本信息
 EXTRACT_JS = """() => {
     const R = [];
     const sws = document.querySelectorAll(
@@ -29,21 +30,35 @@ EXTRACT_JS = """() => {
                 sale_time: '', presale_type: ''
             };
 
-            // 尝试从 Vue 组件数据中提取 itemID
+            // 尝试多种方式提取 itemID
+            // 1. Vue 组件数据
             try {
-                const vue = iw.__vue__ || (iw.__vue_app__ && iw.__vue_app__._instance);
+                const vue = iw.__vue__;
                 if (vue) {
-                    const d = vue.$data || vue.data || vue;
-                    o.item_id = String(d.itemID || d.itemId || d.item_id || '');
+                    const d = vue.$data || vue._data || vue;
+                    o.item_id = String(d.itemID || d.itemId || d.item_id
+                        || d.goodsId || d.goods_id || '');
+                    // 也检查 props
+                    if (!o.item_id && vue.$props) {
+                        const pp = vue.$props;
+                        o.item_id = String(pp.itemID || pp.itemId
+                            || pp.item_id || pp.goodsId || '');
+                    }
+                    // 检查 item 对象
+                    if (!o.item_id && d.item) {
+                        o.item_id = String(d.item.itemID || d.item.itemId
+                            || d.item.item_id || '');
+                    }
                 }
             } catch(e) {}
 
-            // 尝试从 data-* 属性提取
+            // 2. data-* 属性
             if (!o.item_id) {
-                o.item_id = iw.dataset.itemId || iw.dataset.itemid || '';
+                o.item_id = iw.dataset.itemId || iw.dataset.itemid
+                    || iw.dataset.goodsId || iw.dataset.id || '';
             }
 
-            // 尝试从内部链接提取
+            // 3. 内部链接
             if (!o.item_id) {
                 const a = iw.querySelector('a[href*="itemID"]');
                 if (a) {
@@ -52,6 +67,10 @@ EXTRACT_JS = """() => {
                 }
             }
 
+            // 4. 图片 URL 中可能有商品信息（备用）
+            const img = iw.querySelector('.item_img img');
+            o.img_src = img ? (img.src || img.dataset.src || '') : '';
+
             const te = iw.querySelector('.item_title');
             if (te) o.title = te.textContent.trim();
             const sk = iw.querySelector('.item_sku');
@@ -89,15 +108,12 @@ EXTRACT_JS = """() => {
 async def fetch_cart_presale_items(account_id):
     """
     获取指定账号购物车中的预售商品列表。
-    双重提取：拦截购物车 API 获取 itemID 映射 + DOM 提取预售信息。
+    对于没有 itemID 的商品，通过点击图片获取跳转 URL 来提取。
     返回: (success, items_or_msg)
     """
     if not has_auth(account_id):
         return False, "账号未登录"
 
-    # 用于存储 API 返回的 cart_item_id -> itemID 映射
-    api_item_map = {}
-
     async with async_playwright() as p:
         browser, context = await get_browser_context(
             p, account_id, headless=True
@@ -105,20 +121,6 @@ async def fetch_cart_presale_items(account_id):
         page = await context.new_page()
         await stealth_async(page)
 
-        # 拦截购物车相关 API，提取 itemID
-        async def on_response(response):
-            url = response.url
-            # 购物车 API 通常包含 cart 相关路径
-            if any(k in url for k in ['cart/list', 'cart/query', 'cartList', 'getCart',
-                                       'cart-server', 'newcart']):
-                try:
-                    data = await response.json()
-                    _extract_item_ids(data, api_item_map)
-                except Exception:
-                    pass
-
-        page.on("response", on_response)
-
         try:
             await page.goto(
                 CART_URL, wait_until="networkidle", timeout=20000
@@ -137,74 +139,100 @@ async def fetch_cart_presale_items(account_id):
             await browser.close()
             return False, f"打开购物车失败: {e}"
 
-        # 也尝试从页面内嵌的 JS 变量/window 对象提取
-        try:
-            extra_map = await page.evaluate("""() => {
-                const m = {};
-                // 尝试从 window.__INITIAL_STATE__ 或类似全局变量提取
-                const sources = [
-                    window.__INITIAL_STATE__,
-                    window.__NUXT__,
-                    window.__APP_DATA__,
-                    window.cartData,
-                    window.__data__,
-                ];
-                function walk(obj, depth) {
-                    if (!obj || depth > 5) return;
-                    if (Array.isArray(obj)) {
-                        for (const item of obj) walk(item, depth + 1);
-                    } else if (typeof obj === 'object') {
-                        const cid = String(obj.cartItemId || obj.cart_item_id || obj.cartId || '');
-                        const iid = String(obj.itemID || obj.itemId || obj.item_id || obj.goodsId || '');
-                        if (cid && iid && iid !== cid) m[cid] = iid;
-                        // 也存 itemUrl
-                        if (cid && obj.itemUrl) m[cid + '_url'] = obj.itemUrl;
-                        for (const v of Object.values(obj)) walk(v, depth + 1);
-                    }
-                }
-                for (const s of sources) { if (s) walk(s, 0); }
-                return m;
-            }""")
-            if extra_map:
-                api_item_map.update(extra_map)
-        except Exception:
-            pass
-
+        # 提取基本信息
         raw_items = await page.evaluate(EXTRACT_JS)
+
+        # 筛选预售商品
+        presale = [it for it in raw_items if it.get("is_presale")]
+
+        # 对没有 itemID 的预售商品，通过点击图片获取跳转 URL
+        for item in presale:
+            if item.get('item_id'):
+                continue
+            cid = item.get('cart_item_id', '')
+            if not cid:
+                continue
+
+            item_id = await _get_item_id_by_click(page, context, cid)
+            if item_id:
+                item['item_id'] = item_id
+                item['url'] = f'https://weidian.com/item.html?itemID={item_id}'
+
         await browser.close()
 
-    # 合并 API 数据到 DOM 提取结果
-    for item in raw_items:
-        cid = item.get('cart_item_id', '')
-        if not item.get('item_id') and cid in api_item_map:
-            item['item_id'] = api_item_map[cid]
-        # 检查是否有 URL
-        url_key = cid + '_url'
-        if url_key in api_item_map:
-            item['url'] = api_item_map[url_key]
-
-    # 只返回预售商品
-    presale = [it for it in raw_items if it.get("is_presale")]
     return True, presale
 
 
-def _extract_item_ids(data, result_map):
-    """递归遍历 API 响应 JSON，提取 cart_item_id -> itemID 映射"""
-    if isinstance(data, list):
-        for item in data:
-            _extract_item_ids(item, result_map)
-    elif isinstance(data, dict):
-        # 常见字段名
-        cid = str(data.get('cartItemId', data.get('cart_item_id',
-                  data.get('cartId', ''))))
-        iid = str(data.get('itemID', data.get('itemId',
-                  data.get('item_id', data.get('goodsId', '')))))
-        if cid and iid and cid != iid and iid != 'None':
-            result_map[cid] = iid
-        # 也提取 URL
-        item_url = data.get('itemUrl', data.get('item_url', ''))
-        if cid and item_url:
-            result_map[cid + '_url'] = item_url
-        for v in data.values():
-            if isinstance(v, (dict, list)):
-                _extract_item_ids(v, result_map)
+async def _get_item_id_by_click(page, context, cart_item_id):
+    """
+    通过点击购物车中商品的图片，拦截跳转 URL 来提取 itemID。
+    点击后会打开新 tab，从新 tab 的 URL 中提取 itemID，然后关闭。
+    """
+    try:
+        # 定位商品图片
+        img_locator = page.locator(
+            f'.item_warp[id="{cart_item_id}"] .item_img')
+        if await img_locator.count() == 0:
+            return None
+
+        # 监听新页面打开事件
+        async with context.expect_page(timeout=5000) as new_page_info:
+            await img_locator.first.click()
+
+        new_page = await new_page_info.value
+        # 等待 URL 加载
+        await asyncio.sleep(1)
+        url = new_page.url
+
+        # 从 URL 提取 itemID
+        item_id = _extract_item_id_from_url(url)
+
+        # 关闭新 tab
+        await new_page.close()
+
+        return item_id
+
+    except Exception:
+        # 如果 expect_page 超时，可能是在当前页面跳转了
+        # 检查当前页面 URL
+        try:
+            current_url = page.url
+            item_id = _extract_item_id_from_url(current_url)
+            if item_id:
+                # 跳回购物车
+                await page.goto(CART_URL, wait_until="networkidle",
+                                timeout=15000)
+                await asyncio.sleep(2)
+                return item_id
+        except Exception:
+            pass
+
+        # 确保回到购物车页面
+        try:
+            if 'new-cart' not in page.url:
+                await page.goto(CART_URL, wait_until="networkidle",
+                                timeout=15000)
+                await asyncio.sleep(2)
+        except Exception:
+            pass
+
+        return None
+
+
+def _extract_item_id_from_url(url):
+    """从 URL 中提取 itemID"""
+    if not url:
+        return None
+    # https://weidian.com/item.html?itemID=123456
+    m = re.search(r'itemID=(\d+)', url, re.IGNORECASE)
+    if m:
+        return m.group(1)
+    # https://shop.weidian.com/item/123456
+    m = re.search(r'/item/(\d+)', url)
+    if m:
+        return m.group(1)
+    # https://weidian.com/...?id=123456
+    m = re.search(r'[?&]id=(\d+)', url)
+    if m:
+        return m.group(1)
+    return None

server/services/snatcher.py

@@ -1,3 +1,11 @@
+"""
+抢购核心服务 — 优化版 v2
+策略：两阶段重试
+  阶段1: 商品详情页直接抢购（5次快速重试）
+  阶段2: 若阶段1失败，切换到购物车入口下单（5次重试）
+         从购物车勾选商品 → 结算 → 提交订单
+  这样避免单一页面刷新过多触发风控
+"""
 import asyncio
 from playwright.async_api import async_playwright
 from utils.stealth import stealth_async
@@ -6,6 +14,12 @@ from server.services.auth_service import get_browser_context, has_auth
 from server.database import get_db
 from datetime import datetime
 
+CONCURRENT_TABS = 2
+PHASE1_RETRIES = 5   # 商品页重试次数
+PHASE2_RETRIES = 5   # 购物车重试次数
+BUY_TEXTS = ["立即抢购", "立即购买", "马上抢", "立即秒杀"]
+CART_URL = "https://weidian.com/new-cart/index.php"
+
 
 async def run_snatch(task_id):
     """执行单个抢购任务"""
@@ -19,63 +33,105 @@ async def run_snatch(task_id):
         _update_task(db, task_id, 'failed', '账号未登录')
         return
 
+    target_url = task['target_url']
+    if not target_url or not target_url.strip():
+        _update_task(db, task_id, 'failed',
+                     '商品链接为空，请检查购物车同步是否获取到了 itemID')
+        return
+
     _update_task(db, task_id, 'running', '正在准备...')
 
     timer = PrecisionTimer()
     timer.sync_time()
 
+    cart_item_id = task['item_id'] or ''  # 购物车商品ID，用于阶段2
+
     try:
         async with async_playwright() as p:
-            browser, context = await get_browser_context(p, account_id, headless=True)
+            browser, context = await get_browser_context(
+                p, account_id, headless=True)
+
+            # ── 1. 预热：打开商品页面 ──
+            _update_task(db, task_id, 'running', '预热：打开商品页面...')
             page = await context.new_page()
             await stealth_async(page)
+            await page.goto(target_url, wait_until='networkidle',
+                            timeout=20000)
 
-            target_url = task['target_url']
-
-            # 1. 预热：先打开商品页面
-            _update_task(db, task_id, 'running', '正在打开商品页面...')
-            await page.goto(target_url, wait_until='networkidle', timeout=20000)
-
-            # 检查是否被重定向到登录页
+            # 检查页面状态
             if 'login' in page.url.lower():
                 _update_task(db, task_id, 'failed', '登录态已过期')
                 await browser.close()
                 return
 
-            # 检查商品是否存在
-            page_text = await page.locator('body').text_content()
-            if '商品不存在' in (page_text or '') or '已下架' in (page_text or ''):
-                _update_task(db, task_id, 'failed', f'商品不存在或已下架 (URL: {target_url})')
+            body_text = await page.locator('body').text_content()
+            if '商品不存在' in (body_text or '') or '已下架' in (body_text or ''):
+                _update_task(db, task_id, 'failed', '商品不存在或已下架')
                 await browser.close()
                 return
 
-            if not target_url or target_url.strip() == '':
-                _update_task(db, task_id, 'failed', '商品链接为空，请检查购物车同步是否获取到了 itemID')
-                await browser.close()
-                return
-
-            # 2. 等待抢购时间
+            # ── 2. 等待抢购时间 ──
             snatch_time = task['snatch_time']
             if snatch_time:
-                _update_task(db, task_id, 'running', f'等待抢购时间: {snatch_time}')
-                await timer.wait_until(snatch_time)
+                _update_task(db, task_id, 'running',
+                             f'等待开售: {snatch_time}')
+                await timer.wait_until_early(snatch_time, early_ms=500)
 
-            # 3. 抢购核心逻辑（与 main.py 一致）
-            _update_task(db, task_id, 'running', '开始抢购...')
-            result = await _do_purchase(page)
+            # ══════════════════════════════════════
+            #  阶段1: 商品详情页直接抢购
+            # ══════════════════════════════════════
+            _update_task(db, task_id, 'running', '阶段1: 商品页抢购...')
 
-            if '已提交' in result or '已发送' in result:
+            pages = [page]
+            for _ in range(CONCURRENT_TABS - 1):
+                try:
+                    p2 = await context.new_page()
+                    await stealth_async(p2)
+                    await p2.goto(target_url, wait_until='commit',
+                                  timeout=10000)
+                    pages.append(p2)
+                except Exception:
+                    pass
+
+            tasks_coro = [_phase1_purchase(pg, i) for i, pg in
+                          enumerate(pages)]
+            results = await asyncio.gather(*tasks_coro,
+                                          return_exceptions=True)
+
+            result = _pick_success(results)
+
+            # ══════════════════════════════════════
+            #  阶段2: 购物车入口下单（降级策略）
+            # ══════════════════════════════════════
+            if not _is_success(result):
+                _update_task(db, task_id, 'running',
+                             f'阶段1失败({result})，切换购物车下单...')
+
+                # 关闭之前的 tab，开新 tab 去购物车
+                for pg in pages:
+                    try:
+                        await pg.close()
+                    except Exception:
+                        pass
+
+                cart_page = await context.new_page()
+                await stealth_async(cart_page)
+                result = await _phase2_cart_purchase(
+                    cart_page, cart_item_id)
+
+            # ── 记录结果 ──
+            if _is_success(result):
                 _update_task(db, task_id, 'completed', result)
                 db.execute(
-                    'INSERT INTO orders (task_id, account_id, status, detail) VALUES (?, ?, ?, ?)',
-                    (task_id, account_id, 'submitted', result)
-                )
+                    'INSERT INTO orders (task_id, account_id, status, detail)'
+                    ' VALUES (?, ?, ?, ?)',
+                    (task_id, account_id, 'submitted', result))
             else:
                 _update_task(db, task_id, 'failed', result)
                 db.execute(
-                    'INSERT INTO orders (task_id, account_id, status, detail) VALUES (?, ?, ?, ?)',
-                    (task_id, account_id, 'failed', result)
-                )
+                    'INSERT INTO orders (task_id, account_id, status, detail)'
+                    ' VALUES (?, ?, ?, ?)',
+                    (task_id, account_id, 'failed', result))
             db.commit()
 
             await asyncio.sleep(3)
@@ -87,71 +143,190 @@ async def run_snatch(task_id):
         db.close()
 
 
-async def _do_purchase(page):
-    """
-    执行购买流程：
-    1. 刷新页面（预售商品需要刷新才能出现购买按钮）
-    2. 点击"立即购买"/"立即抢购"
-    3. 处理 SKU 选择 -> 点击"确定"
-    4. 进入订单确认页 -> 点击"提交订单"
-    支持多次重试
-    """
-    max_retries = 3
-    for attempt in range(max_retries):
+# ─────────────────────────────────────────────
+#  阶段1: 商品详情页快速抢购
+# ─────────────────────────────────────────────
+async def _phase1_purchase(page, tab_index=0):
+    """商品详情页极速购买"""
+    for attempt in range(PHASE1_RETRIES):
         try:
-            # 刷新页面，让预售按钮变为可点击
-            if attempt > 0:
-                await asyncio.sleep(0.3)
-            await page.reload(wait_until='domcontentloaded', timeout=10000)
-            await asyncio.sleep(0.5)
+            await page.reload(wait_until='commit', timeout=8000)
+            # 等 DOM 关键元素出现，比纯 commit 多等一点
+            await asyncio.sleep(0.5 if attempt == 0 else 0.3)
 
-            # 点击购买按钮（兼容多种文案）
             buy_btn = None
-            for text in ["立即抢购", "立即购买", "马上抢", "立即秒杀"]:
+            for text in BUY_TEXTS:
                 loc = page.get_by_text(text, exact=False)
-                if await loc.count() > 0:
+                try:
+                    await loc.first.wait_for(state="visible", timeout=2000)
                     buy_btn = loc.first
                     break
+                except Exception:
+                    continue
 
             if not buy_btn:
-                if attempt < max_retries - 1:
+                if attempt < PHASE1_RETRIES - 1:
+                    await asyncio.sleep(0.05)
                     continue
-                return "抢购操作失败: 未找到购买按钮"
+                return f"P1-tab{tab_index}: 未找到购买按钮"
 
-            await buy_btn.click(timeout=3000)
+            # force=True 绕过遮罩层/倒计时覆盖层的拦截
+            await buy_btn.click(timeout=3000, force=True)
 
-            # 处理 SKU 选择（如果弹出规格选择框）
-            await asyncio.sleep(0.5)
+            # SKU 弹窗
+            await _handle_sku(page)
+
+            # 提交订单
+            submit_btn = page.get_by_text("提交订单")
+            await submit_btn.wait_for(state="visible", timeout=8000)
+            await submit_btn.click(force=True)
+            return f"P1-tab{tab_index}: 抢购请求已提交"
+
+        except Exception as e:
+            if attempt < PHASE1_RETRIES - 1:
+                await asyncio.sleep(0.05)
+                continue
+            return f"P1-tab{tab_index}: {e}"
+
+    return f"P1-tab{tab_index}: 重试次数用尽"
+
+
+# ─────────────────────────────────────────────
+#  阶段2: 购物车入口下单
+# ─────────────────────────────────────────────
+async def _phase2_cart_purchase(page, cart_item_id):
+    """
+    从购物车下单：
+    1. 打开购物车页面
+    2. 找到目标商品并勾选
+    3. 点击结算
+    4. 提交订单
+    """
+    for attempt in range(PHASE2_RETRIES):
         try:
-                # 检查是否有 SKU 弹窗
-                confirm_btn = page.get_by_text("确定", exact=True)
-                if await confirm_btn.count() > 0 and await confirm_btn.first.is_visible():
-                    # 自动选择第一个可用的 SKU 选项
-                    sku_items = page.locator('.sku-item:not(.disabled), .sku_item:not(.disabled), [class*="sku"] [class*="item"]:not([class*="disabled"])')
-                    if await sku_items.count() > 0:
-                        await sku_items.first.click()
+            # 打开购物车
+            if attempt == 0:
+                await page.goto(CART_URL, wait_until='networkidle',
+                                timeout=15000)
+                await asyncio.sleep(1)
+            else:
+                await page.reload(wait_until='domcontentloaded',
+                                  timeout=10000)
+                await asyncio.sleep(0.5)
+
+            # 尝试勾选目标商品
+            selected = False
+            if cart_item_id:
+                # 通过 cart_item_id 精确定位
+                item_warp = page.locator(f'#\\3{cart_item_id[0]} {cart_item_id[1:]}' if len(cart_item_id) > 1 else f'#{cart_item_id}')
+                # 更可靠的方式：用 data-v + id 属性
+                item_warp = page.locator(f'.item_warp[id="{cart_item_id}"]')
+                if await item_warp.count() > 0:
+                    # 点击商品前面的勾选框
+                    checkbox = item_warp.locator('.checkbox').first
+                    if await checkbox.count() > 0:
+                        await checkbox.click(force=True)
+                        selected = True
+                        await asyncio.sleep(0.3)
+
+            if not selected:
+                # 没有精确定位到，尝试全选
+                try:
+                    select_all = page.get_by_text("全选", exact=False)
+                    if await select_all.count() > 0:
+                        await select_all.first.click(force=True)
+                        selected = True
                         await asyncio.sleep(0.3)
-                    await confirm_btn.first.click(timeout=3000)
                 except Exception:
                     pass
 
-            # 等待进入订单确认页，点击"提交订单"
+            if not selected:
+                # 还是没选中，点击第一个商品的 checkbox
+                try:
+                    first_cb = page.locator(
+                        '.item_warp .checkbox').first
+                    if await first_cb.count() > 0:
+                        await first_cb.click(force=True)
+                        selected = True
+                        await asyncio.sleep(0.3)
+                except Exception:
+                    pass
+
+            # 点击结算按钮
+            settle_btn = None
+            for text in ["结算", "去结算", "立即结算"]:
+                loc = page.get_by_text(text, exact=False)
+                try:
+                    await loc.first.wait_for(state="visible", timeout=2000)
+                    settle_btn = loc.first
+                    break
+                except Exception:
+                    continue
+
+            if not settle_btn:
+                if attempt < PHASE2_RETRIES - 1:
+                    await asyncio.sleep(0.1)
+                    continue
+                return "P2: 未找到结算按钮"
+
+            await settle_btn.click(timeout=3000, force=True)
+
+            # 等待跳转到订单确认页
+            await asyncio.sleep(1)
+
+            # 提交订单
             submit_btn = page.get_by_text("提交订单")
             await submit_btn.wait_for(state="visible", timeout=8000)
-            await submit_btn.click()
-            return "抢购请求已提交"
+            await submit_btn.click(force=True)
+            return "P2-购物车: 抢购请求已提交"
 
         except Exception as e:
-            if attempt < max_retries - 1:
+            if attempt < PHASE2_RETRIES - 1:
+                await asyncio.sleep(0.1)
                 continue
-            return f"抢购操作失败: {e}"
+            return f"P2-购物车: {e}"
 
-    return "抢购操作失败: 重试次数用尽"
+    return "P2-购物车: 重试次数用尽"
+
+
+# ─────────────────────────────────────────────
+#  公共工具
+# ─────────────────────────────────────────────
+async def _handle_sku(page):
+    """处理 SKU 选择弹窗"""
+    try:
+        confirm_btn = page.get_by_text("确定", exact=True)
+        await confirm_btn.first.wait_for(state="visible", timeout=2000)
+        sku_sel = ('.sku-item:not(.disabled), '
+                   '.sku_item:not(.disabled), '
+                   '[class*="sku"] [class*="item"]'
+                   ':not([class*="disabled"])')
+        sku_items = page.locator(sku_sel)
+        if await sku_items.count() > 0:
+            await sku_items.first.click(force=True)
+            await asyncio.sleep(0.1)
+        await confirm_btn.first.click(timeout=2000, force=True)
+    except Exception:
+        pass
+
+
+def _is_success(result):
+    return isinstance(result, str) and ('已提交' in result or '已发送' in result)
+
+
+def _pick_success(results):
+    for r in results:
+        if _is_success(r):
+            return r
+    for r in results:
+        if isinstance(r, str):
+            return r
+    return f"全部失败: {results}"
 
 
 def _update_task(db, task_id, status, result):
     db.execute(
         "UPDATE tasks SET status = ?, result = ?, updated_at = ? WHERE id = ?",
-        (status, result, datetime.now().strftime('%Y-%m-%d %H:%M:%S'), task_id)
-    )
+        (status, result, datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
+         task_id))
     db.commit()

templates/base.html

@@ -35,6 +35,8 @@
                 <a class="nav-link" href="/tasks/"><i class="bi bi-list-task"></i> 任务</a>
                 <a class="nav-link" href="/accounts/"><i class="bi bi-people"></i> 账号</a>
                 <a class="nav-link" href="/orders/"><i class="bi bi-receipt"></i> 订单</a>
+                <a class="nav-link text-warning-emphasis" href="/change_password" title="修改密码"><i class="bi bi-key"></i></a>
+                <a class="nav-link text-warning-emphasis" href="/logout" title="退出"><i class="bi bi-box-arrow-right"></i></a>
             </div>
         </div>
     </nav>

templates/change_password.html

@@ -0,0 +1,33 @@
+{% extends "base.html" %}
+{% block content %}
+<div class="row justify-content-center">
+    <div class="col-md-5">
+        <div class="card">
+            <div class="card-body p-4">
+                <h5 class="mb-3"><i class="bi bi-key"></i> 修改密码</h5>
+                {% if error %}
+                <div class="alert alert-danger py-2" style="border-radius:10px;font-size:.85rem">{{ error }}</div>
+                {% endif %}
+                {% if success %}
+                <div class="alert alert-success py-2" style="border-radius:10px;font-size:.85rem">{{ success }}</div>
+                {% endif %}
+                <form method="POST">
+                    <div class="mb-3">
+                        <label class="form-label">原密码</label>
+                        <input type="password" class="form-control" name="old_password" required>
+                    </div>
+                    <div class="mb-3">
+                        <label class="form-label">新密码（至少6位）</label>
+                        <input type="password" class="form-control" name="new_password" required minlength="6">
+                    </div>
+                    <div class="mb-3">
+                        <label class="form-label">确认新密码</label>
+                        <input type="password" class="form-control" name="new_password2" required minlength="6">
+                    </div>
+                    <button type="submit" class="btn btn-primary w-100">确认修改</button>
+                </form>
+            </div>
+        </div>
+    </div>
+</div>
+{% endblock %}

templates/login.html

@@ -0,0 +1,44 @@
+<!DOCTYPE html>
+<html lang="zh-CN">
+<head>
+    <meta charset="UTF-8">
+    <meta name="viewport" content="width=device-width, initial-scale=1.0">
+    <title>登录 - 微店抢购管理</title>
+    <link href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.3/dist/css/bootstrap.min.css" rel="stylesheet">
+    <link href="https://cdn.jsdelivr.net/npm/bootstrap-icons@1.11.3/font/bootstrap-icons.min.css" rel="stylesheet">
+    <style>
+        body { background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); min-height: 100vh;
+               display: flex; align-items: center; justify-content: center;
+               font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif; }
+        .card { width: 360px; border-radius: 18px; border: none; box-shadow: 0 8px 40px rgba(0,0,0,.2); }
+        .card-header { text-align: center; padding: 2rem 2rem 1rem; background: none; border: none; }
+        .card-header i { font-size: 2.5rem; color: #667eea; }
+        .card-header h4 { margin-top: .8rem; font-weight: 600; color: #333; }
+        .card-body { padding: 0 2rem 2rem; }
+        .form-control { border-radius: 10px; padding: .6rem 1rem; }
+        .btn-primary { background: linear-gradient(135deg, #667eea, #764ba2); border: none;
+                       border-radius: 10px; padding: .6rem; font-weight: 500; width: 100%; }
+        .alert { border-radius: 10px; font-size: .85rem; }
+    </style>
+</head>
+<body>
+    <div class="card">
+        <div class="card-header">
+            <i class="bi bi-lightning-charge-fill"></i>
+            <h4>微店抢购管理</h4>
+        </div>
+        <div class="card-body">
+            {% if error %}
+            <div class="alert alert-danger py-2">{{ error }}</div>
+            {% endif %}
+            <form method="POST">
+                <div class="mb-3">
+                    <input type="password" class="form-control" name="password"
+                           placeholder="请输入访问密码" autofocus required>
+                </div>
+                <button type="submit" class="btn btn-primary">登 录</button>
+            </form>
+        </div>
+    </div>
+</body>
+</html>

templates/setup.html

@@ -0,0 +1,50 @@
+<!DOCTYPE html>
+<html lang="zh-CN">
+<head>
+    <meta charset="UTF-8">
+    <meta name="viewport" content="width=device-width, initial-scale=1.0">
+    <title>初始设置 - 微店抢购管理</title>
+    <link href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.3/dist/css/bootstrap.min.css" rel="stylesheet">
+    <link href="https://cdn.jsdelivr.net/npm/bootstrap-icons@1.11.3/font/bootstrap-icons.min.css" rel="stylesheet">
+    <style>
+        body { background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); min-height: 100vh;
+               display: flex; align-items: center; justify-content: center;
+               font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif; }
+        .card { width: 380px; border-radius: 18px; border: none; box-shadow: 0 8px 40px rgba(0,0,0,.2); }
+        .card-header { text-align: center; padding: 2rem 2rem 1rem; background: none; border: none; }
+        .card-header i { font-size: 2.5rem; color: #667eea; }
+        .card-header h4 { margin-top: .8rem; font-weight: 600; color: #333; }
+        .card-header p { color: #888; font-size: .85rem; }
+        .card-body { padding: 0 2rem 2rem; }
+        .form-control { border-radius: 10px; padding: .6rem 1rem; }
+        .btn-primary { background: linear-gradient(135deg, #667eea, #764ba2); border: none;
+                       border-radius: 10px; padding: .6rem; font-weight: 500; width: 100%; }
+        .alert { border-radius: 10px; font-size: .85rem; }
+    </style>
+</head>
+<body>
+    <div class="card">
+        <div class="card-header">
+            <i class="bi bi-shield-lock-fill"></i>
+            <h4>首次使用设置</h4>
+            <p>请设置管理密码（至少6位）</p>
+        </div>
+        <div class="card-body">
+            {% if error %}
+            <div class="alert alert-danger py-2">{{ error }}</div>
+            {% endif %}
+            <form method="POST">
+                <div class="mb-3">
+                    <input type="password" class="form-control" name="password"
+                           placeholder="设置密码" autofocus required minlength="6">
+                </div>
+                <div class="mb-3">
+                    <input type="password" class="form-control" name="password2"
+                           placeholder="确认密码" required minlength="6">
+                </div>
+                <button type="submit" class="btn btn-primary">确认设置</button>
+            </form>
+        </div>
+    </div>
+</body>
+</html>

utils/timer.py

@@ -3,47 +3,67 @@ import asyncio
 import ntplib
 from datetime import datetime
 
+
 class PrecisionTimer:
     def __init__(self):
         self.offset = 0  # 服务器时间 - 本地时间
 
     def sync_time(self):
-        """
-        同步 NTP 时间，计算偏移量
-        """
+        """多次 NTP 同步取中位数，提高精度"""
+        offsets = []
+        servers = ['ntp.aliyun.com', 'ntp.tencent.com', 'pool.ntp.org']
+        for server in servers:
             try:
                 client = ntplib.NTPClient()
-            response = client.request('pool.ntp.org', version=3)
-            self.offset = response.tx_time - time.time()
-            print(f"时间同步完成，偏移量: {self.offset:.3f}s")
-        except Exception as e:
-            print(f"NTP同步失败: {e}，将使用系统时间")
+                resp = client.request(server, version=3)
+                offsets.append(resp.tx_time - time.time())
+            except Exception:
+                continue
+        if offsets:
+            offsets.sort()
+            self.offset = offsets[len(offsets) // 2]
+            print(f"时间同步完成，偏移量: {self.offset:.3f}s (采样{len(offsets)}个)")
+        else:
+            print("NTP同步失败，将使用系统时间")
 
     def get_server_time(self):
         return time.time() + self.offset
 
     async def wait_until(self, target_time_str):
-        """
-        等待直到目标时间 (格式: 2026-02-01 10:00:00)
-        """
+        """等待直到目标时间，最后阶段忙等保证精度"""
         target_dt = datetime.strptime(target_time_str, "%Y-%m-%d %H:%M:%S")
-        target_timestamp = target_dt.timestamp()
+        target_ts = target_dt.timestamp()
 
-        print(f"正在等待目标时间: {target_time_str}")
+        print(f"等待目标时间: {target_time_str}")
 
         while True:
-            current_time = self.get_server_time()
-            remaining = target_timestamp - current_time
+            now = self.get_server_time()
+            remaining = target_ts - now
 
             if remaining <= 0:
-                print("目标时间已到！触发抢购！")
+                print("目标时间已到！")
                 break
-            
-            # 动态调整调整休眠时间以节省 CPU 并保持精度
-            if remaining > 1:
-                await asyncio.sleep(remaining - 0.5)
+            elif remaining > 10:
+                await asyncio.sleep(remaining - 10)
+            elif remaining > 2:
+                await asyncio.sleep(0.5)
+            elif remaining > 0.1:
+                await asyncio.sleep(0.01)
+            # remaining <= 0.1: 忙等，不 sleep
+
+    async def wait_until_early(self, target_time_str, early_ms=500):
+        """提前 early_ms 毫秒触发，用于需要预操作的场景"""
+        target_dt = datetime.strptime(target_time_str, "%Y-%m-%d %H:%M:%S")
+        target_ts = target_dt.timestamp() - (early_ms / 1000.0)
+
+        while True:
+            now = self.get_server_time()
+            remaining = target_ts - now
+            if remaining <= 0:
+                break
+            elif remaining > 10:
+                await asyncio.sleep(remaining - 10)
+            elif remaining > 2:
+                await asyncio.sleep(0.5)
             elif remaining > 0.1:
                 await asyncio.sleep(0.01)
-            else:
-                # 最后一刻进入忙等以获取最高精度
-                pass