Reduce the waiting time
import logging
import sys
import time
from enum import Enum
from typing import Optional

from actions.order import OrderSessionPool
from common.schemas import DeliverySchema, ShopSchema, OrderSchema, OrderDeliverySchema
from libs.notifications import NotificationBase
from libs.requests import Request

logger = logging.getLogger(__name__)

apple_api_host = "https://www.apple.com"


class DeliveryStatusEnum(str, Enum):
    AVAILABLE = "available"
    INELIGIBLE = "ineligible"


class InventoryMonitor(object):
    def __init__(self) -> None:
        super().__init__()
        self.session = Request(apple_api_host)
        self.is_stop = False
        self.order_pool: Optional[OrderSessionPool] = None

    def start(
        self,
        shop_data: ShopSchema,
        order: bool = False,
        delivery_data: Optional[OrderDeliverySchema] = None,
        notification_providers: Optional[list[NotificationBase]] = None,
        interval: int = 5,
        order_notice_count: int = 1,
    ):
        logger.info(f"Start monitoring, query interval: {interval}s")
        order_data: Optional[OrderSchema] = None
        if order:
            # only support one product
            # fixme Is there a better way to obtain the model code?
            order_data = OrderSchema(
                model=shop_data.models[0],
                model_code=shop_data.code,
                country=shop_data.country,
            )
            self.enable_order(order_data)

        ignore_wait = False
        while not self.is_stop:
            try:
                ignore_wait = False
                inventory_data = self.get_data(
                    shop_data.country,
                    shop_data.models,
                    shop_data.location,
                )
                pickup_lists = self.parse_data(inventory_data)
                pickup_lists = [i for i in pickup_lists if not shop_data.store_filters or any(
                        [
                            True
                            for ii in shop_data.store_filters
                            if ii in i.store_name
                        ]
                    )]

                if not pickup_lists:
                    logger.warning("No available stores found")
                    time.sleep(interval)
                    continue

                for pickup in pickup_lists:
                    logger.info(pickup.intro())

                available_lists = [i for i in pickup_lists if i.status == DeliveryStatusEnum.AVAILABLE]
                if available_lists and notification_providers:
                    self.push_notifications(available_lists, notification_providers)

                if available_lists and order:
                    for pickup in available_lists:
                        order_data.store_number = pickup.store_number
                        order_data.state = pickup.state
                        order_data.city = pickup.city
                        order_data.district = pickup.district
                        order_data.delivery = delivery_data

                        order_result = self.start_order(
                            order_data,
                            notification_providers,
                            notice_count=order_notice_count,
                        )
                        if order_result is False:
                            ignore_wait = True

            except Exception as e:
                logging.exception(
                    "Failed to retrieve inventory data with error: ", exc_info=e
                )
                ignore_wait = True
            if not ignore_wait:
                time.sleep(interval)

    def enable_order(self, data: OrderSchema):
        self.order_pool = OrderSessionPool()
        self.order_pool.start(data)

    def start_order(
        self,
        data: OrderSchema,
        notification_providers: list[NotificationBase],
        notice_count: int,
    ):
        order_obj = self.order_pool.get()
        order_result = order_obj.start_order(data)
        if order_result:
            for provider in notification_providers:
                title, content = (
                    "Order success notification",
                    "Check your email for detailed information.",
                )
                provider.repeat_push(title, content, max_count=notice_count)
            logger.info(
                "The order has been successfully placed, and the program will automatically exit."
            )
            self.stop()
        return order_result

    def push_notifications(
        self, pickup_lists: list[DeliverySchema], providers: list[NotificationBase]
    ):
        title = "Apple inventory notification"
        buffers = []
        for pickup in pickup_lists:
            buffers.append(pickup.intro())

        if not buffers:
            return

        for provider in providers:
            try:
                provider.push(
                    title,
                    "\r\n".join(buffers),
                    key=f"inventory_monitor_{provider.name}",
                    min_interval=60,
                )
            except Exception as e:
                logging.exception(
                    "Inventory information push failed with error: ", exc_info=e
                )

    def get_data(
        self,
        country: str,
        models: list[str],
        location: str = "",
        postal_code: str = "",
        state: str = "",
    ):
        parts = {f"parts.{idx}": i for idx, i in enumerate(models)}
        search_params = {
            "searchNearby": "true",
            "pl": "true",
            "mts.0": "regular",
            "mts.1": "compact",
        } | parts
        if location:
            search_params["location"] = location
        if postal_code:
            search_params["postalCode"] = postal_code
        if state:
            search_params["state"] = state

        resp = self.session.get(
            f"/{country}/shop/fulfillment-messages", params=search_params
        )

        return resp.json()

    def parse_data(self, data: dict):
        pickup_message = data["body"]["content"]["pickupMessage"]
        if not pickup_message.get("stores"):
            logger.error("No stores found")
            return []
        deliveries = []
        for store in pickup_message["stores"]:
            parts = store["partsAvailability"].values()
            for part in parts:
                address = store["retailStore"]["address"]
                model_name = part["messageTypes"]["regular"][
                    "storePickupProductTitle"
                ].replace("\xa0", " ")
                deliveries.append(
                    DeliverySchema(
                        state=address["state"],
                        city=address["city"],
                        district=address["district"],
                        store_name=store["storeName"],
                        store_number=store["storeNumber"],
                        model_name=model_name,
                        pickup_quote=part["pickupSearchQuote"],
                        model=part["partNumber"],
                        status=part["pickupDisplay"],
                        pickup_type=part["pickupType"],
                    )
                )

        return deliveries

    def stop(self):
        self.is_stop = True
        self.order_pool and self.order_pool.stop()
        sys.exit(0)

