damai/ticket_script.py

# coding: utf-8
"""
大麦网抢票核心脚本
支持：大麦网 (damai.cn) 移动端页面
"""

import json
import logging
import pickle
import sys
import time
from dataclasses import dataclass, field
from enum import Enum, auto
from os.path import exists
from pathlib import Path
from typing import List, Optional

from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.common.by import By
from selenium.webdriver.remote.webelement import WebElement
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.support.ui import WebDriverWait
from selenium.common.exceptions import (
    NoSuchElementException,
    TimeoutException,
    WebDriverException,
)

logger = logging.getLogger("damai")


# ── 状态机 ───────────────────────────────────────────────────────────────

class State(Enum):
    INIT = auto()
    LOGGED_IN = auto()
    PAGE_LOADED = auto()
    TICKET_SELECTED = auto()
    ORDER_CONFIRMED = auto()
    SUCCESS = auto()
    FAILED = auto()


# ── 配置 ─────────────────────────────────────────────────────────────────

@dataclass
class TicketConfig:
    date: List[int]           # 日期优先级序号 (1-based)
    sess: List[int]           # 场次优先级序号 (1-based)
    price: List[int]          # 票价优先级序号 (1-based)
    ticket_num: int           # 购买张数
    viewer_person: List[int]  # 观影人序号 (1-based)
    nick_name: str            # 用户昵称
    damai_url: str            # 大麦首页
    target_url: str           # 目标演出 URL
    driver_path: str          # chromedriver 路径
    cookies_file: str = "cookies.pkl"
    max_retries: int = 180
    retry_delay: float = 0.3
    headless: bool = False
    mobile_emulation: bool = True


# ── 工具函数 ─────────────────────────────────────────────────────────────

def safe_click(driver, element: WebElement, delay: float = 0.05):
    """带滚动和延迟的安全点击"""
    driver.execute_script("arguments[0].scrollIntoView({block:'center'});", element)
    time.sleep(delay)
    element.click()


def has_class(element: WebElement, class_name: str) -> bool:
    """检查元素是否包含指定 class"""
    classes = element.get_attribute("class") or ""
    return class_name in classes.split()


def find_child(element: WebElement, by: str, value: str) -> Optional[WebElement]:
    """安全查找子元素，找不到返回 None"""
    try:
        return element.find_element(by=by, value=value)
    except NoSuchElementException:
        return None


def find_children(element: WebElement, by: str, value: str) -> List[WebElement]:
    """安全查找子元素列表"""
    try:
        return element.find_elements(by=by, value=value)
    except NoSuchElementException:
        return []


# ── 核心类 ───────────────────────────────────────────────────────────────

