Check if new dict values has been added

Question

I have been working on creating a scraper that auto detects if there is any changes on a website as example as there is a release date or a increas/decrease of size value.

I have created couple of files where I have:

product_monitoring.py - The brain behind to check if there is any restock or not

#!/usr/bin/python3
# -*- coding: utf-8 -*-

import random
import time
from datetime import datetime, timedelta
from threading import Thread
from typing import Dict, List

import pendulum
from loguru import logger

import lib.notifications as discord
from lib.scraper import generate_session, load_store
from lib.utils import ExceptionCounter

simpleException = ExceptionCounter()


class ProductMonitoring:

    def __init__(self):
        self._requests: Dict[str, datetime] = {}

    def monitor_stock(self):
        """
        Monitoring product restocks

        :return:
        """

        store: str = "footlockerse"
        link: str = "https://www.footlocker.se/en/product/~/314102888204.html"

        store_class = load_store(store)

        session = generate_session()

        try:
            product_sizes = store_class.products.get_info(link, session)
        except Exception as err:
            print(err)
            simpleException.check(
                exception=Exception,
                allowed_count=0,
                msg={
                    'Title': 'Unexpected Error -> Scraping Data',
                    'Reason': str(err),
                    'Payload': str({"store": store, "link": link})
                }
            )

        while True:

            try:
                get_product_info = store_class.products.get_info(link, session)
                print(get_product_info)
            except Exception as err:
                simpleException.check(
                    exception=Exception,
                    allowed_count=0,
                    msg={
                        'Title': 'Unexpected Error -> Scraping Data',
                        'Reason': str(err),
                        'Payload': str({"store": store, "link": link})
                    }
                )

            if get_product_info.get('releaseDate') and "Footlocker" not in get_product_info["store"]:
                delta_seconds = (get_product_info["releaseDate"].subtract(seconds=10)) - pendulum.now()
                if not delta_seconds.seconds:
                    logger.info(
                        f'Release date enabled | Will sleep to -> {(get_product_info["releaseDate"].subtract(seconds=10)).to_datetime_string()}')
                    time.sleep(delta_seconds.seconds)

            if "Footlocker" in product_sizes["store"] and get_product_info["displayCountDownTimer"] and \
                    get_product_info["displayCountDownTimer"] != product_sizes["displayCountDownTimer"]:
                logger.info(
                    f'Detected new timer change -> Name: {get_product_info["name"]} | Display Time: {get_product_info["displayCountDownTimer"]}')

                get_product_info["status"] = "Timer change!"
                get_product_info["keyword"] = True

                Thread(
                    target=discord.create_embed,
                    args=(
                        get_product_info,
                    )
                ).start()

                product_sizes["displayCountDownTimer"] = get_product_info["displayCountDownTimer"]

            if get_product_info["sizes"] != product_sizes["sizes"]:
                if self.spam_filter(get_product_info["delay"], get_product_info["sizes"]):
                    logger.info(
                        f'Detected restock -> Name: {get_product_info["name"]} | Sizes: {get_product_info["sizes"]}')

                    get_product_info["status"] = "Restock!"
                    get_product_info["keyword"] = True

                    Thread(
                        target=discord.create_embed,
                        args=(
                            get_product_info,
                        )
                    ).start()

                    product_sizes["sizes"] = get_product_info["sizes"]

            else:
                logger.info(
                    f'[Product Name -> {get_product_info["name"]}]'
                    f'[Sizes In Stock -> {", ".join(get_product_info["sizes"]) if get_product_info["sizes"] else "No sizes found"}]'
                    f'[Scraped Site -> {get_product_info["link"]}]'
                )

                product_sizes["sizes"] = get_product_info["sizes"]
                time.sleep(random.randint(1, 3))

    def spam_filter(self, delay: int, requests: List[str]) -> List[str]:
        """
        Filter requests to only those that haven't been made previously within our defined cooldown period.

        :param delay: Delta seconds
        :param requests:
        :return:
        """
        # Get filtered set of requests.
        filtered = [
            r for r in list(set(requests))
            if (
                    r not in self._requests
                    or datetime.now() - self._requests[r] >= timedelta(seconds=delay)
            )
        ]
        # Refresh timestamps for requests we're actually making.
        for r in filtered:
            self._requests[r] = datetime.now()

        return filtered