src/actions/inventory_monitoring.py
ToolGallery/APEStoreAssistant
import dataclasses
import json
import logging
import random
import re
import threading
import time
from datetime import datetime
from typing import Optional
from urllib.parse import urlparse, parse_qsl, quote_plus

from common.schemas import OrderSchema
from libs.requests import Request

apple_api_host = "https://www.apple.com"

logger = logging.getLogger(__name__)


class Order(object):
    def __init__(self, country: str) -> None:
        super().__init__()
        # only support cn yet
        assert country == "cn", "Only support cn yet"
        api_host = apple_api_host + ".cn"
        self.session = Request(
            api_host,
            headers={
                "Referer": f"https://www.apple.com/{country}/shop/bag",
                "Content-Type": "application/x-www-form-urlencoded",
            },
        )
        self.secure_host = ""

    def init_order(self, order_data: OrderSchema):
        self.add_to_cart(order_data.model, order_data.model_code)
        cart_item_id = self.get_cart_item_id()

        signin_url, signin_params, secure_api_host = self.start_checkout(
            cart_item_id,
            order_data.country,
            order_data.state,
            order_data.city,
            order_data.district,
        )
        self.secure_host = secure_api_host

        self.get_page_with_meta(signin_url, None)

        signin_data = self.signin(signin_params)

        logger.debug("Access '/bg' page")
        self.get_page_with_meta(signin_data["head"]["data"]["url"], None)

        self.update_delivery_method()

    def start_order(self, order_data: OrderSchema):
        logger.info(
            f"Order starting with {order_data.model_code} {order_data.model} {order_data.state} {order_data.city}..."
        )

        address_data = self.fill_address(
            order_data.store_number,
            order_data.country,
            order_data.state,
            order_data.city,
            order_data.district,
        )

        selected_window = self.get_select_window(address_data)
        if not selected_window:
            return False

        self.fill_contact(
            selected_window,
            order_data.store_number,
            order_data.country,
            order_data.state,
            order_data.city,
            order_data.district,
        )

        self.fill_recipient(
            order_data.delivery.first_name,
            order_data.delivery.last_name,
            order_data.delivery.email,
            order_data.delivery.phone,
            order_data.delivery.idcard,
        )
        self.fill_pay_method(
            order_data.delivery.payment, order_data.delivery.payment_number
        )
        self.finish_checkout()

        return True

    def get_cart_item_id(self):
        logger.info("Getting cart id...")
        page_data = self.get_page_with_meta("/shop/bag", None)
        item_id = page_data["shoppingCart"]["items"]["c"].pop()
        logger.debug(f"Cart item id: {item_id}")
        return item_id

    def add_to_cart(self, model_number: str, phone_model: str):
        logger.info("Adding to cart...")
        resp_atb = self.session.get("/shop/beacon/atb")
        atb_str: str = resp_atb.cookies.get("as_atb")
        atb_token = atb_str.split("|")[-1]

        params = {
            "product": model_number,
            "purchaseOption": "fullPrice",
            "step": "select",
            # "ao.add_iphone15_ac_iup": "none", # fixme apple care?
            "ams": "0",
            "atbtoken": atb_token,
            "igt": "true",
            "add-to-cart": "add-to-cart",
        }

        resp = self.session.get(
            f"/shop/buy-iphone/iphone-{phone_model}/{model_number}#",
            params=params,
        )
        assert resp.status_code == 200

    def start_checkout(
        self, item_id: str, country: str, state: str, city: str, district: str
    ) -> (dict, str):
        logger.info("Starting checkout...")
        data = {
            "shoppingCart.recommendations.recommendedItem.part": "",
            f"shoppingCart.items.{item_id}.isIntentToGift": "false",
            f"shoppingCart.items.{item_id}.itemQuantity.quantity": "1",
            f"shoppingCart.items.{item_id}.delivery.lineDeliveryOptions.address.provinceCityDistrictTabs.city": city,
            f"shoppingCart.items.{item_id}.delivery.lineDeliveryOptions.address.provinceCityDistrictTabs.state": state,
            f"shoppingCart.items.{item_id}.delivery.lineDeliveryOptions.address.provinceCityDistrictTabs.provinceCityDistrict": f"{state} {city} {district}",
            f"shoppingCart.items.{item_id}.delivery.lineDeliveryOptions.address.provinceCityDistrictTabs.countryCode": country,
            f"shoppingCart.items.{item_id}.delivery.lineDeliveryOptions.address.provinceCityDistrictTabs.district": district,
            "shoppingCart.locationConsent.locationConsent": "false",
            "shoppingCart.summary.promoCode.promoCode": "",
            "shoppingCart.actions.fcscounter": "",
            "shoppingCart.actions.fcsdata": "",
        }
        resp = self.session.post(
            "/shop/bagx/checkout_now",
            params={
                "_a": "checkout",
                "_m": "shoppingCart.actions",
            },
            data=data,
        )
        resp_json = resp.json()
        signin_url = resp_json["head"]["data"]["url"]
        url_parsed = urlparse(signin_url)

        signin_params = dict(parse_qsl(url_parsed.query))
        secure_api_host = f"{url_parsed.scheme}://{url_parsed.hostname}"

        logger.debug(f"Secure api host: {secure_api_host}")
        return signin_url, signin_params, secure_api_host

    def signin(self, signin_params: dict):
        timestamp = str(int(time.time() * 1000))
        now = datetime.now()
        data = {
            "signIn.consentOverlay.policiesAccepted": "true",
            "signIn.consentOverlay.dataHandleByApple": "true",
            "signIn.consentOverlay.dataOutSideMyCountry": "true",
            "signIn.guestLogin.deviceID": f'TF1;015;;;;;;;;;;;;;;;;;;;;;;Mozilla;Netscape;5.0 (Macintosh);20100101;undefined;true;Intel Mac OS X 10.15;true;MacIntel;undefined;Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:{random.randrange(100, 109)}.0) Gecko/20100101 Firefox/{random.randrange(100, 117)}.0;en-US;undefined;secure7.www.apple.com.cn;undefined;undefined;undefined;undefined;false;false;{timestamp};8;6/7/2005, 9:33:44 PM;2560;1440;;;;;;;;-480;-480;{now.strftime("%-m/%-d/%Y")}, {now.strftime("%-I:%M:%S %p")};30;2560;1415;0;25;;;;;;;;;;;;;;;;;;;25;',
        }
        data["signIn.guestLogin.deviceID"] = ";".join(
            [quote_plus(i) for i in data["signIn.guestLogin.deviceID"].split(";")]
        )

        sign_resp = self.session.post(
            self.secure_host + "/shop/signInx",
            params=signin_params | {"_a": "guestLogin", "_m": "signIn.guestLogin"},
            data=data,
        )
        sign_resp_json = sign_resp.json()
        start_response = self.session.post(
            sign_resp_json["head"]["data"]["url"],
            data=sign_resp_json["head"]["data"]["args"],
        )
        start_response_json = start_response.json()
        assert start_response_json["head"]["status"] == 302
        return start_response_json

    def checkout_request(
        self,
        url: str,
        params: Optional[dict],
        data: Optional[dict] = None,
        assert_code: int = 200,
    ):
        resp = self.session.post(url, params=params, data=data)

        resp_json = resp.json()
        if assert_code:
            logger.debug(f"Url {url} response {resp_json}")
            resp_status = resp_json["head"]["status"]
            if resp_status != assert_code:
                return False
            assert (
                resp_status == assert_code
            ), f"Expected {assert_code}, actually {resp_status}"

        return resp_json

    def update_delivery_method(self):
        # Pick up at physical store
        pick_store_data = self.checkout_request(
            self.secure_host + "/shop/checkoutx",
            params={
                "_a": "selectFulfillmentLocationAction",
                "_m": "checkout.fulfillment.fulfillmentOptions",
            },
            data={
                "checkout.fulfillment.fulfillmentOptions.selectFulfillmentLocation": "RETAIL",
            },
        )

    def fill_address(
        self,
        store_number: str,
        country: str,
        state: str,
        city: str,
        district: str,
    ):
        logger.info("Starting fill address...")

        data = {
            "checkout.fulfillment.pickupTab.pickup.storeLocator.showAllStores": "false",
            "checkout.fulfillment.pickupTab.pickup.storeLocator.selectStore": store_number,
            "checkout.fulfillment.pickupTab.pickup.storeLocator.searchInput": f"{state} {city} {district}",
            "checkout.fulfillment.pickupTab.pickup.storeLocator.address.stateCitySelectorForCheckout.city": city,
            "checkout.fulfillment.pickupTab.pickup.storeLocator.address.stateCitySelectorForCheckout.state": state,
            "checkout.fulfillment.pickupTab.pickup.storeLocator.address.stateCitySelectorForCheckout.provinceCityDistrict": f"{state} {city} {district}",
            "checkout.fulfillment.pickupTab.pickup.storeLocator.address.stateCitySelectorForCheckout.countryCode": country,
            "checkout.fulfillment.pickupTab.pickup.storeLocator.address.stateCitySelectorForCheckout.district": district,
        }
        store_resp = self.checkout_request(
            self.secure_host + "/shop/checkoutx",
            params={
                "_a": "search",
                "_m": "checkout.fulfillment.pickupTab.pickup.storeLocator",
            },
            data=data,
        )

        return store_resp

    def get_select_window(self, address_data: dict):
        pickups = address_data["body"]["checkout"]["fulfillment"]["pickupTab"]["pickup"]
        if "timeSlot" not in pickups:
            logger.info("No available pickup time")
            return None
        pick_data = pickups["timeSlot"]["dateTimeSlots"]["d"]
        selected_window = {}
        for idx, window in enumerate(pick_data["timeSlotWindows"]):
            assert isinstance(window, dict)
            deep_windows = list(window.values())[0] if window else None
            if not deep_windows:
                continue
            for deep_window in deep_windows:
                if not deep_window["isRestricted"]:
                    selected_window = pick_data
                    selected_window["window"] = deep_window
                    pick_date = pick_data["pickUpDates"][idx]
                    selected_window["date"] = pick_date
                    logger.debug(f"Delivery raw: {selected_window}")
                    logger.info(
                        f"Delivery information: {pick_date.get('dayOfWeek')} {deep_window.get('Label')}"
                    )
                    break
            if selected_window:
                return selected_window
        logger.debug(f"Pick_data: {pick_data}")
        assert selected_window, "No pickup options found."

    def fill_contact(
        self,
        selected_window: dict,
        store_number: str,
        country: str,
        state: str,
        city: str,
        district: str,
    ):
        logger.info("Starting fill contact...")
        pickup_prefix = "checkout.fulfillment.pickupTab.pickup"
        dt_prefix = "checkout.fulfillment.pickupTab.pickup.timeSlot.dateTimeSlots"
        data = {
            "checkout.fulfillment.fulfillmentOptions.selectFulfillmentLocation": "RETAIL",
            f"{pickup_prefix}.storeLocator.showAllStores": "false",
            f"{pickup_prefix}.storeLocator.selectStore": store_number,
            f"{pickup_prefix}.storeLocator.searchInput": f"{state} {city} {district}",
            f"{pickup_prefix}.storeLocator.address.stateCitySelectorForCheckout.city": city,
            f"{pickup_prefix}.storeLocator.address.stateCitySelectorForCheckout.state": state,
            f"{pickup_prefix}.storeLocator.address.stateCitySelectorForCheckout.provinceCityDistrict": f"{state} {city} {district}",
            f"{pickup_prefix}.storeLocator.address.stateCitySelectorForCheckout.countryCode": country,
            f"{pickup_prefix}.storeLocator.address.stateCitySelectorForCheckout.district": district,
            f"{dt_prefix}.startTime": selected_window["window"]["checkInStart"],
            f"{dt_prefix}.displayEndTime": selected_window["displayEndTime"],
            f"{dt_prefix}.isRecommended": str(selected_window["isRecommended"]).lower(),
            f"{dt_prefix}.endTime": selected_window["window"]["checkInEnd"],
            f"{dt_prefix}.date": selected_window["date"]["date"],
            f"{dt_prefix}.timeSlotId": selected_window["window"]["SlotId"],
            f"{dt_prefix}.signKey": selected_window["window"]["signKey"],
            f"{dt_prefix}.timeZone": selected_window["window"]["timeZone"],
            f"{dt_prefix}.timeSlotValue": selected_window["window"]["timeSlotValue"],
            f"{dt_prefix}.dayRadio": selected_window["dayRadio"],
            f"{dt_prefix}.isRestricted": selected_window["isRestricted"] or "",
            f"{dt_prefix}.displayStartTime": selected_window["displayStartTime"],
        }

        contact_data = self.checkout_request(
            self.secure_host + "/shop/checkoutx",
            params={
                "_a": "continueFromFulfillmentToPickupContact",
                "_m": "checkout.fulfillment",
            },
            data=data,
        )

    def fill_recipient(
        self,
        first_name: str,
        last_name: str,
        email: str,
        phone: str,
        idcard: str,
    ):
        logger.info("Starting fill recipient...")

        data = {
            "checkout.pickupContact.selfPickupContact.selfContact.address.lastName": last_name,
            "checkout.pickupContact.selfPickupContact.selfContact.address.firstName": first_name,
            "checkout.pickupContact.selfPickupContact.selfContact.address.emailAddress": email,
            "checkout.pickupContact.selfPickupContact.selfContact.address.fullDaytimePhone": phone,
            "checkout.pickupContact.selfPickupContact.nationalIdSelf.nationalIdSelf": idcard,
            "checkout.pickupContact.eFapiaoSelector.selectFapiao": "none",  # e_personal
        }
        review_bill_data = self.checkout_request(
            self.secure_host + "/shop/checkoutx",
            params={
                "_a": "continueFromPickupContactToBilling",
                "_m": "checkout.pickupContact",
            },
            data=data,
        )

    def fill_pay_method(self, payment: str, number: int):
        logger.info("Starting fill pay methods...")
        data = {
            "checkout.billing.billingOptions.selectBillingOption": payment,
            "checkout.locationConsent.locationConsent": "false",
        }
        bill_option_data = self.checkout_request(
            self.secure_host + "/shop/checkoutx/billing",
            params={
                "_a": "selectBillingOptionAction",
                "_m": "checkout.billing.billingOptions",
            },
            data=data,
        )
        data = {
            "checkout.billing.billingOptions.selectBillingOption": payment,
            "checkout.billing.billingOptions.selectedBillingOptions.installments.installmentOptions.selectInstallmentOption": str(
                number
            ),
        }
        bill_confirm_data = self.checkout_request(
            self.secure_host + "/shop/checkoutx/billing",
            params={
                "_a": "continueFromBillingToReview",
                "_m": "checkout.billing",
            },
            data=data,
        )

    def finish_checkout(self, show_cookie: bool = False):
        logger.info("Starting final checkout...")

        place_order_data = self.get_place_order_data()

        self.get_page_with_meta(
            self.secure_host + place_order_data["head"]["data"]["url"], None
        )

        status_data = self.get_checkout_status_x()

        if show_cookie:
            cookie_str = "; ".join(
                [f"{k}={v}" for k, v in self.session.session.cookies.get_dict().items()]
            )

            logger.info("Order page cookies: %s", cookie_str)

        while True:
            thank_data = self.get_page_with_meta(
                self.secure_host + place_order_data["head"]["data"]["url"], None
            )
            thank_you_interstitial = thank_data.get("thankYouInterstitial") or {}
            order_data = thank_you_interstitial.get("d") or {}
            order_number = order_data.get("orderNumber")
            if order_number:
                logger.info(f"Order done, order number: {order_number}.")
                break
            time.sleep(1)

    def get_place_order_data(self):
        logger.info("Get order data...")
        place_order_data = self.checkout_request(
            self.secure_host + "/shop/checkoutx",
            params={
                "_a": "continueFromReviewToProcess",
                "_m": "checkout.review.placeOrder",
            },
            assert_code=302,
        )
        if place_order_data is False:
            time.sleep(1)
            return self.get_place_order_data()
        return place_order_data

    def get_checkout_status_x(self):
        logger.info("Get order status...")
        status_data = self.checkout_request(
            self.secure_host + "/shop/checkoutx/statusX",
            params={
                "_a": "checkStatus",
                "_m": "spinner",
            },
            assert_code=302,
        )
        if status_data is False:
            time.sleep(1)
            return self.get_checkout_status_x()
        return status_data

    def get_page_with_meta(self, url, params, data: Optional[dict] = None):
        page_resp = self.session.get(url, params=params, data=data)
        assert page_resp.status_code == 200
        page_content = page_resp.text
        assert "x-aos-stk" in page_content
        cart_meta_match = re.search(
            r"<script id=\"init_data\" type=\"application/json\">(.+?)</script>",
            page_content,
            flags=re.DOTALL,
        )
        assert cart_meta_match
        meta_json = cart_meta_match.group(1)
        meta_json_data = json.loads(meta_json.strip())
        headers = meta_json_data["meta"]["h"]
        self.session.session.headers.update(headers)

        return meta_json_data


