法国NVIDIA官网购入5090 FE - Chiphell - 分享与交流用户体验

import requests
import time
import random
import json
# --- PushDeer 配置 ---
PUSHDEER_KEY = "按照https://www.pushdeer.com/official.html申请key和安装app" # 你的 PushDeer Key
PUSHDEER_API_URL = "https://api2.pushdeer.com/message/push" # PushDeer 公共服务器地址
def send_pushdeer_notification(product_info):
"""
通过 PushDeer 发送消息通知
:param product_info: 包含产品信息的字典，现在会包含更多的上下文信息
"""
# 提取更具体的产品信息用于消息
sku = product_info.get('fe_sku', product_info.get('monitored_sku', 'N/A'))
locale = product_info.get('locale', product_info.get('monitored_locale', 'N/A'))
price = product_info.get('price', 'N/A')
product_url = product_info.get('product_url', 'N/A')
product_name = product_info.get('monitored_name', 'N/A')
message_title = f"NVIDIA 产品库存更新：{sku} ({locale}) 已可用！"
message_text = (
f"产品 {product_name} ({locale}) 已变为可用！\n"
f"价格: {price}\n"
f"产品URL: {product_url}"
)
payload = {
"pushkey": PUSHDEER_KEY,
"text": message_title,
"desp": message_text,
"type": "markdown"
}
try:
response = requests.post(PUSHDEER_API_URL, data=payload, timeout=5)
response.raise_for_status()
pushdeer_response = response.json()
if pushdeer_response.get("code") == 0:
print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] PushDeer 消息发送成功！")
else:
print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] PushDeer 消息发送失败: {pushdeer_response.get('message', '未知错误')}")
print(f"PushDeer 原始响应: {json.dumps(pushdeer_response, ensure_ascii=False)}")
except requests.exceptions.RequestException as e:
print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 发送 PushDeer 消息时网络错误: {e}")
except json.JSONDecodeError:
print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] PushDeer 响应不是有效 JSON。")
if 'response' in locals() and response.text: # 确保response对象和text存在
print(f"响应内容: {response.text[:200]}...")
except Exception as e:
print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 发送 PushDeer 消息时发生未知错误: {e}")
# --- 要监控的产品列表 ---
# 每个字典代表一个需要监控的产品配置
PRODUCTS_TO_MONITOR = [
{
"base_url": "https://api.store.nvidia.com/partner/v1/feinventory",
"skus": "LCFEGF50LD90",
"locale": "fr-fr",
"name": "RTX 5090 Founders Edition (法国)" # 可以添加一个友好的名称以便识别
},
{
"base_url": "https://api.store.nvidia.com/partner/v1/feinventory",
"skus": "PROFESHOP5090", # 新增的SKU
"locale": "de-de", # 新增的区域
"name": "RTX 5090 Founders Edition (德国)" # 可以添加一个友好的名称以便识别
}
]
# --- HTTP 请求的通用 Headers ---
headers = {
"Accept": "*/*",
"Accept-Encoding": "gzip, deflate, br, zstd",
"Accept-Language": "zh-CN,zh-TW;q=0.9,zh;q=0.8,en;q=0.7",
"Origin": "https://marketplace.nvidia.com",
"Referer": "https://marketplace.nvidia.com/",
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-site",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/127.0.0.0 Safari/537.36"
}
print("开始持续监控 NVIDIA 产品库存...")
response = requests.post(PUSHDEER_API_URL, data={"pushkey": PUSHDEER_KEY, "text": "监控开始"}, timeout=5)
response.raise_for_status()
while True:
for product_config in PRODUCTS_TO_MONITOR:
monitor_url = product_config["base_url"]
monitor_skus = product_config["skus"]
monitor_locale = product_config["locale"]
product_name = product_config.get("name", f"{monitor_skus} ({monitor_locale})") # 获取友好的产品名称
params = {
"status": "1",
"skus": monitor_skus,
"locale": monitor_locale
}
print(f"\n[{time.strftime('%Y-%m-%d %H:%M:%S')}] 正在检查产品: {product_name}...")
try:
response = requests.get(monitor_url, params=params, headers=headers, timeout=10)
if response.status_code == 200:
try:
data = response.json()
if data and data.get("success") and "listMap" in data and len(data["listMap"]) > 0:
product_data_from_api = data["listMap"][0]
is_active = product_data_from_api.get("is_active")
print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {product_name} - is_active: {is_active}")
if is_active == "true":
print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {product_name} - is_active 为 true，发送 PushDeer 消息！")
# 组合产品信息，以便通知中包含 SKU 和区域
notification_data = {
**product_data_from_api, # API返回的信息
"monitored_sku": monitor_skus, # 额外加上我们监控的SKU和区域，以防API返回的FE_SKU不同
"monitored_locale": monitor_locale,
"monitored_name": product_name
}
send_pushdeer_notification(notification_data)
# 如果您希望每个产品只通知一次后就停止监控，可以在这里添加逻辑
# 例如，从 PRODUCTS_TO_MONITOR 列表中移除这个产品，或者设置一个已通知的标志。
# 为了演示持续监控，这里不添加 break 或退出。
else:
print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {product_name} - is_active 为 false。")
else:
print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {product_name} - JSON 响应格式不符合预期或 listMap 为空。")
if data:
print(f"原始响应数据: {json.dumps(data, ensure_ascii=False)}")
except json.JSONDecodeError:
print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {product_name} - 响应不是有效的 JSON 格式。")
print(f"响应内容: {response.text[:200]}...")
except KeyError as ke:
print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {product_name} - JSON 数据缺少预期的键: {ke}")
if data:
print(f"原始响应数据: {json.dumps(data, ensure_ascii=False)}")
else:
print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {product_name} - 请求失败，状态码: {response.status_code}")
print(f"响应内容: {response.text[:200]}...")
except requests.exceptions.RequestException as e:
print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {product_name} - 网络请求发生错误: {e}")
except Exception as e:
print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {product_name} - 发生未知错误: {e}")
# 随机延迟 30 到 60 秒
sleep_time = random.uniform(30, 60)
print(f"\n[{time.strftime('%Y-%m-%d %H:%M:%S')}] 完成一轮所有产品的检查。等待 {sleep_time:.2f} 秒后进行下一次检查...")
time.sleep(sleep_time)

复制代码