if __name__ == "__main__":
    ProductMonitoring().monitor_stock()

---

lib.utils - Utils where we use diffrent utils functions that are needed or not

#!/usr/bin/python3
# -*- coding: utf-8 -*-
import re
import time
import uuid

from lib.discord_reporter import report


def normalize_input(input_string: str) -> str:
    """
    Strip string from special characters, different \n \t and more...
    """
    return re.sub(r'\s+', ' ', re.sub(r'[^\w\s]', ' ', input_string))


# -------------------------------------------------------------------------
# Exception checker
# -------------------------------------------------------------------------

class ExceptionCounter:
    """
    Counter to check if we get exceptions x times in a row.
    """

    def __init__(self):
        self.exception_count = {}

    def check(self, exception, msg, allowed_count):
        exception_name = exception.__name__

        # the dict.get() method will return the value from a dict if exists, else it will return the value provided
        self.exception_count[exception_name] = self.exception_count.get(exception_name, 0) + 1

        if self.exception_count[exception_name] >= allowed_count:
            report(msg)
            raise exception(msg)

    def reset(self):
        self.exception_count.clear()


# -------------------------------------------------------------------------
# Walk through JSON
# -------------------------------------------------------------------------

_SENTINEL = object()


class PrettyExitException(BaseException):
    pass


def walk(obj, path, default=_SENTINEL):
    try:
        for segment in path:
            obj = obj[segment]
        return obj
    except LookupError:
        if default is _SENTINEL:
            raise LookupError(f"couldn't walk path; {path}") from None
        else:
            return default


def get(obj, *path):
    return walk(obj, path, default={})

---

lib.scraper - The heart where I handle all the requests such as adding headers, cookies or whatever else it could be regarding to do a request. I'm using cloudscraper instead of requests package, as it automatically creates a header for you.

#!/usr/bin/python3
# -*- coding: utf-8 -*-
import importlib
import random
import sys
import time
from typing import Dict, Optional

import cloudscraper
import tldextract as urlparse
from loguru import logger
from proxymanager import ProxyManager
from requests.exceptions import ConnectionError, ProxyError, ReadTimeout, RequestException, Timeout

from lib.exceptions import (
    TooManyFailedRequests,
    TooManyTimedOut
)
from lib.utils import ExceptionCounter

# -------------------------------------------------------------------------
# From loguru to be able to see the colors in pm2
# -------------------------------------------------------------------------
logger.remove()
logger.add(sys.stdout, filter=lambda record: record["level"].name == "INFO")
logger.add(sys.stderr, filter=lambda record: record["level"].name != "INFO")

# -------------------------------------------------------------------------
# Proxies
# -------------------------------------------------------------------------
PROXIES_PATH: Dict[str, ProxyManager] = {
    "rotating": ProxyManager("./proxies/rotating.txt"),
}

# -------------------------------------------------------------------------
# Exception counter
# -------------------------------------------------------------------------
simpleException = ExceptionCounter()


def get_random_proxy(proxy_type: str) -> Dict:
    """
    Return a proxy dict from a given file.
    """
    try:
        return PROXIES_PATH[proxy_type.lower()].random_proxy().get_dict()
    except:  # noqa
        return {}


def generate_session() -> Dict:
    """
    Create cloudscraper session
    :return:
    """
    return cloudscraper.create_scraper(
        browser={
            "browser": "chrome",
            "mobile": False,
            "platform": "windows"
        }
    )


def reset_session(session: cloudscraper) -> Dict:
    """
    Reset cookies and headers - Usually needed after failed request
    :param session:
    :return:
    """
    # Clear the cookies in case we loop again after a exception or incorrect status code
    session.headers = cloudscraper.User_Agent(
        browser={
            "browser": "chrome",
            "mobile": False,
            "platform": "windows"
        }
    ).headers
    session.cookies.clear()
    time.sleep(random.randint(1, 3))
    return session