@dataclasses.dataclass()
class PoolData(object):
    order: Order
    timestamp: float
    available: bool = True


class OrderSessionPool(object):
    def __init__(self, timeout: int = 60 * 30) -> None:
        super().__init__()
        self.timeout = timeout
        self.pools: list[PoolData] = []
        self.redundant_time = 60 * 5
        self.lock = threading.Lock()
        self.is_stop = False

    def start(self, order_data: OrderSchema):
        thread = threading.Thread(
            target=self.handle_pool, args=(order_data,), name="OrderPool"
        )
        thread.start()

    def handle_pool(self, order_data: OrderSchema, max_count: int = 3):
        timeout = self.timeout - self.redundant_time
        logger.info("Start maintaining the order session pool...")
        while not self.is_stop:
            for pool in self.pools:
                if time.time() - pool.timestamp >= timeout:
                    pool.available = False
            with self.lock:
                self.pools = [i for i in self.pools if i.available]

            logger.info(f"Number of available order session pools: {len(self.pools)}")

            while max_count - len(self.pools) > 0:
                pool_data = self.new(order_data)
                with self.lock:
                    self.pools.append(pool_data)
            time.sleep(30)

    def new(self, order_data: OrderSchema) -> PoolData:
        create_timestamp = time.time()
        order = Order(order_data.country)
        order.init_order(order_data)
        return PoolData(order=order, timestamp=create_timestamp)

    def stop(self):
        self.is_stop = True

    def get(self) -> Order:
        while True:
            if not self.pools:
                time.sleep(0.1)
                continue
            with self.lock:
                pool_data = self.pools.pop(0)
            return pool_data.order

