淘宝“秒杀”脚本

python + selenium 实现，定时自动结算购物车，邮件提醒结果。

说是秒杀脚本，但其实根本做不到毫秒级（看很多文章写毫秒级也是跪了）。自己在mac上实测，大约10s左右会收到邮件。用selenium秒杀是不要想了，用作自动提交订单，还算ok。

整个流程比较简单，扫码登陆，跳转到购物车界面，不停刷新，等时间差不多到了，全选，提交订单，做个练手脚本，简单实用。

通知截图：

整个代码如下，邮箱相关参数读取部分懒得写全了，就这样吧。

# -*- coding: UTF-8 -*-
import argparse
from selenium import webdriver
from selenium.webdriver.common.action_chains import ActionChains
from selenium.webdriver.support.wait import WebDriverWait
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.by import By
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
import datetime
import time
import logging
import sys
from retry import retrylogging.basicConfig(level=logging.INFO, format=&＃39;%(asctime)s - %(funcName)s - %(levelname)s - %(message)s&＃39;)
logger = logging.getLogger(&＃39;Buyer&＃39;)class EmailInfo(object):def __init__(self):self.port = 25self.mail_host = &＃39;xxxxxxxxx&＃39;self.mail_user = &＃39;xxxxxxxxx&＃39;self.mail_pass = &＃39;xxxxxxx&＃39;self.sender = &＃39;xxxxxx&＃39;self.receivers = &＃39;xxxxxxxxx&＃39;class retryTimes():def __init__(self, retry_times=3):self.retry_times = retry_timesclass BUY_MASHINE():def __init__(self, buy_time, max_retry=3):self.BUY_TIME = buy_timeself.BUY_TIME_OBJECT = datetime.datetime.strptime(self.BUY_TIME, &＃39;%Y-%m-%d %H:%M:%S&＃39;)self.LOG_STATUS = 0 # 未登录self.MAX_RETRY = max_retryself.TAOBAO = "https://login.taobao.com/"self.MAX_SUBMIT = 50#self.BUY_TYPE = buy_type # 0：秒杀 1：到点改价self.Email = EmailInfo()def init(self):self.option = Options()self.option.add_argument(&＃39;disable-infobars&＃39;) # 隐藏 启动浏览器出现‘Chrome正在受到自动软件的控制&＃39;self.option.add_argument(&＃39;log-level=3&＃39;) # LOG_FATALself.option.add_experimental_option(&＃39;excludeSwitches&＃39;, [&＃39;enable-automation&＃39;])# prefs = {"profile.managed_default_content_settings.images": 2}# self.option.add_experimental_option("prefs", prefs)self.driver = webdriver.Chrome(chrome_optiOns=self.option)#self.driver.maximize_window()def switch_to_qr_load(self):# 点击扫码界面locator_qr = (By.ID, &＃39;J_Static2Quick&＃39;)WebDriverWait(self.driver, 30, 0.5).until(EC.presence_of_element_located(locator_qr))logger.info(&＃39;二维码已载入&＃39;)qr = self.driver.find_element_by_id(&＃39;J_Static2Quick&＃39;)if qr.is_displayed():logger.info(&＃39;账号密码已显示&＃39;)ActionChains(self.driver).move_to_element_with_offset(qr, 0, 0).click().perform()# ActionChains(dr).move_to_element(ma).click().perform() # 此方法无效logger.info(&＃39;已切换为二维码登录&＃39;)# 请点击刷新# 扫描成功def login(self):logger.info("开始登陆===>>>")self.driver.get(self.TAOBAO)time.sleep(1)max_tries=retryTimes(self.MAX_RETRY)@retry(tries=self.MAX_RETRY, delay=1)def login_with_retry(**kargs):cnt = 0max_tries.retry_times -= 1for key in kargs.keys():kargs[key]while True:if cnt == 60: # 1min 刷新if max_tries.retry_times > 0:logger.info("retrying...")raise NotImplementedError(&＃39;login&＃39;)else:return Falseif self.driver.current_url != self.TAOBAO:logger.info(&＃39;扫描成功&＃39;)# page_source = self.driver.page_source# logger.debug(page_source)return Truetime.sleep(1)cnt += 1return login_with_retry(func1=self.driver.refresh(), func2=self.switch_to_qr_load())def working(self):self.driver.get("https://cart.taobao.com/cart.htm")time.sleep(1)self.wait()self.driver.get("https://cart.taobao.com/cart.htm")time.sleep(1)# 点击购物车里全选按钮if self.driver.find_element_by_id("J_SelectAll1"):self.driver.find_element_by_id("J_SelectAll1").click()logger.info("已经选中购物车中全部商品 ...")self.buying()def buying(self):max_submit = retryTimes(self.MAX_SUBMIT)# 60～31@retry(tries=-1)def submit_with_retry(driver):try:driver.find_element_by_link_text(&＃39;提交订单&＃39;).click()logger.info("已经点击提交订单按钮")return Trueexcept Exception as e:if max_submit.retry_times <= 0:logger.info("提交订单失败...")return Falselogger.info("没发现提交订单按钮，可能页面还没加载出来，重试...")raise NotImplementedError(&＃39;submit&＃39;)finally:max_submit.retry_times -= 1@retry(tries=-1)def buying_with_retry(driver, deadline, ending=60):now = datetime.datetime.now()if now >= self.BUY_TIME_OBJECT:# 点击结算按钮if self.driver.find_element_by_id("J_Go"):self.driver.find_element_by_id("J_Go").click()logger.info("已经点击结算按钮...")return submit_with_retry(driver)if (now - deadline).seconds >= ending:logger.info("Orz...buying failed at {}".format(now))return Falseelse:logger.info(&＃39;finally waiting...&＃39;)raise NotImplementedError(&＃39;buying&＃39;)ret = buying_with_retry(self.driver, self.BUY_TIME_OBJECT)self.sendMail(ret, datetime.datetime.now())return retdef wait(self):while True:currentTime = datetime.datetime.now()if (self.BUY_TIME_OBJECT - currentTime).seconds > 90:self.driver.refresh()logger.info("刷新购物车界面，防止登录超时...")time.sleep(60)else:logger.info("抢购时间点将近，停止自动刷新，准备进入抢购阶段...")breakdef sendMail(self, Result, ok_time):class retry_times():num = 3b_succeed = Falsetries = retry_times()message = MIMEMultipart(&＃39;mixed&＃39;)message[&＃39;Subject&＃39;] = &＃39;秒杀通知&＃39;message[&＃39;From&＃39;] = self.Email.sendermessage[&＃39;To&＃39;] = self.Email.receiversif Result:text = &＃39;{} 秒杀成功，请尽快付款～&＃39;.format(ok_time)else:text = &＃39;{} 秒杀失败，省钱啦～&＃39;.format(ok_time)text_plain = MIMEText(text, &＃39;plain&＃39;, &＃39;utf-8&＃39;)message.attach(text_plain)@retry(tries=3, delay=30, backoff=2)def sendMailKernel():tries.num -= 1try:tries.b_succeed = TruesmtpObj = smtplib.SMTP()smtpObj.connect(self.Email.mail_host, self.Email.port)smtpObj.ehlo()smtpObj.starttls()smtpObj.login(self.Email.mail_user, self.Email.mail_pass)smtpObj.sendmail(self.Email.sender, self.Email.receivers, message.as_string())smtpObj.quit()logger.info(&＃39;send mail succeed.&＃39;)except smtplib.SMTPException as e:tries.b_succeed = Falselogger.info(&＃39;send mail failed&＃39;)if tries.num > 0:logger.info(&＃39;SendEmail retry&＃39;)raise NotImplementedError(&＃39;Email&＃39;)sendMailKernel()return tries.b_succeedif __name__ == &＃39;__main__&＃39;:parser = argparse.ArgumentParser()parser.add_argument(&＃39;--buy_time&＃39;, type=str, help=&＃39;buy time like \&＃39;2019-11-11 00:00:00\&＃39;&＃39;)parser.add_argument(&＃39;--login_retry&＃39;, type=int, help=&＃39;max retry login times.(default: 3)&＃39;, default=3)parser.add_argument(&＃39;--email_reminder&＃39;, type=bool, help=&＃39;use email to notice result.(default: False)&＃39;, default=False)parser.add_argument(&＃39;--port&＃39;, type=int, help=&＃39;set email port.(default: 25)&＃39;, default=25)parser.add_argument(&＃39;--mail_host&＃39;, type=str, help=&＃39;email host.(default: 163--\&＃39;smtp.163.com\&＃39;)&＃39;, default=&＃39;smtp.163.com&＃39;)parser.add_argument(&＃39;--mail_count&＃39;, type=str, help=&＃39;email count.&＃39;)parser.add_argument(&＃39;--mail_pass&＃39;, type=str, help=&＃39;email password.&＃39;)parser.add_argument(&＃39;--mail_receivers&＃39;, type=str, help=&＃39;receive email count.(default: sender)&＃39;)args = parser.parse_args()if not args.buy_time:logger.info("buy_time must specify.")sys.exit(-1)buy_time = args.buy_timemax_retry = args.login_retrybuyer = BUY_MASHINE(buy_time, max_retry)buyer.init()buyer.login()buyer.working()

参考链接：https://www.jianshu.com/p/5fe9bd8f7674