def setup_scraper(site_url: str, session: cloudscraper, proxies: Optional[str] = None, headers: Optional[Dict] = None,
                  cookies: Optional[Dict] = None, post_request: Optional[Dict] = False, ) -> str:
    """
    Setup for Cloudscraper including different parameters such as headers, cookies and more
    :param site_url:
    :param session:
    :param proxies:
    :param headers:
    :param cookies:
    :param post_request:
    :param cloudflare:
    :return:
    """

    # Store domain
    store_domain: str = urlparse.extract(str(site_url)).domain

    while True:

        # Add proxies if we need too - Which we should always do but not for Supreme etc
        if proxies:
            session.proxies = get_random_proxy(proxies)

        # Add custom headers if needed
        if headers:
            for i in headers.keys():
                session.headers[i] = headers[i]

        # Add custom headers if needed
        if cookies:
            for i in cookies.keys():
                session.cookies[i] = cookies[i]

        try:
            # If post_request is true -> do POST else GET request
            response = session.get(site_url, timeout=10) if not post_request else session.post(site_url,
                                                                                               json=post_request,
                                                                                               timeout=10
                                                                                               )

            if response.ok or response.status_code in [302, 400, 401, 404] or (
                    store_domain in ["shelta", "aplace"] and response.status_code == 403):
                # Reset the counter
                simpleException.reset()
                return response

            else:
                logger.debug(
                    f'[Response -> {response.status_code}]'
                    f'[Response Url -> {response.url}]'
                    f'[Proxies -> {session.proxies["https"] if session.proxies else "No proxies used"}]'
                )

                if response.status_code in {429, 403}:
                    simpleException.check(
                        exception=TooManyFailedRequests,
                        allowed_count=20,
                        msg={
                            'Title': 'Too many failed requests',
                            'Reason': f"Too many {response.status_code} response requests",
                            'URL': str(site_url),
                            'Proxies': str(
                                session.proxies["https"]
                                if session.proxies
                                else "No proxies used"
                            ),
                            'Headers': str(session.headers),
                            'Cookies': str(session.cookies),
                            'Response text': str(response.text)
                        }
                    )

                reset_session(session)
                continue

        except (ReadTimeout, Timeout, ConnectionError, ProxyError) as err:

            logger.info(f'{site_url} | {session.proxies["https"] if session.proxies else "No proxies used"} | {err}')

            if "503 Service Unavailable" not in str(err):
                simpleException.check(
                    exception=TooManyTimedOut,
                    allowed_count=30,
                    msg={
                        'Title': 'Too many timed out in a row -> Total: 30',
                        'Reason': str(err),
                        'URL': str(site_url),
                        'Proxies': str(
                            session.proxies["https"]
                            if session.proxies
                            else "No proxies used"
                        ),
                        'Headers': str(session.headers),
                        'Cookies': str(session.cookies)
                    }
                )

            reset_session(session)
            continue

        except RequestException as err:
            logger.info(
                f'{site_url} | Response {response.status_code} | {session.proxies["https"] if session.proxies else "No proxies used"} | {err}')
            reset_session(session)
            continue

        except Exception as unexpected:
            logger.debug(
                f'Unexpected | {site_url} | Proxies -> {session.proxies["https"] if session.proxies else "No proxies used"} | {unexpected}')

            simpleException.check(
                exception=Exception,
                allowed_count=2,
                msg={
                    'Title': 'Unexpected error',
                    'Reason': str(unexpected),
                    'URL': str(site_url),
                    'Proxies': str(
                        session.proxies["https"]
                        if session.proxies
                        else "No proxies used"
                    ),
                    'Headers': str(session.headers),
                    'Cookies': str(session.cookies)
                }
            )

            reset_session(session)
            continue