src/actions/order.py
ToolGallery/APEStoreAssistant
import logging
import sys


def init_logging():
    logging.basicConfig(
        stream=sys.stdout,
        level=logging.INFO,
        format="%(asctime)s.%(msecs)03d %(levelname)s %(threadName)s.%(module)s/%(funcName)s: %(message)s",
    )


init_logging()

src/common/__init__.py
ToolGallery/APEStoreAssistant
import dataclasses
from typing import Optional


@dataclasses.dataclass()
class ShopSchema(object):
    country: str
    models: list[str]
    location: str = ""
    postal_code: str = ""
    state: str = ""
    code: str = ""
    store_filters: list[str] = dataclasses.field(default_factory=lambda: [])


@dataclasses.dataclass()
class DeliverySchema(object):
    state: str
    city: str
    district: str
    store_name: str
    store_number: str
    model_name: str
    pickup_quote: str
    model: str
    status: str
    pickup_type: str

    def intro(self) -> str:
        return " ".join(
            [
                self.store_name,
                self.model_name,
                self.pickup_type,
                self.pickup_quote,
            ]
        )


@dataclasses.dataclass()
class ProductSchema(object):
    model: str
    type: str
    color: str
    capacity: str
    color_display: str
    price: float
    price_display: str
    price_currency: str
    carrier_model: str = ""

    def key(self):
        return "-".join([self.type, self.capacity, self.color])

    def intro(self):
        buffers = [
            i
            for i in [
                self.model,
                self.type,
                self.capacity,
                self.carrier_model,
                self.color_display,
                self.price_display,
            ]
            if i
        ]
        return " ".join(buffers)


