import collections import ctypes import io import pywintypes import struct import sys import time import win32api import win32con import win32file import winerror import winioctlcon import wmi from ctypes import wintypes from functools import reduce kernel32 = ctypes.WinDLL('kernel32') # , use_last_error=True) kernel32.FindFirstVolumeW.restype = wintypes.HANDLE kernel32.FindNextVolumeW.argtypes = (wintypes.HANDLE, wintypes.LPWSTR, wintypes.DWORD) kernel32.FindVolumeClose.argtypes = (wintypes.HANDLE,) def FindFirstVolume(): volume_name = ctypes.create_unicode_buffer(" " * 255) h = kernel32.FindFirstVolumeW(volume_name, 255) if h == win32file.INVALID_HANDLE_VALUE: raise RuntimeError("FindFirstVolume() returned an invalid handle.") return h, volume_name.value def FindNextVolume(hSearch): volume_name = ctypes.create_unicode_buffer(" " * 255) if kernel32.FindNextVolumeW(hSearch, volume_name, 255) != 0: return volume_name.value else: errno = ctypes.GetLastError() if errno == winerror.ERROR_NO_MORE_FILES: FindVolumeClose(hSearch) return None raise RuntimeError("FindNextVolume failed (%s)" % errno) def FindVolumeClose(hSearch): """Close a search handle opened by FindFirstVolume, typically after the last volume has been returned. """ if kernel32.FindVolumeClose(hSearch) == 0: raise RuntimeError("FindVolumeClose() failed.") def findAvailableDrives(): return [(d, win32file.GetDriveType(d)) for d in win32api.GetLogicalDriveStrings().rstrip('\0').split('\0')] def findNewDriveLetter(used_letters): all_letters = set([chr(i) for i in range(ord('C'), ord('Z')+1)]) return min(list(all_letters - set([s[0] for s in used_letters]))) def _openHandle(path, bWriteAccess, bWriteShare, logfunc = lambda s: None): TIMEOUT, NUM_RETRIES = 10, 20 for retry_count in range(6): try: access_flag = win32con.GENERIC_READ | \ (bWriteAccess and win32con.GENERIC_WRITE or 0) share_flag = win32con.FILE_SHARE_READ | \ (bWriteShare and win32con.FILE_SHARE_WRITE or 0) handle = win32file.CreateFile( path, access_flag, share_flag, None, win32con.OPEN_EXISTING, win32con.FILE_ATTRIBUTE_NORMAL, None) nth = { 0: 'first', 1:'second', 2:'third'} logfunc("Opening [%s]: success at the %s iteration" % (path, nth.get(retry_count, '%sth' % (retry_count+1)))) return handle except pywintypes.error as e: logfunc('Exception=>'+str(e)) if NUM_RETRIES/3 < retry_count: bWriteShare = True time.sleep(TIMEOUT / float(NUM_RETRIES)) else: raise RuntimeError("Couldn't open handle for %s." % path) def _closeHandle(h): x = win32file.CloseHandle(h) assert x != win32file.INVALID_HANDLE_VALUE return x class openHandle: def __init__(self, path, bWriteAccess, bWriteShare, logfunc = lambda s: None): self.path = path self.bWriteAccess = bWriteAccess self.bWriteShare = bWriteShare self.logfunc = logfunc self.h = None def __enter__(self): self.h = _openHandle(self.path, self.bWriteAccess, self.bWriteShare, self.logfunc) return self def __exit__(self, type_, value, traceback_): _closeHandle(self.h) def assert_physical_drive(self): if self.path.lower().find('physicaldrive')<0: raise RuntimeError("Handle is not one of a physical drive.") def LockPhysicalDrive(self): self.assert_physical_drive() lockPhysicalDrive(self.h, self.logfunc) self.logfunc("Successfully locked '%s'" % self.path) def ReadFile(self, size): return win32file.ReadFile(self.h, size, None) def WriteFile(self, b): return win32file.WriteFile(self.h, b, None) geometory_tuple = collections.namedtuple( 'DiskGeometory', ['number_of_cylinders', 'media_type', 'tracks_per_cylinder', 'sectors_per_track', 'bytes_per_sector', 'disk_size']) def DiskGeometory(self): self.assert_physical_drive() o = win32file.DeviceIoControl( self.h, winioctlcon.IOCTL_DISK_GET_DRIVE_GEOMETRY_EX, None, 256, None) return self.geometory_tuple(*struct.unpack('%s' % x) x = win32file.DeleteVolumeMountPoint(volume_path+'\\') log_func('DeleteVolumeMountPoint=>%s' % x) else: log_func('No volumes on %s' % target_drive) add1MB = False hDrive.ZapMBRGPT(geom.disk_size, geom.bytes_per_sector, add1MB) if __name__ == '__main__': # used_letters = [d for d in # win32api.GetLogicalDriveStrings().rstrip('\0').split('\0')] # print (used_letters) # print (findNewDriveLetter(used_letters)) # TargetDrive = 2 # vinfo_list = [x for x in findVolumeGuids() # if x.Extents and x.Extents[0].DiskNumber==TargetDrive] TargetDrive = 5 with openHandle('\\\\.\\PhysicalDrive%d' % TargetDrive, True, False, lambda s:sys.stdout.write(s+'\n')) as hDrive: hDrive.CopyFrom('c:/Users/shinj/Downloads/salitaz-rolling_iso', lambda b: None)