def load_store(store: str) -> importlib:
    mod = importlib.import_module(f"lib.vendors.{store}")
    class_pointer = getattr(mod, store)()
    return class_pointer

---

lib.exceptions - My own cute exceptions :D

# -*- coding: utf-8 -*-

# ------------------------------------------------------------------------------- #

"""
exceptions
~~~~~~~~~~~~~~~~~~~
This module contains the set of exceptions.
"""


# ------------------------------------------------------------------------------- #


class RequestsException(Exception):
    """
    Base exception class for Requests
    """


class TooManyFailedRequests(RequestsException):
    """
    Raise an exception for FailedRequests
    """


class TooManyTimedOut(RequestsException):
    """
    Raise an exception for TimedOut
    """


class IncorrectProductID(RequestsException):
    """
    Raise an exception for incorrect product id for MESH
    """


# ------------------------------------------------------------------------------- #

class DiscordException(Exception):
    """
    Base exception class for Discord
    """


class IncorrectPayload(DiscordException):
    """
    Raise an exception for incorrect payload to discord embed
    """


class UnexpectedDiscord(DiscordException):
    """
    Raise an exception for unexpected discord issues
    """

---

lib.vendors.footlockerse.init

#!/usr/bin/python3
# -*- coding: utf-8 -*-
from lib.scraper import setup_scraper
from lib.utils import normalize_input
from .products import Products


class footlockerse:

    def __init__(self):
        self.display_name = "footlockerse".lower()
        self.base_url = "https://www.footlocker.se"

        self.site_catalog = [
            f"{self.base_url}/api/products/search?query=%3Arelevance%3AisNewarrival%3ANEW%26sort%3DnewArrivals%3Aproducttype%3AShoes&currentPage=0&pageSize=10&timestamp=0&sort=newArrivals",
            f"{self.base_url}/apigate/release-calendar?timestamp=2&sort=newArrivals"
        ]

        self.scraper = setup_scraper

        self.clean_string = normalize_input

        self.products = Products(self)

---

lib.vendors.footlockerse.products - Where I scrape the data on given product given

#!/usr/bin/python3
# -*- coding: utf-8 -*-

import json
import random
import re
import string
import time
from typing import Dict, Union
from urllib.parse import urlparse

import cloudscraper
import pendulum
from simplejson import JSONDecodeError

from config import configuration
from lib.utils import get