@dataclasses.dataclass()
class PaymentSchema(object):
    label: str
    key: str
    value: str
    numbers: list[int]

    def intro(self):
        return " ".join(
            [self.value, self.label]
            + (
                ["support numbers: ", ",".join(map(str, self.numbers))]
                if self.numbers
                else []
            )
        )


@dataclasses.dataclass()
class OrderDeliverySchema(object):
    first_name: str
    last_name: str
    email: str
    phone: str
    idcard: str
    payment: str
    payment_number: int = 0


@dataclasses.dataclass()
class OrderSchema(object):
    model: str
    model_code: str
    country: str
    state: str = ""
    city: str = ""
    district: str = ""
    store_number: str = ""
    delivery: Optional[OrderDeliverySchema] = None

src/common/schemas.py
ToolGallery/APEStoreAssistant
import requests


def get_address(country: str, filter_str: str = ""):
    default_headers = {
        "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:109.0) Gecko/20100101 Firefox/117.0",
    }
    filters = filter_str.split(" ")
    params = {
        "state": filters[0] if len(filters) > 0 else None,
        "city": filters[1] if len(filters) > 1 else None,
        "district": filters[2] if len(filters) > 2 else None,
    }
    resp = requests.get(
        f"https://www.apple.com/{country}/shop/address-lookup",
        params=params,
        headers=default_headers,
    )
    resp_json = resp.json()
    assert resp_json["head"]["status"] == "200"
    address_data = resp_json["body"].popitem()[1]
    if isinstance(address_data, dict):
        addresses = [i["value"] for i in address_data["data"]]
    else:
        addresses = [address_data]
    return addresses

src/libs/address.py
ToolGallery/APEStoreAssistant
import abc
import logging
import time
from typing import Optional
from urllib.parse import quote_plus

import requests


class NotificationBase(object):
    name: str

    def __init__(self, token: Optional[str] = None) -> None:
        super().__init__()
        self.token = token
        self.last_push_maps: dict[str, float] = {}

    def push(
        self, title: str, content: str, key: str = "default", min_interval: int = 0
    ):
        if (
            min_interval
            and self.last_push_maps.get(key)
            and (time.time() - self.last_push_maps[key]) < min_interval
        ):
            logging.info("Pushing too frequently, wait for the next push")
            return

        self.push_data(title, content)

        self.last_push_maps[key] = time.time()

    def repeat_push(
        self, title: str, content: str, max_count: int = 0, interval: int = 5
    ):
        max_count = 1024 * 1024 if max_count <= 0 else max_count
        for i in range(0, max_count):
            self.push_data(title, content)
            time.sleep(interval)

    @abc.abstractmethod
    def push_data(self, title: str, content: str):
        pass


class DingTalkNotification(NotificationBase):
    name = "dingtalk"

    def push_data(self, title: str, content: str):
        assert self.token, "Access_token credentials must be provided"
        url = f"https://oapi.dingtalk.com/robot/send?access_token={self.token}"
        resp = requests.post(
            url,
            json={
                "msgtype": "text",
                "text": {"content": title + "\r\n\r\n" + content},
                "at": {"isAtAll": 0},
            },
        )
        resp_json = resp.json()

        assert resp_json.get("errcode") == 0, resp_json.get("errmsg")


