pythonでメモリ上にzipを作る。

[mtamaki]

昨日の続き。メモリ上にzip形式のファイルオブジェクトを作る関数。 zipしたいファイルのリストかフォルダパスを指定するだけだから使いやすい。。。かな？

def zip_create_obj( path_or_paths, base_path = None ):
    "base_pathからの相対パスとしてファイルを格納したzip形式のファイルオブジェクトを返す。"
    def path_get_files( path ):
        import stat, os, fnmatch
        if not ( os.access( path, os.F_OK ) and stat.S_ISDIR( os.stat( path )[stat.ST_MODE] ) ):
            #ファイル名部分を抜き出す。
            file_name_filter = os.path.split(path)[1]
            path = os.path.dirname( path )
        else:
            file_name_filter = "*"

        files = os.listdir(path)
        results = []
        for file in files:
            child_path = os.path.join(path, file)
            #ディレクトリではなく、
            if not ( os.access( child_path, os.F_OK ) and stat.S_ISDIR( os.stat( child_path )[stat.ST_MODE] ) ):
                #フィルタにマッチするとき
                if fnmatch.fnmatch( os.path.split(path)[1], file_name_filter ):
                    #追加
                    results.append(child_path)
        return results

    def path_get_folders(path):
        import os, stat
        results = []
        try:
            files = os.listdir(path)
            for file in files:
                child_path = os.path.join(path, file)
                if os.access( child_path, os.F_OK ) and stat.S_ISDIR( os.stat( child_path )[stat.ST_MODE] ):
                    #追加
                    results.append(child_path)
        except WindowsError:
            pass
        return results

    def path_get_recursive_files( path ):
        import stat, os
        #フォルダを指定していたら最後に*をつける。
        if os.access( path, os.F_OK ) and stat.S_ISDIR( os.stat( path )[stat.ST_MODE] ):
            path += "/*"
        #このフォルダのファイルを取得
        results = path_get_files( path )
        #サブフォルダを取得
        for sub_folder in path_get_folders( os.path.dirname(path) ):
            results += path_get_recursive_files( sub_folder + "/" + os.path.split(path)[1] )
        return results
    def zip_raw_write(zip_file, zinfo_or_arcname, obj):
        #base: zipfile.ZipFile.writestr
        from zipfile import ZipInfo, ZIP_DEFLATED
        import time
        if not isinstance(zinfo_or_arcname, ZipInfo):
            zinfo = ZipInfo(filename=zinfo_or_arcname,
                            date_time=time.localtime(time.time()))
            zinfo.compress_type = zip_file.compression
        else:
            zinfo = zinfo_or_arcname
        zip_file._writecheck(zinfo)
        read_obj = obj.read()
        zinfo.file_size = obj.tell()            # Uncompressed size
        import binascii
        zinfo.CRC = binascii.crc32(read_obj)       # CRC-32 checksum
        if zinfo.compress_type == ZIP_DEFLATED:
            import zlib
            co = zlib.compressobj(zlib.Z_DEFAULT_COMPRESSION,
                 zlib.DEFLATED, -15)
            read_obj = co.compress(read_obj) + co.flush()
            zinfo.compress_size = len(read_obj)    # Compressed size
        else:
            zinfo.compress_size = zinfo.file_size
        zinfo.header_offset = zip_file.fp.tell()    # Start of header obj
        zip_file.fp.write(zinfo.FileHeader())
        zinfo.file_offset = zip_file.fp.tell()      # Start of file obj
        zip_file.fp.write(read_obj)
        if zinfo.flag_bits & 0x08:
            # Write CRC and file sizes after the file data
            zip_file.fp.write(struct.pack("<lLL", zinfo.CRC, zinfo.compress_size,
                  zinfo.file_size))
        zip_file.filelist.append(zinfo)
        zip_file.NameToInfo[zinfo.filename] = zinfo

    import os, sys, stat
    #まず、第一引数が単一のファイルかファイルのリストかを調べる
    if isinstance( path_or_paths, (list, tuple) ):
        zip_files = path_or_paths
    elif os.access( path_or_paths, os.F_OK ) and stat.S_ISDIR( os.stat( path_or_paths )[stat.ST_MODE] ):
        #zipするファイルをリストアップ
        zip_files = path_get_recursive_files(path_or_paths)
    else:
        #zipするファイルをリストアップ
        zip_files = [path_or_paths]
    #ベースフォルダを決定
    if not base_path:
        base_path = os.path.commonprefix(zip_files)
    #末尾に\\をつける。
    if os.path.isdir(base_path):
        base_path = os.path.join(base_path, "1")[:-1]
    #zipファイルを作る。
    import zipfile, StringIO
    zip_file_obj = StringIO.StringIO()
    try:
        # zlib がない場合，この呼び出しは失敗して RuntimeError が返る
        zip_file = zipfile.ZipFile(zip_file_obj, "w", zipfile.ZIP_DEFLATED )
    except RuntimeError:
        # ZIP_STORED (非圧縮) がデフォルトの設定
        zip_file = zipfile.ZipFile(zip_file_obj, "w") #, zipfile.ZIP_STORED)
    #ファイルをzipに追加
    encoding = sys.getfilesystemencoding()
    for file in zip_files:
        arc_name = file[len(base_path):]
        if isinstance( arc_name, unicode ):
            arc_name = arc_name.encode( encoding )
        zip_file.write( file, arc_name )
    zip_file.close()
    zip_file_obj.seek(0,0)
    return zip_file_obj

if __name__ == '__main__':
        zip_obj = zip_create_obj( r"C:\WINDOWS\system32\drivers\etc" )
        import os
        zip_file_obj = open(os.path.splitext(__file__)[0] + ".zip", "wb")
        zip_file_obj.write( zip_obj.read() )
        zip_file_obj.close()

サンプルでetcをzipしてるのはわりと誰のPCにでもあって程よい大きさのフォルダだから。。。バイナリファイルが含まれてればベストなんだけどなー。

(080620追記:Python2.5でzip_raw_writeがエラーになります。 PythonZipOnMemory にあるzip_raw_writeと差し替えてください。)