某东抢购某台脚本

代码如下

import concurrent
import datetime
from concurrent.futures import ProcessPoolExecutor
import json
import sys
import time
import requests
import jpype
import urllib3
from log import logger

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
cookie_str = ''

sub_order_time = '2024-05-01 18:07:59.500'
skuId = 124234234234234232343

# 创建一个Session对象
session = requests.session()

jump_url = 'https://un.m.jd.com/cgi-bin/app/appjmp'
token_body = '{"action":"to","to":"https://divide.jd.com/user_routing?skuId=%s"}' % skuId
token_function_id = ''
client = ''
clientVersion = ''
uuid = ''

params = {
    "tokenKey": 'AAEAMAje64z7n1vGvN0nq_FH_PgEBZMFBUlqW7rVS-_W2kNwPpykXwjaGg7QDHS7TzrFPQ1',
    "to": "https://divide.jd.com/user_routing?skuId=%s" % skuId
}

fp = ''
eid = ''
ep = ''
token_payload = {}

order_rul = 'https://marathon.jd.com/seckillnew/orderService/submitOrder.action?skuId=' + str(skuId)
data = {}

is_first_init_data = True

def make_reserve():
    """商品预约"""
    url = 'https://api.m.jd.com/client.action'
    body = '{"autoAddCart":"0","bsid":"","check":"0","ctext":"","isShowCode":"0","mad":"0","skuId":"%s","type":"4"}' % skuId
    payload = {}
    JDClass = jpype.JClass("com.jdsdk.jd_main")
    jd = JDClass()
    sing = jd.runJni(['appoint', body, uuid, client, clientVersion])
    sign = str(sing)
    data_array = sign.split('&')
    data_dict = {}
    # 遍历分割后的数组
    for item in data_array:
        key, value = item.split('=')
        data_dict[key] = value
    payload['st'] = data_dict['st']
    payload['sign'] = data_dict['sign']
    payload['sv'] = 111
    resp = session.post(url=url, params=payload)
    resp_json = parse_json(resp.text)
    print(resp_json)
    try:
        if resp_json['title'] == '您已成功预约，无需重复预约' or resp_json['title'] == '预约成功！':
            logger.info(resp_json['title'])
    except Exception as e:
        logger.error('预约失败正在重试...')


def sub_order():
    global is_first_init_data
    while True:
        """
            init.action 拿到生成订单的信息
            替换订单模版的某些信息
        """
        if is_first_init_data:
            order_init_url = 'https://marathon.jd.com/seckillnew/orderService/init.action'
            order_init_data = {
                'sku': skuId,
                'num': 1,
            }
            init_data = session.post(order_init_url, params=order_init_data)
            if init_data.text == 'null':
                print('init_data == null')
                continue
            try:
                init_data = parse_json(init_data.text)
                if init_data.get('islogin'):
                    print('islogin.no')
                    continue
            except Exception as e:
                logger.info('抢购失败，返回信息:{}'.format(init_data.text[0: 128]))
                continue
        """
            提交订单
        """

        for _ in range(1000):
            order_resp = session.post(url=order_rul, params=data)
            try:
                resp_json = parse_json(order_resp.text)
            except Exception as e:
                logger.info('抢购失败，返回信息:{}'.format(order_resp.text[0: 128]))
                continue
            if resp_json.get('success'):
                order_id = resp_json.get('orderId')
                total_money = resp_json.get('totalMoney')
                pay_url = 'https:' + resp_json.get('pcUrl')
                print('抢购成功，订单号:{}, 总价:{}, 电脑端付款链接:{}'.format(order_id, total_money, pay_url))
                logger.info('抢购成功，订单号:{}, 总价:{}, 电脑端付款链接:{}'.format(order_id, total_money, pay_url))
            else:
                logger.info('抢购失败，返回信息:{}'.format(resp_json))
                continue
            """
                休息10毫秒
            """
            time.sleep(0.01)


