simple_stock_monitor.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
简单苹果库存监控工具
输入型号代码即可开始监控，地区固定为宁波
"""

import requests
import time
import json
import logging
from datetime import datetime
from typing import Dict, Optional
import os
import subprocess
import platform

class SimpleStockMonitor:
    """简单库存监控器"""
    
    def __init__(self):
        self.api_url = "https://www.apple.com.cn/shop/fulfillment-messages"
        self.ningbo_store_code = "R531"  # 宁波门店代码
        self.session = requests.Session()
        self.setup_logging()
    
    def play_sound(self):
        """播放提醒声音"""
        try:
            system = platform.system()
            if system == "Darwin":  # macOS
                os.system("afplay /System/Library/Sounds/Glass.aiff")
            elif system == "Windows":
                import winsound
                winsound.Beep(1000, 500)  # 频率1000Hz，持续500ms
            elif system == "Linux":
                os.system("paplay /usr/share/sounds/alsa/Front_Left.wav 2>/dev/null || echo -e '\a'")
            else:
                print("\a")  # 系统铃声
        except Exception as e:
            print("\a")  # 如果播放失败，使用系统铃声
        
        # 设置请求头
        self.session.headers.update({
            "accept": "*/*",
            "accept-language": "zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6",
            "priority": "u=1, i",
            "sec-ch-ua": '"Not;A=Brand";v="99", "Microsoft Edge";v="139", "Chromium";v="139"',
            "sec-ch-ua-mobile": "?0",
            "sec-ch-ua-platform": '"macOS"',
            "sec-fetch-dest": "empty",
            "sec-fetch-mode": "cors",
            "sec-fetch-site": "same-origin",
            "x-aos-ui-fetch-call-1": "4h8ytuzso9-mf0t96nt"
        })
    
    def setup_logging(self):
        """设置日志"""
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('simple_stock_monitor.log', encoding='utf-8'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
    
    def check_stock(self, part_number: str) -> Dict:
        """检查指定产品在宁波的库存
        
        Args:
            part_number: 产品型号代码，如 'MYTQ3CH/A'
            
        Returns:
            包含库存信息的字典
        """
        try:
            # 构建API请求参数
            params = {
                'fae': 'true',
                'pl': 'true', 
                'mts.0': 'regular',
                'mts.1': 'compact',
                'cppart': 'UNLOCKED/WW',
                f'parts.0': part_number,
                'searchNearby': 'true',
                'store': self.ningbo_store_code
            }
            
            # 发送API请求
            response = self.session.get(self.api_url, params=params)
            response.raise_for_status()
            
            data = response.json()
            
            # 解析响应数据
            if 'body' in data and 'content' in data['body'] and 'pickupMessage' in data['body']['content']:
                pickup_message = data['body']['content']['pickupMessage']
                
                if 'stores' in pickup_message and pickup_message['stores']:
                    store_info = pickup_message['stores'][0]  # 取第一个门店
                    
                    if 'partsAvailability' in store_info and part_number in store_info['partsAvailability']:
                        part_info = store_info['partsAvailability'][part_number]
                        
                        return {
                            'success': True,
                            'store_name': store_info.get('storeName', '未知门店'),
                            'store_number': store_info.get('storeNumber', self.ningbo_store_code),
                            'part_number': part_number,
                            'available': part_info.get('pickupDisplay') == 'available',
                            'pickup_quote': part_info.get('pickupSearchQuote', ''),
                            'store_pick_eligible': part_info.get('storePickEligible', False),
                            'buyable': part_info.get('buyability', {}).get('isBuyable', False),
                            'inventory': part_info.get('buyability', {}).get('inventory', 0),
                            'distance': store_info.get('storeDistanceWithUnit', ''),
                            'address': store_info.get('address', {}).get('address2', ''),
                            'phone': store_info.get('phoneNumber', ''),
                            'check_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                        }
            
            return {
                'success': False,
                'error': '无法解析API响应数据',
                'part_number': part_number,
                'check_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            }
            
        except requests.RequestException as e:
            self.logger.error(f"API请求失败: {e}")
            return {
                'success': False,
                'error': f'API请求失败: {str(e)}',
                'part_number': part_number,
                'check_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            }
        except Exception as e:
            self.logger.error(f"检查库存时发生错误: {e}")
            return {
                'success': False,
                'error': f'检查库存时发生错误: {str(e)}',
                'part_number': part_number,
                'check_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            }
    
    def monitor_stock(self, part_number: str, check_interval: int = 30):
        """持续监控库存
        
        Args:
            part_number: 产品型号代码
            check_interval: 检查间隔（秒）
        """
        print(f"🍎 开始监控产品: {part_number}")
        print(f"📍 监控地区: 宁波")
        print(f"⏰ 检查间隔: {check_interval}秒")
        print("按 Ctrl+C 停止监控...")
        print("=" * 50)
        
        check_count = 0
        
        try:
            while True:
                check_count += 1
                result = self.check_stock(part_number)
                
                if result['success']:
                    if result['available']:
                        print(f"🎉 [{result['check_time']}] 第{check_count}次检查 - 有货!")
                        
                        # 发出声音提醒
                        self.play_sound()
                        
                        # 检查是否为天一广场
                        if '天一广场' in result['store_name']:
                            print(f"🏪 天一广场: 有货!")
                            print(f"   门店: {result['store_name']}")
                            print(f"   状态: {result['pickup_quote']}")
                            print(f"   地址: {result['address']}")
                            print(f"   电话: {result['phone']}")
                            print(f"   距离: {result['distance']}")
                            self.logger.info(f"天一广场有货: {part_number} - {result['pickup_quote']}")
                        else:
                            print(f"🏪 天一广场: 暂无库存")
                            print(f"📍 附近门店有货:")
                            print(f"   门店: {result['store_name']}")
                            print(f"   状态: {result['pickup_quote']}")
                            print(f"   地址: {result['address']}")
                            print(f"   电话: {result['phone']}")
                            print(f"   距离: {result['distance']}")
                            self.logger.info(f"附近门店有货: {part_number} 在 {result['store_name']} - {result['pickup_quote']}")
                        
                        # 可以在这里添加通知功能（如发送邮件、微信等）
                        
                    else:
                        print(f"😔 [{result['check_time']}] 第{check_count}次检查 - 暂无库存")
                        print(f"🏪 天一广场: 暂无库存")
                        if result.get('pickup_quote'):
                            print(f"   状态: {result['pickup_quote']}")
                else:
                    print(f"❌ [{result['check_time']}] 第{check_count}次检查 - 查询失败: {result.get('error', '未知错误')}")
                    self.logger.error(f"库存查询失败: {result.get('error', '未知错误')}")
                
                # 等待下次检查
                time.sleep(check_interval)
                
        except KeyboardInterrupt:
            print("\n监控已停止")
            self.logger.info(f"监控停止，共检查了 {check_count} 次")
        except Exception as e:
            print(f"\n监控过程中发生错误: {e}")
            self.logger.error(f"监控过程中发生错误: {e}")

def main():
    """主函数"""
    print("🍎 简单苹果库存监控工具")
    print("📍 监控地区: 宁波")
    print("=" * 50)
    
    # 获取用户输入的产品型号代码
    part_number = input("请输入产品型号代码 (如 MYTQ3CH/A): ").strip()
    
    if not part_number:
        print("❌ 产品型号代码不能为空")
        return
    
    # 获取检查间隔
    try:
        interval_input = input("请输入检查间隔（秒，默认30秒）: ").strip()
        check_interval = int(interval_input) if interval_input else 30
        if check_interval < 5:
            print("⚠️ 检查间隔不能少于5秒，已设置为5秒")
            check_interval = 5
    except ValueError:
        print("⚠️ 输入无效，使用默认间隔30秒")
        check_interval = 30
    
    # 创建监控器并开始监控
    monitor = SimpleStockMonitor()
    
    # 先测试一次查询
    print("\n🔍 正在测试查询...")
    test_result = monitor.check_stock(part_number)
    
    if test_result['success']:
        print(f"✅ 查询成功，开始监控...")
        monitor.monitor_stock(part_number, check_interval)
    else:
        print(f"❌ 查询失败: {test_result.get('error', '未知错误')}")
        print("请检查产品型号代码是否正确")

if __name__ == '__main__':
    main()