web_stock_monitor.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
苹果库存监控工具 - Web版本
基于Flask和WebSocket的网页界面
"""

import requests
import time
import logging
from datetime import datetime
import json
import os
import platform
import threading
from flask import Flask, render_template, request, jsonify
from flask_socketio import SocketIO, emit
from typing import Dict, Optional

class WebStockMonitor:
    def __init__(self, debug=False):
        self.base_url = "https://www.apple.com.cn/shop/fulfillment-messages"
        self.location = "浙江 宁波 鄞州区"
        self.session = requests.Session()
        self.debug = debug
        self.setup_logging()
        self.monitoring = False
        self.monitor_thread = None
        self.app = Flask(__name__)
        self.app.config['SECRET_KEY'] = 'apple_stock_monitor_secret'
        self.socketio = SocketIO(self.app, cors_allowed_origins="*")
        
        # 设置请求头
        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_routes()
    
    def setup_logging(self):
        """设置日志记录"""
        handlers = [logging.StreamHandler()]
        
        # 只有在debug模式下才添加文件日志处理器
        if self.debug:
            handlers.append(logging.FileHandler('web_stock_monitor.log', encoding='utf-8'))
        
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=handlers
        )
        self.logger = logging.getLogger(__name__)
    
    def setup_routes(self):
        """设置Web路由"""
        @self.app.route('/')
        def index():
            return render_template('index.html')
        
        @self.app.route('/api/test', methods=['POST'])
        def test_query():
            data = request.get_json()
            model_code = data.get('model_code', '').strip()
            
            if not model_code:
                return jsonify({'success': False, 'error': '请输入产品型号代码'})
            
            result = self.check_stock(model_code)
            return jsonify({'success': True, 'result': result})
        
        @self.socketio.on('start_monitoring')
        def handle_start_monitoring(data):
            model_code = data.get('model_code', '').strip()
            interval = data.get('interval', 30)
            
            if not model_code:
                emit('error', {'message': '请输入产品型号代码'})
                return
            
            if interval < 5:
                emit('error', {'message': '检查间隔不能少于5秒'})
                return
            
            if self.monitoring:
                emit('error', {'message': '监控已在运行中'})
                return
            
            self.start_monitoring_thread(model_code, interval)
            emit('monitoring_started', {'model_code': model_code, 'interval': interval})
        
        @self.socketio.on('stop_monitoring')
        def handle_stop_monitoring():
            self.stop_monitoring()
            emit('monitoring_stopped')
        
        @self.socketio.on('connect')
        def handle_connect():
            emit('connected', {'message': '连接成功'})
        
        @self.socketio.on('disconnect')
        def handle_disconnect():
            print('客户端断开连接')
    
    def play_urgent_sound_web(self):
        """通过WebSocket发送声音提醒指令"""
        self.socketio.emit('play_sound', {'type': 'urgent'})
    
    def emit_log(self, message, level="info"):
        """发送日志消息到Web界面"""
        timestamp = datetime.now().strftime("%H:%M:%S")
        self.socketio.emit('log_message', {
            'timestamp': timestamp,
            'message': message,
            'level': level
        })
        
        # 同时记录到文件
        if level == "info":
            self.logger.info(message)
        elif level == "error":
            self.logger.error(message)
        elif level == "warning":
            self.logger.warning(message)
    
    def emit_tianyi_status(self, status, has_stock=False):
        """发送天一广场状态更新"""
        self.socketio.emit('tianyi_status', {
            'status': status,
            'has_stock': has_stock
        })
    
    def check_stock(self, part_number: str) -> Dict:
        """检查库存状态"""
        try:
            params = {
                'fae': 'true',
                'pl': 'true',
                'mts.0': 'regular',
                'mts.1': 'compact',
                'cppart': 'UNLOCKED/WW',
                'parts.0': part_number,
                'searchNearby': 'true',
                'store': 'R531'  # 使用天一广场门店ID
            }
            
            # 更新请求头以包含更多验证信息
            headers = {
                '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': 'dkjnqd4y3g-mf0ugw72',
                'referer': 'https://www.apple.com.cn/shop/buy-iphone/iphone-16-pro/MYTX3CH/A',
                'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36 Edg/139.0.0.0'
            }
            
            # 记录请求参数
            self.logger.info(f"=== API请求开始 ===")
            self.logger.info(f"请求URL: {self.base_url}")
            self.logger.info(f"请求参数: {json.dumps(params, ensure_ascii=False, indent=2)}")
            self.logger.info(f"请求头: {json.dumps(headers, ensure_ascii=False, indent=2)}")
            
            response = self.session.get(self.base_url, params=params, headers=headers, timeout=10)
            response.raise_for_status()
            
            data = response.json()
            
            # 记录完整响应数据
            self.logger.info(f"=== API响应数据 ===")
            self.logger.info(f"响应状态码: {response.status_code}")
            self.logger.info(f"完整响应数据: {json.dumps(data, ensure_ascii=False, indent=2)}")
            self.logger.info(f"=== API响应结束 ===")
            
            if 'body' in data and 'content' in data['body']:
                content = data['body']['content']
                self.logger.info(f"Content结构: {list(content.keys()) if isinstance(content, dict) else type(content)}")
                
                if 'pickupMessage' in content:
                    pickup_info = content['pickupMessage']
                    
                    # 检查是否有错误消息
                    if 'errorMessage' in pickup_info:
                        self.logger.error(f"API返回错误: {pickup_info['errorMessage']}")
                        return {
                            'available': False,
                            'stores': [],
                            'check_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
                            'message': pickup_info['errorMessage']
                        }
                    
                    # 获取门店信息
                    stores = pickup_info.get('stores', [])
                    availability_stores = pickup_info.get('availabilityStores', '')
                    
                    self.logger.info(f"找到 {len(stores)} 家门店")
                    if availability_stores:
                        self.logger.info(f"可用门店ID: {availability_stores}")
                    
                    # 如果没有门店信息，输出完整的pickupMessage用于调试
                    if len(stores) == 0:
                        self.logger.error(f"pickupMessage为空，完整数据: {pickup_info}")
                        
                        # 如果有availabilityStores但没有stores数组，创建虚拟门店信息
                        if availability_stores:
                            store_ids = availability_stores.split(',')
                            stores = []
                            for store_id in store_ids:
                                stores.append({
                                    'storeNumber': store_id.strip(),
                                    'storeName': f'门店 {store_id.strip()}',
                                    'pickupDisplay': '有货',
                                    'distance': '未知'
                                })
                            self.logger.info(f"根据availabilityStores创建了 {len(stores)} 家门店信息")
                    
                    # 检查是否有门店有库存
                    has_stock = False
                    for store in stores:
                        # 从partsAvailability中获取库存信息
                        parts_availability = store.get('partsAvailability', {})
                        part_info = parts_availability.get(part_number, {})
                        pickup_display = part_info.get('pickupDisplay', '无法提供')
                        
                        # 记录每个门店的库存状态
                        store_name = store.get('storeName', '未知门店')
                        self.logger.info(f"门店: {store_name} - pickupDisplay: {pickup_display}")
                        
                        if pickup_display != '无法提供' and '暂无库存' not in pickup_display and pickup_display != 'unavailable':
                            has_stock = True
                            self.logger.info(f"✅ {store_name} - 有库存: {pickup_display}")
                        else:
                            self.logger.info(f"❌ {store_name} - 无库存: {pickup_display}")
                    
                    return {
                        'available': has_stock,
                        'stores': stores,
                        'check_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                    }
                else:
                    self.logger.warning(f"未找到pickupMessage，content包含: {list(content.keys())}")
                    # 输出完整的content数据用于调试
                    self.logger.error(f"完整的content数据: {content}")
            else:
                self.logger.warning(f"API响应结构异常，data包含: {list(data.keys()) if isinstance(data, dict) else 'non-dict'}")
                # 输出完整的API响应数据用于调试
                self.logger.error(f"完整的API响应数据: {data}")
            
            return {
                'available': False,
                'stores': [],
                'check_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
                'message': '暂无库存'
            }
            
        except Exception as e:
            self.logger.error(f"API调用异常: {str(e)}")
            return {
                'available': False,
                'stores': [],
                'error': str(e),
                'check_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            }
    
    def start_monitoring_thread(self, model_code, interval):
        """启动监控线程"""
        self.monitoring = True
        
        def monitor_thread():
            check_count = 0
            self.emit_log(f"🍎 开始监控产品: {model_code}")
            self.emit_log(f"📍 监控地区: {self.location}")
            self.emit_log(f"⏰ 检查间隔: {interval}秒")
            
            while self.monitoring:
                check_count += 1
                result = self.check_stock(model_code)
                
                if result.get('error'):
                    self.emit_log(f"第{check_count}次检查失败: {result['error']}", "error")
                    self.emit_tianyi_status("查询失败", False)
                else:
                    stores = result.get('stores', [])
                    tianyi_store = None
                    available_stores = []
                    unavailable_stores = []
                    
                    for store in stores:
                        # 从partsAvailability中获取库存状态
                        parts_availability = store.get('partsAvailability', {})
                        part_info = parts_availability.get(model_code, {})
                        pickup_display = part_info.get('pickupDisplay', '无法提供')
                        is_available = pickup_display != '无法提供' and '暂无库存' not in pickup_display and pickup_display != 'unavailable'
                        
                        if '天一广场' in store['storeName']:
                            tianyi_store = store
                        elif is_available:
                            available_stores.append(store)
                        else:
                            unavailable_stores.append(store)
                    
                    if result['available']:
                        self.emit_log(f"🎉 第{check_count}次检查 - 有货!")
                        # 发送声音提醒
                        self.play_urgent_sound_web()
                    else:
                        self.emit_log(f"❌ 第{check_count}次检查 - 暂无库存")
                    
                    # 显示天一广场状态
                    if tianyi_store:
                        # 从partsAvailability中获取天一广场的库存状态
                        tianyi_parts = tianyi_store.get('partsAvailability', {})
                        tianyi_part_info = tianyi_parts.get(model_code, {})
                        tianyi_pickup = tianyi_part_info.get('pickupDisplay', '无法提供')
                        tianyi_available = tianyi_pickup != '无法提供' and '暂无库存' not in tianyi_pickup and tianyi_pickup != 'unavailable'
                        
                        if tianyi_available:
                            self.emit_tianyi_status(f"有货 - {tianyi_pickup}", True)
                            self.emit_log(f"✅ 天一广场: {tianyi_store['storeName']} - {tianyi_pickup}")
                            if 'address' in tianyi_store:
                                self.emit_log(f"   地址: {tianyi_store['address'].get('street', '地址信息不可用')}")
                            if 'phoneNumber' in tianyi_store:
                                self.emit_log(f"   电话: {tianyi_store['phoneNumber']}")
                        else:
                            self.emit_tianyi_status("暂无库存", False)
                            self.emit_log(f"❌ 天一广场: {tianyi_store['storeName']} - {tianyi_pickup}")
                    else:
                        self.emit_tianyi_status("暂无库存", False)
                        self.emit_log("❌ 天一广场: 未找到门店信息")
                    
                    # 发送所有门店状态更新事件
                    all_stores = available_stores + unavailable_stores
                    self.socketio.emit('all_stores_status', {
                        'stores': all_stores,
                        'model_code': model_code
                    })
                    
                    # 显示附近有货门店
                    if available_stores:
                        self.emit_log(f"📍 附近门店有货 ({len(available_stores)}家):")
                        for store in available_stores:
                            distance = store.get('distance', '未知')
                            # 从partsAvailability中获取库存信息
                            parts_availability = store.get('partsAvailability', {})
                            part_info = parts_availability.get(model_code, {})
                            pickup_display = part_info.get('pickupDisplay', '未知')
                            pickup_quote = part_info.get('pickupSearchQuote', '')
                            display_text = f"{pickup_display} - {pickup_quote}" if pickup_quote else pickup_display
                            self.emit_log(f"   ✅ {store['storeName']} - {display_text} (距离: {distance} km)")
                            if 'address' in store:
                                self.emit_log(f"      地址: {store['address'].get('street', '地址信息不可用')}")
                            if 'phoneNumber' in store:
                                self.emit_log(f"      电话: {store['phoneNumber']}")
                    
                    # 显示附近无货门店（仅显示前3家）
                    if unavailable_stores:
                        display_count = min(3, len(unavailable_stores))
                        self.emit_log(f"📍 附近门店暂无库存 (显示{display_count}/{len(unavailable_stores)}家):")
                        for store in unavailable_stores[:display_count]:
                            distance = store.get('distance', '未知')
                            # 从partsAvailability中获取库存信息
                            parts_availability = store.get('partsAvailability', {})
                            part_info = parts_availability.get(model_code, {})
                            pickup_display = part_info.get('pickupDisplay', '无法提供')
                            self.emit_log(f"   ❌ {store['storeName']} - {pickup_display} (距离: {distance} km)")
                
                # 等待指定间隔
                for i in range(interval):
                    if not self.monitoring:
                        break
                    time.sleep(1)
            
            self.emit_log(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.emit_log("监控已停止")
    
    def run(self, host='127.0.0.1', port=5000, debug=False):
        """运行Web应用"""
        print(f"🍎 苹果库存监控工具 - Web版本")
        print(f"🌐 访问地址: http://{host}:{port}")
        print(f"📱 请在浏览器中打开上述地址")
        
        # 创建模板目录
        os.makedirs('templates', exist_ok=True)
        
        self.socketio.run(self.app, host=host, port=port, debug=debug)

def main():
    """主函数"""
    import argparse
    
    parser = argparse.ArgumentParser(description='苹果库存监控工具 - Web版本')
    parser.add_argument('--debug', action='store_true', help='开启调试模式，记录日志文件')
    args = parser.parse_args()
    
    monitor = WebStockMonitor(debug=args.debug)
    print("🍎 苹果库存监控工具 - Web版本")
    print("🌐 访问地址: http://127.0.0.1:5000")
    print("📱 请在浏览器中打开上述地址")
    if args.debug:
        print("🐛 调试模式已开启，日志将记录到 web_stock_monitor.log")
    else:
        print("📝 日志记录已关闭，使用 --debug 参数开启")
    monitor.run()

if __name__ == "__main__":
    main()