2
\$\begingroup\$

After I have been doing some improvements from my Previous code review. I have taken the knowledge to upgrade and be a better coder but now im here again asking for Code review where I think it could be better.

The purpose of this code is a monitoring that checks for a special site every random 30 to 120 seconds. If there has been a changes then it goes through some if statements as you can see and it will then print to my discord if there has been a changed made.

This is what I have created:

monitoring.py

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

import concurrent.futures
import random
import time
from datetime import datetime, timedelta
from typing import Any, Dict, List

import pendulum
from loguru import logger

from scrape_values import Product

store: str = "shelta"
link: str = "https://shelta.se/sneakers/nike-air-zoom-type-whiteblack-cj2033-103"

# -------------------------------------------------------------------------
# Utils
# -------------------------------------------------------------------------
_size_filter: Dict[str, datetime] = {}


def monitor_stock():
    """
    Function that checks if there has happen a restock or countdown change on the website
    """
    payload = Product.from_page(url=link).payload

    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        while True:

            # Request for new product information
            new_payload = Product.from_page(url=link).payload

            # Release sleep
            release_date_sleeper(new_payload)

            # Check countdown timer comparision
            if countdown_timer_comparision(payload, new_payload):
                # Send notification to discord
                executor.submit(send_notification, new_payload, "Timer change!")
                # Replace list
                payload["displayCountDownTimer"] = new_payload["displayCountDownTimer"]

            # Check sizes comparision
            if sizes_comparision(payload, new_payload):
                # Send notification to discord
                executor.submit(send_notification, new_payload, "Restock!")
                # Replace list
                payload["sizes"] = new_payload["sizes"]

            else:
                # No changes happen
                logger.info("No changes made")

                payload["sizes"] = new_payload["sizes"]
                time.sleep(random.randint(30, 120))


def release_date_sleeper(payload) -> None:
    """
    Check if there is a release date on the website. We should sleep if there is to save resources
    :param payload:
    """
    if payload.get('releaseDate'):
        delta_seconds = (payload["releaseDate"].subtract(seconds=10)) - pendulum.now()
        if not delta_seconds.seconds:
            logger.info(f'Release date enabled | Will sleep to -> {(payload["releaseDate"].subtract(seconds=10)).to_datetime_string()}')
            time.sleep(delta_seconds.seconds)


def countdown_timer_comparision(payload, new_payload) -> bool:
    """
    Compare the first requests with the latest request and see if the countdown timer has been changed on the website
    :param payload: First request made
    :param new_payload: Latest request made
    :return: bool
    """
    if new_payload.get("displayCountDownTimer") and payload["displayCountDownTimer"] != new_payload[
        "displayCountDownTimer"]:
        logger.info(f'Detected new timer change -> Name: {new_payload["name"]} | Display Time: {new_payload["displayCountDownTimer"]}')
        return True


def sizes_comparision(payload, new_payload) -> bool:
    """
    Compare the first requests with the latest request and see if the sizes has been changed on the website
    :param payload: First request made
    :param new_payload: Latest request made
    :return: bool
    """
    if payload["sizes"] != new_payload["sizes"]:
        if spam_filter(new_payload["delay"], new_payload["sizes"]):
            logger.info(f'Detected restock -> Name: {new_payload["name"]} | Sizes: {new_payload["sizes"]}')
            return True


def send_notification(payload, status) -> Any:
    """
    Send to discord
    :param payload: Payload of the product
    :param status: Type of status that being sent to discord
    """
    payload["status"] = status
    payload["keyword"] = True
    # FIXME: call create_embed(payload) for post to discord
    # See more here https://codereview.stackexchange.com/questions/260043/creating-embed-for-discord-reading-from-dictionary


def spam_filter(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 _size_filter
                or datetime.now() - _size_filter[r] >= timedelta(seconds=delay)
        )
    ]
    # Refresh timestamps for requests we're actually making.
    for r in filtered:
        _size_filter[r] = datetime.now()

    return filtered


if __name__ == "__main__":
    monitor_stock()

scrape_values.py

import json
import re
from dataclasses import dataclass
from typing import List, Optional

import requests
from bs4 import BeautifulSoup


