I have been working with where I use watchdogs to be able read whenever I have created a file by the name -> kill_.json and it contains:
*kill_*.json & *start_*.json contains:
{"store": "twitch", "link": "https://www.twitch.tv/xqcow", "id": "2710"}
and currently what my script does is to read the json and the filename. if it contains any of the if elif statement then it will start or kill depending on the filename else we will add it to my database which I dont think its needed to be handed here since its not the code review I wish for... for now :P but if there is any question regarding it I will be answering and adding if needed of course! :)
# !/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
I wonder if there is a smarter or maybe even cleaner way to improve this code both performance and whatever it can be?