class BarkNotification(NotificationBase):
    name = "bark"

    def __init__(
        self,
        token: Optional[str] = None,
        host: Optional[str] = None,
    ) -> None:
        super().__init__(token=token)
        self.host = (host or "https://api.day.app").rstrip("/")

    def push_data(self, title: str, content: str):
        assert self.token, "Token credentials must be provided"
        title, content = quote_plus(title), quote_plus(content)
        url = f"{self.host}/{self.token}/{title}/{content}"
        resp = requests.get(url)
        resp_json = resp.json()

        assert resp_json.get("code") == 200, resp_json.get("message")


class FeishuNotification(NotificationBase):
    name = "feishu"

    def push_data(self, title: str, content: str):
        assert self.token, "token credentials must be provided"
        url = f"https://open.feishu.cn/open-apis/bot/v2/hook/{self.token}"
        resp = requests.post(
            url,
            json={
                "msg_type": "text",
                "content": {"text": title + "\r\n\r\n" + content},
            },
        )
        resp_json = resp.json()

        assert resp_json.get("code") == 0, resp_json.get("msg")

src/libs/notifications.py
ToolGallery/APEStoreAssistant
import json
import logging
import os.path

from common.schemas import PaymentSchema


def get_payments(country: str):
    # current only support cn yet
    payments = []
    payments_path = os.path.abspath("statics/payments")
    file_path = payments_path + f"/{country}.json"
    if not os.path.isfile(file_path):
        logging.error(f"Payment methods does not support {country} yet.")
        return []
    with open(file_path, "r") as f:
        f_content = f.read()
        payments_json = json.loads(f_content)
        for payment in payments_json:
            payments.append(
                PaymentSchema(
                    label=payment.get("label", payment.get("labelImageAlt", "")),
                    key=payment["moduleKey"],
                    value=payment["value"],
                    numbers=payment.get("numbers", [])
                )
            )

    return payments

src/libs/payments.py
ToolGallery/APEStoreAssistant
import json
import re

import requests

from common.schemas import ProductSchema


def get_products(code: str, country: str):
    default_headers = {
        "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:109.0) Gecko/20100101 Firefox/117.0",
    }
    resp = requests.get(
        f"https://www.apple.com/{country}/shop/buy-iphone/iphone-{code}",
        headers=default_headers,
    )
    content = resp.text
    assert "productSelectionData" in resp.text
    return parse_products(content)


def parse_products(content):
    select_match = re.search(
        r"window.PRODUCT_SELECTION_BOOTSTRAP = (.+?)</script>", content, flags=re.DOTALL
    )
    assert select_match
    select_text = (
        select_match.group(1)
        .strip()
        .replace("productSelectionData", '"productSelectionData"')
    )
    select_data = json.loads(select_text)["productSelectionData"]
    products = []
    prices_data = select_data["displayValues"]["prices"]
    colors_data = select_data["displayValues"]["dimensionColor"]
    for product in select_data["products"]:
        price_tag = product["fullPrice"]
        price_data = prices_data[price_tag]
        products.append(
            ProductSchema(
                type=product["familyType"],
                model=product["partNumber"],
                color=product["dimensionColor"],
                capacity=product["dimensionCapacity"],
                color_display=colors_data[product["dimensionColor"]]["value"],
                price=float(price_data["currentPrice"]["raw_amount"]),
                price_display=price_data["currentPrice"]["amount"],
                price_currency=price_data["priceCurrency"],
                carrier_model=product.get("carrierModel", ""),
            )
        )
    return sorted(products, key=lambda x: (x.price, x.color))

src/libs/products.py
ToolGallery/APEStoreAssistant
from typing import Optional

import requests


class Request(object):
    default_headers = {
        "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.14; rv:108.0) Gecko/20100101 Firefox/116.0",
    }

    def __init__(
        self, host: str, headers: Optional[dict] = None, timeout: int = 5
    ) -> None:
        super().__init__()
        self.session = requests.Session()
        self.request_host = host

        self.session.headers.update(self.default_headers)
        headers and self.session.headers.update(headers)

        self.default_timeout = timeout

    def request(self, method: str, *args, **kwargs):
        kwargs = {k: v for k, v in kwargs.items() if v is not None}
        if "timeout" in kwargs:
            kwargs["timeout"] = self.default_timeout
        return self.session.request(method, *args, **kwargs)

    def get(
        self,
        path: str,
        params: Optional[dict] = None,
        data: Optional[dict] = None,
        headers: Optional[dict] = None,
    ):
        return self.request(
            "GET", self.get_url(path), params=params, data=data, headers=headers
        )

    def get_url(self, path: str):
        if path.startswith("http"):
            return path
        return self.request_host + path

    def post(
        self,
        path: str,
        params: Optional[dict] = None,
        data: Optional[dict] = None,
        headers: Optional[dict] = None,
        fetch_header: bool = True,
    ):
        headers = headers if headers else self.session.headers
        if fetch_header:
            headers = dict(headers) | {"X-Requested-With": "Fetch"}
        return self.request(
            "POST", self.get_url(path), params=params, data=data, headers=headers
        )