@dataclass
class Product:
    name: Optional[str] = None
    price: Optional[str] = None
    image: Optional[str] = None
    sizes: List[str] = None

    @staticmethod
    def get_sizes(doc: BeautifulSoup) -> List[str]:
        pat = re.compile(
            r'^<script>var JetshopData='
            r'(\{.*\})'
            r';</script>$',
        )
        for script in doc.find_all('script'):
            match = pat.match(str(script))
            if match is not None:
                break
        else:
            return []

        data = json.loads(match[1])
        return [
            variation
            for get_value in data['ProductInfo']['Attributes']['Variations']
            if get_value.get('IsBuyable')
            for variation in get_value['Variation']
        ]

    @classmethod
    def from_page(cls, url: str) -> Optional['Product']:
        with requests.get(url) as response:
            if not response.ok:
                return None
            doc = BeautifulSoup(response.text, 'html.parser')

        name = doc.select_one('h1.product-page-header')
        price = doc.select_one('span.price')
        image = doc.select_one('meta[property="og:image"]')

        return cls(
            name=name and name.text.strip(),
            price=price and price.text.strip(),
            image=image and image['content'],
            sizes=cls.get_sizes(doc),
        )

    @property
    def payload(self) -> dict:
        return {
            "name": self.name or "Not found",
            "price": self.price or "Not found",
            "image": self.image or "Not found",
            "sizes": self.sizes,
        }

My concern is that I might have done it incorrectly where I have split it into multiple functions that maybe is not necessary to do? Im not sure and I do hope I will get some cool feedbacks! Looking forward

\$\endgroup\$
4
  • \$\begingroup\$ This code will not run. You have imbalanced parens. \$\endgroup\$
    – Reinderien
    Apr 22 at 17:19
  • \$\begingroup\$ @Reinderien Sorry, was a typo miss in sleep :) Fixed now \$\endgroup\$ Apr 22 at 17:22
  • \$\begingroup\$ Why is the monitor_stock running at random intervals? \$\endgroup\$
    – Mast
    Apr 29 at 16:44
  • \$\begingroup\$ @Mast Im not sure why I used random. Thought it would be cooler :) That could be changed to 120 sec or even more :) the idea is to check again and compare with previous requests if there has been a change. If there has been then it checks for the if statements inside the monitor_stock \$\endgroup\$ Apr 29 at 16:46
1
+50
\$\begingroup\$

I believe you are still facing some issues with Real Classes comment from the previous post. Why do I say this? Your Product class is not even necessary. It has only functions that can be without a class.

  1. Using Named Tuple instead of data class.

    # Basic example
    >>> Point = namedtuple('Point', ['x', 'y'])
    >>> p = Point(11, y=22)     # instantiate with positional or keyword arguments
    >>> p[0] + p[1]             # indexable like the plain tuple (11, 22)
    33
    >>> x, y = p                # unpack like a regular tuple
    >>> x, y
    (11, 22)
    >>> p.x + p.y               # fields also accessible by name
    33
    >>> p                       # readable __repr__ with a name=value style
    Point(x=11, y=22)
    

    Return named tuple from payload and from_page

  2. Creating variable delays.

    Instead of:

    def monitor_stock():
        ...
        time.sleep(random.randint(30, 120))
    

    Use something like:

    # Global Scope
    delays = [30, 60, 90, 120]
    curr_delay = 0
    
    def get_delay():
        res = curr_delay
        curr_delay += 1
        if curr_delay >= len(delays):
            curr_delay = 0
        return res
    
    def monitor_stock():
        ...
        time.sleep(delays[get_delay()])
    
\$\endgroup\$
3
  • \$\begingroup\$ Hi! Im kinda curious regarding the named tuples, what is the reason why I dont need the class if I might ask? Is it more like im doing too much sort of? - Regarding the variable delays, that is pretty neat! Very cool! \$\endgroup\$ May 2 at 17:39
  • \$\begingroup\$ @ProtractorNewbie, you are not doing anything with the class. If you want to have data class, use named tuples. You are not using 'O' (object) in OOPs, i.e. there are no objects or any need in your case. \$\endgroup\$
    – mangupt
    May 3 at 3:54
  • \$\begingroup\$ Named tuples are great! That said, it's my opinion that typing.NamedTuple is better than collections.namedtuple, and in practice dataclasses.dataclass is even better for nearly all applications. Dataclasses take a little more reading to understand what all the pieces do, but they're worth learning and @dataclass(frozen=True) is basically a drop-in replacement for NamedTuple. \$\endgroup\$ May 6 at 12:54

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.