class Products:

    def __init__(self, parent):
        self.parent = parent
        self.letters = string.ascii_uppercase

    # ------------------------------------------------------------------------
    # Scrape all information about one product (Name, Price, Image, Sizes etc)
    # ------------------------------------------------------------------------
    def get_info(self, site_url: str, session: cloudscraper) -> Dict[str, Union[str, int, list, bool]]:

        payload = {
            "store": "Footlocker SE",
            "name": self.parent.clean_string(urlparse(site_url).path.split("/", 3)[-1].replace(".html", "")).title(),
            "price": None,
            "sku": re.search('\/(\d+)', site_url).group(1),
            "image": None,
            "link": site_url,
            "displayCountDownTimer": False,
            "sizes": [],
            "shortcut": [
                f'[Cart]({self.parent.base_url}/en/cart)'
            ],
            "webhook": "swedish",
            "delay": 3600
        }

        headers = {
            'Authority': 'www.footlocker.se',
            'Accept': 'application/json',
            'X-Api-Lang': 'en-GB',
            'Accept-Language': 'en-GB,en;q=0.9',
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/83.0.4103.116 Safari/537.36',
            'Referer': site_url,
        }

        with self.parent.scraper(
                site_url=f"{self.parent.base_url}/api/products/pdp/{payload['sku']}?timestamp={int(time.time())}?q={''.join(random.choice(self.letters) for i in range(10))}",
                session=session, headers=headers, proxies="rotating") as response:

            if response.status_code == 400:
                # Check if its in release date
                with self.parent.scraper(
                        site_url=f"{self.parent.base_url}/apigate/release-calendar?timestamp=2&sort=newArrivalsf&q={time.time()}",
                        session=session, headers=headers, proxies="rotating") as response:
                    if not response.ok:
                        return payload

                    try:
                        doc = json.loads(response.text)
                        release_calendar = True
                    except JSONDecodeError:
                        return payload

            elif not response.ok:
                return payload

            else:
                # Check for restocks
                try:
                    doc = json.loads(response.text)
                    release_calendar = False
                except JSONDecodeError:
                    return payload

        if release_calendar:
            self.release_calendar(site_url=site_url, doc=doc, payload=payload)

        else:
            self.product_api(doc=doc, payload=payload)

        return payload

    # ------------------------------------------------------------------------
    # Scrape info for release_calendar
    # ------------------------------------------------------------------------

    def release_calendar(self, site_url, doc, payload):
        for product_link in get(doc, "releaseCalendarProducts"):
            if site_url.split(self.parent.base_url)[1] == get(product_link, "pdpLink"):

                product_name = get(product_link, "name")
                product_brand = get(product_link, "brandName")
                product_style = get(product_link, "style")

                product_has_image = get(product_link, "hasImage")
                product_id = get(product_link, "id")

                product_display_countdown_timer = get(product_link, "displayCountDownTimer")
                product_release_date = get(product_link, "skuLaunchDate")

                product_has_stock = get(product_link, "hasStock")
                product_has_manufacturerSku = get(product_link, "manufacturerSku")

                if product_brand and product_name and product_style:
                    payload["name"] = f"{product_brand} {product_name} {product_style}"

                if product_has_image and product_id:
                    payload["image"] = f"https://images.footlocker.com/is/image/FLEU/{product_id}_01?wid=640&hei=640&fmt=png-alpha"

                if product_display_countdown_timer:
                    payload["displayCountDownTimer"] = product_display_countdown_timer

                if product_release_date:
                    date = pendulum.parse(product_release_date, strict=False)
                    payload["releaseDate"] = date.astimezone(pendulum.tz.timezone("Europe/Stockholm"))

                if product_has_stock:
                    payload["stockLoaded"] = product_has_stock

                if product_has_manufacturerSku:
                    payload["styleCode"] = product_has_manufacturerSku
                    payload["shortcut"].append(f"[StockX](https://stockx.com/search?s={product_has_manufacturerSku})")

                # Stop looking after it in loop if we already found it
                break

        return payload

    # ------------------------------------------------------------------------
    # Scrape info for product_api
    # ------------------------------------------------------------------------

    def product_api(self, doc, payload):

        product_name = get(doc, "name")
        product_variant = get(doc, "variantAttributes", 0)
        product_id = payload["sku"]
        product_size = get(doc, "sellableUnits")

        if product_name:
            payload['name'] = self.parent.clean_string(product_name)

        if product_id:
            payload[
                'image'] = f"https://images.footlocker.com/is/image/FLEU/{product_id}_01?wid=640&hei=640&fmt=png-alpha"

        if product_size:
            payload["sizes"] = [get(get_sizes, "attributes", 0, "value") for get_sizes in product_size if
                                get(get_sizes, "sku")[:-3] == product_id]

        if product_variant:

            product_price = get(product_variant, "price", "value")
            product_currency = get(product_variant, "price", "currencyIso")
            product_release_date = get(product_variant, "skuLaunchDate")
            product_display_countdown_timer = get(product_variant, "displayCountDownTimer")
            product_has_stock = get(product_variant, "stockLevelStatus")

            if product_price and product_currency:
                payload["price"] = f"{product_currency} {product_price}"

            if product_display_countdown_timer:
                payload["displayCountDownTimer"] = product_display_countdown_timer

            if product_release_date:
                date = pendulum.parse(product_release_date, strict=False)
                payload["releaseDate"] = date.astimezone(pendulum.tz.timezone("Europe/Stockholm"))

            if product_has_stock:
                payload["stockLoaded"] = product_has_stock

        return payload

I have added small description of each block of script where you can see at each start of the code.