src/libs/requests.py
ToolGallery/APEStoreAssistant
import logging
import argparse
import os
import sys

from common.schemas import ShopSchema, DeliverySchema, OrderDeliverySchema
from actions.inventory_monitoring import InventoryMonitor
from libs.address import get_address
from libs.notifications import (
    DingTalkNotification,
    NotificationBase,
    BarkNotification,
    FeishuNotification,
)
from libs.payments import get_payments
from libs.products import get_products


def get_notification_providers() -> list[NotificationBase]:
    providers = []
    dingtalk_token = os.environ.get("DINGTALK_TOKEN")
    bark_host = os.environ.get("BARK_HOST")
    bark_token = os.environ.get("BARK_TOKEN")
    feishu_token = os.environ.get("FEISHU_TOKEN")

    if dingtalk_token:
        providers.append(DingTalkNotification(dingtalk_token))
    if bark_token:
        providers.append(BarkNotification(bark_token, host=bark_host))
    if feishu_token:
        providers.append(FeishuNotification(feishu_token))
    return providers


def get_delivery_data() -> DeliverySchema:
    data = OrderDeliverySchema(
        first_name=os.environ.get("DELIVERY_FIRST_NAME"),
        last_name=os.environ.get("DELIVERY_LAST_NAME"),
        email=os.environ.get("DELIVERY_EMAIL"),
        phone=os.environ.get("DELIVERY_PHONE"),
        idcard=os.environ.get("DELIVERY_IDCARD"),
        payment=os.environ.get("DELIVERY_PAYMENT"),
        payment_number=int(os.environ.get("DELIVERY_PAYMENT_NUMBER") or 0),
    )
    assert (
        data.first_name
        and data.last_name
        and data.email
        and data.phone
        and data.idcard
        and data.payment
    ), "Please check the delivery information"
    return data


def get_args():
    parser = argparse.ArgumentParser()
    parser.add_argument("-p", "--products", nargs="+", default=[], type=str, help="")
    parser.add_argument("-l", "--location", type=str, default="", help="")
    parser.add_argument("-pc", "--postal-code", type=str, default="", help="")
    parser.add_argument("--state", type=str, default="", help="")
    parser.add_argument("-lp", "--list-products", action="store_true", help="")
    parser.add_argument("-la", "--list-address", action="store_true", help="")
    parser.add_argument("-lpa", "--list-payments", action="store_true", help="")
    parser.add_argument("-o", "--order", action="store_true", help="")
    parser.add_argument("-onc", "--order-notice-count", type=int, default=1, help="")
    parser.add_argument(
        "-c", "--country", type=str, required=True, help="cn|hk-zh|sg|jp"
    )
    parser.add_argument("--code", type=str, default="", help="15|15-pro")
    parser.add_argument("-i", "--interval", type=int, default=5, help="Query interval")
    parser.add_argument("-ft", "--filter", type=str, default="", help="")
    parser.add_argument(
        "-sft", "--store-filter", nargs="+", type=str, default=[], help=""
    )
    return parser.parse_args()


def main():
    args = get_args()

    if args.list_products:
        assert args.country and args.code, "Lack of key information"
        products = get_products(args.code, args.country)
        for product in products:
            logging.info(product.intro())
        sys.exit(0)
    if args.list_address:
        assert args.country, "Lack of key information"
        addresses = get_address(args.country, args.filter)
        for address in addresses:
            logging.info(address)
        sys.exit(0)
    if args.list_payments:
        assert args.country, "Lack of key information"
        payments = get_payments(args.country)
        for payment in payments:
            logging.info(payment.intro())
        sys.exit(0)
    delivery_data = None
    if args.order:
        delivery_data = get_delivery_data()
        assert args.code, "Lack of key information"

    shop_data = ShopSchema(
        args.country,
        models=args.products,
        location=args.location,
        postal_code=args.postal_code,
        state=args.state,
        code=args.code,
        store_filters=args.store_filter
    )
    InventoryMonitor().start(
        shop_data,
        order=args.order,
        delivery_data=delivery_data,
        notification_providers=get_notification_providers(),
        interval=args.interval,
        order_notice_count=args.order_notice_count,
    )


if __name__ == "__main__":
    main()

src/main.py
ToolGallery/APEStoreAssistant