This is a follow up of a previous question

I have written a file-copy routine, as I experienced unreliable results using standard Python when copying file over very slow networks.

The requirements are as follows:

  • run on Windows (other OS's are not a requirement at the moment, however not limiting at the moment)
  • not to be limited by the 256 character limit of Windows if source or target file is located on a (standard) mounted-drive-letter
  • configurable when to use hash checksum checking (source / target)
  • configurable "buffer size" or "chunk size" (it seems that smaller chunks are more reliable on unreliable/slow networks)
  • file only available on target file system AFTER all content has copied. If you copy files, the target filename already exists on the target filesystem, hence a remote application may "see" this file and try to open it while copy still in progress. This is implemented with a temp-target-filename in the code below.
  • configurable console output

So this is what I have, and it tested to be working OK.

I am looking for tips/tricks to improve this code


import ccopy
source = r'c:\test1\testfile.ext'
target = r'\\someserver\test2\testfile.ext'
ccopy.filecopy(source, target, True, 'md4', 1024, True, True)


import os
import uuid
import hashlib
import platform

def norm_file_path(path):
    Filenames and paths have a default limitation of 256 characters in Windows.
    By inserting '\\\\?\\' at the start of the path it removes this limitation.

    This function inserts '\\\\?\\' at the start of the path, on Windows only
    Only if the path starts with '<driveletter>:\\' e.g 'C:\\'.

    It will also normalise the characters/case of the path.

    On Windows:
    path = 'C:\\Test\\TestFile.txt'
    returns '\\\\?\\c:\\test\\testfile.txt'
    if _check_if_windows():
        if path[1:3] == ':\\':
            return ''.join([u'\\\\?\\', os.path.normcase(path)])
    return os.path.normcase(path)        

def hashfile(filepath, hashtype='md4', buffersize=1024, consoleoutput = False):
    This routine returns a byte value haslib-checksum of the contents of a file.

    filepath = should be a existing file.
    hastype = should be ['md4'| 'md5' | 'sha1'] but can be any value the hashlib module supports.
    buffersize = should be of type 'int', smaller or equal to 1MB, indicating the size of chunks, smaller chunks for slower networks.
    buffersize = 0 will set buffer size equal to size of file, e.g. it will try to read the whole file into memory.
    consoleoutput = boolean: determines if print messages is enabled or not.

    filepath = norm_file_path(filepath)

    if buffersize == 0:
        buffersize = os.stat(filepath).st_size     

    _cprint(['Reading file: ' + filepath,
             'Calculating haslib/' + hashtype,
             ''], consoleoutput)
    ha = hashlib.new(hashtype)
    f = open(filepath, 'rb')
    while True:
        chunk = f.read(buffersize)
        if not chunk: 
    _cprint(['hashlib/' + hashtype + ' = ' + str(ha.digest()), 
             ''], consoleoutput)
    return ha.digest() 

def filecopy(source, target, hashcheck = False, hashtype = 'md4', buffersize = 1024, overwrite = True, consoleoutput = False): 
    This copies a source file to target file, in chunks of the size of buffersize.
    It will calculate and compare hashlib checksums (if specified)
    It will either overwrite a target if it exists or not (if specified)

    It will utilize a temporary file as target until all data is copied.
    When hash checksum is matched (if used) then it will rename the temp file to the target file name.

    source = source filename
    target = target filename
    hashcheck = boolean: compare source and target checksums after copying
    hashtype = type of hash chechsum should be ['md4'| 'md5' | 'sha1'] but can be any value the hashlib module supports
    buffersize = should be of type 'int', smaller or equal to 1MB, indicating the size of chunks, smaller chunks for slower networks
    buffersize = 0 will set buffer size equal to size of file, e.g. it will try to read the whole file into memory
    overwrite = boolean: determines if target file is overwritten when exists
    consoleoutput = boolean: determines if print messages to console
    if hashcheck: 

    targetfilepath = norm_file_path(target)
    _check_target_isfile_and_not_exists(targetfilepath, overwrite)
    targetdir = norm_file_path(os.path.dirname(targetfilepath))

    sourcefilepath = norm_file_path(source)

    _cprint(['Buffersize = ' + str(buffersize), 

    tempfilename = '.'.join([str(uuid.uuid4()), str(uuid.uuid4())])
    tempfilepath = os.path.join(targetdir, tempfilename)

    if hashcheck:
        sourcefilehash = hashfile(sourcefilepath, hashtype, buffersize, consoleoutput)

    _copy_file_in_chunks(sourcefilepath, tempfilepath, buffersize, consoleoutput)

    if hashcheck:
        targetfilehash = hashfile(tempfilepath, hashtype, buffersize, consoleoutput)
        if not sourcefilehash == targetfilehash:
            _cprint(['Haslib/' + hashtype + ' mismatch'], consoleoutput) 
            raise Exception('File not successfully copied: haslib/md4 mismatch')
            _cprint(['Source File and Temp Target File Hashlib/' + hashtype + ' match',
                     ''], consoleoutput)

    _rename_file(tempfilepath, targetfilepath, consoleoutput)
    _cprint(['Done'], consoleoutput)
    return True

def _check_target_isfile_and_not_exists(filepath, overwrite):
    if os.path.isdir(filepath): raise Exception('Target must be a File not a Directory')
    if os.path.exists(filepath) and (not overwrite): raise Exception('Target File exists')

def _check_sourcefile(sourcefilepath):

def _check_symbolic_links_in_path(path):
    if not os.path.realpath(path) == path: raise Exception('The use of symbolic links or ".." is not allowed')

def _check_path_exists(path):
    if not os.path.exists(path): raise Exception('Path not Found ' + path)

def _check_buffersize(buffersize):
    if not isinstance(buffersize, int): raise Exception('Buffer Size must be int')
    if buffersize > 1048576: raise Exception('Buffer Size > 1MB (1048576 bytes)')
    if buffersize < 0: raise Exception('Buffer Size < 0')

def _check_hashtype(hashtype):
    if hashtype in hashlib.algorithms_available:
        return True
    raise Exception('hashtype must be of type: ' + str(hashlib.algorithms_available))

def _check_if_windows():
    return platform.system() == 'Windows'

def _remove_file(filepath, consoleoutput = False):
    _cprint(['Removing File: ' + filepath], consoleoutput) 

def _rename_file(source, target, consoleoutput = False):
    _cprint(['Renaming File',
             'Source Filename: ' + source,
             'Target Filename: ' + target], consoleoutput) 
    os.replace(source, target)    

def _copy_file_in_chunks(source, target, buffersize = 1024, consoleoutput = False):
    if buffersize == 0:
        buffersize = os.stat(source).st_size     
    copysource = open(source, 'rb')
    copytarget = open(target, 'wb') 
    _cprint(['Copying File',
             'Source File: ' + source,
             'Target File: ' + target, 
             ''], consoleoutput)
    while True:
        chunk = copysource.read(buffersize)
        if not chunk:
    _cprint(['File copied',
             ''], consoleoutput)    

def _cprint(text, consoleoutput):
    if consoleoutput:
        for line in text:

def main():
    print('NOT to be used interactively')

if __name__ == '__main__':
  • \$\begingroup\$ would anyone like to comment and/or answer ? \$\endgroup\$ Nov 25, 2016 at 11:52


Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Browse other questions tagged or ask your own question.