Detect file creation with watchdog
-
25-06-2021 - |
Domanda
I am trying to detect when a file with a given name is created in a directory. I am doing it thanks to watchdog. The creation is correctly detected but I don't know how to terminate the application properly once the detection is done.
My piece of code is the following:
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import logging
import sys
import time
from watchdog.events import FileSystemEventHandler
from watchdog.observers import Observer
logging.basicConfig(level=logging.ERROR)
class MyEventHandler(FileSystemEventHandler):
def __init__(self, observer, filename):
self.observer = observer
self.filename = filename
def on_created(self, event):
print "e=", event
if not event.is_directory and event.src_path.endswith(self.filename):
print "file created"
self.observer.unschedule_all()
self.observer.stop()
def main(argv=None):
path = argv[1]
filename = argv[2]
observer = Observer()
event_handler = MyEventHandler(observer, filename)
observer.schedule(event_handler, path, recursive=False)
observer.start()
observer.join()
return 0
if __name__ == "__main__":
sys.exit(main(sys.argv))
I am new to python and I cannot figure out what is wrong. The detection seems to be scheduled in a dedicated thread and the join() method is waiting for this thread to terminate. Thus, I suppose that I am not calling the right method on the observer to stop waiting/looping, but the watchdog documentation seems really not clear to point out what are the methods that may be used.
Does someone have an idea how I can achieve my goal?
Soluzione
Finally, after taking a look at the watchdog implementation, it is not necessary to call unschedule_all
before stop
, this is done automatically. Removing the line containing this method call fixes the issue and the application is running perfectly.
Altri suggerimenti
Also, the below script is used to observe filename
at a specific path
using the PatternMatchingEventHandler
.
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from watchdog.events import PatternMatchingEventHandler
import sys
class Watcher:
def __init__(self, path, filename):
self.observer = Observer()
self.path = path
self.filename = filename
def run(self):
event_handler = Handler(self.filename)
self.observer.schedule(event_handler, self.path, recursive=True)
self.observer.start()
try:
while True:
time.sleep(1)
except:
self.observer.stop()
print("Error")
self.observer.join()
class Handler(PatternMatchingEventHandler):
def __init__(self, filename):
super(Handler, self).__init__(
patterns=[filename],
ignore_patterns=["*.tmp"],
ignore_directories=True,
case_sensitive=False,
)
def on_any_event(self, event):
print(
"[{}] noticed: [{}] on: [{}] ".format(
time.asctime(), event.event_type, event.src_path
)
)
if __name__ == "__main__":
path = "."
filename = "test.csv"
w = Watcher(path, filename)
w.run()
output:
[Tue Feb 9 01:55:38 2021] noticed: [created] on: [/Users/mt/Documents/stackoverflow/test.csv]
[Tue Feb 9 01:55:44 2021] noticed: [modified] on: [/Users/mt/Documents/stackoverflow/test.csv]
[Tue Feb 9 01:56:01 2021] noticed: [deleted] on: [/Users/mt/Documents/stackoverflow/test.csv]
It is also possible to determine the creation of a new file without installing additional libraries.
import os
import time
def watch_file(filename, time_limit=3600, check_interval=60):
"""Return true if filename exists, if not keep checking once every check_interval seconds for time_limit seconds.
time_limit defaults to 1 hour
check_interval defaults to 1 minute
"""
now = time.time()
last_time = now + time_limit
while time.time() <= last_time:
if os.path.exists(filename):
return True
else:
# Wait for check interval seconds, then check again.
time.sleep(check_interval)
return False
if __name__ == "__main__":
filename = "test.csv"
time_limit = 60
check_interval = 1
if watch_file(filename, time_limit, check_interval):
print(f"File created: {os.path.abspath(filename)}")
else:
print(
f"File {filename} not found after waiting: {time_limit} seconds!"
)
output:
File created: /Users/mt/Documents/stackoverflow/test.csv