def kill_mt():
    global marathon_location
    """
        appjmp
        获取跳转Location
     """
    response = session.get(url=jump_url, params=params, allow_redirects=False, verify=False)
    location = response.headers.get('Location')
    """
        divide验证登陆信息
    """
    divide_response = session.get(location, allow_redirects=False, verify=False)
    if divide_response.headers.get('Location') == 'https://marathon.jd.com/mobile/koFail.html':
        print("第二步出错！！wskey过期")
        die()
    else:
        marathon_location = divide_response.headers.get('Location')
    """
        获取第三步跳转连接
        当活动开始的时候才会订单生成地址
        成功网址 = https://marathon.jd.com/seckillM/seckill.action?skuId=100012043978&num=1&rid=1621310648
        失败网址 = https://marathon.jd.com/mobile/koFail.html 非抢购时间都会失败
    """
    marathon = session.get(marathon_location, allow_redirects=False, verify=False)
    get_pay_location = marathon.headers.get('Location')
    """
        seckill.action
        在app上就是填写订单
        只有在抢购期间内才会请求成功，否则都是失败的
    """
    if get_pay_location == 'https://marathon.jd.com/mobile/koFail.html' or get_pay_location is None:
        die()
    session.get(get_pay_location, allow_redirects=False, verify=False)
    sub_order()


def parse_json(s):
    begin = s.find('{')
    end = s.rfind('}') + 1
    return json.loads(s[begin:end])


def die():
    sys.exit(1)

def getToken():
    logger.info('开始获取 token')
    JDClass = jpype.JClass("com.jdsdk.jd_main")
    jd = JDClass()
    sing = jd.runJni([token_function_id, token_body, uuid, client, clientVersion])
    sign = str(sing)
    data_array = sign.split('&')
    data_dict = {}
    # 遍历分割后的数组
    for item in data_array:
        key, value = item.split('=')
        data_dict[key] = value
    token_payload['st'] = data_dict['st']
    token_payload['sign'] = data_dict['sign']
    token_payload['sv'] = 111
    url = 'https://api.m.jd.com/client.action'
    token_resp = requests.post(url, params=token_payload)
    if token_resp.status_code != 200:
        print("请求出错！！！")
    json = parse_json(token_resp.text)
    if json.get('echo'):
        print("错误信息")
        print(json['echo'])
        die()
    params['tokenKey'] = json['tokenKey']
    logger.info('获取 token完毕')
    logger.info('获取到token:' + json['tokenKey'])


if __name__ == "__main__":
    # 添加Jar包到类路径
    jvmPath = jpype.getDefaultJVMPath()
    d = './out/artifacts/xxx_jar/xxx.jar'  # 对应jar地址
    jpype.startJVM(jvmPath, "-ea", "-Djava.class.path=" + d + "")
    make_reserve()
    getToken()
    kill_mt()

说明

这个脚本是用来抢购京东商品的Python脚本，其中包含了许多自动化操作来模拟用户行为。下面是脚本的主要部分及其功能的详细解释：

1. 初始化和导入库:

   – 导入必要的库，如`concurrent`, `datetime`, `requests`, `jpype`等，用于网络请求、日期处理和Java代码交互。

   – 使用`urllib3`禁用了HTTPS警告，这是常见的在发起HTTPS请求时避免不必要警告的做法。

2. 全局变量设置:

   – 设置cookie字符串、预定时间、商品ID、以及一些URL和header参数。

3. Session对象:

   – 创建了一个requests的Session对象用于维持一系列请求的持久连接。

4. 商品预约(make_reserve):

   – 发起POST请求预约商品，使用了封装的加密算法生成请求签名，并处理返回的预约结果。

5. 提交订单(sub_order):

   – 通过不断检查和提交订单尝试抢购商品。使用了一个循环来持续尝试直到成功或达到尝试次数限制。

6. 抢购主函数(kill_mt):

   – 包含了获取token、处理登录跳转、验证登录信息，最后根据响应跳转到抢购页面。

   – 如果抢购时间到了，会调用`sub_order()`函数来提交订单。

7. 辅助功能:

   – `parse_json`函数用于解析JSON格式的字符串。

   – `die()`函数用于在发生错误时退出脚本。

   – `getToken()`用于从服务器获取必要的token以继续进行抢购。

8. JPype的使用:

   – 使用JPype来启动Java虚拟机，允许Python脚本运行Java代码。这通常用于调用一些复杂的加密函数，这些加密函数是用Java实现的，可能涉及到登录和提交订单的过程。

9. 入口点:

   – 脚本的执行从这里开始，首先设置JPype的环境，然后进行商品预约，获取token，并尝试抢购。

这个脚本的目的是自动化地参与京东的商品抢购活动，通过模拟用户的登录和下单过程来尽可能地提高成功率。请注意，使用这类脚本可能违反京东的使用条款，并可能导致账号被封禁。在使用之前，请确保了解相关的法律和政策限制。

注意：

本文部分变量已做脱敏处理，仅用于测试和学习研究，禁止用于商业用途，不能保证其合法性，准确性，完整性和有效性，请根据情况自行判断。技术层面需要提供帮助，可以通过打赏的方式进行探讨。