def setUp(self):
self.watcher_threads = []
self.watcher_thread_changes = []
self.dir_names = []
self.dir_handles = []
for i in range(self.num_test_dirs):
td = tempfile.mktemp("-test-directory-changes-%d" % i)
os.mkdir(td)
self.dir_names.append(td)
hdir = win32file.CreateFile(td,
ntsecuritycon.FILE_LIST_DIRECTORY,
win32con.FILE_SHARE_READ,
None, # security desc
win32con.OPEN_EXISTING,
win32con.FILE_FLAG_BACKUP_SEMANTICS |
win32con.FILE_FLAG_OVERLAPPED,
None)
self.dir_handles.append(hdir)
changes = []
t = threading.Thread(target=self._watcherThreadOverlapped,
args=(td, hdir, changes))
t.start()
self.watcher_threads.append(t)
self.watcher_thread_changes.append(changes)
python类FILE_FLAG_BACKUP_SEMANTICS的实例源码
def setUp(self):
self.watcher_threads = []
self.watcher_thread_changes = []
self.dir_names = []
self.dir_handles = []
for i in range(self.num_test_dirs):
td = tempfile.mktemp("-test-directory-changes-%d" % i)
os.mkdir(td)
self.dir_names.append(td)
hdir = win32file.CreateFile(td,
ntsecuritycon.FILE_LIST_DIRECTORY,
win32con.FILE_SHARE_READ,
None, # security desc
win32con.OPEN_EXISTING,
win32con.FILE_FLAG_BACKUP_SEMANTICS |
win32con.FILE_FLAG_OVERLAPPED,
None)
self.dir_handles.append(hdir)
changes = []
t = threading.Thread(target=self._watcherThreadOverlapped,
args=(td, hdir, changes))
t.start()
self.watcher_threads.append(t)
self.watcher_thread_changes.append(changes)
def watchos():
#get path or maintain current path of app
FILE_LIST_DIRECTORY = 0x0001
try: path_to_watch = myos.get() or "."
except: path_to_watch = "."
path_to_watch = os.path.abspath(path_to_watch)
textbox.insert(END, "Watching %s at %s" % (path_to_watch, time.asctime()) + "\n\n")
# FindFirstChangeNotification sets up a handle for watching
# file changes.
while 1:
hDir = win32file.CreateFile (
path_to_watch,
FILE_LIST_DIRECTORY,
win32con.FILE_SHARE_READ | win32con.FILE_SHARE_WRITE,
None,
win32con.OPEN_EXISTING,
win32con.FILE_FLAG_BACKUP_SEMANTICS,
None
)
change_handle = win32file.ReadDirectoryChangesW (
hDir,
1024,
True,#Heap Size include_subdirectories,
win32con.FILE_NOTIFY_CHANGE_FILE_NAME |
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,
None,
None
)
# Loop forever, listing any file changes. The WaitFor... will
# time out every half a second allowing for keyboard interrupts
# to terminate the loop.
ACTIONS = {
1 : "Created",
2 : "Deleted",
3 : "Updated",
4 : "Renamed from something",
5 : "Renamed to something"
}
results = change_handle
for action, files in results:
full_filename = os.path.join(path_to_watch, files)
theact = ACTIONS.get(action, "Unknown")
textbox.insert(END, str(full_filename) + "\t" + str(theact) +"\n")
def TestDeviceNotifications(dir_names):
wc = win32gui.WNDCLASS()
wc.lpszClassName = 'test_devicenotify'
wc.style = win32con.CS_GLOBALCLASS|win32con.CS_VREDRAW | win32con.CS_HREDRAW
wc.hbrBackground = win32con.COLOR_WINDOW+1
wc.lpfnWndProc={win32con.WM_DEVICECHANGE:OnDeviceChange}
class_atom=win32gui.RegisterClass(wc)
hwnd = win32gui.CreateWindow(wc.lpszClassName,
'Testing some devices',
# no need for it to be visible.
win32con.WS_CAPTION,
100,100,900,900, 0, 0, 0, None)
hdevs = []
# Watch for all USB device notifications
filter = win32gui_struct.PackDEV_BROADCAST_DEVICEINTERFACE(
GUID_DEVINTERFACE_USB_DEVICE)
hdev = win32gui.RegisterDeviceNotification(hwnd, filter,
win32con.DEVICE_NOTIFY_WINDOW_HANDLE)
hdevs.append(hdev)
# and create handles for all specified directories
for d in dir_names:
hdir = win32file.CreateFile(d,
winnt.FILE_LIST_DIRECTORY,
winnt.FILE_SHARE_READ | winnt.FILE_SHARE_WRITE | winnt.FILE_SHARE_DELETE,
None, # security attributes
win32con.OPEN_EXISTING,
win32con.FILE_FLAG_BACKUP_SEMANTICS | # required privileges: SE_BACKUP_NAME and SE_RESTORE_NAME.
win32con.FILE_FLAG_OVERLAPPED,
None)
filter = win32gui_struct.PackDEV_BROADCAST_HANDLE(hdir)
hdev = win32gui.RegisterDeviceNotification(hwnd, filter,
win32con.DEVICE_NOTIFY_WINDOW_HANDLE)
hdevs.append(hdev)
# now start a message pump and wait for messages to be delivered.
print("Watching", len(hdevs), "handles - press Ctrl+C to terminate, or")
print("add and remove some USB devices...")
if not dir_names:
print("(Note you can also pass paths to watch on the command-line - eg,")
print("pass the root of an inserted USB stick to see events specific to")
print("that volume)")
while 1:
win32gui.PumpWaitingMessages()
time.sleep(0.01)
win32gui.DestroyWindow(hwnd)
win32gui.UnregisterClass(wc.lpszClassName, None)
def initialize(self):
self.hDir = win32file.CreateFile(
self.path_to_watch,
self.FILE_LIST_DIRECTORY,
win32con.FILE_SHARE_READ | win32con.FILE_SHARE_WRITE | win32con.FILE_SHARE_DELETE,
None,
win32con.OPEN_EXISTING,
win32con.FILE_FLAG_BACKUP_SEMANTICS,
None)
def TestDeviceNotifications(dir_names):
wc = win32gui.WNDCLASS()
wc.lpszClassName = 'test_devicenotify'
wc.style = win32con.CS_GLOBALCLASS|win32con.CS_VREDRAW | win32con.CS_HREDRAW
wc.hbrBackground = win32con.COLOR_WINDOW+1
wc.lpfnWndProc={win32con.WM_DEVICECHANGE:OnDeviceChange}
class_atom=win32gui.RegisterClass(wc)
hwnd = win32gui.CreateWindow(wc.lpszClassName,
'Testing some devices',
# no need for it to be visible.
win32con.WS_CAPTION,
100,100,900,900, 0, 0, 0, None)
hdevs = []
# Watch for all USB device notifications
filter = win32gui_struct.PackDEV_BROADCAST_DEVICEINTERFACE(
GUID_DEVINTERFACE_USB_DEVICE)
hdev = win32gui.RegisterDeviceNotification(hwnd, filter,
win32con.DEVICE_NOTIFY_WINDOW_HANDLE)
hdevs.append(hdev)
# and create handles for all specified directories
for d in dir_names:
hdir = win32file.CreateFile(d,
winnt.FILE_LIST_DIRECTORY,
winnt.FILE_SHARE_READ | winnt.FILE_SHARE_WRITE | winnt.FILE_SHARE_DELETE,
None, # security attributes
win32con.OPEN_EXISTING,
win32con.FILE_FLAG_BACKUP_SEMANTICS | # required privileges: SE_BACKUP_NAME and SE_RESTORE_NAME.
win32con.FILE_FLAG_OVERLAPPED,
None)
filter = win32gui_struct.PackDEV_BROADCAST_HANDLE(hdir)
hdev = win32gui.RegisterDeviceNotification(hwnd, filter,
win32con.DEVICE_NOTIFY_WINDOW_HANDLE)
hdevs.append(hdev)
# now start a message pump and wait for messages to be delivered.
print "Watching", len(hdevs), "handles - press Ctrl+C to terminate, or"
print "add and remove some USB devices..."
if not dir_names:
print "(Note you can also pass paths to watch on the command-line - eg,"
print "pass the root of an inserted USB stick to see events specific to"
print "that volume)"
while 1:
win32gui.PumpWaitingMessages()
time.sleep(0.01)
win32gui.DestroyWindow(hwnd)
win32gui.UnregisterClass(wc.lpszClassName, None)
