I want to watch 2 different directory for receiving 2 different names of PDF files and on receive of any PDF file, I want to execute some Command prompt script in the Event_Handler. I can create 2 separate files for my problem, but I want to keep watching these files using single python code. could you please guide me how and where we can put If-Else condition or other approach to handle this problem in my below code.
with 2 separate code file, I am able to archive this. I can understand this code will work only for one class because of **while loop - time.sleep(5). but not able to think alternative approach
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import subprocess
class Watcher1:
DIRECTORY_TO_WATCH1 = "C:\\Users\\BPM_admin\\Desktop\\FinalOCR\\Diageo\\SampleInvoices"
def __init__(self):
self.observer = Observer()
def run1(self):
event_handler1 = Handler1()
self.observer.schedule(event_handler1, self.DIRECTORY_TO_WATCH1, recursive=True)
self.observer.start()
try:
while True:
time.sleep(5)
except:
self.observer.stop()
print("Error")
self.observer.join()
class Handler1(FileSystemEventHandler):
@staticmethod
def on_any_event(event):
if event.is_directory:
return None
elif event.event_type == 'created':
subprocess.call([
'C:\\Users\\BPM_admin\\AppData\\Local\\UiPath\\app-19.7.0\\UiRobot.exe',
'/file',
'C:\\Users\\BPM_admin\\Desktop\\FinalOCR\\InvoiceAutomation\\PDFReadwithNative.xaml'
])
class Watcher2:
DIRECTORY_TO_WATCH2 = "C:\\Users\\BPM_admin\\Desktop\\SecondOCR\\Diageo\\SampleInvoices"
def __init__(self):
self.observer = Observer()
def run2(self):
event_handler2 = Handler2()
self.observer.schedule(event_handler2, self.DIRECTORY_TO_WATCH2, recursive=True)
self.observer.start()
try:
while True:
time.sleep(5)
except:
self.observer.stop()
print("Error")
self.observer.join()
class Handler2(FileSystemEventHandler):
@staticmethod
def on_any_event(event):
if event.is_directory:
return None
elif event.event_type == 'created':
subprocess.call([
'C:\\Users\\BPM_admin\\AppData\\Local\\UiPath\\app-19.7.0\\UiRobot.exe',
'/file',
'C:\\Users\\BPM_admin\\Desktop\\SecondOCR\\InvoiceAutomation\\PDFReadwithNative.xaml'
Like this I am call the Main method:
if __name__ == '__main__':
w1 = Watcher1()
w1.run1()
w2 = Watcher2()
w2.run2()
I would move while True from class to __main__
if __name__ == '__main__':
w1 = Watcher1()
w2 = Watcher2()
w1.start()
w2.start()
try:
while True:
time.sleep(5)
except:
w1.stop()
w2.stop()
print("Error")
w1.join()
w2.join()
And there is no need to use numbers 1 2 in separated classes for event_handler1/event_handler2 and DIRECTORY_TO_WATCH1/DIRECTORY_TO_WATCH2
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import subprocess
class Handler1(FileSystemEventHandler):
@staticmethod
def on_any_event(event):
if event.is_directory:
return None
elif event.event_type == 'created':
subprocess.call([
'C:\\Users\\BPM_admin\\AppData\\Local\\UiPath\\app-19.7.0\\UiRobot.exe',
'/file',
'C:\\Users\\BPM_admin\\Desktop\\FinalOCR\\InvoiceAutomation\\PDFReadwithNative.xaml'
])
class Handler2(FileSystemEventHandler):
@staticmethod
def on_any_event(event):
if event.is_directory:
return None
elif event.event_type == 'created':
subprocess.call([
'C:\\Users\\BPM_admin\\AppData\\Local\\UiPath\\app-19.7.0\\UiRobot.exe',
'/file',
'C:\\Users\\BPM_admin\\Desktop\\SecondOCR\\InvoiceAutomation\\PDFReadwithNative.xaml'
])
class Watcher1:
DIRECTORY_TO_WATCH = "C:\\Users\\BPM_admin\\Desktop\\FinalOCR\\Diageo\\SampleInvoices"
def __init__(self):
self.observer = Observer()
def start(self):
event_handler = Handler1()
self.observer.schedule(event_handler, self.DIRECTORY_TO_WATCH, recursive=True)
self.observer.start()
def stop(self)
self.observer.stop()
def join(self)
self.observer.join()
class Watcher2:
DIRECTORY_TO_WATCH = "C:\\Users\\BPM_admin\\Desktop\\SecondOCR\\Diageo\\SampleInvoices"
def __init__(self):
self.observer = Observer()
def start(self):
event_handler = Handler2()
self.observer.schedule(event_handler, self.DIRECTORY_TO_WATCH, recursive=True)
self.observer.start()
def stop(self)
self.observer.stop()
def join(self)
self.observer.join()
if __name__ == '__main__':
w1 = Watcher1()
w2 = Watcher2()
w1.start()
w2.start()
try:
while True:
time.sleep(5)
except:
w1.stop()
w2.stop()
print("Error")
w1.join()
w2.join()
Because classes are very similar so it could be one class with different arguments
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import subprocess
class Handler1(FileSystemEventHandler):
@staticmethod
def on_any_event(event):
if event.is_directory:
return None
elif event.event_type == 'created':
subprocess.call([
'C:\\Users\\BPM_admin\\AppData\\Local\\UiPath\\app-19.7.0\\UiRobot.exe',
'/file',
'C:\\Users\\BPM_admin\\Desktop\\FinalOCR\\InvoiceAutomation\\PDFReadwithNative.xaml'
])
class Handler2(FileSystemEventHandler):
@staticmethod
def on_any_event(event):
if event.is_directory:
return None
elif event.event_type == 'created':
subprocess.call([
'C:\\Users\\BPM_admin\\AppData\\Local\\UiPath\\app-19.7.0\\UiRobot.exe',
'/file',
'C:\\Users\\BPM_admin\\Desktop\\SecondOCR\\InvoiceAutomation\\PDFReadwithNative.xaml'
])
class Watcher:
def __init__(self, directory, handler):
self.directory = directory
self.handler = handler
self.observer = Observer()
def start(self):
self.observer.schedule(self.handler, self.directory, recursive=True)
self.observer.start()
def stop(self)
self.observer.stop()
def join(self)
self.observer.join()
if __name__ == '__main__':
w1 = Watcher("C:\\Users\\BPM_admin\\Desktop\\FinalOCR\\Diageo\\SampleInvoices", Handler1())
w2 = Watcher("C:\\Users\\BPM_admin\\Desktop\\SecondOCR\\Diageo\\SampleInvoices", Handler2())
w1.start()
w2.start()
try:
while True:
time.sleep(5)
except:
w1.stop()
w2.stop()
print("Error")
w1.join()
w2.join()
Handlers are also very similar so probably you could reduce them to one handler with different arguments
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import subprocess
class Handler(FileSystemEventHandler):
def __init__(self, path):
super().__init__()
self.path = path
def on_any_event(event):
if event.is_directory:
return None
elif event.event_type == 'created':
subprocess.call([
'C:\\Users\\BPM_admin\\AppData\\Local\\UiPath\\app-19.7.0\\UiRobot.exe',
'/file',
self.path
])
class Watcher:
def __init__(self, directory, handler):
self.directory = directory
self.handler = handler
self.observer = Observer()
def start(self):
self.observer.schedule(self.handler, self.directory, recursive=True)
self.observer.start()
def stop(self)
self.observer.stop()
def join(self)
self.observer.join()
if __name__ == '__main__':
handler1 = Handler('C:\\Users\\BPM_admin\\Desktop\\FinalOCR\\InvoiceAutomation\\PDFReadwithNative.xaml')
handler2 = Handler('C:\\Users\\BPM_admin\\Desktop\\SecondOCR\\InvoiceAutomation\\PDFReadwithNative.xaml')
w1 = Watcher("C:\\Users\\BPM_admin\\Desktop\\FinalOCR\\Diageo\\SampleInvoices", handler1)
w2 = Watcher("C:\\Users\\BPM_admin\\Desktop\\SecondOCR\\Diageo\\SampleInvoices", handler2)
w1.start()
w2.start()
try:
while True:
time.sleep(5)
except:
w1.stop()
w2.stop()
print("Error")
w1.join()
w2.join()