主要代码如下
import base64
import random
import time
import requests
import functools
import json,sys,hashlib
import os
import pickle
import urllib.parse
import multiprocessing
from encrpty import *
from mi_logger import logger
from timer import Timer
from config import global_config
from concurrent.futures import ProcessPoolExecutor
from util import (
parse_json,
send_wechat,
wait_some_time,
get_random_useragent
)
"""
主要用到的加密算法有两个：
1、登录接口的加密算法，包括请求头和body，具体加密逻辑在其他文件。
2、抢购、预约的入参加密，包括 yp-srs、 yp-ss、 yp-srt三个值。
具体加密逻辑比较敏感，暂不公开。
"""
class MiSeckill(object):
def __init__(self , user_name):
# 初始化信息
self.act_id = global_config.getRaw('config', 'act_id')
self.timers = Timer()
self.spider_session = SpiderSession(user_name)
self.spider_session.load_cookies_from_local()
self.login = Login(self.spider_session,user_name)
self.session = self.spider_session.get_session()
self.user_agent = self.spider_session.user_agent
self.locationUrl = None
def login_by_accountAndPassword(self):
"""
登陆
:return:
"""
if self.login.is_login:
logger.info('登录成功')
return
self.login.login_by_accountAndPassword()
if self.login.is_login:
#将cookie记录在本地
self.spider_session.save_cookies_to_local(self.login.nickName)
else:
raise logger.error("账号密码登录失败！")
def check_login(func):
"""
用户登陆态校验装饰器。若用户未登陆，则调用登陆
"""
@functools.wraps(func)
def new_func(self, *args, **kwargs):
if not self.login.is_login:
self.login_by_accountAndPassword()
return func(self, *args, **kwargs)
return new_func
@check_login
def appoint(self):
"""
预约
"""
self._appoint()
def getSignByData(self,data):
#核心加密，隐藏
pass
re ={
'yp-srt':xxx,
'yp-srs': xxx,
'yp-ss': xxx
}
return re
def isAppoint(self):
# 核心代码隐藏
pass
re = self.session.post("xxxxxxxxxxxxx",
data=json.dumps(data), headers=header)
if re.json().get('code') == 0 and re.json().get('success') and re.json().get('data').get('userStatusInfo'):
return re.json().get('data').get('userStatusInfo').get('ordered')
else:
logger.error('获取今日商品信息出错，程序继续执行。')
return False
def _appoint(self):
self.getTodayActid()
if self.isAppoint() :
logger.info("---------您已预约成功，无需再次预约！----------")
return True
nowtime = int(round(time.time() * 1000))
data = [{}, {"actId": self.act_id, "tel": ""}]
#获取校验所需加密参数
signDict = self.getSignByData(data)
headerPhone = {
"accept": "*/*"
, "accept-encoding": "gzip, deflate, br"
, "accept-language": "zh-CN,zh;q=0.9"
, "content-type": "application/json"
,"referer": self.locationUrl
,"user-agent": self.user_agent
, "X-Requested-With": "XMLHttpRequest"
, "yp-srs": signDict['yp-srs']
, "yp-ss": signDict['yp-ss']
, "yp-srt": signDict['yp-srt']
}
re = self.session.post("https://m.xiaomiyoupin.com/mtop/act/orderspike/order?_=" + str(nowtime),
data=json.dumps(data),headers=headerPhone)
if json.loads(re.text).get("code") == 0 and json.loads(re.text).get("success") == True:
logger.info("---------预约成功，记得按时抢购哦！----------")
logger.info("当前预约人数为：{}".format(json.loads(re.text).get("data").get("orderCount") ))
return True
else:
logger.info("预约失败，返回结果为：{}".format(json.loads(re.text)))
return False
@check_login
def seckill(self):
"""
抢购
"""
self._seckill()
def seckill_by_proc_pool(self, work_count=1):
"""
多进程进行抢购
work_count：进程数量
"""
multiprocessing.freeze_support()
with ProcessPoolExecutor(work_count) as pool:
for i in range(work_count):
pool.submit(self.seckill)
def getUserlevel(self):
try :
# 核心代码隐藏
pass
re = self.session.post("xxxx", data=json.dumps(data),
headers=header)
re = re.json().get("data").get("userLevelVO")
# logger.info(re)
if re:
logger.info("获取用户品值为："+str(re.get("score"))+", 活跃度："+str(re.get("activity"))+
", 名誉值："+str(re.get("reputation")))
except Exception as e :
logger.error("获取用户品值出现异常【{}】".format(e))
def getTodayActid(self):
try :
#核心代码隐藏
pass
###返回302转跳链接
if re.headers.get('Location'):
self.locationUrl = re.headers.get('Location')
urlparse = urllib.parse.urlparse(self.locationUrl)
query_dict = urllib.parse.parse_qs(urlparse.query)
self.act_id = query_dict.get('actId')[0]
logger.info("获取到今日飞天茅台商品id为 -----> {}".format(self.act_id))
##访问一次抢购页面
self.session.get(self.locationUrl)
except Exception as e :
logger.error("获取当天商品id出错：{}".format(e))
def _seckill(self):
"""
抢购
"""
a=0
while True:
try:
self.getUserlevel()
self.getTodayActid()
self.timers.start()
while True:
logger.info("第{}次抢购。".format(a))
flag = self.startKill()
#抢购成功直接返回，失败继续
if flag :
return
a = a + 1
wait_some_time()
if a>15 :
logger.info('抢购失败,明天继续努力。')
return
except Exception as e:
logger.info('抢购发生异常，稍后继续执行！', e)
wait_some_time()
def startKill(self):
data = [{}, {"actId": self.act_id,
"token": self.act_id}]
reDict = self.getSignByData(data)
trustType = xxx
headerPhone = {
"accept": "*/*"
, "accept-encoding": "gzip, deflate, br"
, "accept-language": "zh-CN,zh;q=0.9"
, "content-type": "application/json"
,"referer": self.locationUrl
,"user-agent": self.user_agent
, "X-Requested-With": "XMLHttpRequest"
, "c": xxxx
, "d": xxxx
, "yp-srs": reDict['yp-srs']
, "yp-ss": reDict['yp-ss']
, "yp-srt": reDict['yp-srt']
}
re = self.session.post("https://m.xiaomiyoupin.com/mtop/act/orderspike/ekips?_=" + str(int(time.time() * 1000)),
data=json.dumps(data), headers=headerPhone)
if json.loads(re.text).get("code") == 0 :
if json.loads(re.text).get("data").get("success"):
logger.info("-------------成功啦，请作者喝杯咖啡吧！！！---------------")
return True
logger.info("抢购失败:【{}】".format(re.text.strip()))
return False说明
这个脚本是一个用于小米有品的秒杀助手，基于 Python 编写，使用了多个库和工具来处理登录、预约、和秒杀流程。下面详细解析各部分的功能和工作机制：
导入的库和模块
– `base64`, `random`, `time`, `requests`, `functools`, `json`, `sys`, `hashlib`, `os`, `pickle`, `urllib.parse`, `multiprocessing`: 这些是Python标准库，用于处理数据编码、时间操作、网络请求、功能扩展、数据序列化和并行计算。
– `encrpty`, `mi_logger`, `timer`, `config`, `util`: 这些是自定义模块，分别处理加密、日志记录、计时、配置管理和一些实用功能，如解析JSON、发送微信通知、等待随机时间、获取随机用户代理等。
核心类：`MiSeckill`
这个类是秒杀流程的主体，包含了登录、预约、秒杀等功能的实现。
– 初始化 (`__init__`): 设置活动ID、计时器、会话管理、用户代理等。
– 登录 (`login_by_accountAndPassword`): 尝试通过账号和密码登录，并在成功后保存cookie到本地。
– 登录校验装饰器 (`check_login`): 一个装饰器，用于确保在执行任何需要登录的操作前用户已经登录。
– 预约 (`appoint`): 执行预约动作，如果用户已经预约则不再重复预约。
– 获取签名 (`getSignByData`): 用于生成请求的签名，这是秒杀中防止请求伪造的关键步骤，实现细节未公开。
– 检查预约状态 (`isAppoint`): 检查用户是否已经成功预约。
– 执行预约 (`_appoint`): 包含获取当日活动ID和检查预约状态的逻辑。
– 秒杀 (`seckill`): 执行秒杀动作。
– **多进程秒杀** (`seckill_by_proc_pool`): 使用多进程来同时执行多个秒杀任务，提高成功率。
– 获取用户等级 (`getUserlevel`): 获取用户在小米有品上的等级和其他相关信息，可能用于优先级判断。
– 获取当日活动ID (`getTodayActid`): 获取当天的活动ID，这是进行秒杀所必需的。
– 实际执行秒杀 (`_seckill`): 包括用户等级获取、活动ID获取、以及反复尝试秒杀的循环逻辑。
– 开始秒杀 (`startKill`): 根据提供的数据生成签名，构造请求头，然后发送秒杀请求。
实际使用
这个脚本配置了多线程运行方式，可以为每个账户创建一个独立的线程，从而同时执行多个秒杀任务。使用了ProcessPoolExecutor来管理进程池，这是Python `concurrent.futures`模块的一部分，用于简化并行执行的复杂性。
注意事项
– 该脚本涉及加密和网络请求，这可能受到法律和道德的限制，特别是在自动化购买（秒杀）和自动登录等领域。使用此类脚本可能违反服务提供商的使用条款。- 秒杀行为可能对服务器造成压力，且在竞争激烈的情况下可能会导致账户被暂时或永久封禁。
建议
仅用于学习和研究目的。
© 版权声明
1. 转载请在文内以超链形式注明出处，谢谢合作！
2. 本站除原创内容，其余所有内容均收集自互联网，仅限用于学习和研究目的，本站不对其内容的合法性承担任何责任。如有版权内容，请通知我们或作者删除，其版权均归原作者所有，本站虽力求保存原有版权信息，但因众多资源经多次转载，已无法确定其真实来源，或已将原有信息丢失，所以敬请原作者谅解！
3. 本站用户所发布的一切资源内容不代表本站立场，并不代表本站赞同其观点和对其真实性负责，若您对本站所载资源作品版权归属存有异议，请留言附说明联系邮箱，我们将在第一时间予以处理 ，同时向您表示歉意！为尊重作者版权，请购买原版作品，支持您喜欢的作者，谢谢！
4. 本站一律禁止以任何方式发布或转载任何违法的相关信息，访客如有发现请立即向站长举报；本站资源文件大多存储在云盘，如发现链接或图片失效，请联系作者或站长及时更新。
THE END
暂无评论内容