|
|
分享一段ai写的自动监控和使用pushdeer给手机发信息提醒的代码。
需要事先按照https://www.pushdeer.com/official.html的指示申请key和安装pushdeer的app。
祝愿人人都有5090fe。
- 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)
|
|
楼主: linkzero
310112100042806
发表于 2025-6-20 16:54
