Function Compute:関数計算を使用して OSS にアップロードされた ZIP ファイルを自動的に解凍する

使用上の注意

• 文字化けや解凍の中断を避けるため、ファイルやフォルダの命名には UTF-8 または GB2312 エンコーディングを使用することをお勧めします。

• アーカイブファイルまたはコールドアーカイブファイルの場合は、解凍する前にファイルをリストアする必要があります。

• ZIP パッケージの解凍に許容される最大時間は 2 時間です。この制限時間を超えた解凍タスクは失敗します。

• 解凍の失敗を避けるため、ZIP パッケージ内の単一ファイルのサイズは 1 GB 未満にしておくことをお勧めします。

• 関数のタイムアウトは 2 時間以上 24 時間以下に設定することをお勧めします。

• 関数は、関連付けられている OSS バケットと同じリージョンに作成することをお勧めします。

前提条件

• 関数計算がアクティブ化されていること。詳細については、「前提条件」をご参照ください。

• OSS がアクティブ化されており、バケットが作成されていること。詳細については、「OSS コンソールを使用して開始する」をご参照ください。

手順

ステップ 1: 関数を作成する

1. 関数計算コンソール にログインします。左側のナビゲーションウィンドウで、[関数] をクリックします。

2. 上部のナビゲーションバーで、リージョンを選択します。[関数] ページで、[関数を作成] をクリックします。

3. [関数を作成] ページで、関数作成の方法を選択し、必要な項目を設定して、[作成] をクリックします。

   主要な設定項目の概要を以下に示します。その他の項目の詳細については、「関数を作成する」をご参照ください。

   • ランタイム: 関数のランタイムとして Python 3.10 を選択します。

   • 関数のロール: 既存のロールを選択するか、新しいロールを作成して、関数が OSS バケットにアクセスするために必要な権限を持っていることを確認します。このロールには AliyunOSSFullAccess ポリシーをアタッチできます。

4. 作成した関数の [関数の詳細] ページで、[コード] タブをクリックします。[コード] タブで、コードファイルを作成し、コードエディタでコードを記述して、[デプロイ] をクリックします。

   サンプルコードは以下のとおりです。

   • index.py ファイル

     サンプルコードを表示するにはクリックしてください

     # -*- coding: utf-8 -*-
     '''
     ステートメント:
     この関数は、ファイルとフォルダに次のように名前を付け、エンコードします。
     1. MAC/Linux の場合、デフォルトで UTF-8 エンコーディングが使用されます。
     2. Windows の場合、デフォルトで GB2312 または UTF-8 エンコーディングが使用されます。
     
     その他のエンコーディングについては、chardet ライブラリを使用してエンコーディング検出が実行されます。
     ただし、100% の精度は保証されません。
     この関数は必要な場合にのみ書き直し、デバッグに合格することを確認してください。
     '''
     
     import helper
     import oss2
     import json
     import os
     import time
     import logging
     import chardet
     
     """
     source/ プレフィックスオブジェクトが OSS バケットに配置されると、オブジェクトが解凍され、processed/ プレフィックスとしてバケットに格納されることが期待されます。
     たとえば、source/a.zip は processed/a/... として処理されます。
     「Source /」、「processed/」は必要に応じて変更できます。
     """
     # 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('現在の関数 [%s] の実行時間は %.2f です' %
                         (func.__name__, time.time() - local_time))
             return ret
         return wrapper
     
     
     def get_zipfile_name(origin_name):  # 中国語の文字化けの問題を解決する
         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):
         """
         OSS からのオブジェクトは自動的に解凍されます。
         param: event:   OSS イベント json 文字列。OSS オブジェクト URI などの情報が含まれます。
     
         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 object_name == "":
                 raise RuntimeError('{} は無効なシンボリックリンクファイルです'.format(
                     evt['oss']['object']['key']))
     
         file_type = os.path.splitext(object_name)[1]
     
         if file_type != ".zip":
             raise RuntimeError('{} filetype is not zip'.format(object_name))
     
         LOGGER.info("zip ファイル = {} の解凍を開始します".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)
         # ファイルのようなオブジェクト
         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 ファイルの読み書き。
     
     UTF-8 への XXX 参照については、さらに調査が必要です。
     """
     # ... (rest of the ossZipfile.py code remains in English)
     

