Function Compute を使用して、OSS にアップロードした圧縮ファイルを自動的に解凍できます。指定された解凍ルールに一致する ZIP ファイルが Object Storage Service にアップロードされると、Function Compute が自動的にトリガーされてファイルが解凍されます。解凍が完了すると、Function Compute は解凍されたファイルを Object Storage Service の指定されたディレクトリにアップロードします。
使用法
ファイル名やフォルダ名には UTF-8 または GB 2312 エンコーディングを使用してください。これにより、文字化けや解凍の失敗を防ぐことができます。
アーカイブストレージまたはコールドアーカイブストレージのファイルは、解凍前に復元する必要があります。
単一の ZIP パッケージの最大解凍時間は 2 時間です。この制限を超えたタスクは失敗します。
解凍の失敗を防ぐため、ZIP パッケージ内の各ファイルが 1 GB を超えないようにしてください。
関数のタイムアウトは 2 時間以上に設定してください。最大 24 時間まで設定可能です。
関数は Object Storage Service バケットと同じリージョンに作成してください。
前提条件
操作手順
ステップ 1：関数の作成
Function Compute コンソールにログインし、左側のナビゲーションウィンドウで関数管理 > 関数リストを選択します。
トップナビゲーションバーでリージョンを選択し、関数リスト ページで 関数を作成 をクリックします。
関数を作成 ページで、作成方法を選択し、以下の設定を行い、作成 をクリックします。
このセクションでは、主要な設定について説明します。その他の設定については、「関数の作成」をご参照ください。
ランタイム環境: Python 3.10 を選択します。
関数ロール：既存のロールを選択するか、新しいロールを作成します。AliyunOSSFullAccess ポリシーをアタッチして、ロールに OSS バケットへのアクセス権限があることを確認してください。
「関数の詳細」ページで、コード タブを選択し、ファイルを作成し、コードを入力して、デプロイメントコード をクリックします。
サンプルコードは次のとおりです。
index.pyファイルサンプルコード
# -*- coding: utf-8 -*- """ 注： この関数は、ZIP アーカイブ内のファイル名とフォルダ名の文字エンコーディングを処理します。 エンコーディングは、アーカイブを作成したオペレーティングシステムによって異なります。 - macOS または Linux で作成されたアーカイブの場合、ファイル名は通常 UTF-8 でエンコードされます。 - Windows で作成されたアーカイブの場合、ファイル名は GB2312 でエンコードされることが多いですが、UTF-8 の場合もあります。 これらの違いに対応するため、この関数は 'chardet' ライブラリを使用してエンコーディングを自動的に検出します。 ただし、検出が 100% 正確であるとは限りません。解凍後にファイル名が文字化けする問題が発生した場合は、 この関数のエンコーディングロジックを修正して再デプロイする必要があります。 """ import helper import oss2 import json import os import time import logging import chardet """ この関数は、.zip ファイルが OSS バケットの指定されたソースプレフィックスにアップロードされると実行されます。 ファイルを解凍し、コンテンツを宛先プレフィックスにアップロードします。 たとえば、`source/archive.zip` としてアップロードされたファイルは `processed/archive/` に解凍されます。 ソースと宛先のプレフィックスは、環境変数を使用して設定できます。 """ # OSS SDK によって出力される情報ログを抑制します。 logging.getLogger("oss2.api").setLevel(logging.ERROR) logging.getLogger("oss2.auth").setLevel(logging.ERROR) LOGGER = logging.getLogger() # 関数の実行時間を出力します。 def print_excute_time(func): def wrapper(*args, **kwargs): local_time = time.time() ret = func(*args, **kwargs) LOGGER.info('current Function [%s] excute time is %.2f' % (func.__name__, time.time() - local_time)) return ret return wrapper def get_zipfile_name(origin_name): # ファイル名の潜在的な文字化けを処理します (特に非 UTF-8 エンコーディングの場合)。 name = origin_name try: name_bytes = origin_name.encode(encoding="cp437") except: name_bytes = origin_name.encode(encoding="utf-8") # 検出は文字列が長いほど正確になります。 detect = chardet.detect(name_bytes) confidence = detect["confidence"] detect_encoding = detect["encoding"] if confidence > 0.75 and (detect_encoding.lower() in ["gb2312", "gbk", "gb18030", "ascii", "utf-8"]): try: if detect_encoding.lower() in ["gb2312", "gbk", "gb18030"]: detect_encoding = "gb18030" name = name_bytes.decode(detect_encoding) except: name = name_bytes.decode(encoding="gb18030") else: try: name = name_bytes.decode(encoding="gb18030") except: name = name_bytes.decode(encoding="utf-8") # パスの一貫性を保つため、Windows スタイルのバックスラッシュをフォワードスラッシュに置き換えます。 name = name.replace("\\", "/") return name @print_excute_time def handler(event, context): """ Function Compute によって実行されるメインハンドラ関数です。OSS バケットから .zip オブジェクトを自動的に解凍します。 :param event: OSS イベントを表す JSON 文字列。オブジェクトのキーやその他のメタデータが含まれます。 :param context: 関数コンテキスト。認証情報とランタイム情報を提供します。 """ evt_lst = json.loads(event) creds = context.credentials auth = oss2.StsAuth( creds.access_key_id, creds.access_key_secret, creds.security_token) evt = evt_lst['events'][0] bucket_name = evt['oss']['bucket']['name'] endpoint = 'oss-' + evt['region'] + '-internal.aliyuncs.com' bucket = oss2.Bucket(auth, endpoint, bucket_name) object_name = evt['oss']['object']['key'] if "ObjectCreated:PutSymlink" == evt['eventName']: object_name = bucket.get_symlink(object_name).target_key if not object_name: raise RuntimeError('{} is an invalid symlink file'.format( evt['oss']['object']['key'])) file_type = os.path.splitext(object_name)[1] if file_type != ".zip": raise RuntimeError('{} is not a .zip file'.format(object_name)) LOGGER.info("Starting to decompress zip file: {}".format(object_name)) lst = object_name.split("/") zip_name = lst[-1] PROCESSED_DIR = os.environ.get("PROCESSED_DIR", "") RETAIN_FILE_NAME = os.environ.get("RETAIN_FILE_NAME", "") if PROCESSED_DIR and PROCESSED_DIR[-1] != "/": PROCESSED_DIR += "/" if RETAIN_FILE_NAME == "false": newKey = PROCESSED_DIR else: newKey = PROCESSED_DIR + zip_name zip_fp = helper.OssStreamFileLikeObject(bucket, object_name) newKey = newKey.replace(".zip", "/") with helper.zipfile_support_oss.ZipFile(zip_fp) as zip_file: for name in zip_file.namelist(): with zip_file.open(name) as file_obj: name = get_zipfile_name(name) bucket.put_object(newKey + name, file_obj)helper.pyファイルサンプルコード
# -*- coding: utf-8 -*- import oss2 from oss2 import utils, models import ossZipfile as zipfile zipfile_support_oss = zipfile # ファイルライクオブジェクトとして OSS へのアップロードをサポートします。 def make_crc_adapter(data, init_crc=0): data = utils.to_bytes(data) # file-like object if hasattr(data, 'read'): return utils._FileLikeAdapter(data, crc_callback=utils.Crc64(init_crc)) utils.make_crc_adapter = make_crc_adapter class OssStreamFileLikeObject(object): def __init__(self, bucket, key): super(OssStreamFileLikeObject, self).__init__() self._bucket = bucket self._key = key self._meta_data = self._bucket.get_object_meta(self._key) @property def bucket(self): return self._bucket @property def key(self): return self._key @property def filesize(self): return self._meta_data.content_length def get_reader(self, begin, end): begin = begin if begin >= 0 else 0 end = end if end > 0 else self.filesize - 1 end = end if end < self.filesize else self.filesize - 1 begin = begin if begin < end else end return self._bucket.get_object(self._key, byte_range=(begin, end)) def get_content_bytes(self, begin, end): reader = self.get_reader(begin, end) return reader.read() def get_last_content_bytes(self, offset): return self.get_content_bytes(self.filesize-offset, self.filesize-1)ossZipfile.pyサンプルコード
""" ZIP ファイルの読み取りと書き込みを行います。 XXX UTF-8 への参照については、さらなる調査が必要です。 """ import io import os import importlib.util import sys import time import stat import shutil import struct import binascii import threading try: import zlib # その圧縮メソッドが必要になる場合があります crc32 = zlib.crc32 except ImportError: zlib = None crc32 = binascii.crc32 try: import bz2 # その圧縮メソッドが必要になる場合があります except ImportError: bz2 = None try: import lzma # その圧縮メソッドが必要になる場合があります except ImportError: lzma = None __all__ = ["BadZipFile", "BadZipfile", "error", "ZIP_STORED", "ZIP_DEFLATED", "ZIP_BZIP2", "ZIP_LZMA", "is_zipfile", "ZipInfo", "ZipFile", "PyZipFile", "LargeZipFile"] class BadZipFile(Exception): pass class LargeZipFile(Exception): """ ZIP64 拡張が必要な ZIP ファイルを書き込む際に、 それらの拡張が無効になっている場合に発生します。 """ error = BadZipfile = BadZipFile # 3.2 以前の互換性名 ZIP64_LIMIT = (1 << 31) - 1 ZIP_FILECOUNT_LIMIT = (1 << 16) - 1 ZIP_MAX_COMMENT = (1 << 16) - 1 # ZIP ファイル圧縮メソッドの定数 ZIP_STORED = 0 ZIP_DEFLATED = 8 ZIP_BZIP2 = 12 ZIP_LZMA = 14 # 他の ZIP 圧縮メソッドはサポートされていません DEFAULT_VERSION = 20 ZIP64_VERSION = 45 BZIP2_VERSION = 46 LZMA_VERSION = 63 # そのバージョンまでのすべての機能を認識します (必ずしもサポートするわけではありません) MAX_EXTRACT_VERSION = 63 # 以下は、struct モジュールを使用してヘッダーを読み書きするための # いくつかのフォーマットと関連データです。ヘッダー/レコードの名前と構造は、 # ZIP ファイル形式の PKWARE の説明で使用されているものです： # http://www.pkware.com/documents/casestudies/APPNOTE.TXT # (URL は 2008 年 1 月時点で有効) # 「中央ディレクトリの終わり」構造体、マジックナンバー、サイズ、およびインデックス # (フォーマットドキュメントのセクション V.I) structEndArchive = b"<4s4H2LH" stringEndArchive = b"PK\005\006" sizeEndCentDir = struct.calcsize(structEndArchive) _ECD_SIGNATURE = 0 _ECD_DISK_NUMBER = 1 _ECD_DISK_START = 2 _ECD_ENTRIES_THIS_DISK = 3 _ECD_ENTRIES_TOTAL = 4 _ECD_SIZE = 5 _ECD_OFFSET = 6 _ECD_COMMENT_SIZE = 7 # これら最後の 2 つのインデックスは、 # 仕様で定義されている構造の一部ではありませんが、便宜上このモジュール内で使用されます _ECD_COMMENT = 8 _ECD_LOCATION = 9 # 「中央ディレクトリ」構造体、マジックナンバー、サイズ、および # 構造体内のエントリのインデックス (フォーマットドキュメントのセクション V.F) structCentralDir = "<4s4B4HL2L5H2L" stringCentralDir = b"PK\001\002" sizeCentralDir = struct.calcsize(structCentralDir) # 中央ディレクトリ構造体内のエントリのインデックス _CD_SIGNATURE = 0 _CD_CREATE_VERSION = 1 _CD_CREATE_SYSTEM = 2 _CD_EXTRACT_VERSION = 3 _CD_EXTRACT_SYSTEM = 4 _CD_FLAG_BITS = 5 _CD_COMPRESS_TYPE = 6 _CD_TIME = 7 _CD_DATE = 8 _CD_CRC = 9 _CD_COMPRESSED_SIZE = 10 _CD_UNCOMPRESSED_SIZE = 11 _CD_FILENAME_LENGTH = 12 _CD_EXTRA_FIELD_LENGTH = 13 _CD_COMMENT_LENGTH = 14 _CD_DISK_NUMBER_START = 15 _CD_INTERNAL_FILE_ATTRIBUTES = 16 _CD_EXTERNAL_FILE_ATTRIBUTES = 17 _CD_LOCAL_HEADER_OFFSET = 18 # 「ローカルファイルヘッダー」構造体、マジックナンバー、サイズ、およびインデックス # (フォーマットドキュメントのセクション V.A) structFileHeader = "<4s2B4HL2L2H" stringFileHeader = b"PK\003\004" sizeFileHeader = struct.calcsize(structFileHeader) _FH_SIGNATURE = 0 _FH_EXTRACT_VERSION = 1 _FH_EXTRACT_SYSTEM = 2 _FH_GENERAL_PURPOSE_FLAG_BITS = 3 _FH_COMPRESSION_METHOD = 4 _FH_LAST_MOD_TIME = 5 _FH_LAST_MOD_DATE = 6 _FH_CRC = 7 _FH_COMPRESSED_SIZE = 8 _FH_UNCOMPRESSED_SIZE = 9 _FH_FILENAME_LENGTH = 10 _FH_EXTRA_FIELD_LENGTH = 11 # 「Zip64 中央ディレクトリの終わりのロケーター」構造体、マジックナンバー、およびサイズ structEndArchive64Locator = "<4sLQL" stringEndArchive64Locator = b"PK\x06\x07" sizeEndCentDir64Locator = struct.calcsize(structEndArchive64Locator) # 「Zip64 中央ディレクトリの終わり」レコード、マジックナンバー、サイズ、およびインデックス # (フォーマットドキュメントのセクション V.G) structEndArchive64 = "<4sQ2H2L4Q" stringEndArchive64 = b"PK\x06\x06" sizeEndCentDir64 = struct.calcsize(structEndArchive64) _CD64_SIGNATURE = 0 _CD64_DIRECTORY_RECSIZE = 1 _CD64_CREATE_VERSION = 2 _CD64_EXTRACT_VERSION = 3 _CD64_DISK_NUMBER = 4 _CD64_DISK_NUMBER_START = 5 _CD64_NUMBER_ENTRIES_THIS_DISK = 6 _CD64_NUMBER_ENTRIES_TOTAL = 7 _CD64_DIRECTORY_SIZE = 8 _CD64_OFFSET_START_CENTDIR = 9 def _check_zipfile(fp): try: if _EndRecData(fp): return True # ファイルには正しいマジックナンバーがあります except OSError: pass return False def is_zipfile(filename): """マジックナンバーをチェックして、ファイルが ZIP ファイルかどうかを素早く判断します。 filename 引数は、ファイルまたはファイルライクオブジェクトです。 """ result = False try: if hasattr(filename, "read"): result = _check_zipfile(fp=filename) else: with open(filename, "rb") as fp: result = _check_zipfile(fp) except OSError: pass return result def _EndRecData64(fpin, offset, endrec): """ ZIP64 のアーカイブ終了レコードを読み取り、それらを使用して endrec を更新します。 """ if hasattr(fpin, 'bucket'): data = fpin.get_content_bytes( fpin.filesize+offset-sizeEndCentDir64Locator, fpin.filesize+offset-1) else: try: fpin.seek(offset - sizeEndCentDir64Locator, 2) except OSError: # シークが失敗した場合、ファイルは ZIP64 # アーカイブ終了レコードを含むほど大きくないため、与えられた終了レコードをそのまま返します。 return endrec data = fpin.read(sizeEndCentDir64Locator) if len(data) != sizeEndCentDir64Locator: return endrec sig, diskno, reloff, disks = struct.unpack(structEndArchive64Locator, data) if sig != stringEndArchive64Locator: return endrec if diskno != 0 or disks != 1: raise BadZipFile("複数ディスクにまたがる ZIP ファイルはサポートされていません") # 'zip64 拡張可能データ' がないと仮定します if hasattr(fpin, 'bucket'): data = fpin.get_content_bytes( fpin.filesize + offset - sizeEndCentDir64Locator - sizeEndCentDir64, fpin.filesize+offset-sizeEndCentDir64Locator-1) else: fpin.seek(offset - sizeEndCentDir64Locator - sizeEndCentDir64, 2) data = fpin.read(sizeEndCentDir64) if len(data) != sizeEndCentDir64: return endrec sig, sz, create_version, read_version, disk_num, disk_dir, \ dircount, dircount2, dirsize, diroffset = \ struct.unpack(structEndArchive64, data) if sig != stringEndArchive64: return endrec # ZIP64 レコードのデータを使用して元の endrec を更新します endrec[_ECD_SIGNATURE] = sig endrec[_ECD_DISK_NUMBER] = disk_num endrec[_ECD_DISK_START] = disk_dir endrec[_ECD_ENTRIES_THIS_DISK] = dircount endrec[_ECD_ENTRIES_TOTAL] = dircount2 endrec[_ECD_SIZE] = dirsize endrec[_ECD_OFFSET] = diroffset return endrec def _EndRecData(fpin): """「中央ディレクトリの終わり」レコードからデータを返すか、None を返します。 データは、ZIP の「中央ディレクトリの終わり」レコードの 9 つの項目のリストで、 その後に 10 番目の項目としてこのレコードのファイルシークオフセットが続きます。""" # ファイルサイズを決定します if hasattr(fpin, 'bucket'): filesize = fpin.filesize data = fpin.get_last_content_bytes(sizeEndCentDir) else: fpin.seek(0, 2) filesize = fpin.tell() # これがアーカイブコメントのない ZIP ファイルかどうかを確認します ( # この場合、「中央ディレクトリの終わり」構造体はファイルの最後の項目である必要があります)。 try: fpin.seek(-sizeEndCentDir, 2) except OSError: return None data = fpin.read() if (len(data) == sizeEndCentDir and data[0:4] == stringEndArchive and data[-2:] == b"\000\000"): # 署名は正しく、コメントはありません。構造体をアンパックします endrec = struct.unpack(structEndArchive, data) endrec = list(endrec) # 空のコメントとレコード開始オフセットを追加します endrec.append(b"") endrec.append(filesize - sizeEndCentDir) # 「Zip64 中央ディレクトリの終わり」構造体を読み取ろうとします return _EndRecData64(fpin, -sizeEndCentDir, endrec) # これは ZIP ファイルではないか、アーカイブコメント付きの ZIP ファイルです。 # ファイルの末尾で「中央ディレクトリの終わり」レコード署名を検索します。 # コメントは ZIP ファイルの最後の項目で、最大 64K の長さになる場合があります。 # 「中央ディレクトリの終わり」のマジックナンバーはコメントには現れないと想定されます。 maxCommentStart = max(filesize - (1 << 16) - sizeEndCentDir, 0) if hasattr(fpin, 'bucket'): data = fpin.get_content_bytes(maxCommentStart, -1) else: fpin.seek(maxCommentStart, 0) data = fpin.read() start = data.rfind(stringEndArchive) if start >= 0: # マジックナンバーが見つかりました。アンパックと解釈を試みます recData = data[start:start+sizeEndCentDir] if len(recData) != sizeEndCentDir: # ZIP ファイルが破損しています。 return None endrec = list(struct.unpack(structEndArchive, recData)) commentSize = endrec[_ECD_COMMENT_SIZE] # ZIP ファイルが主張するとおり comment = data[start+sizeEndCentDir:start+sizeEndCentDir+commentSize] endrec.append(comment) endrec.append(maxCommentStart + start) # 「Zip64 中央ディレクトリの終わり」構造体を読み取ろうとします return _EndRecData64(fpin, maxCommentStart + start - filesize, endrec) # 有効な中央ディレクトリの終わり構造体を見つけることができません return None class ZipInfo (object): """ZIP アーカイブ内の各ファイルを記述する属性を持つクラス。""" __slots__ = ( 'orig_filename', 'filename', 'date_time', 'compress_type', 'comment', 'extra', 'create_system', 'create_version', 'extract_version', 'reserved', 'flag_bits', 'volume', 'internal_attr', 'external_attr', 'header_offset', 'CRC', 'compress_size', 'file_size', '_raw_time', ) def __init__(self, filename="NoName", date_time=(1980, 1, 1, 0, 0, 0)): self.orig_filename = filename # アーカイブ内の元のファイル名 # 最初のヌルバイトでファイル名を終了させます。ファイル名内のヌルバイトは、 # アーカイブ内のウイルスによるトリックとして使用されます。 null_byte = filename.find(chr(0)) if null_byte >= 0: filename = filename[0:null_byte] # これは、生成された ZIP ファイル内のパスが常に # ZIP 形式仕様で要求されるように、ディレクトリ区切り文字として # フォワードスラッシュを使用することを保証するために使用されます。 if os.sep != "/" and os.sep in filename: filename = filename.replace(os.sep, "/") self.filename = filename # 正規化されたファイル名 self.date_time = date_time # 年、月、日、時、分、秒 if date_time[0] < 1980: raise ValueError('ZIP は 1980 年より前のタイムスタンプをサポートしていません') # 標準値： self.compress_type = ZIP_STORED # ファイルの圧縮タイプ self.comment = b"" # 各ファイルのコメント self.extra = b"" # ZIP 拡張データ if sys.platform == 'win32': self.create_system = 0 # ZIP アーカイブを作成したシステム else: # それ以外はすべて unix 系と仮定します self.create_system = 3 # ZIP アーカイブを作成したシステム self.create_version = DEFAULT_VERSION # ZIP アーカイブを作成したバージョン self.extract_version = DEFAULT_VERSION # アーカイブを解凍するために必要なバージョン self.reserved = 0 # ゼロでなければなりません self.flag_bits = 0 # ZIP フラグビット self.volume = 0 # ファイルヘッダーのボリューム番号 self.internal_attr = 0 # 内部属性 self.external_attr = 0 # 外部ファイル属性 # その他の属性はクラス ZipFile によって設定されます： # header_offset ファイルヘッダーへのバイトオフセット # CRC 非圧縮ファイルの CRC-32 # compress_size 圧縮ファイルのサイズ # file_size 非圧縮ファイルのサイズ def __repr__(self): result = ['<%s filename=%r' % (self.__class__.__name__, self.filename)] if self.compress_type != ZIP_STORED: result.append(' compress_type=%s' % compressor_names.get(self.compress_type, self.compress_type)) hi = self.external_attr >> 16 lo = self.external_attr & 0xFFFF if hi: result.append(' filemode=%r' % stat.filemode(hi)) if lo: result.append(' external_attr=%#x' % lo) isdir = self.is_dir() if not isdir or self.file_size: result.append(' file_size=%r' % self.file_size) if ((not isdir or self.compress_size) and (self.compress_type != ZIP_STORED or self.file_size != self.compress_size)): result.append(' compress_size=%r' % self.compress_size) result.append('>') return ''.join(result) def FileHeader(self, zip64=None): """ファイルごとのヘッダーを文字列として返します。""" dt = self.date_time dosdate = (dt[0] - 1980) << 9 | dt[1] << 5 | dt[2] dostime = dt[3] << 11 | dt[4] << 5 | (dt[5] // 2) if self.flag_bits & 0x08: # ファイルデータの後に書き込むため、これらをゼロに設定します CRC = compress_size = file_size = 0 else: CRC = self.CRC compress_size = self.compress_size file_size = self.file_size extra = self.extra min_version = 0 if zip64 is None: zip64 = file_size > ZIP64_LIMIT or compress_size > ZIP64_LIMIT if zip64: fmt = '<hhqq'> ZIP64_LIMIT or compress_size > ZIP64_LIMIT: if not zip64: raise LargeZipFile("ファイルサイズには ZIP64 拡張が必要です") # ファイルが 4 バイト整数に収まるサイズより大きい場合、 # ZIP64 拡張にフォールバックします。 file_size = 0xffffffff compress_size = 0xffffffff min_version = ZIP64_VERSION if self.compress_type == ZIP_BZIP2: min_version = max(BZIP2_VERSION, min_version) elif self.compress_type == ZIP_LZMA: min_version = max(LZMA_VERSION, min_version) self.extract_version = max(min_version, self.extract_version) self.create_version = max(min_version, self.create_version) filename, flag_bits = self._encodeFilenameFlags() header = struct.pack(structFileHeader, stringFileHeader, self.extract_version, self.reserved, flag_bits, self.compress_type, dostime, dosdate, CRC, compress_size, file_size, len(filename), len(extra)) return header + filename + extra def _encodeFilenameFlags(self): try: return self.filename.encode('ascii'), self.flag_bits except UnicodeEncodeError: return self.filename.encode('utf-8'), self.flag_bits | 0x800 def _decodeExtra(self): # 拡張フィールドをデコードしようとします。 extra = self.extra unpack = struct.unpack while len(extra) >= 4: tp, ln = unpack('<hh',> len(extra): raise BadZipFile( "破損した拡張フィールド %04x (size=%d)" % (tp, ln)) if tp == 0x0001: if ln >= 24: counts = unpack('<qqq',>> 1) ^ 0xEDB88320 else: crc >>= 1 return crc # ZIP はパスワードベースの暗号化形式をサポートしています。既知の # 平文攻撃が発見されていますが、それでも # そのようなファイルからデータを取り出すことができるのは便利です。 # # 使用法： # zd = _ZipDecrypter(mypwd) # plain_bytes = zd(cipher_bytes) def _ZipDecrypter(pwd): key0 = 305419896 key1 = 591751049 key2 = 878082192 global _crctable if _crctable is None: _crctable = list(map(_gen_crc, range(256))) crctable = _crctable def crc32(ch, crc): """1 バイトに対して CRC32 プリミティブを計算します。""" return (crc >> 8) ^ crctable[(crc ^ ch) & 0xFF] def update_keys(c): nonlocal key0, key1, key2 key0 = crc32(c, key0) key1 = (key1 + (key0 & 0xFF)) & 0xFFFFFFFF key1 = (key1 * 134775813 + 1) & 0xFFFFFFFF key2 = crc32(key1 >> 24, key2) for p in pwd: update_keys(p) def decrypter(data): """bytes オブジェクトを復号します。""" result = bytearray() append = result.append for c in data: k = key2 | 2 c ^= ((k * (k ^ 1)) >> 8) & 0xFF update_keys(c) append(c) return bytes(result) return decrypter class LZMACompressor: def __init__(self): self._comp = None def _init(self): props = lzma._encode_filter_properties({'id': lzma.FILTER_LZMA1}) self._comp = lzma.LZMACompressor(lzma.FORMAT_RAW, filters=[ lzma._decode_filter_properties(lzma.FILTER_LZMA1, props) ]) return struct.pack('<bbh',><< 31 - 1 # 圧縮ファイルから 4k ブロックで読み取ります。 MIN_READ_SIZE = 4096 def __init__(self, fileobj, mode, zipinfo, decrypter=None, close_fileobj=False): self._fileobj = fileobj self._decrypter = decrypter self._close_fileobj = close_fileobj self._compress_type = zipinfo.compress_type self._compress_left = zipinfo.compress_size self._left = zipinfo.file_size self._decompressor = _get_decompressor(self._compress_type) self._eof = False self._readbuffer = b'' self._offset = 0 self.newlines = None # 暗号化されたファイルの場合、最初の 12 バイトは # 暗号化/パスワード情報用なので、読み取りサイズを調整します。 if self._decrypter is not None: self._compress_left -= 12 self.mode = mode self.name = zipinfo.filename if hasattr(zipinfo, 'CRC'): self._expected_crc = zipinfo.CRC self._running_crc = crc32(b'') else: self._expected_crc = None def __repr__(self): result = ['<%s.%s' % (self.__class__.__module__, self.__class__.__qualname__)] if not self.closed: result.append(' name=%r mode=%r' % (self.name, self.mode)) if self._compress_type != ZIP_STORED: result.append(' compress_type=%s' % compressor_names.get(self._compress_type, self._compress_type)) else: result.append(' [closed]') result.append('>') return ''.join(result) def readline(self, limit=-1): """ストリームから 1 行を読み取って返します。 limit が指定されている場合、最大で limit バイトが読み取られます。 """ if limit < 0: # 一般的なケースのショートカット：バッファ内で改行が見つかった場合。 i = self._readbuffer.find(b'\n', self._offset) + 1 if i > 0: line = self._readbuffer[self._offset: i] self._offset = i return line return io.BufferedIOBase.readline(self, limit) def peek(self, n=1): """位置を進めずにバッファリングされたバイトを返します。""" if n > len(self._readbuffer) - self._offset: chunk = self.read(n) if len(chunk) > self._offset: self._readbuffer = chunk + self._readbuffer[self._offset:] self._offset = 0 else: self._offset -= len(chunk) # タイトなループでの割り当てオーバーヘッドを減らすために、最大 512 バイトを返します。 return self._readbuffer[self._offset: self._offset + 512] def readable(self): return True def read(self, n=-1): """最大 n バイトを読み取って返します。 引数が省略されたり、None、または負の場合は、EOF に達するまでデータが読み取られて返されます。 """ if n is None or n < 0: buf = self._readbuffer[self._offset:] self._readbuffer = b'' self._offset = 0 while not self._eof: buf += self._read1(self.MAX_N) return buf end = n + self._offset if end < len(self._readbuffer): buf = self._readbuffer[self._offset:end] self._offset = end return buf n = end - len(self._readbuffer) buf = self._readbuffer[self._offset:] self._readbuffer = b'' self._offset = 0 while n > 0 and not self._eof: data = self._read1(n) if n < len(data): self._readbuffer = data self._offset = n buf += data[:n] break buf += data n -= len(data) return buf def _update_crc(self, newdata): # 与えられたデータを使用して CRC を更新します。 if self._expected_crc is None: # 参照値がない場合は CRC を計算する必要はありません。 return self._running_crc = crc32(newdata, self._running_crc) # ファイルの終わりに達したら CRC をチェックします。 if self._eof and self._running_crc != self._expected_crc: raise BadZipFile("Bad CRC-32 for file %r" % self.name) def read1(self, n): """最大 1 回の read() システムコールで最大 n バイトを読み取ります。""" if n is None or n < 0: buf = self._readbuffer[self._offset:] self._readbuffer = b'' self._offset = 0 while not self._eof: data = self._read1(self.MAX_N) if data: buf += data break return buf end = n + self._offset if end < len(self._readbuffer): buf = self._readbuffer[self._offset:end] self._offset = end return buf n = end - len(self._readbuffer) buf = self._readbuffer[self._offset:] self._readbuffer = b'' self._offset = 0 if n > 0: while not self._eof: data = self._read1(n) if n < len(data): self._readbuffer = data self._offset = n buf += data[:n] break if data: buf += data break return buf def _read1(self, n): # 最大 1 回の read() システムコールで最大 n バイトの圧縮データを読み取り、 # それらを復号して解凍します。 if self._eof or n <= 0: return b'' # ファイルから読み取ります。 if self._compress_type == ZIP_DEFLATED: # 未消費のデータを処理します。 data = self._decompressor.unconsumed_tail if n > len(data): data += self._read2(n - len(data)) else: data = self._read2(n) if self._compress_type == ZIP_STORED: self._eof = self._compress_left <= 0 elif self._compress_type == ZIP_DEFLATED: n = max(n, self.MIN_READ_SIZE) data = self._decompressor.decompress(data, n) self._eof = (self._decompressor.eof or self._compress_left <= 0 and not self._decompressor.unconsumed_tail) if self._eof: data += self._decompressor.flush() else: data = self._decompressor.decompress(data) self._eof = self._decompressor.eof or self._compress_left <= 0 data = data[:self._left] self._left -= len(data) if self._left <= 0: self._eof = True self._update_crc(data) return data def _read2(self, n): if self._compress_left <= 0: return b'' n = max(n, self.MIN_READ_SIZE) n = min(n, self._compress_left) data = self._fileobj.read(n) self._compress_left -= len(data) if not data: raise EOFError if self._decrypter is not None: data = self._decrypter(data) return data def close(self): try: if self._close_fileobj: self._fileobj.close() finally: super().close() class _ZipWriteFile(io.BufferedIOBase): def __init__(self, zf, zinfo, zip64): self._zinfo = zinfo self._zip64 = zip64 self._zipfile = zf self._compressor = _get_compressor(zinfo.compress_type) self._file_size = 0 self._compress_size = 0 self._crc = 0 @property def _fileobj(self): return self._zipfile.fp def writable(self): return True def write(self, data): if self.closed: raise ValueError('閉じたファイルに対する I/O 操作です。') nbytes = len(data) self._file_size += nbytes self._crc = crc32(data, self._crc) if self._compressor: data = self._compressor.compress(data) self._compress_size += len(data) self._fileobj.write(data) return nbytes def close(self): if self.closed: return super().close() # コンプレッサーからデータをフラッシュし、ヘッダー情報を更新します。 if self._compressor: buf = self._compressor.flush() self._compress_size += len(buf) self._fileobj.write(buf) self._zinfo.compress_size = self._compress_size else: self._zinfo.compress_size = self._file_size self._zinfo.CRC = self._crc self._zinfo.file_size = self._file_size # 更新されたヘッダー情報を書き込みます。 if self._zinfo.flag_bits & 0x08: # ファイルデータの後に CRC とファイルサイズを書き込みます fmt = '<lqq'> ZIP64_LIMIT: raise RuntimeError('ファイルサイズが予期せず ZIP64 ' '制限を超えました') if self._compress_size > ZIP64_LIMIT: raise RuntimeError('圧縮サイズが予期せず ZIP64 ' '制限を超えました') # 後方にシークしてファイルヘッダーを書き込みます (これには # 正しい CRC とファイルサイズが含まれます)。 # ファイル内の現在位置を保持します。 self._zipfile.start_dir = self._fileobj.tell() self._fileobj.seek(self._zinfo.header_offset) self._fileobj.write(self._zinfo.FileHeader(self._zip64)) self._fileobj.seek(self._zipfile.start_dir) self._zipfile._writing = False # 正常に書き込まれました：ファイルをキャッシュに追加します。 self._zipfile.filelist.append(self._zinfo) self._zipfile.NameToInfo[self._zinfo.filename] = self._zinfo class ZipFile: """ZIP ファイルを開く、読み取る、書き込む、閉じる、一覧表示するためのメソッドを持つクラス。 z = ZipFile(file, mode="r", compression=ZIP_STORED, allowZip64=True) file: ファイルへのパス、またはファイルライクオブジェクト。 パスの場合、ファイルは ZipFile によって開閉されます。 mode: 読み取りの場合は 'r'、書き込みの場合は 'w'、排他的作成の場合は 'x'、 または追記の場合は 'a' です。 compression: ZIP_STORED (圧縮なし)、ZIP_DEFLATED (zlib が必要)、 ZIP_BZIP2 (bz2 が必要)、または ZIP_LZMA (lzma が必要)。 allowZip64: True の場合、ZipFile は必要なときに ZIP64 拡張を持つファイルを作成します。 それ以外の場合、ZIP64 が必要な場合は例外を発生させます。 """ fp = None # __del__ がチェックするためここで設定 _windows_illegal_name_trans_table = None def __init__(self, file, mode="r", compression=ZIP_STORED, allowZip64=True): """ZIP ファイルを読み取り ('r')、書き込み ('w')、排他的作成 ('x')、 または追記 ('a') モードで開きます。""" if mode not in ('r', 'w', 'x', 'a'): raise ValueError("ZipFile にはモード 'r'、'w'、'x'、または 'a' が必要です") _check_compression(compression) self._allowZip64 = allowZip64 self._didModify = False self.debug = 0 # 出力レベル：0 から 3 self.NameToInfo = {} # ファイル名を ZipInfo インスタンスにマッピングします self.filelist = [] # アーカイブの ZipInfo インスタンスのリスト self.compression = compression # デフォルトの圧縮メソッド self.mode = mode self.pwd = None self._comment = b'' # ファイルライクオブジェクトが渡されたかどうかを確認します if isinstance(file, os.PathLike): file = os.fspath(file) if isinstance(file, str): # ファイル名です self._filePassed = 0 self.filename = file modeDict = {'r': 'rb', 'w': 'w+b', 'x': 'x+b', 'a': 'r+b', 'r+b': 'w+b', 'w+b': 'wb', 'x+b': 'xb'} filemode = modeDict[mode] while True: try: self.fp = io.open(file, filemode) except OSError: if filemode in modeDict: filemode = modeDict[filemode] continue raise break else: self._filePassed = 1 self.fp = file self.filename = getattr(file, 'name', None) self._fileRefCnt = 1 self._lock = threading.RLock() self._seekable = True self._writing = False try: if mode == 'r': self._RealGetContents() elif mode in ('w', 'x'): # アーカイブにファイルが追加されなくても中央ディレクトリが # 書き込まれるように、変更フラグを設定します。 self._didModify = True try: self.start_dir = self.fp.tell() except (AttributeError, OSError): self.fp = _Tellable(self.fp) self.start_dir = 0 self._seekable = False else: # 一部のファイルライクオブジェクトは tell() を提供できますが、seek() は提供できません。 try: self.fp.seek(self.start_dir) except (AttributeError, OSError): self._seekable = False elif mode == 'a': try: # ファイルが ZIP ファイルかどうかを確認します。 self._RealGetContents() # 中央ディレクトリの開始位置にシークして上書きします。 self.fp.seek(self.start_dir) except BadZipFile: # ファイルは ZIP ファイルではありません。単に追記します。 self.fp.seek(0, 2) # アーカイブにファイルが追加されなくても中央ディレクトリが # 書き込まれるように、変更フラグを設定します。 self._didModify = True self.start_dir = self.fp.tell() else: raise ValueError("モードは 'r'、'w'、'x'、または 'a' でなければなりません") except: fp = self.fp self.fp = None self._fpclose(fp) raise def __enter__(self): return self def __exit__(self, type, value, traceback): self.close() def __repr__(self): result = ['<%s.%s' % (self.__class__.__module__, self.__class__.__qualname__)] if self.fp is not None: if self._filePassed: result.append(' file=%r' % self.fp) elif self.filename is not None: result.append(' filename=%r' % self.filename) result.append(' mode=%r' % self.mode) else: result.append(' [closed]') result.append('>') return ''.join(result) def _RealGetContents(self): """ZIP ファイルから目次を読み取ります。""" fp = self.fp try: endrec = _EndRecData(fp) except OSError: raise BadZipFile("ファイルは ZIP ファイルではありません") if not endrec: raise BadZipFile("ファイルは ZIP ファイルではありません") if self.debug > 1: print(endrec) size_cd = endrec[_ECD_SIZE] # 中央ディレクトリのバイト数 offset_cd = endrec[_ECD_OFFSET] # 中央ディレクトリのオフセット self._comment = endrec[_ECD_COMMENT] # アーカイブコメント # "concat" は、ZIP ファイルが別のファイルに連結されていない限りゼロです。 concat = endrec[_ECD_LOCATION] - size_cd - offset_cd if endrec[_ECD_SIGNATURE] == stringEndArchive64: # ZIP64 拡張構造体が存在する場合、それらを考慮に入れます。 concat -= (sizeEndCentDir64 + sizeEndCentDir64Locator) if self.debug > 2: inferred = concat + offset_cd print("given, inferred, offset", offset_cd, inferred, concat) # self.start_dir: 中央ディレクトリの開始位置。 self.start_dir = offset_cd + concat if hasattr(fp, "bucket"): data = fp.get_content_bytes( self.start_dir, self.start_dir+size_cd-1) else: fp.seek(self.start_dir, 0) data = fp.read(size_cd) fp = io.BytesIO(data) total = 0 while total < size_cd: centdir = fp.read(sizeCentralDir) if len(centdir) != sizeCentralDir: raise BadZipFile("切り捨てられた中央ディレクトリ") centdir = struct.unpack(structCentralDir, centdir) if centdir[_CD_SIGNATURE] != stringCentralDir: raise BadZipFile("中央ディレクトリの不正なマジックナンバー") if self.debug > 2: print(centdir) filename = fp.read(centdir[_CD_FILENAME_LENGTH]) flags = centdir[5] if flags & 0x800: # UTF-8 ファイル名拡張 filename = filename.decode('utf-8') else: # 従来の ZIP ファイル名エンコーディング filename = filename.decode('cp437') # ファイル情報を格納するために ZipInfo インスタンスを作成します。 x = ZipInfo(filename) x.extra = fp.read(centdir[_CD_EXTRA_FIELD_LENGTH]) x.comment = fp.read(centdir[_CD_COMMENT_LENGTH]) x.header_offset = centdir[_CD_LOCAL_HEADER_OFFSET] (x.create_version, x.create_system, x.extract_version, x.reserved, x.flag_bits, x.compress_type, t, d, x.CRC, x.compress_size, x.file_size) = centdir[1:12] if x.extract_version > MAX_EXTRACT_VERSION: raise NotImplementedError("ZIP ファイルバージョン %.1f" % (x.extract_version / 10)) x.volume, x.internal_attr, x.external_attr = centdir[15:18] # 日付/時刻コードを (年, 月, 日, 時, 分, 秒) に変換します。 x._raw_time = t x.date_time = ((d >> 9)+1980, (d >> 5) & 0xF, d & 0x1F, t >> 11, (t >> 5) & 0x3F, (t & 0x1F) * 2) x._decodeExtra() x.header_offset = x.header_offset + concat self.filelist.append(x) self.NameToInfo[x.filename] = x # 中央ディレクトリから読み取った合計バイト数を更新します。 total = (total + sizeCentralDir + centdir[_CD_FILENAME_LENGTH] + centdir[_CD_EXTRA_FIELD_LENGTH] + centdir[_CD_COMMENT_LENGTH]) if self.debug > 2: print("total", total) def namelist(self): """アーカイブ内のファイル名のリストを返します。""" return [data.filename for data in self.filelist] def infolist(self): """アーカイブ内のファイルの ZipInfo インスタンスのリストを返します。""" return self.filelist def printdir(self, file=None): """ZIP ファイルの目次を印刷します。""" print("%-46s %19s %12s" % ("File Name", "Modified ", "Size"), file=file) for zinfo in self.filelist: date = "%d-%02d-%02d %02d:%02d:%02d" % zinfo.date_time[:6] print("%-46s %s %12d" % (zinfo.filename, date, zinfo.file_size), file=file) def testzip(self): """すべてのファイルを読み取り、CRC をチェックします。""" chunk_size = 2 ** 20 for zinfo in self.filelist: try: # 非常に大きな埋め込みファイルで OverflowError または # MemoryError を回避するために、チャンクで読み取ります。 with self.open(zinfo.filename, "r") as f: while f.read(chunk_size): # CRC-32 をチェック pass except BadZipFile: return zinfo.filename def getinfo(self, name): """メンバー `name` の ZipInfo を返します。""" info = self.NameToInfo.get(name) if info is None: raise KeyError( 'アーカイブに %r という名前のアイテムはありません' % name) return info def setpassword(self, pwd): """暗号化されたファイルのデフォルトパスワードを設定します。""" if pwd and not isinstance(pwd, bytes): raise TypeError("pwd: bytes が期待されましたが、%s を受け取りました" % type(pwd).__name__) if pwd: self.pwd = pwd else: self.pwd = None @property def comment(self): """ZIP ファイルに関連付けられたコメントテキスト。""" return self._comment @comment.setter def comment(self, comment): if not isinstance(comment, bytes): raise TypeError("comment: bytes が期待されましたが、%s を受け取りました" % type(comment).__name__) # 有効なコメント長をチェックします。 if len(comment) > ZIP_MAX_COMMENT: import warnings warnings.warn('アーカイブコメントが長すぎます。%d バイトに切り詰めます' % ZIP_MAX_COMMENT, stacklevel=2) comment = comment[:ZIP_MAX_COMMENT] self._comment = comment self._didModify = True def read(self, name, pwd=None): """ファイル `name` のバイトを返します。""" with self.open(name, "r", pwd) as fp: return fp.read() def open(self, name, mode="r", pwd=None, *, force_zip64=False): """`name` のファイルライクオブジェクトを返します。 `name` は ZIP ファイル内のファイル名を指定する文字列、または ZipInfo オブジェクトです。 `mode` はアーカイブからファイルを読み取る場合は 'r'、 アーカイブに新しいファイルを書き込む場合は 'w' でなければなりません。 `pwd` はファイルを復号するためのパスワードです (読み取り時にのみ使用)。 書き込み時に、ファイルサイズが事前にわからないが 2 GiB を超える可能性がある場合、 `force_zip64=True` を渡して、大きなファイルをサポートする ZIP64 形式を使用します。 サイズが事前にわかっている場合は、`zinfo.file_size` を設定した ZipInfo インスタンスを `name` に渡すのが最善です。 """ if mode not in {"r", "w"}: raise ValueError('open() にはモード "r" または "w" が必要です') if pwd and not isinstance(pwd, bytes): raise TypeError("pwd: bytes が期待されましたが、%s を受け取りました" % type(pwd).__name__) if pwd and (mode == "w"): raise ValueError("pwd はファイルの読み取りにのみサポートされています") if not self.fp: raise ValueError( "すでに閉じられた ZIP アーカイブを使用しようとしました") # info オブジェクトがあることを確認します if isinstance(name, ZipInfo): # 'name' はすでに info オブジェクトです zinfo = name elif mode == 'w': zinfo = ZipInfo(name) zinfo.compress_type = self.compression else: # name の info オブジェクトを取得します zinfo = self.getinfo(name) if mode == 'w': return self._open_to_write(zinfo, force_zip64=force_zip64) if self._writing: raise ValueError("ZIP ファイルに開いている書き込みハンドルがある間は " "読み取りできません。 " "読み取る前に書き込みハンドルを閉じてください。") # 読み取り用に開きます： self._fileRefCnt += 1 zef_file = _SharedFile(self.fp, zinfo.header_offset, self._fpclose, self._lock, lambda: self._writing) try: # ファイルヘッダーをスキップします： fheader = zef_file.read(sizeFileHeader) if len(fheader) != sizeFileHeader: raise BadZipFile("切り捨てられたファイルヘッダー") fheader = struct.unpack(structFileHeader, fheader) if fheader[_FH_SIGNATURE] != stringFileHeader: raise BadZipFile("ファイルヘッダーの不正なマジックナンバー") fname = zef_file.read(fheader[_FH_FILENAME_LENGTH]) if fheader[_FH_EXTRA_FIELD_LENGTH]: zef_file.read(fheader[_FH_EXTRA_FIELD_LENGTH]) if zinfo.flag_bits & 0x20: # Zip 2.7: 圧縮パッチデータ raise NotImplementedError( "圧縮パッチデータ (フラグビット 5)") if zinfo.flag_bits & 0x40: # 強力な暗号化 raise NotImplementedError("強力な暗号化 (フラグビット 6)") if zinfo.flag_bits & 0x800: # UTF-8 ファイル名 fname_str = fname.decode("utf-8") else: fname_str = fname.decode("cp437") if fname_str != zinfo.orig_filename: raise BadZipFile( 'ディレクトリ %r とヘッダー %r のファイル名が異なります。' % (zinfo.orig_filename, fname)) # 暗号化フラグをチェックし、パスワードを処理します is_encrypted = zinfo.flag_bits & 0x1 zd = None if is_encrypted: if not pwd: pwd = self.pwd if not pwd: raise RuntimeError("ファイル %r は暗号化されています。解凍には " "パスワードが必要です" % name) zd = _ZipDecrypter(pwd) # 暗号ストリームの最初の 12 バイトは、アルゴリズムを強化するために # 使用される暗号化ヘッダーを形成します。最初の 11 バイトは # 完全にランダムですが、12 番目のバイトには CRC の MSB # またはファイル時刻が含まれ、ヘッダータイプに応じて # パスワードの正しさをチェックするために使用されます。 header = zef_file.read(12) h = zd(header[0:12]) if zinfo.flag_bits & 0x8: # 拡張ローカルヘッダーのファイルタイプと比較します check_byte = (zinfo._raw_time >> 8) & 0xff else: # それ以外の場合は CRC と比較します check_byte = (zinfo.CRC >> 24) & 0xff if h[11] != check_byte: raise RuntimeError("ファイル %r のパスワードが不正です" % name) return ZipExtFile(zef_file, mode, zinfo, zd, True) except: zef_file.close() raise def _open_to_write(self, zinfo, force_zip64=False): if force_zip64 and not self._allowZip64: raise ValueError( "force_zip64 は True ですが、ZIP ファイルを開くときに allowZip64 が " "False でした。" ) if self._writing: raise ValueError("別の書き込みハンドルが開いている間は ZIP ファイルに " "書き込みできません。 " "最初のハンドルを閉じてから別のハンドルを開いてください。") # サイズと CRC は、ファイルの処理後に正しいデータで上書きされます。 if not hasattr(zinfo, 'file_size'): zinfo.file_size = 0 zinfo.compress_size = 0 zinfo.CRC = 0 zinfo.flag_bits = 0x00 if zinfo.compress_type == ZIP_LZMA: # 圧縮データにはストリーム終了 (EOS) マーカーが含まれます zinfo.flag_bits |= 0x02 if not self._seekable: zinfo.flag_bits |= 0x08 if not zinfo.external_attr: zinfo.external_attr = 0o600 << 16 # 権限：?rw------- # 圧縮サイズは非圧縮サイズより大きくなる可能性があります。 zip64 = self._allowZip64 and \ (force_zip64 or zinfo.file_size * 1.05 > ZIP64_LIMIT) if self._seekable: self.fp.seek(self.start_dir) zinfo.header_offset = self.fp.tell() self._writecheck(zinfo) self._didModify = True self.fp.write(zinfo.FileHeader(zip64)) self._writing = True return _ZipWriteFile(self, zinfo, zip64) def extract(self, member, path=None, pwd=None): """アーカイブからメンバーを現在の作業ディレクトリに解凍します。 `member` はファイル名または ZipInfo オブジェクトです。`path` を使用して 別のディレクトリを指定できます。そのファイル情報は 可能な限り正確に解凍されます。 """ if path is None: path = os.getcwd() else: path = os.fspath(path) return self._extract_member(member, path, pwd) def extractall(self, path=None, members=None, pwd=None): """アーカイブからすべてのメンバーを現在の作業 ディレクトリに解凍します。`path` は解凍先の別のディレクトリを指定します。 `members` はオプションで、`namelist()` が返すリストの サブセットでなければなりません。 """ if members is None: members = self.namelist() if path is None: path = os.getcwd() else: path = os.fspath(path) for zipinfo in members: self._extract_member(zipinfo, path, pwd) @classmethod def _sanitize_windows_name(cls, arcname, pathsep): """無効な文字を置換し、部分から末尾のドットを削除します。""" table = cls._windows_illegal_name_trans_table if not table: illegal = ':<>|"?*' table = str.maketrans(illegal, '_' * len(illegal)) cls._windows_illegal_name_trans_table = table arcname = arcname.translate(table) # 末尾のドットを削除 arcname = (x.rstrip('.') for x in arcname.split(pathsep)) # 再結合し、空の部分を削除 arcname = pathsep.join(x for x in arcname if x) return arcname def _extract_member(self, member, targetpath, pwd): """ZipInfo オブジェクト `member` を `targetpath` パス上の 物理ファイルに解凍します。 """ if not isinstance(member, ZipInfo): member = self.getinfo(member) # 宛先パス名を構築し、 # フォワードスラッシュをプラットフォーム固有の区切り文字に置き換えます。 arcname = member.filename.replace('/', os.path.sep) if os.path.altsep: arcname = arcname.replace(os.path.altsep, os.path.sep) # 絶対パス名を相対パスとして解釈し、ドライブ文字または # UNC パス、および冗長な区切り文字、"." および ".." コンポーネントを削除します。 arcname = os.path.splitdrive(arcname)[1] invalid_path_parts = ('', os.path.curdir, os.path.pardir) arcname = os.path.sep.join(x for x in arcname.split(os.path.sep) if x not in invalid_path_parts) if os.path.sep == '\\': # Windows 上の不正な文字をフィルタリングします arcname = self._sanitize_windows_name(arcname, os.path.sep) targetpath = os.path.join(targetpath, arcname) targetpath = os.path.normpath(targetpath) # 必要に応じてすべての上位ディレクトリを作成します。 upperdirs = os.path.dirname(targetpath) if upperdirs and not os.path.exists(upperdirs): os.makedirs(upperdirs) if member.is_dir(): if not os.path.isdir(targetpath): os.mkdir(targetpath) return targetpath with self.open(member, pwd=pwd) as source, \ open(targetpath, "wb") as target: shutil.copyfileobj(source, target) return targetpath def _writecheck(self, zinfo): """アーカイブにファイルを書き込む前にエラーをチェックします。""" if zinfo.filename in self.NameToInfo: import warnings warnings.warn('重複した名前： %r' % zinfo.filename, stacklevel=3) if self.mode not in ('w', 'x', 'a'): raise ValueError("write() にはモード 'w'、'x'、または 'a' が必要です") if not self.fp: raise ValueError( "すでに閉じられた ZIP アーカイブに書き込もうとしました") _check_compression(zinfo.compress_type) if not self._allowZip64: requires_zip64 = None if len(self.filelist) >= ZIP_FILECOUNT_LIMIT: requires_zip64 = "ファイル数" elif zinfo.file_size > ZIP64_LIMIT: requires_zip64 = "ファイルサイズ" elif zinfo.header_offset > ZIP64_LIMIT: requires_zip64 = "ZIP ファイルサイズ" if requires_zip64: raise LargeZipFile(requires_zip64 + " には ZIP64 拡張が必要です") def write(self, filename, arcname=None, compress_type=None): """ファイル `filename` をアーカイブに `arcname` という名前で 書き込みます。""" if not self.fp: raise ValueError( "すでに閉じられた ZIP アーカイブに書き込もうとしました") if self._writing: raise ValueError( "開いている書き込みハンドルが存在する間は ZIP アーカイブに書き込みできません" ) zinfo = ZipInfo.from_file(filename, arcname) if zinfo.is_dir(): zinfo.compress_size = 0 zinfo.CRC = 0 else: if compress_type is not None: zinfo.compress_type = compress_type else: zinfo.compress_type = self.compression if zinfo.is_dir(): with self._lock: if self._seekable: self.fp.seek(self.start_dir) zinfo.header_offset = self.fp.tell() # ヘッダーバイトの開始 if zinfo.compress_type == ZIP_LZMA: # 圧縮データにはストリーム終了 (EOS) マーカーが含まれます zinfo.flag_bits |= 0x02 self._writecheck(zinfo) self._didModify = True self.filelist.append(zinfo) self.NameToInfo[zinfo.filename] = zinfo self.fp.write(zinfo.FileHeader(False)) self.start_dir = self.fp.tell() else: with open(filename, "rb") as src, self.open(zinfo, 'w') as dest: shutil.copyfileobj(src, dest, 1024*8) def writestr(self, zinfo_or_arcname, data, compress_type=None): """アーカイブにファイルを書き込みます。コンテンツは `data` で、 `str` または `bytes` インスタンスです。`str` の場合、 最初に UTF-8 としてエンコードされます。 `zinfo_or_arcname` は ZipInfo インスタンスまたは アーカイブ内のファイル名です。""" if isinstance(data, str): data = data.encode("utf-8") if not isinstance(zinfo_or_arcname, ZipInfo): zinfo = ZipInfo(filename=zinfo_or_arcname, date_time=time.localtime(time.time())[:6]) zinfo.compress_type = self.compression if zinfo.filename[-1] == '/': zinfo.external_attr = 0o40775 << 16 # drwxrwxr-x zinfo.external_attr |= 0x10 # MS-DOS ディレクトリフラグ else: zinfo.external_attr = 0o600 << 16 # ?rw------- else: zinfo = zinfo_or_arcname if not self.fp: raise ValueError( "すでに閉じられた ZIP アーカイブに書き込もうとしました") if self._writing: raise ValueError( "開いている書き込みハンドルが存在する間は ZIP アーカイブに書き込みできません。" ) if compress_type is not None: zinfo.compress_type = compress_type zinfo.file_size = len(data) # 非圧縮サイズ with self._lock: with self.open(zinfo, mode='w') as dest: dest.write(data) def __del__(self): """ユーザーが忘れた場合に備えて "close()" メソッドを呼び出します。""" self.close() def close(self): """ファイルを閉じ、'w'、'x'、'a' モードの場合は終了 レコードを書き込みます。""" if self.fp is None: return if self._writing: raise ValueError("ZIP ファイルに開いている書き込みハンドルがある間は " "閉じることができません。 " "ZIP ファイルを閉じる前に書き込みハンドルを閉じてください。") try: if self.mode in ('w', 'x', 'a') and self._didModify: # 終了レコードを書き込みます with self._lock: if self._seekable: self.fp.seek(self.start_dir) self._write_end_record() finally: fp = self.fp self.fp = None self._fpclose(fp) def _write_end_record(self): for zinfo in self.filelist: # 中央ディレクトリを書き込みます dt = zinfo.date_time dosdate = (dt[0] - 1980) << 9 | dt[1] << 5 | dt[2] dostime = dt[3] << 11 | dt[4] << 5 | (dt[5] // 2) extra = [] if zinfo.file_size > ZIP64_LIMIT \ or zinfo.compress_size > ZIP64_LIMIT: extra.append(zinfo.file_size) extra.append(zinfo.compress_size) file_size = 0xffffffff compress_size = 0xffffffff else: file_size = zinfo.file_size compress_size = zinfo.compress_size if zinfo.header_offset > ZIP64_LIMIT: extra.append(zinfo.header_offset) header_offset = 0xffffffff else: header_offset = zinfo.header_offset extra_data = zinfo.extra min_version = 0 if extra: # 拡張データに ZIP64 フィールドを追加します extra_data = struct.pack( '<hh'> ZIP_FILECOUNT_LIMIT: requires_zip64 = "ファイル数" elif centDirOffset > ZIP64_LIMIT: requires_zip64 = "中央ディレクトリのオフセット" elif centDirSize > ZIP64_LIMIT: requires_zip64 = "中央ディレクトリのサイズ" if requires_zip64: # ZIP64 アーカイブ終了レコードを書き込む必要があります if not self._allowZip64: raise LargeZipFile(requires_zip64 + " には ZIP64 拡張が必要です") zip64endrec = struct.pack( structEndArchive64, stringEndArchive64, 44, 45, 45, 0, 0, centDirCount, centDirCount, centDirSize, centDirOffset) self.fp.write(zip64endrec) zip64locrec = struct.pack( structEndArchive64Locator, stringEndArchive64Locator, 0, pos2, 1) self.fp.write(zip64locrec) centDirCount = min(centDirCount, 0xFFFF) centDirSize = min(centDirSize, 0xFFFFFFFF) centDirOffset = min(centDirOffset, 0xFFFFFFFF) endrec = struct.pack(structEndArchive, stringEndArchive, 0, 0, centDirCount, centDirCount, centDirSize, centDirOffset, len(self._comment)) self.fp.write(endrec) self.fp.write(self._comment) self.fp.flush() def _fpclose(self, fp): assert self._fileRefCnt > 0 self._fileRefCnt -= 1 if not self._fileRefCnt and not self._filePassed: fp.close() class PyZipFile(ZipFile): """Python ライブラリファイルとパッケージで ZIP アーカイブを作成するクラス。""" def __init__(self, file, mode="r", compression=ZIP_STORED, allowZip64=True, optimize=-1): ZipFile.__init__(self, file, mode=mode, compression=compression, allowZip64=allowZip64) self._optimize = optimize def writepy(self, pathname, basename="", filterfunc=None): """`pathname` からすべてのファイルを ZIP アーカイブに追加します。 `pathname` がパッケージディレクトリの場合、ディレクトリと すべてのサブディレクトリを再帰的に検索して *.py ファイルを探し、 モジュールをアーカイブに追加します。 `pathname` が通常のディレクトリの場合、ディレクトリからすべての *.py ファイルを 追加します。 `pathname` が Python の *.py ファイルの場合、このメソッドはモジュールを アーカイブに追加します。このメソッドは常に追加されたモジュールを `module.pyc` として保存します。 このメソッドは、必要に応じて `module.py` を `module.pyc` にコンパイルします。 `filterfunc(pathname)` が提供されている場合、すべてのパスに対して呼び出されます。 `filterfunc` が False を返した場合、このメソッドはファイルまたはディレクトリをスキップします。 """ pathname = os.fspath(pathname) if filterfunc and not filterfunc(pathname): if self.debug: label = 'path' if os.path.isdir(pathname) else 'file' print('%s %r skipped by filterfunc' % (label, pathname)) return dir, name = os.path.split(pathname) if os.path.isdir(pathname): initname = os.path.join(pathname, "__init__.py") if os.path.isfile(initname): # これはパッケージディレクトリなので、追加します if basename: basename = "%s/%s" % (basename, name) else: basename = name if self.debug: print("Adding package in", pathname, "as", basename) fname, arcname = self._get_codename(initname[0:-3], basename) if self.debug: print("Adding", arcname) self.write(fname, arcname) dirlist = os.listdir(pathname) dirlist.remove("__init__.py") # すべての *.py ファイルとパッケージサブディレクトリを追加します for filename in dirlist: path = os.path.join(pathname, filename) root, ext = os.path.splitext(filename) if os.path.isdir(path): if os.path.isfile(os.path.join(path, "__init__.py")): # これはパッケージディレクトリなので、追加します self.writepy(path, basename, filterfunc=filterfunc) # 再帰呼び出し elif ext == ".py": if filterfunc and not filterfunc(path): if self.debug: print('file %r skipped by filterfunc' % path) continue fname, arcname = self._get_codename(path[0:-3], basename) if self.debug: print("Adding", arcname) self.write(fname, arcname) else: # これはパッケージディレクトリではありません。そのファイルをトップレベルに追加します if self.debug: print("Adding files from directory", pathname) for filename in os.listdir(pathname): path = os.path.join(pathname, filename) root, ext = os.path.splitext(filename) if ext == ".py": if filterfunc and not filterfunc(path): if self.debug: print('file %r skipped by filterfunc' % path) continue fname, arcname = self._get_codename(path[0:-3], basename) if self.debug: print("Adding", arcname) self.write(fname, arcname) else: if pathname[-3:] != ".py": raise RuntimeError( 'writepy() で追加されるファイルは ".py" で終わる必要があります') fname, arcname = self._get_codename(pathname[0:-3], basename) if self.debug: print("Adding file", arcname) self.write(fname, arcname) def _get_codename(self, pathname, basename): """与えられたパスに対して (filename, archivename) を返します。 モジュールパスが与えられた場合、正しいファイルパスと アーカイブ名を返し、必要に応じてコンパイルします。たとえば、 `/python/lib/string` が与えられた場合、(`/python/lib/string.pyc`, `string`) を返します。 """ def _compile(file, optimize=-1): import py_compile if self.debug: print("Compiling", file) try: py_compile.compile(file, doraise=True, optimize=optimize) except py_compile.PyCompileError as err: print(err.msg) return False return True file_py = pathname + ".py" file_pyc = pathname + ".pyc" pycache_opt0 = importlib.util.cache_from_source( file_py, optimization='') pycache_opt1 = importlib.util.cache_from_source( file_py, optimization=1) pycache_opt2 = importlib.util.cache_from_source( file_py, optimization=2) if self._optimize == -1: # レガシーモード：存在するファイルを使用します if (os.path.isfile(file_pyc) and os.stat(file_pyc).st_mtime >= os.stat(file_py).st_mtime): # .pyc ファイルを使用します。 arcname = fname = file_pyc elif (os.path.isfile(pycache_opt0) and os.stat(pycache_opt0).st_mtime >= os.stat(file_py).st_mtime): # __pycache__/*.pyc ファイルを使用しますが、アーカイブにはレガシーの pyc # ファイル名で書き込みます。 fname = pycache_opt0 arcname = file_pyc elif (os.path.isfile(pycache_opt1) and os.stat(pycache_opt1).st_mtime >= os.stat(file_py).st_mtime): # __pycache__/*.pyc ファイルを使用しますが、アーカイブにはレガシーの pyc # ファイル名で書き込みます。 fname = pycache_opt1 arcname = file_pyc elif (os.path.isfile(pycache_opt2) and os.stat(pycache_opt2).st_mtime >= os.stat(file_py).st_mtime): # __pycache__/*.pyc ファイルを使用しますが、アーカイブにはレガシーの pyc # ファイル名で書き込みます。 fname = pycache_opt2 arcname = file_pyc else: # .py を PEP 3147 .pyc ファイルにコンパイルします。 if _compile(file_py): if sys.flags.optimize == 0: fname = pycache_opt0 elif sys.flags.optimize == 1: fname = pycache_opt1 else: fname = pycache_opt2 arcname = file_pyc else: fname = arcname = file_py else: # 新しいモード：与えられた最適化レベルを使用します if self._optimize == 0: fname = pycache_opt0 arcname = file_pyc else: arcname = file_pyc if self._optimize == 1: fname = pycache_opt1 elif self._optimize == 2: fname = pycache_opt2 else: msg = "invalid value for 'optimize': {!r}".format( self._optimize) raise ValueError(msg) if not (os.path.isfile(fname) and os.stat(fname).st_mtime >= os.stat(file_py).st_mtime): if not _compile(file_py, optimize=self._optimize): fname = arcname = file_py archivename = os.path.split(arcname)[1] if basename: archivename = "%s/%s" % (basename, archivename) return (fname, archivename) def main(args=None): import argparse description = 'zipfile モジュールのためのシンプルなコマンドラインインターフェイス。' parser = argparse.ArgumentParser(description=description) group = parser.add_mutually_exclusive_group(required=True) group.add_argument('-l', '--list', metavar='<zipfile>', help='ZIP ファイルのリストを表示します') group.add_argument('-e', '--extract', nargs=2, metavar=('<zipfile>', '<output_dir>'), help='ZIP ファイルをターゲットディレクトリに解凍します') group.add_argument('-c', '--create', nargs='+', metavar=('<name>', '<file>'), help='ソースから ZIP ファイルを作成します') group.add_argument('-t', '--test', metavar='<zipfile>', help='ZIP ファイルが有効かどうかをテストします') args = parser.parse_args(args) if args.test is not None: src = args.test with ZipFile(src, 'r') as zf: badfile = zf.testzip() if badfile: print( "The following enclosed file is corrupted: {!r}".format(badfile)) print("Done testing") elif args.list is not None: src = args.list with ZipFile(src, 'r') as zf: zf.printdir() elif args.extract is not None: src, curdir = args.extract with ZipFile(src, 'r') as zf: zf.extractall(curdir) elif args.create is not None: zip_name = args.create.pop(0) files = args.create def addToZip(zf, path, zippath): if os.path.isfile(path): zf.write(path, zippath, ZIP_DEFLATED) elif os.path.isdir(path): if zippath: zf.write(path, zippath) for nm in os.listdir(path): addToZip(zf, os.path.join(path, nm), os.path.join(zippath, nm)) # else: 無視 with ZipFile(zip_name, 'w') as zf: for path in files: zippath = os.path.basename(path) if not zippath: zippath = os.path.basename(os.path.dirname(path)) if zippath in ('', os.curdir, os.pardir): zippath = '' addToZip(zf, path, zippath) if __name__ == "__main__": main()</zipfile></file></name></output_dir></zipfile></zipfile></hh'>
ステップ 2：OSS トリガーの作成
「関数の詳細」ページで、トリガー タブを選択し、トリガーの作成 をクリックします。
「トリガーの作成」パネルで、OSS トリガータイプを選択し、設定を行い、次に OK をクリックします。
主要な設定は以下のとおりです。その他の設定については、「ネイティブ OSS トリガーの設定」をご参照ください。
バケット名：作成したバケットを選択します。
オブジェクトプレフィックス: この例では
srcと入力します。オブジェクトサフィックス: この例では、
zipを入力します。トリガーイベント：この例では、次のイベントを選択します：
oss:ObjectCreated:PutObject, oss:ObjectCreated:PostObject, oss:ObjectCreated:CompleteMultipartUpload, oss:ObjectCreated:PutSymlink。ロール名：関数を呼び出す権限を持つロールを選択します。AliyunFCFullAccess ポリシーをロールにアタッチすることで、これらの権限を付与できます。
作成後、トリガーはトリガーページに表示されます。
ステップ 3：検証
検証設定は 2 つの方法でテストできます：
方法 1：コンソール経由でのファイルのアップロード
Object Storage Service コンソールにログインし、ステップ 2: OSS トリガーを作成するで選択したバケットの src ディレクトリに、code.zip などの ZIP ファイルをアップロードします。アップロードによって関数が自動的にトリガーされ、ZIP ファイルがバケットのルートディレクトリに解凍されます。
方法 2：イベントパラメーターを使用したテスト
イベントパラメーターを設定した後、関数のテスト をクリックして関数を実行します。次に、Object Storage Service コンソールに移動し、ターゲットバケット内のファイルリストを確認して、ZIP ファイルが解凍されていることを確認します。
関数詳細ページで、コード タブをクリックし、関数のテスト の横にある
アイコンをクリックし、ドロップダウンリストから テストパラメーターの設定 を選択します。テストパラメーターの設定 パネルで、イベントテンプレート を選択し、イベント名 とイベントコンテンツを入力して、OK をクリックします。
以下はサンプルイベントです。
{ "events": [ { "eventName": "ObjectCreated:PutObject", "eventSource": "acs:oss", "eventTime": "2023-08-13T06:45:43.000Z", "eventVersion": "1.0", "oss": { "bucket": { "arn": "acs:oss:cn-hangzhou:10343546824****:bucket****", "name": "bucket****", "ownerIdentity": "10343546824****" }, "object": { "deltaSize": 122539, "eTag": "688A7BF4F233DC9C88A80BF985AB****", "key": "src/test.zip", "size": 122539 }, "ossSchemaVersion": "1.0", "ruleId": "9adac8e253828f4f7c0466d941fa3db81161****" }, "region": "cn-hangzhou", "requestParameters": { "sourceIPAddress": "140.205.XX.XX" }, "responseElements": { "requestId": "58F9FF2D3DF792092E12044C" }, "userIdentity": { "principalId": "10343546824****" } } ] }イベントパラメーターのフィールドの説明については、「関数の入力パラメーターの設定」をご参照ください。
重要前述のイベントコンテンツは、デモンストレーション専用です。実際の情報に基づいていくつかのパラメーターを変更し、指定されたファイル (この例では
src/test.zip) が設定済みのバケットに存在することを確認してください。そうしないと、関数がトリガーされないか、関数の実行が失敗します。次のパラメーターは、実際の構成に基づいて変更する必要があります：
bucket.arn： 例:acs:oss:<region>:<your_account_id>:<your_bucket>。<region>を関数を作成したリージョンに、<your_account_id>をお客様の Alibaba Cloud アカウント ID に、<your_bucket>を同じリージョンに作成したバケットの名前に置き換えます。 お客様の Alibaba Cloud アカウント ID は、[Function Compute コンソール] の [概要] ページの [共通情報] セクションで確認できます。bucket.name：ご利用のバケットの名前に置き換えます。bucket.ownerIdentity：ご利用の Alibaba Cloud アカウント ID に置き換えます。object.key：ターゲットバケット内のオブジェクトのキーに置き換えます。region：関数のリージョンに置き換えます。userIdentity.principalId：ご利用の Alibaba Cloud アカウント ID に置き換えます。
コード タブで、関数のテスト をクリックしてテストを実行します。
操作が完了したら、Object Storage Service コンソールにログインし、ターゲットバケット内のターゲット ZIP ファイル (この例では
src/test.zip) が解凍されているかどうかを確認します。詳細については、「OSS コンソールを使用してファイルを照会する」をご参照ください。
テストが終了したら、不要な料金を避けるために、不要になったアプリケーションとそれに関連するリソースを削除してください。
関連ドキュメント
OSS トリガーの詳細については、「OSS トリガーの概要」をご参照ください。
Function Compute を使用して OSS からファイルをパッケージ化してダウンロードするには、「複数のファイルを ZIP パッケージとしてダウンロード」をご参照ください。