updated: 2025-05-01T17:40:03
aliases:
tags: Python,qinglong
author: Aura Service
使用方法
抓包
如图：
青龙配置如图
示例：
不抢仅签到：
sss123-3441-4444-4444-4444|
要签到要券：
sss113-23129-123123-123123|1
sss234-23129-123123-234312|222
AI总结的脚本功能概述
执行主流程：start()
[ol]
解析环境变量 env
逐账号签到并获取积分
[/ol]
r = checkin(tk) # 执行签到
point = api_get_my_point(tk) # 获取当前积分
判断 point > 5 且 “监控标志” 不为空 → 将该 tk 加入监控列表
[/ol]
start_monitoring(monitor_tk)
主要函数说明
checkin(tk)
功能：调用签到接口
接口：POST https://sss-web.tastientech.com/api/sign/member/signV2
返回：响应对象
api_get_my_point(tk)
功能：获取当前积分
接口：POST https://sss-web.tastientech.com/api/wx/point/myPoint
返回：含 result.point 的响应
get_orders(tk)
功能：拉取“积分秒杀”订单列表
接口：POST https://sss-web.tastientech.com/api/wx/point/coupon/activity/queryAppletActivityList
处理：
[ol]
过滤 groupName == '积分秒杀'
提取每个 activity 的 id、name、exchangePoint、open_times
[/ol]
返回：[{id, name, exchangePoint, open_times}, …]
monitor_order(tk, order)
功能：在每个 open_time 前后 ±10 分钟内，尝试秒杀
步骤：
[ol]
验证并解析 open_times 为 datetime.time 列表
检查是否在任一时间点的 ±10 分钟区间内
若在关键区间（±5 秒）内，每 0.5 秒调用 api_create_order(tk, order_id)，最多尝试 10 秒
若超出 ±1 分钟区间，每 10 秒检查一次；否则每 30 秒检查一次
所有时间点均超出 ±10 分钟后，退出监控
[/ol]
api_create_order(tk, activityId)
功能：发起抢券请求
接口：POST https://sss-web.tastientech.com/api/c/pointOrder/create
返回：响应对象
流程图（Mermaid）
flowchart TD
A[读取 env 内容] --> B[解析每行 token和flag]
B --> C[逐账号签到 checkin]
C --> D[获取积分 api_get_my_point]
D --> E{积分 > 5 且标志不为空?}
E -- 是 --> F[加入监控列表]
E -- 否 --> G[输出“积分不足或未监控”]
F --> H[所有账号处理完?]
G --> H
H -- 是 --> I{监控列表非空?}
H -- 否 --> K[结束]
I -- 是 --> J[start_monitoring]
I -- 否 --> K
J --> L[线程: 获取订单 get_orders]
L --> M[遍历每个订单 monitor_order]
M --> N[解析 open_times]
N --> O{当前在 ±10 分钟区间?}
O -- 否 --> P[等待（±1 分钟内:10s; 否:30s） → 重试]
O -- 是 --> Q{在 ±5 秒区间?}
Q -- 否 --> P
Q -- 是 --> R[每0.5 秒调用 api_create_order（最长 10s）]
R --> S{返回 code==200?}
S -- 是 --> T[打印“创建成功”并退出线程]
S -- 否 --> P
源代码
# -*- coding: utf-8 -*-
"""
@file : tasting-monitor.py
@AuThor : aura service
@Time : 2025/5/1 17:18
@Description:
使用方法：
修改env 的值改为抓包小程序 的 user-token
每行一个账号
账号1的tk|是否需要抢券，这里随便填值不为空则表示要抢
账号2的tk|是否需要抢券
示例：
sss113-23129-123123-123123|1
sss234-23129-123123-234312|222
"""
import os
import threading
import time
import requests
import json
from datetime import datetime, timedelta
# 直接跑程序使用的变量
env = """
账号1的toekn|为空则跳过抢券，任意值表示需要抢券
sss123-3441-4444-4444-4444444444|1
"""
# 青龙面板使用 取消下面注释
# env = os.getenv('tst_tk_env')
debug_mode = False
def log_debug(msg):
if debug_mode:
print(f"[debug]{msg}")
else:
pass
def checkin(tk):
import requests
import json
url = "https://sss-web.tastientech.com/api/sign/member/signV2"
payload = {
"activityId": 59,
"memberName": "",
"memberPhone": ""
}
headers = {
'User-Agent': "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36 MicroMessenger/7.0.20.1781(0x6700143B) NetType/WIFI MiniProgramEnv/Windows WindowsWechat/WMPF WindowsWechat(0x63090a13) UnifiedPCWindowsWechat(0xf254032b) XWEB/13655",
'Content-Type': "application/json",
'version': "3.16.0",
'xweb_xhr': "1",
'user-token': tk,
'channel': "1"
}
response = requests.post(url, data=json.dumps(payload), headers=headers)
log_debug(response.text)
return response
def api_create_order(tk, activityId=338):
"""
- 兑换
:param tk:
:return:
"""
url = "https://sss-web.tastientech.com/api/c/pointOrder/create"
payload = {
"requestId": "0d4e4367d8c",
"activityId": activityId
}
headers = {
'User-Agent': "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36 MicroMessenger/7.0.20.1781(0x6700143B) NetType/WIFI MiniProgramEnv/Windows WindowsWechat/WMPF WindowsWechat(0x63090a13) UnifiedPCWindowsWechat(0xf254032b) XWEB/13655",
'Content-Type': "application/json",
'version': "3.16.0",
'xweb_xhr': "1",
'user-token': tk,
'channel': "1",
'Sec-Fetch-Site': "cross-site",
'Sec-Fetch-Mode': "cors",
'Sec-Fetch-Dest': "empty",
'Accept-Language': "zh-CN,zh;q=0.9",
}
response = requests.post(url, data=json.dumps(payload), headers=headers)
log_debug(response.text)
return response
def api_get_order_list(tk):
"""
获取订单列表
:param tk:
:return:
"""
url = "https://sss-web.tastientech.com/api/wx/point/coupon/activity/queryAppletActivityList"
payload = {
"platform": "exchangeActivities"
}
headers = {
'User-Agent': "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36 MicroMessenger/7.0.20.1781(0x6700143B) NetType/WIFI MiniProgramEnv/Windows WindowsWechat/WMPF WindowsWechat(0x63090a13) UnifiedPCWindowsWechat(0xf254032b) XWEB/13655",
'Content-Type': "application/json",
'version': "3.16.0",
'user-token': tk,
'channel': "1"
}
response = requests.post(url, data=json.dumps(payload), headers=headers)
log_debug(response.text)
return response
def get_orders(tk):
"""
获取订单数据
:return:
"""
# 过滤秒杀order
orders = [i for i in api_get_order_list(tk).json().get("result", []) if
i.get('groupId') == 41 and i.get('groupName') == '积分秒杀']
orders = [i for i in orders[0].get('activities', [])]
re_orders = []
for order in orders:
id = order.get('id')
name = order.get('name')
exchangePoint = order.get('exchangePoint')
open_times = [i['openTime'].split('-')[0] for i in order.get('timeActivityInfo', [])]
re_orders.append({
"id": id,
"name": name,
"exchangePoint": exchangePoint,
"open_times": open_times
})
print(f"{id}|{name}|{exchangePoint}|{open_times}")
return re_orders
def api_get_my_point(tk):
url = "https://sss-web.tastientech.com/api/wx/point/myPoint"
headers = {
'User-Agent': "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36 MicroMessenger/7.0.20.1781(0x6700143B) NetType/WIFI MiniProgramEnv/Windows WindowsWechat/WMPF WindowsWechat(0x63090a13) UnifiedPCWindowsWechat(0xf254032b) XWEB/13655",
'Content-Type': "application/json",
'version': "3.16.0",
'xweb_xhr': "1",
'user-token': tk,
'channel': "1",
}
response = requests.post(url, data=json.dumps({}), headers=headers)
print(response.text)
return response
def monitor_order(tk, order):
order_id = order.get('id')
open_times = order.get('open_times', [])
order_name = order.get('name')
# 验证 open_times 格式
valid_open_times = []
for open_time in open_times:
try:
valid_time = datetime.strptime(open_time, "%H:%M").time()
valid_open_times.append(valid_time)
except ValueError:
print(f"无效的时间格式: {open_time}，跳过该时间。")
if not valid_open_times:
print(f"没有有效时间，订单 {order_id} 不再监控。")
return
now = datetime.now().time()
in_any_range = any(
(datetime.combine(datetime.today(), t) - timedelta(minutes=10)).time() datetime.combine(now.date(), t) + timedelta(minutes=10)
for t in valid_open_times
):
print(f"所有 open_time 已超过 10 分钟，订单 {order_id} 退出监控。")
break
# 启动监控进程
def start_monitoring(tks):
threads = []
for tk in tks:
orders = get_orders(tk)
for order in orders:
monitor_thread = threading.Thread(target=monitor_order, args=(tk, order,))
threads.append(monitor_thread)
monitor_thread.start()
# 可选：等待所有线程完成
for thread in threads:
thread.join()
def filter_lines(input_string):
# 按行分割字符串
lines = input_string.splitlines()
# 过滤空行并检查每行是否包含且仅包含一个 '|'
filtered_lines = [
line for line in lines
if line.strip() and line.count('|') == 1
]
return filtered_lines
def start():
monitor_tk = []
print("===== 开始签到 =====")
for user in filter_lines(env):
tk, is_monitor = user.split("|")
print(f"==={tk}")
# 签到任务
r = checkin(tk)
if r.json().get("code") == 200:
print("签到成功")
else:
print(r.json().get("msg", r.text))
point = api_get_my_point(tk).json().get("result", {}).get('point')
print(f"积分：{point}")
if point > 5 and is_monitor:
monitor_tk.append(tk)
continue
else:
print("积分不足或未设置监控")
if monitor_tk:
print("===== 开始监控时间抢0元券 =====")
start_monitoring(monitor_tk)
if __name__ == '__main__':
start()
蓝奏下载链接：https://wwcy.lanzouq.com/b01yo0gmed 密码：52pj
需要自己改成 .py后缀， lanzou 因为不让传.py
