Чтение созданных файлов json внутри заданной папки

Я работал с тем, где я использую сторожевые таймеры, чтобы иметь возможность читать всякий раз, когда я создавал файл по имени -> убийство_.json и содержит:

*kill_*.json & *start_*.json contains:

{"store": "twitch", "link": "https://www.twitch.tv/xqcow", "id": "2710"}

и в настоящее время мой скрипт читает json и имя файла. если он содержит какой-либо из операторов if elif, он будет запускаться или завершаться в зависимости от имени файла, иначе мы добавим его в мою базу данных, которую я не думаю, что ее нужно передавать здесь, поскольку это не тот обзор кода, который я желаю … на данный момент: P но если есть какие-либо вопросы по этому поводу, я отвечу и добавлю, если нужно, конечно! 🙂

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

import json
import os
import os.path
import sys
import time

import watchdog.events
import watchdog.observers
from loguru import logger
from watchdog.events import PatternMatchingEventHandler

import lib.database as database


def start_script(payload):
    payload = database.get_product_data(payload["store"], payload["link"])
    logger.info(f'Starting script -> Name: {payload["store"]}_{payload["id"]} | Link: {payload["link"]}')

class DeleteAutomatic(PatternMatchingEventHandler):

    def __enter__(self):
        self.observer = watchdog.observers.Observer()
        self.observer.schedule(
            self,
            path="./flags/"
            recursive=True
        )
        self.observer.start()
        while True:
            time.sleep(1)

    def __exit__(self):
        self.observer.stop()
        self.observer.join()

    def __init__(self):
        super(DeleteAutomatic, self).__init__(
            [
                "*kill_*.json",
                "*start_*.json",
                "*new_url.json"
            ]
        )

    def on_created(self, event):
        while True:
            try:

                with open(event.src_path) as f:
                    payload = json.load(f)

                if "start" in event.src_path:
                    logger.info(f'Starting -> Store: {payload["store"]} | Link: {payload["link"]} | ID: {payload["id"]}')
                    
                elif "kill" in event.src_path:
                    logger.info(f'Deleting -> Store: {payload["store"]} | Link: {payload["link"]} | ID: {payload["id"]}')

                else:
                    for payload in database.get_all_manual_links():

                        logger.info(
                            f'New Manual Url Found -> Store: {payload["store"]} | Link: {payload["link"]} | ID: {payload["id"]}')

                        if not database.link_exists(payload["store"], payload["link"]):
                            database.register_products(payload["store"], product_info)
                            start_script(payload)

                        elif database.links_deactivated(payload["store"], payload["link"]):
                            database.update_products(payload["store"], payload["link"])
                            start_script(payload)

                        else:
                            logger.info(f'Product already monitored: {payload["store"]}_{payload["id"]} | Link: {payload["link"]}')
                            pass

                        database.delete_manual_links(payload["store"], payload["link"])

                break

            except Exception as err:
                logger.debug(f"Error -> {err}")
                time.sleep(1)
                continue

        while True:
            try:
                os.remove(event.src_path)
                break
            except Exception as err:
                logger.debug(f"Error deleting-> {err}")
                time.sleep(1)
                continue

with DeleteAutomatic():
    pass

Интересно, есть ли более умный или, может быть, даже более чистый способ улучшить этот код как производительность, так и все, что он может быть?

Индивидуальные параметры

Лучше заменить payload параметр для start_script с индивидуальным store: str, id: str и link: str. Или, если эту тройку достаточно часто носить с собой, переместите ее в @dataclass что есть start_script(self) метод.

Порядок метода

Это незначительно, но обычно __init__ должен появиться как первый метод в вашем классе.

Инициализация члена

    self.observer = watchdog.observers.Observer()
    self.observer.schedule(
        self,
        path="./flags/"
        recursive=True
    )

