基于API下单,稳定快速。
使用方法
- 创建python环境,安装相关依赖。
- 修改机器人的token
- 使用python让脚本在后台运行
请注意本脚本依赖redis,如果没有redis无法运行。
脚本收100鸡腿,没鸡腿的要攒攒了。
抢购脚本
#!/usr/bin/python
# coding=utf-8
import json
import random
import time
import redis
import requests
from ccbot import add_to_redis
# 创建 Redis 连接
r = redis.Redis(host='localhost', port=6379, db=0)
# 获取最早插入的数据
def get_latest_data():
# 获取列表中的最后一条数据
chat_id = r.lindex('mylist', -1)
if chat_id:
# 使用 chat_id 从 Redis 的 Hash 中获取用户信息和数据
user_info = r.hgetall(chat_id)
if user_info:
return chat_id.decode(), user_info[b'app_secret'].decode(), user_info[b'hash'].decode()
return None, None, None
def orderService():
# 设置 Telegram Bot Token
bot_token = "xxxx:xxxx" # 替换为你的 Bot Token
# 发送消息到群组
bot_url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
while True:
chat_id, apikey, apihash = get_latest_data()
timestamp = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
print(timestamp + '--->用户--->' + chat_id + "--->")
isorder = False
url = 'https://app.cloudcone.com/blackfriday/offers' # cc链接
try:
req_header = {
"Host": "app.cloudcone.com",
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-origin",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/83.0.4103.97 Safari/537.36",
"X-Requested-With": "XMLHttpRequest",
"Cookie": "CCM19=3tr0n1froibpmg8reaefgsfucl; tz=Asia/Shanghai; _gid=GA1.2.1027245438.1700494458; "
"crisp-client%2Fsession%2Fb4a6582f-f407-4054-b73c-d6e4bf698b1e=session_8ba052a6-7a30-4aeb-bf4c-d4d6361b0f41; "
"crisp-client%2Fsession%2Fb4a6582f-f407-4054-b73c-d6e4bf698b1e"
"%2F268227ff9cfa7e50bbb0cb6443561a4e932e702380a2b73e594bd80f69c25429=session_8ba052a6-7a30-4aeb-bf4c-d4d6361b0f41; "
"_ga_V6Z1CMJLWW=GS1.1.1700495224.1.1.1700495355.0.0.0; _ga_CS73BZKZYP=GS1.1.1700494458.1.1.1700495393.0.0.0; "
"_ga=GA1.1.1402042357.1700494458"
}
response = requests.get(url, req_header)
html1 = str(response.content, "utf-8")
alldata = json.loads(html1)['__data']
vps_data = alldata.get('vps_data')
sc2_data = alldata.get('sc2_data')
data_list = [] # 创建一个空列表
if vps_data is not False:
data_list.append([data for data in vps_data if data['usd_price'] < 20 and data['ram'] > 0.4])
if sc2_data is not False:
data_list.extend([data for data in sc2_data if data['usd_price'] < 20 and data['ram'] > 0.4])
for jsondata in data_list:
if isinstance(jsondata, (list, dict)):
for num in jsondata:
timestamp = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
print(timestamp + " " + str(jsondata[num]['usd_price']) + " 有货")
groupmessage = timestamp + " " + str(jsondata[num]['usd_price']) + " 有货了"
grouppayload = {
"chat_id": '@nodeloc_notify',
"text": groupmessage
}
requests.post(bot_url, json=grouppayload)
url = "https://api.cloudcone.com/api/v2/compute/create"
plan_id = num
cpu = jsondata[num]['cpu']
ram = jsondata[num]['ram']
disk = jsondata[num]['disk']
payload = {
"billing_plan_id": plan_id,
"cpu":cpu,
"ram": ram,
"disk": disk,
"ip_count": 1,
"hostname": "nodeloc.com",
"os_id": 1007
}
headers = {
"accept": "application/json",
"content-type": "application/x-www-form-urlencoded",
"App-Secret": apikey,
"Hash": apihash
}
response = requests.post(url, data=payload, headers=headers)
if response.status_code == 200:
isorder = True
# 从 Redis 列表中移除最后一条数据
r.lpop('mylist')
bot_payload = {
"chat_id": chat_id,
"text": plan_id +"抢购成功!请到产品页面查看,你的api已经从api池中移除"
}
requests.post(bot_url, json=bot_payload)
elif response.status_code == 400:
data = json.loads(response)
if data["__data"]["error"] == "auth":
isorder = True
# 错误为 "auth" 的处理逻辑
r.lpop('mylist')
bot_payload = {
"chat_id": chat_id,
"text": "抢购失败!原因 API key或hash有误,您已被从队列移除"
}
requests.post(bot_url, json=bot_payload)
else:
# 其他错误的处理逻辑
bot_payload = {
"chat_id": chat_id,
"text": "抢购失败!原因:" + data["message"]
}
requests.post(bot_url, json=bot_payload)
else:
timestamp = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
print(timestamp + "暂时无货!")
if isorder is False:
add_to_redis(chat_id, apikey, apihash)
except Exception as e:
print(e.__str__())
continue
time.sleep(2.5 * random.randint(1, 3))
if __name__ == '__main__':
orderService()
机器人脚本
import requests
import redis
import telegram
from telegram.ext import *
# 创建 Redis 连接
r = redis.Redis(host='localhost', port=6379, db=0)
# 处理接收到的文本消息
def handle_text_message(update, context):
message = update.message
text = message.text
chat_id = message.chat_id
# 校验格式
if not validate_format(text):
# 格式不正确,发送提示消息
context.bot.send_message(chat_id=chat_id, text="格式不正确,请重新发送。")
return
parts = text.split(":")
app_secret = parts[0]
hash_value = parts[1]
# 校验API
url = "https://api.cloudcone.com/api/v2/compute/instances"
headers = {
"accept": "application/json",
"App-Secret": app_secret,
"Hash": hash_value
}
response = requests.get(url, headers=headers)
if response.status_code == 200:
# API校验通过,将chat_id、app_secret和hash数据加入到Redis中
add_to_redis(chat_id, app_secret, hash_value)
context.bot.send_message(chat_id=chat_id, text="API校验通过,已将数据加入Redis。")
else:
# API校验失败,发送提示消息
context.bot.send_message(chat_id=chat_id, text="API有误,请检查后重新发送。")
# 校验格式
def validate_format(text):
parts = text.split(":")
if len(parts) != 2:
return False
if not parts[0] or not parts[1]:
return False
return True
# 将数据加入Redis
def add_to_redis(chat_id, app_secret, hash_value):
# 将数据添加到列表的头部
r.lpush('mylist', chat_id)
# 将用户信息和数据存储到 Redis 的 Hash 中
r.hset(chat_id, 'app_secret', app_secret)
r.hset(chat_id, 'hash', hash_value)
# 可选:打印添加的数据
print(f"Added to Redis: {chat_id} - {app_secret} - {hash_value}")
# 获取最早插入的数据
def get_earliest_data():
# 获取列表中的第一个元素(用户 ID)
chat_id = r.lindex('mylist', 0)
if chat_id:
# 使用 chat_id 从 Redis 的 Hash 中获取用户信息和数据
user_info = r.hgetall(chat_id)
if user_info:
return chat_id.decode(), user_info[b'app_secret'].decode(), user_info[b'hash'].decode()
return None, None, None
if __name__ == '__main__':
# 设置机器人的API令牌
TOKEN = "xxxx:xxxx"
application = Application.builder().token(TOKEN).build()
# 创建一个MessageHandler,处理接收到的文本消息
text_handler = MessageHandler(filters.TEXT & ~filters.COMMAND, handle_text_message)
# Commands
application.add_handler(text_handler)
# Run bot
application.run_polling(1.0)