Python 看门狗用于 2 个不同的目录并执行 2 个不同的处理程序

问题描述 投票：0回答：2

我想观看 2 个不同的目录来接收 2 个不同名称的 PDF 文件，并且在接收任何 PDF 文件时，我想在 Event_Handler 中执行一些命令提示脚本。我可以为我的问题创建 2 个单独的文件，但我想继续使用单个 python 代码监视这些文件。您能否指导我在下面的代码中如何以及在何处放置 If-Else 条件或其他方法来处理此问题。

使用 2 个单独的代码文件，我可以将其存档。 我可以理解，由于 **while 循环 - time.sleep(5)，这段代码仅适用于一个类。但无法想到替代方法

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'

像这样我调用Main方法：

 if __name__ == '__main__': 
       w1 = Watcher1() 
       w1.run1() 
       w2 = Watcher2() 
       w2.run2()
python python-3.x watchdog python-watchdog
2个回答
2
投票

我会将 

while True
从班级转到 
__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()

并且无需在单独的类中使用数字 

1
2
event_handler1
/
event_handler2
和 
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()

因为类非常相似，所以它可能是一个具有不同参数的类

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()

处理程序也非常相似，因此您可能可以将它们简化为具有不同参数的一个处理程序

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()
0
投票

我还需要使用 python watchdog 监视 2 个不同的目录。搜索了 stackoverflow，其中这篇文章似乎是最新的。在尝试了不同的解决方案后，由于其简单性，我使用以下解决方案。我想我会把它发布给任何其他遇到这个线程并需要一个简单解决方案的人。

import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler, FileSystemEvent


class MyEventHandler(FileSystemEventHandler):
    def __init__(self, callback) -> None:
        super().__init__()
        self.callback = callback

    def on_modified(self, event: FileSystemEvent):
        #  TODO add your own logic here
        self.callback()

    #  TODO add more methods here if needed.
    #  See docs for which method names to use for event handling:
    #  https://python-watchdog.readthedocs.io/en/stable/api.html#watchdog.events.FileSystemEventHandler


if __name__ == "__main__":
    #  You don't need to subclass Observer for simple use cases.
    observer = Observer()

    #  Just call observer.schedule for each separate directory you need to watch.
    #  One instance of Observer can watch several directories if you don't need more threads for performance.
    #  TODO replace callable1/callable2 with functions of your own.
    #  TODO replace "path/to/dir1" and "path/to/dir2" with paths to dirs to watch.
    observer.schedule(MyEventHandler(callable1), path="path/to/dir1", recursive=False)
    observer.schedule(MyEventHandler(callable2), path="path/to/dir2", recursive=False)
    observer.start()

    try:
        while True:
            time.sleep(300)
    except KeyboardInterrupt:
        print("\nUser abort intercepted!")
    finally:
        observer.stop()
        observer.join()
最新问题
© www.soinside.com 2019 - 2024. All rights reserved.