最近很火的KS-A抢购代码开源了: ovh api 抢购代码(go语言)
本来不想发,有点扰乱秩序,不过自从let上那人发了流程说明以后,大家都在写脚本了。那我就为大家省时间吧。
代码在我的博客上,我就不贴过来了,需要的自取。
戳我
感谢15楼的哥们发了gpt改写的python版。有需要的也可以参考。 *经mjj测试,调用api时改用官方库 https://github.com/ovh/python-ovh 方可正常运行,否则会401鉴权错误。 同时务必请注意,这个版本它在购买成功后没有停止循环,所以,会不停一直购买!切记修改!
如果是加区帐号,常量定义里,
zone请改成 CA,region改成 ovh-ca。
代码是抢购的ks-le-b ssd版。(25skleb01 bandwidth-300-25skle ram-32g-ecc-2400-25skle softraid-2x450nvme-25skle)
如果要抢5o,参数相应改成
plancode: 24ska01
options: bandwidth-100-24sk ram-64g-noecc-2133-24ska01 softraid-1x480ssd-24ska01
代码是设置的下单完成后即支付,如果你只是想下单晚点再支付,请把checkout的autoPayWithPreferredPaymentMethod自动支付改成false即可。
下面是GPT转Python的代码:
import requests
import json
import time
import logging
# 常量定义
APP_KEY = "ovh appkey" # OVH的应用key
APP_SECRET = "ovh appsecret" # OVH的应用secret
CONSUMER_KEY = "ovh consumerkey" # OVH的消费者key
REGION = "ovh-eu" # 区域设置为
TG_TOKEN = "telegrambot token" # 你的Telegram Bot Token
TG_CHAT_ID = "telegram chatid" # 你希望发送消息的Telegram Chat ID
IAM = "go-ovh-ie" # 你的标识
ZONE = "IE" # OVH子公司区域设置
# 设置日志输出
logging.basicConfig(level=logging.INFO)
def send_telegram_msg(bot_token, chat_id, message):
"""向指定的 Telegram 发送消息"""
url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
payload = {
"chat_id": chat_id,
"text": message
}
headers = {"Content-Type": "application/json"}
try:
# 发送POST请求
response = requests.post(url, json=payload, headers=headers)
response.raise_for_status() # 检查请求是否成功
if response.status_code != 200:
logging.error(f"发送消息失败: {response.status_code}, {response.text}")
return False
return True
except requests.exceptions.RequestException as e:
logging.error(f"发送Telegram消息时发生错误: {e}")
return False
def run_task():
"""执行任务,检查OVH的可用性并处理购买流程"""
# 创建OVH API客户端
try:
# 获取数据中心的可用性信息
url = f"https://api.ovh.com/1.0/dedicated/server/datacenter/availabilities"
headers = {
"X-Ovh-Application": APP_KEY,
"X-Ovh-Consumer": CONSUMER_KEY
}
response = requests.get(url, headers=headers)
response.raise_for_status()
result = response.json() # 解析返回的JSON数据
except requests.exceptions.RequestException as e:
logging.error(f"获取OVH数据中心可用性时失败: {e}")
return
found_available = False
fqn, plan_code, datacenter = None, None, None
for item in result:
if item["planCode"] == "25skleb01": # 如果找到了符合条件的计划
fqn = item["fqn"]
plan_code = item["planCode"]
datacenters = item["datacenters"]
for dc_info in datacenters:
availability = dc_info["availability"]
datacenter = dc_info["datacenter"]
logging.info(f"FQN: {fqn}")
logging.info(f"Availability: {availability}")
logging.info(f"Datacenter: {datacenter}")
logging.info("------------------------")
if availability != "unavailable": # 如果该数据中心是可用的
found_available = True
break
if found_available:
logging.info(f"正在继续操作,FQN: {fqn} Datacenter: {datacenter}")
break
if not found_available:
logging.info("没有可购买的记录")
return
msg = f"{IAM}: 在 {datacenter} 找到 ks-le-b 可用"
if not send_telegram_msg(TG_TOKEN, TG_CHAT_ID, msg):
logging.error("发送Telegram消息失败")
return
# 创建购物车
try:
logging.info("创建购物车")
url = f"https://api.ovh.com/1.0/order/cart"
payload = {"ovhSubsidiary": ZONE}
response = requests.post(url, json=payload, headers=headers)
response.raise_for_status()
cart_result = response.json()
except requests.exceptions.RequestException as e:
logging.error(f"创建购物车失败: {e}")
return
cart_id = cart_result["cartId"]
logging.info(f"购物车ID: {cart_id}")
# 绑定购物车
try:
logging.info("绑定购物车")
url = f"https://api.ovh.com/1.0/order/cart/{cart_id}/assign"
response = requests.post(url, headers=headers)
response.raise_for_status()
except requests.exceptions.RequestException as e:
logging.error(f"绑定购物车失败: {e}")
return
# 将商品添加到购物车
try:
logging.info("将商品添加到购物车")
url = f"https://api.ovh.com/1.0/order/cart/{cart_id}/eco"
payload = {
"planCode": plan_code,
"pricingMode": "default",
"duration": "P1M", # 持续时间为1个月
"quantity": 1
}
response = requests.post(url, json=payload, headers=headers)
response.raise_for_status()
item_result = response.json()
except requests.exceptions.RequestException as e:
logging.error(f"将商品添加到购物车失败: {e}")
return
item_id = item_result["itemId"]
logging.info(f"商品ID: {item_id}")
# 获取必需的配置
try:
logging.info("检查必需的配置")
url = f"https://api.ovh.com/1.0/order/cart/{cart_id}/item/{item_id}/requiredConfiguration"
response = requests.get(url, headers=headers)
response.raise_for_status()
required_config = response.json()
except requests.exceptions.RequestException as e:
logging.error(f"获取必需配置失败: {e}")
return
dedicated_os = "none_64.en"
region_value = None
for config in required_config:
if config["label"] == "region":
allowed_values = config["allowedValues"]
if allowed_values:
region_value = allowed_values[0]
configurations = [
{"label": "dedicated_datacenter", "value": datacenter},
{"label": "dedicated_os", "value": dedicated_os},
{"label": "region", "value": region_value},
]
# 配置购物车中的商品
for config in configurations:
try:
logging.info(f"配置 {config['label']}")
url = f"https://api.ovh.com/1.0/order/cart/{cart_id}/item/{item_id}/configuration"
payload = {"label": config["label"], "value": config["value"]}
response = requests.post(url, json=payload, headers=headers)
response.raise_for_status()
except requests.exceptions.RequestException as e:
logging.error(f"配置 {config['label']} 失败: {e}")
return
# 添加附加选项
options = [
"bandwidth-300-25skle",
"ram-32g-ecc-2400-25skle",
"softraid-2x450nvme-25skle"
]
for option in options:
try:
logging.info(f"添加选项 {option}")
url = f"https://api.ovh.com/1.0/order/cart/{cart_id}/eco/options"
payload = {
"duration": "P1M", # 选项持续时间为1个月
"itemId": int(item_id),
"planCode": option,
"pricingMode": "default",
"quantity": 1
}
response = requests.post(url, json=payload, headers=headers)
response.raise_for_status()
except requests.exceptions.RequestException as e:
logging.error(f"添加选项 {option} 失败: {e}")
return
# 进行结账
try:
logging.info("结账")
url = f"https://api.ovh.com/1.0/order/cart/{cart_id}/checkout"
response = requests.get(url, headers=headers)
response.raise_for_status()
checkout_result = response.json()
# 提交结账请求
url = f"https://api.ovh.com/1.0/order/cart/{cart_id}/checkout"
payload = {
"autoPayWithPreferredPaymentMethod": True,
"waiveRetractationPeriod": True # 自动支付并跳过撤销期
}
response = requests.post(url, json=payload, headers=headers)
response.raise_for_status()
except requests.exceptions.RequestException as e:
logging.error(f"结账失败: {e}")
return
logging.info("完成购买!")
def main():
"""主函数:循环执行任务"""
while True:
run_task() # 执行任务
time.sleep(20) # 每20秒执行一次任务
if __name__ == "__main__":
main()
阅读剩余
版权声明:
作者:John
链接:https://vps.la/2024/11/08/%e6%9c%80%e8%bf%91%e5%be%88%e7%81%ab%e7%9a%84ks-a%e6%8a%a2%e8%b4%ad%e4%bb%a3%e7%a0%81%e5%bc%80%e6%ba%90%e4%ba%86-ovh-api-%e6%8a%a2%e8%b4%ad%e4%bb%a3%e7%a0%81go%e8%af%ad%e8%a8%80/
文章版权归作者所有,未经允许请勿转载。
THE END