Apple Store Pickup Minitor

apple-pickup-monitor-ca.py

import os
import sys
import time
import importlib.util

requirements = {
    'requests': 'requests',
    'yaml': 'pyyaml',
}

for name, package in requirements.items():

    if importlib.util.find_spec(name) is None:
        os.system(f'pip3 install {package}')

    globals()[name] = importlib.import_module(name=name)


class ApplePickupMonitor:

    SEARCH_URL = 'https://www.apple.com/ca/shop/fulfillment-messages'

    def __init__(self, location: str, models: list[str], cities: list[str] = None, interval: int = 30, notify_url: str = None) -> None:
        self.location = location
        self.models = models
        self.cities = cities
        self.interval = interval
        self.notify_url = notify_url

    @classmethod
    def search(cls, location: str, models: list[str]) -> dict:

        with requests.get(url=cls.SEARCH_URL, params={
            "mts.0": "regular",
            "mts.1": "compact",
            "pl": "true",
            "location": location,
            **{f'parts.{i}': model for i, model in enumerate(models)},
        }) as r:
            return r.json()

    def do_search(self) -> dict:

        search_result = self.search(self.location, self.models)
        assert search_result['head']['status'] == '200'

        data = {}
        for store in search_result['body']['content']['pickupMessage']['stores']:
            for model, info in store['partsAvailability'].items():
                storeState = store['retailStore']['address']['state']
                storeCity = store['retailStore']['address']['city']
                storeName = store['retailStore']['address']['companyName']

                msgType = info["messageTypes"]
                _info = msgType["compact"] if "compact" in msgType else msgType["regular"]

                searchQuote = info['pickupSearchQuote']
                productTitle = _info['storePickupProductTitle']

                _ = model

                if self.cities and storeCity not in self.cities:
                    continue

                if _info['storeSelectionEnabled']:
                    data.setdefault(productTitle, [])
                    data[productTitle].append(
                        f'{storeName} <i>({storeCity}, {storeState})</i>\nPickup time: {searchQuote}')
        return data

    def request(self):

        try:
            print(f'Request Info: {self.models} in {self.location}')
            msg = "\n\n".join('<b>{}</b>\n\n{}'.format(k, "\n\n".join(v))
                              for k, v in self.do_search().items())
            if msg.strip() and self.notify_url:
                if self.notify_url.startswith("print://"):
                    print(msg)
                else:
                    with requests.post(url=self.notify_url,
                                       data=msg.encode('utf-8'),
                                       headers={'Content-type': 'text/plain; charset=utf-8'}) as r:
                        r.raise_for_status()
                print(f'Notification has been sent to: {self.notify_url}')

        except Exception as e:
            print(f'Request Error: {e}')

    def forever(self) -> None:

        while True:
            self.request()
            time.sleep(self.interval)


def main():

    with open(sys.argv[1], 'rb') as f:
        config = yaml.safe_load(f)

    m = ApplePickupMonitor(**config)
    m.forever()


if __name__ == '__main__':
    try:
        main()
    except KeyboardInterrupt:
        exit()

apple-pickup-monitor-cn.py

import os
import sys
import time
import importlib.util

requirements = {
    'requests': 'requests',
    'yaml': 'pyyaml',
}

for name, package in requirements.items():

    if importlib.util.find_spec(name) is None:
        os.system(f'pip3 install {package}')

    globals()[name] = importlib.import_module(name=name)


class ApplePickupMonitor:

    SEARCH_URL = 'https://www.apple.com.cn/shop/fulfillment-messages'

    def __init__(self, location: str, models: list[str], cities: list[str] = None, interval: int = 30, notify_url: str = None) -> None:
        self.location = location
        self.models = models
        self.cities = cities
        self.interval = interval
        self.notify_url = notify_url

    @classmethod
    def search(cls, location: str, models: list[str]) -> dict:

        with requests.get(url=cls.SEARCH_URL, params={
            "mt": "regular",
            "little": "false",
            "pl": "true",
            "location": location,
            **{f'parts.{i}': model for i, model in enumerate(models)},
        }) as r:
            return r.json()

    def do_search(self) -> dict:

        search_result = self.search(self.location, self.models)
        assert search_result['head']['status'] == '200'

        data = {}
        for store in search_result['body']['content']['pickupMessage']['stores']:
            for model, info in store['partsAvailability'].items():
                productTitle = info['storePickupProductTitle']
                searchQuote = info['pickupSearchQuote']
                storeState = store['retailStore']['address']['state']
                storeCity = store['retailStore']['address']['city']
                storeName = store['retailStore']['address']['companyName']

                _ = model

                if self.cities and storeCity not in self.cities:
                    continue

                if info['storeSelectionEnabled']:
                    data.setdefault(productTitle, [])
                    data[productTitle].append(
                        f'{storeName} <i>({storeState}, {storeCity})</i>\n取货时间: {searchQuote}')
        return data

    def request(self):

        try:
            print(f'Request Info: {self.models} in {self.location}')
            msg = "\n\n".join('<b>{}</b>\n\n{}'.format(k, "\n\n".join(v))
                              for k, v in self.do_search().items())
            if msg.strip() and self.notify_url:
                with requests.post(url=self.notify_url,
                                   data=msg.encode('utf-8'),
                                   headers={'Content-type': 'text/plain; charset=utf-8'}) as r:
                    r.raise_for_status()
                print(f'Notification has been sent to: {self.notify_url}')

        except Exception as e:
            print(f'Request Error: {e}')

    def forever(self) -> None:

        while True:
            self.request()
            time.sleep(self.interval)


def main():

    with open(sys.argv[1], 'rb') as f:
        config = yaml.safe_load(f)

    m = ApplePickupMonitor(**config)
    m.forever()


if __name__ == '__main__':
    try:
        main()
    except KeyboardInterrupt:
        exit()

config-ca.yml

interval: 30

location: L7L 0A5

cities:
  - Burlington
  - Toronto

models:
  - MTUC3VC/A # iPhone 15 Pro 256GB Black Titanium
  - MTUF3VC/A # iPhone 15 Pro 256GB Natural Titanium
  - MTUG3VC/A # iPhone 15 Pro 256GB Blue Titanium
  - MTUD3VC/A # iPhone 15 Pro 256GB White Titanium

notify_url: print://

config-cn.yml

interval: 30

location: 浙江 杭州 西湖区

cities:
  - 杭州

models:
  - MLE03CH/A

notify_url: https://xxxx

docker-compose.yml

version: '2.4'

services:

  apple-monitor:
    image: python:alpine
    volumes:
      - ./apple-pickup-monitor-ca.py:/apple-pickup-monitor-ca.py:ro
      - ./config-ca.yml:/config-ca.yml:ro
    # dns:
    #   - 198.18.0.1
    # network_mode: bridge
    restart: always
    container_name: apple-monitor
    command:
      [
        'sh',
        '-euc',
        'pip3 install -U pyyaml requests && python3 apple-pickup-monitor-ca.py config-ca.yml'
      ]