-
Notifications
You must be signed in to change notification settings - Fork 4
/
taobaosnap-3.2.6.py
222 lines (216 loc) · 11.6 KB
/
taobaosnap-3.2.6.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
'''
Function:
淘宝抢购脚本
Author:
仰望·星空，K龙，Charles
微信公众号:
红客突击队
邮箱：
hsc_2019@163.com
'''
import requests
import re
import time
import datetime
import json
import urllib
import sys
from selenium import webdriver
from prettytable import PrettyTable
from requests.packages.urllib3 import disable_warnings
from requests.cookies import cookiejar_from_dict
def help_info():
help = """
usage : python {0}
--time Buying time and format: 00:00:00:00000000.
--interval Buying time interval.
--l Buying frequency.
""".format(sys.argv[0])
print(help)
def cookie_info():
chrome_options = webdriver.ChromeOptions()
chrome_options.add_argument("start-maximized")
login_url = 'https://login.taobao.com/member/login.jhtml?spm=a21bo.jianhua.201864-2.d1.5af911d9lhGWni&f=top&redirectURL=http%3A%2F%2Fwww.taobao.com%2F'
driver = webdriver.Chrome(options=chrome_options)
print("请尽快扫码！")
driver.get(login_url)
time.sleep(15) # 预留了安全验证的时间
driver.refresh() # 刷新页面
c = driver.get_cookies()
sessions = dict()
for cookie in c:
sessions[cookie['name']] = cookie['value']
#driver.quit()
return sessions
'''淘宝抢购脚本'''
class TaobaoSnap():
def __init__(self, trybuy_interval=None, Seconds_kill_time=None, number=None):
self.session = requests.Session()
self.cookie = cookie_info()
self.session.cookies = cookiejar_from_dict(self.cookie)
self.trybuy_interval = float(trybuy_interval)
self.Seconds_kill_time = Seconds_kill_time
self.number = int(number)
def run_info(self):
# 获得购物车信息
cart_infos, user_id = self.buycartinfo()
# 解析购物车信息
if not cart_infos['success']:
raise RuntimeError('获取购物车信息失败, 请尝试删除cookie缓存文件后重新扫码登录')
if len(cart_infos['list']) == 0:
raise RuntimeError('购物车是空的, 请在购物车中添加需要抢购的商品')
good_infos = {}
for idx, item in enumerate(cart_infos['list']):
good_info = {
'title': item['bundles'][0]['orders'][0]['title'],
'cart_id': item['bundles'][0]['orders'][0]['cartId'],
'cart_params': item['bundles'][0]['orders'][0]['cartActiveInfo']['cartBcParams'],
'item_id': item['bundles'][0]['orders'][0]['itemId'],
'sku_id': item['bundles'][0]['orders'][0]['skuId'],
'seller_id': item['bundles'][0]['orders'][0]['sellerId'],
'to_buy_info': item['bundles'][0]['orders'][0]['toBuyInfo'],
}
good_infos[str(idx)] = good_info
# 打印并选择想要抢购的商品信息
title, items = ['id', 'title'], []
for key, value in good_infos.items():
items.append([key, value['title']])
self.printTable(title, items)
good_id = input('请选择想要抢购的商品编号(例如"0"): ')
assert good_id in good_infos, '输入的商品编号有误'
# 根据选择尝试购买商品
print(f'[{time.strftime("%H:%M:%S", time.localtime())} INFO]: 正在尝试抢购商品***{good_infos[good_id]["title"]}***')
#time_seckill = "2022-00-00 00:00:00:00000000"
# 对比时间，时间到的话就点击结算
while True:
now = datetime.datetime.now().strftime('%H:%M:%S.%f')
print(now)
if now > self.Seconds_kill_time:
for i in range(self.number):
try:
is_success = self.buygood(good_infos[good_id], user_id)
except Exception as err:
crawler = re.findall("'NoneType' object has no attribute 'group'",str(err))
if "'NoneType' object has no attribute 'group'" in crawler:
print("已触发反爬虫机制，请稍后尝试! 错误信息如下：\n{0}\n".format(err))
# is_success = False
break
else:
print(f'[{time.strftime("%H:%M:%S", time.localtime())} INFO]: 抢购失败, 错误信息如下: \n{err}\n将在{self.trybuy_interval}秒后重新尝试.')
is_success = False
if i == self.number-1 and is_success == False:
print("很遗憾，抢购失败，下次继续加油~")
break
elif is_success:
print(f'[{time.strftime("%H:%M:%S", time.localtime())} INFO]: 抢购***{good_infos[good_id]["title"]}***成功, 已为您自动提交订单, 请尽快前往淘宝完成付款.')
# 电脑语音提示
for _ in range(5):
time.sleep(self.trybuy_interval)
break
break
'''打印表格'''
def printTable(self, title, items):
assert isinstance(title, list) and isinstance(items, list), 'title and items should be list...'
table = PrettyTable(title)
for item in items: table.add_row(item)
print(table)
return table
# 获取购物车
def buycartinfo(self):
url = 'https://cart.taobao.com/cart.htm'
headers = {
'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.88 Safari/537.36',
'sec-fetch-dest': 'document', 'sec-fetch-mode': 'navigate', 'sec-fetch-site': 'none', 'sec-fetch-user': '?1',
'upgrade-insecure-requests': '1',
'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9',
'accept-encoding': 'gzip, deflate, br', 'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8',
'cache-control': 'max-age=0'
}
response = self.session.get(url, headers=headers)
# print(response.text)
response_json = re.search('try{var firstData = (.*?);}catch', response.text).group(1)
response_json = json.loads(response_json)
user_id = re.search('\|\^taoMainUser:(.*?):\^', response.headers['s_tag']).group(1)
return response_json, user_id
# 购买商品
def buygood(self, info, user_id):
# 发送结算请求
url = 'https://buy.taobao.com/auction/order/confirm_order.htm?spm=a1z0d.6639537.0.0.undefined'
headers = {
'cache-control': 'max-age=0', 'upgrade-insecure-requests': '1',
'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.88 Safari/537.36',
'origin': 'https://cart.taobao.com', 'content-type': 'application/x-www-form-urlencoded',
'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9',
'sec-fetch-site': 'same-site', 'sec-fetch-mode': 'navigate', 'sec-fetch-user': '?1',
'sec-fetch-dest': 'document', 'referer': 'https://cart.taobao.com/',
'accept-encoding': 'gzip, deflate, br', 'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8'
}
cart_id, item_id, sku_id, seller_id, cart_params, to_buy_info = info['cart_id'], info['item_id'], info['sku_id'], info['seller_id'], info['cart_params'], info['to_buy_info']
data = {
'item': f'{cart_id}_{item_id}_1_{sku_id}_{seller_id}_0_0_0_{cart_params}_{urllib.parse.quote(str(to_buy_info))}__0',
'buyer_from': 'cart',
'source_time': ''.join(str(int(time.time() * 1000)))
}
disable_warnings()
response = self.session.post(url = url, data = data, headers = headers, verify = False)
order_info = re.search('orderData= (.*?);\n</script>', response.text).group(1)
order_info = json.loads(order_info)
# 发送提交订单请求
token = self.session.cookies['_tb_token_']
endpoint = order_info['endpoint']
data = order_info['data']
structure = order_info['hierarchy']['structure']
hierarchy = order_info['hierarchy']
linkage = order_info['linkage']
linkage.pop('url')
submitref = order_info['data']['submitOrderPC_1']['hidden']['extensionMap']['secretValue']
sparam1 = order_info['data']['submitOrderPC_1']['hidden']['extensionMap']['sparam1']
input_charset = order_info['data']['submitOrderPC_1']['hidden']['extensionMap']['input_charset']
event_submit_do_confirm = order_info['data']['submitOrderPC_1']['hidden']['extensionMap']['event_submit_do_confirm']
url = f'https://buy.taobao.com/auction/confirm_order.htm?x-itemid={item_id}&x-uid={user_id}&submitref={submitref}&sparam1={sparam1}'
data_submit = {}
for key, value in data.items():
if value.get('submit') == 'true' or value.get('submit'):
data_submit[key] = value
data = {
'action': '/order/multiTerminalSubmitOrderAction',
'_tb_token_': token,
'event_submit_do_confirm': '1',
'praper_alipay_cashier_domain': 'cashierrz54',
'input_charset': 'utf-8',
'endpoint': urllib.parse.quote(json.dumps(endpoint)),
'data': urllib.parse.quote(json.dumps(data_submit)),
'hierarchy': urllib.parse.quote(json.dumps({"structure": structure})),
'linkage': urllib.parse.quote(json.dumps(linkage))
}
headers = {
'cache-control': 'max-age=0', 'upgrade-insecure-requests': '1', 'origin': 'https://buy.taobao.com',
'content-type': 'application/x-www-form-urlencoded',
'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.88 Safari/537.36',
'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9',
'sec-fetch-site': 'same-origin', 'sec-fetch-mode': 'navigate', 'sec-fetch-user': '?1',
'sec-fetch-dest': 'document',
'referer': 'https://buy.taobao.com/auction/order/confirm_order.htm?spm=a1z0d.6639537.0.0.undefined',
'accept-encoding': 'gzip, deflate, br', 'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8'
}
response = self.session.post(url, data=data, headers=headers, verify = False)
if response.status_code == 200: return True
return False
if __name__ == '__main__':
args = help_info()
print("请输入对应参数，格式为：--interval [time interval] --time [Starting time] -l [frequency]\n"
"示例：--interval 0.1 --time 15:59:59.90000000 --l 5")
parameter = input("\n参数：")
parameter_list = []
for i in parameter.split(' '):
parameter_list.append(i)
for j in range(len(parameter_list)):
if parameter_list[j] == "--interval":
interval = float(parameter_list[j + 1])
elif parameter_list[j] == "--time":
kill_time = parameter_list[j + 1]
elif parameter_list[j] == "--l":
number = int(parameter_list[j + 1])
client = TaobaoSnap(trybuy_interval=interval ,Seconds_kill_time=kill_time ,number=number)
# client = TaobaoSnap(trybuy_interval=1)
client.run_info()