Как сравнить dict с метками времени

Question

Так что я пытаюсь создать словарь, в котором будет каждый продукт , который вы отслеживаете, когда что-то пополнится, он проверит этот словарь и посмотрит, есть ли там элемент, если нет, то добавит его , После того, как он будет добавлен, будет добавлен каждый размер пополнения и отметка времени для каждого размера. При следующем запуске, если что-то пополняется, он принимает эти размеры и проверяет, есть ли они в словаре, если они есть, он сравнивает текущую метку времени с теми, которые находятся внутри словаря. По сути, установка перезарядки для размера.

То, что я сделал до сих пор, является демонстрацией этого:

import random
import sys
import time
import re
from datetime import timedelta, datetime


def randomNumber():
    payload = {
        "sizes": {}
    }
    numbers = {}
    for x in range(10):
        numbers.update({f'US {random.randint(1, 10)}': datetime.now()})

    payload['sizes'] = numbers

    return payload


def demo():
    firstList = randomNumber()

    while True:
        secondList = randomNumber()

        if len(secondList['sizes']) > len(firstList['sizes']):  # Not sure if this is needed but couldn't get another other ideas here to compare

            restockSizes = []
            restocked = False
            for key, value in secondList['sizes'].items():
                if key in firstList['sizes'].keys():
                    timenow = datetime.now()
                    elapsed = timenow - value

                    if elapsed > timedelta(seconds=5):
                        restockSizes.append(f'{key}')
                        restocked = True
                    else:
                        restockSizes.append(f'{key}')
                        time.sleep(random.randint(1, 3))

            if restocked:
                print(f'Restocked: {restockSizes}')
                firstList = secondList

        else:
            print("No restock.")
            firstList = secondList

demo()

Есть некоторая обеспокоенность, что я не уверен, что она действительно работает, как я описал , На данный момент есть кое-что, что, как мне кажется, мне непонятно: есть ли у меня какая-либо причина для firstList = secondList в else statement

Также я не уверен, если len(secondList['sizes']) > len(firstList['sizes']): также нужен в моем случае, если мне нужно сравнить между firstList и secondList?

Надеюсь, кто-нибудь увидит fl aws и поможет мне!

Допустим, наша дельта минутная = 5

Первый запрос:

США 3: 22:23, США 4: 22:23, США 5 22: 23

Второй запрос (1 мин после первого запроса):

США 3: 22:23, США 4: 22:23, США 5 22:23, США 6: 22:24, США 8: 22 : 24

Третий запрос (5 ​​минут после первого запроса):

США 3: 22:28, США 4: 22:28, США 5 22: 28 , США 6: 22:24, США 8: 22: 24

Четвертый запрос (6 минут после первого запроса):

США 3: 22:28, США 4: 22:28 , США 5 22:28, США 6: 22:29, США 8: 22: 29

Answer 1

Я думаю, что это демонстрирует общее представление о том, что вы ищете:

from datetime import datetime, timedelta
import random
from typing import Dict, List


class RequestFilter:
    """Tracks requests and filters them to prevent hammering."""

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

    def filter(self, requests: List[str], time: datetime) -> List[str]:
        """Filter requests to only those that haven't been made
        previously within our defined cooldown period."""
        # Get filtered set of requests.
        filtered = [
            r for r in list(set(requests))
            if (
                r not in self._requests
                or time - self._requests[r] >= self._cooldown
            )
        ]
        # Refresh timestamps for requests we're actually making.
        for r in filtered:
            self._requests[r] = time
        return filtered


if __name__ == '__main__':
    time = datetime.now()
    request_filter = RequestFilter(timedelta(minutes=5))
    for _ in range(100):
        requests = [
            f"US {random.randint(1, 10)}"
            for _ in range(5)
        ]
        print(request_filter.filter(requests, time), time)
        time += timedelta(minutes=1)

Обратите внимание, что я заставил функцию фильтрации брать временную метку вместо того, чтобы фактически использовать datetime.now() для внутреннего использования - таким образом, вы может выполнить его модульное тестирование с использованием поддельных временных меток вместо необходимости исправлять datetime, или фактически заставить его спать в течение минуты между поддельными запросами. :)

В реальной жизни вы просто скажете request_filter.filter(requests, datetime.now()), чтобы использовать его в реальном времени. Или вы можете использовать это как вашу __main__ функцию:

if __name__ == '__main__':
    from time import sleep
    request_filter = RequestFilter(timedelta(minutes=5))
    for _ in range(100):
        requests = [
            f"US {random.randint(1, 10)}"
            for _ in range(5)
        ]
        print(request_filter.filter(requests, datetime.now()), datetime.now())
        sleep(60)