gui_stock_monitor.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
苹果库存监控工具 - GUI版本
支持图形化界面显示和急促声音提醒
"""

import requests
import time
import logging
from datetime import datetime
import json
import os
import platform
import threading
import tkinter as tk
from tkinter import ttk, messagebox, scrolledtext
from typing import Dict, Optional

class GUIStockMonitor:
    def __init__(self):
        self.base_url = "https://www.apple.com.cn/shop/fulfillment-messages"
        self.location = "宁波"
        self.session = requests.Session()
        self.setup_logging()
        self.monitoring = False
        self.monitor_thread = None
        
        # 设置请求头
        self.session.headers.update({
            'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
            'Accept': 'application/json, text/javascript, */*; q=0.01',
            'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
            'Accept-Encoding': 'gzip, deflate, br',
            'Connection': 'keep-alive',
            'Referer': 'https://www.apple.com.cn/',
            'X-Requested-With': 'XMLHttpRequest'
        })
        
        self.setup_gui()
    
    def setup_logging(self):
        """设置日志记录"""
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('gui_stock_monitor.log', encoding='utf-8'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
    
    def setup_gui(self):
        """设置图形界面"""
        self.root = tk.Tk()
        self.root.title("🍎 苹果库存监控工具")
        self.root.geometry("800x600")
        self.root.resizable(True, True)
        
        # 设置样式
        style = ttk.Style()
        style.theme_use('clam')
        
        # 主框架
        main_frame = ttk.Frame(self.root, padding="10")
        main_frame.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S))
        
        # 配置网格权重
        self.root.columnconfigure(0, weight=1)
        self.root.rowconfigure(0, weight=1)
        main_frame.columnconfigure(1, weight=1)
        main_frame.rowconfigure(4, weight=1)
        
        # 标题
        title_label = ttk.Label(main_frame, text="🍎 苹果库存监控工具", font=('Arial', 16, 'bold'))
        title_label.grid(row=0, column=0, columnspan=3, pady=(0, 20))
        
        # 输入区域
        ttk.Label(main_frame, text="产品型号:").grid(row=1, column=0, sticky=tk.W, padx=(0, 10))
        self.model_var = tk.StringVar(value="MYTX3CH/A")
        model_entry = ttk.Entry(main_frame, textvariable=self.model_var, width=20)
        model_entry.grid(row=1, column=1, sticky=(tk.W, tk.E), padx=(0, 10))
        
        ttk.Label(main_frame, text="检查间隔(秒):").grid(row=2, column=0, sticky=tk.W, padx=(0, 10))
        self.interval_var = tk.StringVar(value="30")
        interval_entry = ttk.Entry(main_frame, textvariable=self.interval_var, width=10)
        interval_entry.grid(row=2, column=1, sticky=tk.W, padx=(0, 10))
        
        # 控制按钮
        button_frame = ttk.Frame(main_frame)
        button_frame.grid(row=3, column=0, columnspan=3, pady=20)
        
        self.start_button = ttk.Button(button_frame, text="开始监控", command=self.start_monitoring)
        self.start_button.pack(side=tk.LEFT, padx=(0, 10))
        
        self.stop_button = ttk.Button(button_frame, text="停止监控", command=self.stop_monitoring, state=tk.DISABLED)
        self.stop_button.pack(side=tk.LEFT, padx=(0, 10))
        
        self.test_button = ttk.Button(button_frame, text="测试查询", command=self.test_query)
        self.test_button.pack(side=tk.LEFT)
        
        # 状态显示区域
        status_frame = ttk.LabelFrame(main_frame, text="监控状态", padding="10")
        status_frame.grid(row=4, column=0, columnspan=3, sticky=(tk.W, tk.E, tk.N, tk.S), pady=(0, 10))
        status_frame.columnconfigure(0, weight=1)
        status_frame.rowconfigure(1, weight=1)
        
        # 天一广场状态
        self.tianyi_status = ttk.Label(status_frame, text="🏪 天一广场: 等待查询...", font=('Arial', 12, 'bold'))
        self.tianyi_status.grid(row=0, column=0, sticky=tk.W, pady=(0, 10))
        
        # 日志显示
        self.log_text = scrolledtext.ScrolledText(status_frame, height=15, width=70)
        self.log_text.grid(row=1, column=0, sticky=(tk.W, tk.E, tk.N, tk.S))
        
        # 底部状态栏
        self.status_var = tk.StringVar(value="就绪")
        status_bar = ttk.Label(main_frame, textvariable=self.status_var, relief=tk.SUNKEN)
        status_bar.grid(row=5, column=0, columnspan=3, sticky=(tk.W, tk.E), pady=(10, 0))
    
    def play_urgent_sound(self):
        """播放急促的提醒声音"""
        def play_sound_thread():
            try:
                system = platform.system()
                # 播放3次急促的提醒音
                for i in range(3):
                    if system == "Darwin":  # macOS
                        os.system("afplay /System/Library/Sounds/Glass.aiff")
                        time.sleep(0.2)
                        os.system("afplay /System/Library/Sounds/Glass.aiff")
                        time.sleep(0.2)
                        os.system("afplay /System/Library/Sounds/Glass.aiff")
                    elif system == "Windows":
                        import winsound
                        for j in range(3):
                            winsound.Beep(1500, 200)  # 高频短促
                            time.sleep(0.1)
                    elif system == "Linux":
                        for j in range(3):
                            os.system("paplay /usr/share/sounds/alsa/Front_Left.wav 2>/dev/null || echo -e '\a'")
                            time.sleep(0.1)
                    else:
                        for j in range(3):
                            print("\a")
                            time.sleep(0.1)
                    
                    if i < 2:  # 不是最后一次
                        time.sleep(0.5)
            except Exception as e:
                # 如果播放失败，使用系统铃声
                for i in range(5):
                    print("\a")
                    time.sleep(0.2)
        
        # 在单独线程中播放声音，避免阻塞GUI
        sound_thread = threading.Thread(target=play_sound_thread, daemon=True)
        sound_thread.start()
    
    def log_message(self, message, level="INFO"):
        """在GUI中显示日志消息"""
        timestamp = datetime.now().strftime("%H:%M:%S")
        log_entry = f"[{timestamp}] {message}\n"
        
        # 在GUI线程中更新文本
        self.root.after(0, lambda: self._update_log_text(log_entry, level))
        
        # 同时记录到文件
        if level == "INFO":
            self.logger.info(message)
        elif level == "ERROR":
            self.logger.error(message)
        elif level == "WARNING":
            self.logger.warning(message)
    
    def _update_log_text(self, message, level):
        """更新日志文本框"""
        self.log_text.insert(tk.END, message)
        
        # 根据级别设置颜色
        if level == "ERROR":
            self.log_text.tag_add("error", f"end-{len(message)}c", "end-1c")
            self.log_text.tag_config("error", foreground="red")
        elif level == "WARNING":
            self.log_text.tag_add("warning", f"end-{len(message)}c", "end-1c")
            self.log_text.tag_config("warning", foreground="orange")
        
        # 自动滚动到底部
        self.log_text.see(tk.END)
    
    def update_tianyi_status(self, status, has_stock=False):
        """更新天一广场状态显示"""
        if has_stock:
            status_text = f"🏪 天一广场: ✅ {status}"
            self.tianyi_status.config(foreground="green")
        else:
            status_text = f"🏪 天一广场: ❌ {status}"
            self.tianyi_status.config(foreground="red")
        
        self.root.after(0, lambda: self.tianyi_status.config(text=status_text))
    
    def check_stock(self, part_number: str) -> Dict:
        """检查库存状态"""
        try:
            params = {
                'little': 'false',
                'parts.0': part_number,
                'location': self.location
            }
            
            response = self.session.get(self.base_url, params=params, timeout=10)
            response.raise_for_status()
            
            data = response.json()
            
            if 'body' in data and 'content' in data['body']:
                content = data['body']['content']
                if 'pickupMessage' in content:
                    pickup_info = content['pickupMessage']
                    stores = pickup_info.get('stores', [])
                    
                    if stores:
                        return {
                            'available': True,
                            'stores': stores,
                            'check_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                        }
            
            return {
                'available': False,
                'check_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
                'message': '暂无库存'
            }
            
        except Exception as e:
            return {
                'available': False,
                'error': str(e),
                'check_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            }
    
    def test_query(self):
        """测试查询功能"""
        model_code = self.model_var.get().strip()
        if not model_code:
            messagebox.showerror("错误", "请输入产品型号代码")
            return
        
        self.status_var.set("正在测试查询...")
        self.log_message(f"开始测试查询产品: {model_code}")
        
        def test_thread():
            result = self.check_stock(model_code)
            
            if result.get('error'):
                self.log_message(f"查询失败: {result['error']}", "ERROR")
                self.root.after(0, lambda: self.status_var.set("查询失败"))
            elif result['available']:
                stores = result['stores']
                tianyi_store = None
                other_stores = []
                
                for store in stores:
                    if '天一广场' in store['storeName']:
                        tianyi_store = store
                    else:
                        other_stores.append(store)
                
                if tianyi_store:
                    self.update_tianyi_status(f"有货 - {tianyi_store['pickupDisplay']}", True)
                    self.log_message(f"✅ 天一广场有货: {tianyi_store['pickupDisplay']}")
                else:
                    self.update_tianyi_status("暂无库存", False)
                
                if other_stores:
                    self.log_message(f"📍 附近门店有货 ({len(other_stores)}家):")
                    for store in other_stores:
                        self.log_message(f"   {store['storeName']} - {store['pickupDisplay']}")
                
                self.root.after(0, lambda: self.status_var.set("查询完成 - 有库存"))
            else:
                self.update_tianyi_status("暂无库存", False)
                self.log_message("❌ 暂无库存")
                self.root.after(0, lambda: self.status_var.set("查询完成 - 无库存"))
        
        thread = threading.Thread(target=test_thread, daemon=True)
        thread.start()
    
    def start_monitoring(self):
        """开始监控"""
        model_code = self.model_var.get().strip()
        if not model_code:
            messagebox.showerror("错误", "请输入产品型号代码")
            return
        
        try:
            interval = int(self.interval_var.get())
            if interval < 5:
                messagebox.showerror("错误", "检查间隔不能少于5秒")
                return
        except ValueError:
            messagebox.showerror("错误", "请输入有效的检查间隔")
            return
        
        self.monitoring = True
        self.start_button.config(state=tk.DISABLED)
        self.stop_button.config(state=tk.NORMAL)
        self.status_var.set(f"正在监控 {model_code}...")
        
        self.log_message(f"🍎 开始监控产品: {model_code}")
        self.log_message(f"📍 监控地区: {self.location}")
        self.log_message(f"⏰ 检查间隔: {interval}秒")
        
        def monitor_thread():
            check_count = 0
            while self.monitoring:
                check_count += 1
                result = self.check_stock(model_code)
                
                if result.get('error'):
                    self.log_message(f"第{check_count}次检查失败: {result['error']}", "ERROR")
                elif result['available']:
                    stores = result['stores']
                    tianyi_store = None
                    other_stores = []
                    
                    for store in stores:
                        if '天一广场' in store['storeName']:
                            tianyi_store = store
                        else:
                            other_stores.append(store)
                    
                    self.log_message(f"🎉 第{check_count}次检查 - 有货!")
                    
                    # 播放急促提醒音
                    self.play_urgent_sound()
                    
                    if tianyi_store:
                        self.update_tianyi_status(f"有货 - {tianyi_store['pickupDisplay']}", True)
                        self.log_message(f"✅ 天一广场: {tianyi_store['storeName']} - {tianyi_store['pickupDisplay']}")
                        self.log_message(f"   地址: {tianyi_store['address']['street']}")
                        self.log_message(f"   电话: {tianyi_store['phoneNumber']}")
                    else:
                        self.update_tianyi_status("暂无库存", False)
                    
                    if other_stores:
                        self.log_message(f"📍 附近门店有货 ({len(other_stores)}家):")
                        for store in other_stores:
                            self.log_message(f"   {store['storeName']} - {store['pickupDisplay']} (距离: {store['distance']} km)")
                else:
                    self.log_message(f"❌ 第{check_count}次检查 - 暂无库存")
                    self.update_tianyi_status("暂无库存", False)
                
                # 等待指定间隔
                for i in range(interval):
                    if not self.monitoring:
                        break
                    time.sleep(1)
            
            self.log_message(f"监控停止，共检查了 {check_count} 次")
        
        self.monitor_thread = threading.Thread(target=monitor_thread, daemon=True)
        self.monitor_thread.start()
    
    def stop_monitoring(self):
        """停止监控"""
        self.monitoring = False
        self.start_button.config(state=tk.NORMAL)
        self.stop_button.config(state=tk.DISABLED)
        self.status_var.set("监控已停止")
        self.log_message("监控已停止")
    
    def run(self):
        """运行GUI应用"""
        self.root.protocol("WM_DELETE_WINDOW", self.on_closing)
        self.root.mainloop()
    
    def on_closing(self):
        """关闭窗口时的处理"""
        if self.monitoring:
            if messagebox.askokcancel("退出", "监控正在运行，确定要退出吗？"):
                self.monitoring = False
                self.root.destroy()
        else:
            self.root.destroy()

def main():
    """主函数"""
    app = GUIStockMonitor()
    app.run()

if __name__ == "__main__":
    main()