class TicketBot:
    def __init__(self, config: TicketConfig):
        self.config = config
        self.driver: Optional[webdriver.Chrome] = None
        self.state = State.INIT
        self.attempts = 0
        self.start_time = 0.0
        self.end_time = 0.0

    # ── 浏览器管理 ──

    def _build_driver(self) -> webdriver.Chrome:
        options = Options()

        if self.config.mobile_emulation:
            options.add_experimental_option(
                "mobileEmulation", {"deviceName": "Nexus 6"}
            )

        options.add_argument("--disable-blink-features=AutomationControlled")
        options.add_experimental_option("excludeSwitches", ["enable-automation"])
        options.add_experimental_option("useAutomationExtension", False)

        # 性能优化：不加载图片
        prefs = {
            "profile.managed_default_content_settings.images": 2,
            "profile.managed_default_content_settings.fonts": 2,
        }
        options.add_experimental_option("prefs", prefs)

        # 页面加载策略：不等完全加载
        options.page_load_strategy = "eager"

        if self.config.headless:
            options.add_argument("--headless=new")

        if sys.platform != "win32":
            options.add_argument("--no-sandbox")
            options.add_argument("--disable-dev-shm-usage")

        service = Service(executable_path=self.config.driver_path)
        driver = webdriver.Chrome(service=service, options=options)

        # 隐藏 webdriver 特征
        driver.execute_cdp_cmd(
            "Page.addScriptToEvaluateOnNewDocument",
            {
                "source": """
                Object.defineProperty(navigator, 'webdriver', {get: () => undefined});
                window.navigator.chrome = {runtime: {}};
                Object.defineProperty(navigator, 'plugins', {get: () => [1,2,3,4,5]});
                Object.defineProperty(navigator, 'languages', {get: () => ['zh-CN','en-US','en']});
            """
            },
        )
        return driver

    def start_browser(self):
        """启动浏览器"""
        logger.info("启动浏览器...")
        self.driver = self._build_driver()

    def close_browser(self):
        """关闭浏览器"""
        if self.driver:
            try:
                self.driver.quit()
            except Exception:
                pass
            self.driver = None

    # ── Cookie 管理 ──

    def _save_cookies(self):
        cookies_path = Path(self.config.cookies_file)
        with open(cookies_path, "wb") as f:
            pickle.dump(self.driver.get_cookies(), f)
        logger.info(f"Cookie 已保存到 {cookies_path}")

    def _load_cookies(self) -> bool:
        cookies_path = Path(self.config.cookies_file)
        if not cookies_path.exists():
            return False
        try:
            with open(cookies_path, "rb") as f:
                cookies = pickle.load(f)
            self.driver.get(self.config.damai_url)
            time.sleep(1)
            for cookie in cookies:
                # selenium 要求 domain 不能带当前域之外的 cookie
                cookie.pop("sameSite", None)
                try:
                    self.driver.add_cookie(cookie)
                except Exception:
                    continue
            logger.info("Cookie 已加载")
            return True
        except Exception as e:
            logger.warning(f"加载 Cookie 失败: {e}")
            return False

    # ── 登录 ──

    def login(self) -> bool:
        """登录大麦网，需要用户扫码"""
        logger.info("开始登录流程...")
        self.driver.get(self.config.damai_url)
        time.sleep(2)

        # 尝试加载已有 cookie
        if self._load_cookies():
            self.driver.get(self.config.target_url)
            time.sleep(2)
            # 检查是否已登录
            if self._is_logged_in():
                logger.info("Cookie 登录成功")
                self.state = State.LOGGED_IN
                return True

        # Cookie 失效，需要扫码登录
        logger.info("请扫码登录...")
        self.driver.get(self.config.damai_url)
        time.sleep(1)

        try:
            login_btn = WebDriverWait(self.driver, 10).until(
                EC.element_to_be_clickable((By.CLASS_NAME, "login-user"))
            )
            login_btn.click()
        except TimeoutException:
            logger.error("找不到登录按钮")
            return False

        # 等待用户扫码完成（标题变化）
        logger.info("等待扫码登录...")
        original_title = self.driver.title
        timeout = 120  # 最多等 2 分钟
        start = time.time()
        while time.time() - start < timeout:
            if self.driver.title != original_title and "登录" not in self.driver.title:
                break
            time.sleep(1)
        else:
            logger.error("扫码超时")
            return False

        self._save_cookies()
        self.state = State.LOGGED_IN
        logger.info("登录成功")
        return True

    def _is_logged_in(self) -> bool:
        """检查是否已登录"""
        try:
            self.driver.find_element(By.CLASS_NAME, "login-user")
            return False
        except NoSuchElementException:
            return True

    # ── 选票核心逻辑 ──

    def enter_target_page(self):
        """进入目标演出页面"""
        logger.info(f"打开目标页面: {self.config.target_url}")
        self.driver.get(self.config.target_url)
        try:
            WebDriverWait(self.driver, 15).until(
                EC.presence_of_element_located((By.ID, "app"))
            )
        except TimeoutException:
            raise RuntimeError("页面加载超时")
        self.state = State.PAGE_LOADED

    def _select_date(self, container: WebElement) -> bool:
        """选择日期"""
        try:
            date_section = WebDriverWait(self.driver, 3).until(
                EC.presence_of_element_located(
                    (By.CLASS_NAME, "bui-dm-sku-calendar")
                )
            )
        except TimeoutException:
            logger.debug("没有日期选择器，跳过")
            return True

        date_items = find_children(
            date_section, By.CLASS_NAME, "bui-calendar-day-box"
        )
        if not date_items:
            logger.warning("未找到日期选项")
            return False

        for idx in self.config.date:
            if idx <= len(date_items):
                safe_click(self.driver, date_items[idx - 1])
                logger.info(f"选择日期: 第{idx}个")
                time.sleep(0.3)
                return True

        return False

    def _select_session(self) -> bool:
        """选择场次"""
        try:
            session_section = WebDriverWait(self.driver, 3).until(
                EC.presence_of_element_located(
                    (By.CLASS_NAME, "sku-times-card")
                )
            )
        except TimeoutException:
            logger.debug("没有场次选择器")
            return True

        session_items = find_children(
            session_section, By.CLASS_NAME, "bui-dm-sku-card-item"
        )
        if not session_items:
            logger.warning("未找到场次选项")
            return False

        for idx in self.config.sess:
            if idx > len(session_items):
                continue
            item = session_items[idx - 1]
            tag = find_child(item, By.CLASS_NAME, "item-tag")

            if tag:
                tag_text = tag.text.strip()
                if tag_text == "无票":
                    logger.debug(f"场次{idx}: 无票，跳过")
                    continue
                # 预售/惠/有票 → 可选

            safe_click(self.driver, item)
            logger.info(f"选择场次: 第{idx}个")
            time.sleep(0.3)
            return True

        logger.warning("所有场次均无票")
        return False

    def _select_price(self) -> bool:
        """选择票价档位"""
        try:
            price_section = WebDriverWait(self.driver, 3).until(
                EC.presence_of_element_located(
                    (By.CLASS_NAME, "sku-tickets-card")
                )
            )
        except TimeoutException:
            logger.warning("找不到票价选择区域")
            return False

        price_items = find_children(
            price_section, By.CLASS_NAME, "bui-dm-sku-card-item"
        )
        if not price_items:
            logger.warning("未找到票价选项")
            return False

        for idx in self.config.price:
            if idx > len(price_items):
                continue
            item = price_items[idx - 1]
            tag = find_child(item, By.CLASS_NAME, "item-tag")

            if tag and tag.text.strip() in ("缺货登记", "notticket"):
                logger.debug(f"票价{idx}: 缺货，跳过")
                continue

            safe_click(self.driver, item)
            logger.info(f"选择票价: 第{idx}个")
            time.sleep(0.3)
            return True

        logger.warning("所有票价档位均缺货")
        return False

    def _click_buy_button(self) -> str:
        """点击购买按钮，返回按钮文字"""
        try:
            box = WebDriverWait(self.driver, 5).until(
                EC.presence_of_element_located((By.ID, "app"))
            )
        except TimeoutException:
            raise RuntimeError("页面刷新失败")

        # 关闭实名弹窗
        popups = find_children(
            box, By.XPATH, "//div[@class='realname-popup']"
        )
        if popups:
            try:
                btn = popups[0].find_element(
                    By.XPATH, ".//div[@class='operate']//div[@class='button']"
                )
                btn.click()
                time.sleep(0.3)
                logger.info("已关闭实名弹窗")
            except NoSuchElementException:
                pass

        try:
            buy_btn = box.find_element(By.CLASS_NAME, "buy__button")
            time.sleep(0.5)
            btn_text = buy_btn.text.strip()
        except NoSuchElementException:
            raise RuntimeError("找不到购买按钮")

        if "即将开抢" in btn_text:
            raise RuntimeError("尚未开售，刷新等待")

        if "缺货" in btn_text:
            raise RuntimeError("已缺货")

        buy_btn.click()
        logger.info(f"点击购买按钮: {btn_text}")
        return btn_text

    def _select_tickets_in_popup(self) -> str:
        """在弹窗中选票，返回最终按钮文字"""
        try:
            box = WebDriverWait(self.driver, 3).until(
                EC.presence_of_element_located(
                    (By.CSS_SELECTOR, ".sku-pop-wrapper")
                )
            )
        except TimeoutException:
            raise RuntimeError("选票弹窗未出现")

        self._select_date(box)
        self._select_session()
        self._select_price()

        # 等待计数器出现
        try:
            WebDriverWait(self.driver, 3).until(
                EC.presence_of_element_located(
                    (By.CLASS_NAME, "bui-dm-sku-counter")
                )
            )
        except TimeoutException:
            raise RuntimeError("票数选择器未出现")

        # 查找最终提交按钮
        try:
            submit_btn = box.find_element(
                By.CLASS_NAME, "sku-footer-buy-button"
            )
            submit_text = submit_btn.text.strip()
        except NoSuchElementException:
            raise RuntimeError("找不到提交按钮")

        if not submit_text:
            raise RuntimeError("提交按钮文字为空，请调整重试间隔")

        if submit_text == "选座购买":
            submit_btn.click()
            logger.info("进入选座模式，请手动选择座位")
            return "选座"

        if submit_text == "提交缺货登记":
            raise RuntimeError("票已售罄")

        if submit_text in ("立即预订", "立即购买", "确定"):
            # 增加票数
            plus_btn = find_child(box, By.CLASS_NAME, "plus-enable")
            if plus_btn and self.config.ticket_num > 1:
                for _ in range(self.config.ticket_num - 1):
                    plus_btn.click()
                    time.sleep(0.05)
                logger.info(f"设置购票数量: {self.config.ticket_num}")

            time.sleep(1.0)
            submit_btn.click()
            logger.info(f"点击: {submit_text}")
            return submit_text

        raise RuntimeError(f"未知按钮文字: {submit_text}")

    # ── 订单确认 ──

    def _confirm_order(self) -> bool:
        """选择观影人并提交订单"""
        try:
            WebDriverWait(self.driver, 10).until(
                EC.presence_of_element_located(
                    (
                        By.XPATH,
                        '//*[@id="dmViewerBlock_DmViewerBlock"]/div[2]/div/div',
                    )
                )
            )
        except TimeoutException:
            logger.error("订单确认页面加载超时")
            return False

        # 选择观影人
        people = self.driver.find_elements(
            By.XPATH, '//*[@id="dmViewerBlock_DmViewerBlock"]/div[2]/div/div'
        )
        time.sleep(0.3)

        selected = 0
        for idx in self.config.viewer_person:
            if idx > len(people):
                logger.warning(f"观影人序号{idx}超出范围(共{len(people)}人)")
                continue
            people[idx - 1].click()
            selected += 1
            time.sleep(0.05)

        if selected == 0:
            logger.error("未选中任何观影人")
            return False

        logger.info(f"已选择 {selected} 位观影人")

        # 提交订单
        try:
            submit_btn = WebDriverWait(self.driver, 5).until(
                EC.element_to_be_clickable(
                    (
                        By.XPATH,
                        '//*[@id="dmOrderSubmitBlock_DmOrderSubmitBlock"]/div[2]/div/div[2]/div[3]/div[2]',
                    )
                )
            )
            time.sleep(0.5)
            submit_btn.click()
            logger.info("已提交订单")
        except TimeoutException:
            logger.error("找不到提交订单按钮")
            return False

        return True

    def _wait_payment(self, timeout: int = 30) -> bool:
        """等待跳转到支付页面"""
        logger.info("等待跳转支付页面...")
        start = time.time()
        while time.time() - start < timeout:
            title = self.driver.title
            if "支付宝" in title or "支付" in title:
                logger.info("已跳转到支付页面")
                return True
            time.sleep(1)

        logger.warning(f"等待支付页面超时({timeout}s)，当前标题: {self.driver.title}")
        return False

    # ── 主循环 ──

    def choose_and_buy(self) -> bool:
        """选票 + 下单流程"""
        self.attempts += 1

        try:
            self.enter_target_page()
            self._click_buy_button()
            self._select_tickets_in_popup()

            # 等待跳转确认页
            try:
                WebDriverWait(self.driver, 5).until(EC.title_contains("确认"))
            except TimeoutException:
                logger.warning("未跳转到确认页，重试...")

            self.state = State.TICKET_SELECTED

            if self._confirm_order():
                self.state = State.ORDER_CONFIRMED
                self._wait_payment()
                self.state = State.SUCCESS
                return True

        except RuntimeError as e:
            logger.warning(f"第{self.attempts}次尝试失败: {e}")
            # 返回目标页重试
            try:
                self.driver.get(self.config.target_url)
            except WebDriverException:
                pass
            return False
        except WebDriverException as e:
            logger.error(f"浏览器异常: {e}")
            return False

        return False

    def run(self):
        """运行抢票主流程"""
        self.start_time = time.time()

        try:
            self.start_browser()

            if not self.login():
                logger.error("登录失败，退出")
                self.state = State.FAILED
                return

            logger.info(
                f"开始抢票 | 最大重试: {self.config.max_retries}次 | "
                f"重试间隔: {self.config.retry_delay}s"
            )

            for _ in range(self.config.max_retries):
                if self.choose_and_buy():
                    self.end_time = time.time()
                    elapsed = round(self.end_time - self.start_time, 1)
                    logger.info(
                        f"🎉 抢票成功！共 {self.attempts} 轮，耗时 {elapsed}s"
                    )
                    self.state = State.SUCCESS
                    # 不关闭浏览器，让用户自行支付
                    input("按回车键关闭浏览器...")
                    return

                time.sleep(self.config.retry_delay)

            logger.error(f"达到最大重试次数 ({self.config.max_retries})，抢票失败")
            self.state = State.FAILED

        except KeyboardInterrupt:
            logger.info("用户中断")
            self.state = State.FAILED
        except Exception as e:
            logger.error(f"未预期的错误: {e}", exc_info=True)
            self.state = State.FAILED
        finally:
            self.close_browser()


# ── 入口 ─────────────────────────────────────────────────────────────────

def load_config(path: str = "config.json") -> TicketConfig:
    """从 JSON 配置文件加载"""
    with open(path, "r", encoding="utf-8") as f:
        data = json.load(f)
    return TicketConfig(
        date=data["date"],
        sess=data["sess"],
        price=data["price"],
        ticket_num=data["ticket_num"],
        viewer_person=data["viewer_person"],
        nick_name=data["nick_name"],
        damai_url=data["damai_url"],
        target_url=data["target_url"],
        driver_path=data["driver_path"],
        max_retries=data.get("max_retries", 180),
        retry_delay=data.get("retry_delay", 0.3),
    )


if __name__ == "__main__":
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s [%(levelname)s] %(message)s",
        datefmt="%H:%M:%S",
    )

    config_path = sys.argv[1] if len(sys.argv) > 1 else "config.json"
    config = load_config(config_path)

    bot = TicketBot(config)
    bot.run()