следует переместить в ваш __init__, поскольку у этого члена есть срок жизни экземпляра. Линтерс скажет вам, что инициализировать новых участников за пределами __init__. В start() может остаться в твоем __enter__.

Стандартные параметры

Параметры для __exit__ неверны и должны соответствовать описанию в документация.

Подвешивание метода ввода

Ваша петля сна не должна быть в вашем __enter__. После запуска ваш метод ввода должен передать управление вызывающему, а вместо этого вызывающий должен иметь цикл сна.

Супер без параметров ()

super(DeleteAutomatic, self)

это стиль Python 2; для Python 3 эти параметры можно не указывать.

Логика запуска / завершения

            if "start" in event.src_path:
                logger.info(f'Starting -> Store: {payload["store"]} | Link: {payload["link"]} | ID: {payload["id"]}')
                
            elif "kill" in event.src_path:
                logger.info(f'Deleting -> Store: {payload["store"]} | Link: {payload["link"]} | ID: {payload["id"]}')

должен измениться двумя способами: исключить общий код ведения журнала в один оператор, при этом в обоих случаях будет различаться только строка «Запуск / удаление». Кроме того, распакуйте свою полезную нагрузку либо в отдельные переменные, либо, возможно, в @dataclass.

Одно преимущество для @dataclass подход заключается в том, что этот код:

            if "start" in event.src_path:
                logger.info(f'Starting -> Store: {payload["store"]} | Link: {payload["link"]} | ID: {payload["id"]}')
                
            elif "kill" in event.src_path:
                logger.info(f'Deleting -> Store: {payload["store"]} | Link: {payload["link"]} | ID: {payload["id"]}')

            else:
                for payload in database.get_all_manual_links():

                    logger.info(
                        f'New Manual Url Found -> Store: {payload["store"]} | Link: {payload["link"]} | ID: {payload["id"]}')

                    if not database.link_exists(payload["store"], payload["link"]):
                        database.register_products(payload["store"], product_info)
                        start_script(payload)

                    elif database.links_deactivated(payload["store"], payload["link"]):
                        database.update_products(payload["store"], payload["link"])
                        start_script(payload)

                    else:
                        logger.info(f'Product already monitored: {payload["store"]}_{payload["id"]} | Link: {payload["link"]}')
                        pass

                    database.delete_manual_links(payload["store"], payload["link"])

может перейти к методу класса; и инициализация класса может быть простой Payload(**payload) kwargs.

Повторные попытки

        except Exception as err:
            logger.debug(f"Error -> {err}")
            time.sleep(1)
            continue

рискованно. Вы говорите, что если любой возникает исключение, повесьте на секунду и повторите попытку. Это включает в себя режимы сбоя, которые не следует повторять разумно, например MemoryError. Вам следует подумать о более узком наборе исключений, которые действительно подходят для повторной попытки.

Синтаксис

Ваш код просто не запускается. Этот синтаксис недействителен:

        path="./flags/"
        recursive=True

из-за пропущенной запятой. Я дам вам возможность сомневаться в том, что это ошибка копирования и вставки.

Также, product_info отсутствует полностью. Итак, ваш код неполный.

Ссылки на полезную нагрузку

Этот:

def start_script(payload):
    payload = database.get_product_data(payload["store"], payload["link"])

очень подозрительно. Вы передаете полезную нагрузку, используя ее свойства store и link для поиска в базе данных того, что мне кажется та же полезная нагрузка. Вы уверены, что хотите этим заниматься?

Аналогично это:

            with open(event.src_path) as f:
                payload = json.load(f)

            if "start" in event.src_path:
                logger.info(f'Starting -> Store: {payload["store"]} | Link: {payload["link"]} | ID: {payload["id"]}')
                
            elif "kill" in event.src_path:
                logger.info(f'Deleting -> Store: {payload["store"]} | Link: {payload["link"]} | ID: {payload["id"]}')

            else:
                for payload in database.get_all_manual_links():

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