The point is that I want as much as possible to be dynamic as possible where I in the near feature will add more sites in vendors folder.

It would be nice to get a full review. My biggest concerns are product_monitoring.py as well as lib.scraper. The concern is that I do believe there is lots of code that could be removed as well as re-written I believe and more better (?) performance way as well.

Answer 1

Exceptions

Your exception handling is... a little convoluted. The exception handlers in monitor_stock that all pass allowed_count=0 should not have to go through your ExceptionCounter machinery at all. Instead, re-raise with contextual data in a custom exception type:

try:
    ...
except Exception as e:
    raise ScrapeException('Unexpected error scraping data', store=store, link=link) from e

Don't call Discord from there; don't do exception counting there; don't form a discord message payload dictionary there - just raise. In an upper method, you can except and do the Discord notification.

ExceptionCounter should not know about Discord at all, and should be strictly constrained to exception count limits and re-raising.

ExceptionCounter.exception_count should be

self.exception_counts: DefaultDict[str, int] = defaultdict(int)

(note the plural since this is not a single count.)

then that way,

    self.exception_count[exception_name] = self.exception_count.get(exception_name, 0) + 1

can be

    self.exception_counts[exception_name] += 1

You should not be discarding err here:

    except Exception as err:
        simpleException.check(
            exception=Exception,

Instead,

    except Exception as err:
        simpleException.check(
            exception=err,

then check would be

def check(self, exception: Exception, allowed_count: int):
    exception_name = type(exception).__name__

    self.exception_count[exception_name] += 1
    if self.exception_count[exception_name] >= allowed_count:
        raise

However, it might make more sense to rethink this class to have a member dictionary of exception types (not exception type strings) to maximum counts, and not pass in allowed_count to check at all.

You ask:

I'm not sure how you want me to have a allowed counter without passing it to the check at all?

Do something like:

class ExceptionCounter:
    def __init__(self, limits: Dict[Type[Exception], int]):
        self.limits = limits
        self.exception_counts: DefaultDict[Type[Exception], int] = defaultdict(int)

Pass the limits in as a dictionary per exception type.

normalize_input

This is probably inefficient. Rather than doing a two-pass replace, just do it in one pass:

_norm_replace = re.compile(r'[\W\s]+').sub

def normalize_input(input_string: str) -> str:
    return _norm_replace(' ', input_string)

Real classes

ProductMonitoring currently doesn't deserve to be a class and is basically just a monitor_stock function. To improve this, move store and link (and probably store_class) to members on the class accepted by the constructor; and split up monitor_stock into multiple methods.

You ask:

Where would you split the code in here?

One easy first cut is to move everything in the loop body into a separate method. You can make other methods for change detection, for instance.

Untracked threads

This:

            Thread(
                target=discord.create_embed,
                args=(
                    get_product_info,
                )
            ).start()

is a little bit dangerous. You're creating and starting a thread, and throwing away its reference; so you'll never have an opportunity to join. A more careful approach would have an upper limit to the number of live threads as imposed on a thread pool; and would run monitor_stock in a context manager where the __exit__ would join on all remaining threads.

If you don't have an upper limit on the thread pool and you get more product data than expected, you've basically made a fork bomb.

You ask:

should I read about threading regarding __exit__ and __start__ ?

Object documentation

Weakly-typed dictionaries

Products.get_info needs to be re-thought. Stop using dictionaries for internal data representation; classes are a much better fit for this purpose.

You ask:

I do not know how to pass the payload if we etc. return 404. As you can see in the lib.vendors.footlockerse.products I'm passing the payload to have at least the name from a URL so I can notify myself that there has been some changes in the website so it does let me know which page it is at least as name. But how would you do it in that case?

Don't return a Dict[str, Union[str, int, list, bool]] from get_info. Don't represent payload as a dictionary; represent it as a class instance.

Don't do this:

        elif not response.ok:
            return payload

Instead raise an exception. Why pass a "payload" if there's no useful information in the payload?

There's (much) more but this will have to do for now.