ステップ 2: OSS トリガーを作成する

1. [関数の詳細] ページで、[構成] タブを選択し、左側のナビゲーションウィンドウから [トリガー] タブを選択します。

2. [トリガー] ページで、[トリガーを作成] をクリックします。[トリガーを作成] パネルで、OSS トリガータイプを選択し、必要に応じて設定を構成して、[OK] をクリックします。

   以下に、いくつかの主要な項目の構成の概要を示します。その他の項目の詳細については、「ネイティブ OSS トリガーを構成する」をご参照ください。

   • バケット名: 作成したバケットの名前を選択します。

   • オブジェクトプレフィックス: この例では、src を使用します。

   • オブジェクトサフィックス: この例では、zip を使用します。

   • トリガーイベント: この例では、oss:ObjectCreated:PutObject, oss:ObjectCreated:PostObject, oss:ObjectCreated:CompleteMultipartUpload, oss:ObjectCreated:PutSymlink を使用します。

   • ロール名: ロールを選択し、関数を呼び出すために必要な権限を持っていることを確認します。AliyunFCFullAccess ポリシーをアタッチしてフルアクセスを許可できます。

トリガーが作成されると、[トリガー] ページで表示できます。

ステップ 3: テストと検証

構成は次の 2 つの方法でテストできます。

方法 1: OSS コンソールから手動でファイルをアップロードする

Object Storage Service コンソール にログインします。ステップ 2: OSS トリガーを作成する で選択したバケットの src ディレクトリに、code.zip などの ZIP ファイルをアップロードします。アップロードが完了すると、関数が自動的にトリガーされ、ZIP ファイルがバケットのルートディレクトリに解凍されます。

方法 2: トリガーのイベントパラメータを構成する

この方法では、構成後に [関数のテスト] をクリックして関数を手動でトリガーする必要があることに注意してください。次に、Object Storage Service コンソール にログインして、ターゲットバケットのファイルリストを確認し、ZIP ファイルが正常に解凍されたことを確認します。詳細な手順は以下のとおりです。

1. 「関数詳細」ページで、[コード] タブをクリックし、[関数をテスト] の横にある アイコンをクリックして、ドロップダウンリストから [テストパラメーターの設定] を選択します。

2. [テストパラメータの構成] パネルで、[イベントテンプレート] を選択し、[イベント名] とイベントコンテンツを入力して、[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****"
               }
           }
       ]
   }

   イベントオブジェクトパラメータの詳細については、「ステップ 2: (オプション) 入力パラメータを構成する」をご参照ください。

   重要

   上記のイベントコンテンツは例として提供されています。必要に応じてパラメータを調整してください。指定されたファイル (この例では 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 は、関数計算コンソール の [概要] ページの [参照] セクションにあります。

   • bucket.name: 関数と同じリージョンに作成したバケットの実際の名前に置き換えます。

   • bucket.ownerIdentity: Alibaba Cloud アカウント ID に置き換えます。

   • object.key: 前述のバケットにアップロードされたファイルの名前に置き換えます。

   • region: 関数を作成したときに選択したリージョンに置き換えます。

   • userIdentity.principalId: Alibaba Cloud アカウント ID に置き換えます。

3. [コード] タブの [関数のテスト] をクリックして、テストを実行します。

   関数が正常に実行された後、Object Storage Service コンソール にログインし、指定されたバケット内のターゲット ZIP ファイル (この例では src/test.zip ファイル) が解凍されているかどうかを確認します。詳細な手順については、「OSS コンソールを使用してオブジェクトを照会する」をご参照ください。

テスト後、このアプリケーションが今のところ不要な場合は、不要な課金を避けるため、アプリケーションと関連リソースを必ず削除してください。

参照

• OSS トリガーでできることの詳細については、「OSS トリガーの概要」をご参照ください。

• 関数計算を使用して OSS からファイルをパッケージ化してダウンロードする方法については、「関数計算を使用して複数のオブジェクトをパッケージとしてダウンロードする」をご参照ください。

• このドキュメントで説明されている OSS 操作の詳細については、「バケット」および「オブジェクト」をご参照ください。