Skip to content

Instantly share code, notes, and snippets.

@tiagoad
Created December 9, 2018 17:42
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save tiagoad/0b9d520af6a8803e04df87db61e9e1b3 to your computer and use it in GitHub Desktop.
Save tiagoad/0b9d520af6a8803e04df87db61e9e1b3 to your computer and use it in GitHub Desktop.
Uses ReadDirectoryChangesW asynchronously (overlapped mode) to monitor a directory (and its' children) for changes.
import os
import time
import win32file
import win32con
import pywintypes
import winerror
WATCHED_DIR = "./mon"
FILE_ACTIONS = {
1 : "created",
2 : "deleted",
3 : "updated",
4 : "renamed from",
5 : "renamed to"
}
def main():
ol = pywintypes.OVERLAPPED()
buf = win32file.AllocateReadBuffer(1024)
# get directory handle
hDir = win32file.CreateFile (
WATCHED_DIR, # fileName
0x0001, # FILE_LIST_DIRECTORY # desiredAccess
win32con.FILE_SHARE_READ | # shareMode
win32con.FILE_SHARE_WRITE |
win32con.FILE_SHARE_DELETE,
None, # attributes
win32con.OPEN_EXISTING, # CreationDisposition
win32con.FILE_FLAG_BACKUP_SEMANTICS | # flagsAndAttributes
win32con.FILE_FLAG_OVERLAPPED, # async
None # hTemplateFile
)
while True:
# read directory changes async
win32file.ReadDirectoryChangesW(
hDir, # handle
buf, # buffer
True, # bWatchSubtree
win32con.FILE_NOTIFY_CHANGE_FILE_NAME | # dwNotifyFilter
win32con.FILE_NOTIFY_CHANGE_DIR_NAME |
win32con.FILE_NOTIFY_CHANGE_ATTRIBUTES |
win32con.FILE_NOTIFY_CHANGE_SIZE |
win32con.FILE_NOTIFY_CHANGE_LAST_WRITE |
win32con.FILE_NOTIFY_CHANGE_SECURITY,
ol # overlapped
)
bytes = None
# wait for changes
while bytes == None:
try:
bytes = win32file.GetOverlappedResult(
hDir, # hFile
ol, # overlapped
False # bWait
)
except pywintypes.error as err:
if err.winerror == winerror.ERROR_IO_INCOMPLETE:
# not available yet, wait for a bit
time.sleep(0.1)
else:
# other error...
raise err
# get changes from buffer
result = win32file.FILE_NOTIFY_INFORMATION(buf, bytes)
# print results
for action, filename in result:
print(FILE_ACTIONS.get(action, "Unknown"), filename)
print("---")
if __name__ == "__main__":
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment