スクリプトを使用したファイルまたはディレクトリの更新とプリフェッチ - CDN

Alibaba Cloud CDN は、更新またはプリフェッチタスクをバッチで実行するのに役立つスクリプトを提供します。このスクリプトを使用すると、ファイルやディレクトリを迅速に更新およびプリフェッチできます。これにより、手動でバッチタスクを送信するという面倒なプロセスが自動化されます。このトピックでは、Python スクリプトの使用方法について説明し、Windows オペレーティングシステムを例として使用します。

機能

更新またはプリフェッチする URL のリストを含むファイルを指定すると、スクリプトは指定された同時タスク数に基づいて URL をバッチに分割します。バッチが開始されると、スクリプトは自動的にそのステータスを確認します。スクリプトは、現在のバッチが完了した後にのみ次のバッチを開始します。ロジックは次のとおりです。

バッチ処理: URL リストに 100 個の URL が含まれており、バッチあたりの最大 URL 数を 10 に設定したとします。スクリプトは URL リストを 10 個のミニバッチに分割し、各バッチには 10 個の URL が含まれます。バッチサイズは同時実行数の値に基づいて調整されます。たとえば、同時実行数を 20 に設定した場合、スクリプトは 100 個の URL を 5 つのバッチに分割し、各バッチには 20 個の URL が含まれます。
バッチでのタスクの実行: スクリプトが開始されると、各バッチの更新またはプリフェッチリクエストを順次送信します。各バッチのタスクは同時に実行されます。
次のバッチを開始する前にバッチの完了を待機: 1 つのバッチの更新またはプリフェッチタスクが完了すると、スクリプトは次のバッチに進みます。このプロセスは自動であり、手動での介入は必要ありません。

シナリオ

次のシナリオでスクリプトを使用できます。

開発者リソースの不足により、更新およびプリフェッチタスクが手動で送信され、運用およびメンテナンス (O&M) コストが高くなります。
更新またはプリフェッチする URL の数が多く、手動でバッチ送信するのは非効率です。
更新およびプリフェッチタスクが完了したかどうかを手動またはプログラムで確認する必要があり、時間がかかりリソースを大量に消費します。

制限事項

お使いのオペレーティングシステムに Python 3.x がインストールされていることを確認してください。python --version または python3 --version コマンドを実行して、Python のバージョンを確認できます。

前提条件

Alibaba Cloud アカウントはリソースに対する完全な権限を持っているため、AccessKey ペアが漏洩すると高いセキュリティリスクが生じます。代わりに、Resource Access Management (RAM) ユーザーの AccessKey ペアを作成して使用することをお勧めします。詳細については、「AccessKey ペアの作成」をご参照ください。
RAM ユーザーにドメイン名リソースの管理に必要な権限を付与します。この例では、AliyunDomainFullAccess システムポリシーが使用されます。
1. システムポリシーを使用します。
  - AliyunCDNFullAccess: CDN リソースへのフルアクセスを許可します。
2. カスタムポリシーを使用します。
  カスタムポリシーの作成方法の詳細については、「カスタムポリシーの作成」をご参照ください。

ステップ 1: 依存関係のインストール

次のコマンドを実行して、Python 用の Alibaba Cloud CDN ソフトウェア開発キット (SDK) をインストールします。この例では、バージョン v20180510 を使用します。

pip install alibabacloud_cdn20180510

ステップ 2: URL ファイルの準備

更新またはプリフェッチする URL のリストを含むファイル (例: urllist.txt) を作成します。1 行に 1 つの URL を入力します。各 URL が http:// または https:// で始まり、有効な URL フォーマットであることを確認してください。次のコードは例です。

http://example.com/file1.jpg
http://example.com/file2.jpg
http://example.com/file3.jpg
...
http://example.com/fileN.jpg

ステップ 3: スクリプトの作成

次のコードをスクリプトとして保存し、Refresh.py という名前を付けます。スクリプトにはカスタム名を使用できます。このファイル名は例として使用されます。

サンプルスクリプト

#!/usr/bin/env python3
# coding=utf-8
# __author__ = 'aliyun.cdn'
# __date__ = '2025-08-15'

# SDK インストールコマンド: pip install alibabacloud_cdn20180510

'''パッケージの確認'''
# 必要なライブラリをインポートします。
import re, sys, getopt, time, logging, os

try:
    from alibabacloud_cdn20180510.client import Client as Cdn20180510Client
    from alibabacloud_credentials.models import Config as CreConfig
    from alibabacloud_credentials.client import Client as CredentialClient
    from alibabacloud_tea_openapi.models import Config
    from alibabacloud_cdn20180510 import models as cdn_20180510_models
    from alibabacloud_tea_util import models as util_models

# インポート例外をキャッチします。
except ImportError as e:
    sys.exit(f"[error] Please pip install alibabacloud_cdn20180510. Details: {e}")

# ログを初期化します。
logging.basicConfig(level=logging.DEBUG, filename='./RefreshAndPredload.log')

# AccessKey ID、AccessKey Secret、ファイルディレクトリなどの情報を格納するグローバル変数クラスを定義します。
class Envariable(object):
    LISTS = []
    # エンドポイントについては、https://api.aliyun.com/product/Cdn をご参照ください
    ENDPOINT = 'cdn.aliyuncs.com'
    AK = None
    SK = None
    FD = None
    CLI = None
    TASK_TYPE = None
    TASK_AREA = None
    TASK_OTYPE = None

    # AccessKey ID を設定します。
    @staticmethod
    def set_ak():
        Envariable.AK = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID')

    # AccessKey ID を取得します。
    @staticmethod
    def get_ak():
        return Envariable.AK

    # AccessKey Secret を設定します。
    @staticmethod
    def set_sk():
        Envariable.SK = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET')

    # AccessKey Secret を取得します。
    @staticmethod
    def get_sk():
        return Envariable.SK

    # ファイルディレクトリを設定します。
    @staticmethod
    def set_fd(fd):
        Envariable.FD = fd

    # ファイルディレクトリを取得します。
    @staticmethod
    def get_fd():
        return Envariable.FD

    # タスクタイプを設定します。
    @staticmethod
    def set_task_type(task_type):
        Envariable.TASK_TYPE = task_type

    # タスクタイプを取得します。
    @staticmethod
    def get_task_type():
        return Envariable.TASK_TYPE
    
    # タスクエリアを設定します。
    @staticmethod
    def set_task_area(task_area):
        Envariable.TASK_AREA = task_area

    # タスクエリアを取得します。
    @staticmethod
    def get_task_area():
        return Envariable.TASK_AREA

    # タスクオブジェクトタイプを設定します。
    @staticmethod
    def set_task_otype(task_otype):
        Envariable.TASK_OTYPE = task_otype

    # タスクオブジェクトタイプを取得します。
    @staticmethod
    def get_task_otype():
        return Envariable.TASK_OTYPE

    # 新しいクライアントを作成します。
    @staticmethod
    def set_acs_client():
        try:
            # AccessKey ペアを使用して Credentials クライアントを初期化します。
            credentialsConfig = CreConfig(
                # 認証情報タイプ。
                type='access_key',
                # AccessKey ID に設定します。
                access_key_id=Envariable.get_ak(),
                # AccessKey Secret に設定します。
                access_key_secret=Envariable.get_sk(),
            )
            credentialClient = CredentialClient(credentialsConfig)

            cdnConfig = Config(credential=credentialClient)
            # サービスエンドポイントを設定します。
            cdnConfig.endpoint = Envariable.ENDPOINT
            # CDN クライアントを初期化します。
            Envariable.CLI = Cdn20180510Client(cdnConfig)
        except Exception as e:
            logging.error(f"Failed to create client: {e}")
            raise

    # クライアントを取得します。
    @staticmethod
    def get_acs_client():
        return Envariable.CLI


# モジュールレベルの初期化関数。
def initialize_credentials_and_client():
    """モジュールのロード時に AccessKey ペアとクライアントを初期化します。"""
    try:
        # 環境変数から AccessKey ペアを初期化します。
        Envariable.set_ak()
        Envariable.set_sk()
        
        # AccessKey ペアが取得されたかどうかを確認します。
        if not Envariable.get_ak() or not Envariable.get_sk():
            logging.warning("AK or SK not found in environment variables")
            return False
            
        # クライアントを初期化します。
        Envariable.set_acs_client()
        logging.info("Credentials and client initialized successfully")
        return True
    except Exception as e:
        logging.error(f"Failed to initialize credentials and client: {e}")
        return False


# モジュールのロード時に初期化を実行します。
_initialization_success = initialize_credentials_and_client()





class BaseCheck(object):
    def __init__(self):
        self.invalidurl = ''
        self.lines = 0
        self.urllist = Envariable.get_fd()

    # クォータを確認します。
    def printQuota(self):
        try:
            client = Envariable.get_acs_client()
            if not client:
                raise Exception("CDN client not initialized")
            
            # SDK を使用して呼び出しを行います。
            request = cdn_20180510_models.DescribeRefreshQuotaRequest()
            runtime = util_models.RuntimeOptions()
            response = client.describe_refresh_quota_with_options(request, runtime)
            quotaResp = response.body.to_map()
        except Exception as e:
            logging.error(f"\n[error]: initial Cdn20180510Client failed: {e}\n")
            sys.exit(1)

        if Envariable.TASK_TYPE:
            if Envariable.TASK_TYPE == 'push':
                if self.lines > int(quotaResp['PreloadRemain']):
                    sys.exit("\n[error]:PreloadRemain is not enough {0}".format(quotaResp['PreloadRemain']))
                return True
            if Envariable.TASK_TYPE == 'clear':
                if Envariable.get_task_otype() == 'File' and self.lines > int(quotaResp['UrlRemain']):
                    sys.exit("\n[error]:UrlRemain is not enough {0}".format(quotaResp['UrlRemain']))
                elif Envariable.get_task_otype() == 'Directory' and self.lines > int(quotaResp['DirRemain']):
                    sys.exit("\n[error]:DirRemain is not enough {0}".format(quotaResp['DirRemain']))
                else:
                    return True

    # URL フォーマットを検証します。
    def urlFormat(self):
        try:
            with open(self.urllist, "r") as f:
                for line in f.readlines():
                    self.lines += 1
                    if not re.match(r'^((https)|(http))', line):
                        self.invalidurl = line + '\n' + self.invalidurl
                if self.invalidurl != '':
                    sys.exit("\n[error]: URL format is illegal \n{0}".format(self.invalidurl))
                return True
        except FileNotFoundError:
            sys.exit(f"\n[error]: File not found: {self.urllist}\n")
        except Exception as e:
            sys.exit(f"\n[error]: Failed to read file {self.urllist}: {e}\n")

# URL リストを指定されたサイズの複数のバッチに分割するバッチ処理クラス。
class doTask(object):
    @staticmethod
    def urlencode_pl(inputs_str):
        len_str = len(inputs_str)
        if inputs_str == "" or len_str <= 0:
            return ""
        result_end = ""
        for chs in inputs_str:
            if chs.isalnum() or chs in {":", "/", ".", "-", "_", "*"}:
                result_end += chs
            elif chs == ' ':
                result_end += '+'
            else:
                result_end += f'%{ord(chs):02X}'
        return result_end

    # URL をバッチで処理します。
    @staticmethod
    def doProd():
        gop = 20  # バッチあたりの最大 URL 数を定義します。
        mins = 1
        maxs = gop
        current_batch = []  # グローバル変数の代わりにローカル変数を使用します。
        
        try:
            with open(Envariable.get_fd(), "r") as f:
                for line in f.readlines():
                    line = doTask.urlencode_pl(line.strip()) + "\n"
                    current_batch.append(line)
                    if mins >= maxs:
                        yield current_batch
                        current_batch = []
                        mins = 1
                    else:
                        mins += 1
            if current_batch:
                yield current_batch
        except FileNotFoundError:
            sys.exit(f"\n[error]: File not found: {Envariable.get_fd()}\n")
        except Exception as e:
            sys.exit(f"\n[error]: Failed to read file {Envariable.get_fd()}: {e}\n")

    # 更新またはプリフェッチタスクを実行します。
    @staticmethod
    def doRefresh(lists):
        try:
            client = Envariable.get_acs_client()
            if not client:
                raise Exception("CDN client not initialized")

            runtime = util_models.RuntimeOptions()
            taskID = None
            response_data = None

            if Envariable.get_task_type() == 'clear':
                taskID = 'RefreshTaskId'
                request = cdn_20180510_models.RefreshObjectCachesRequest()
                if Envariable.get_task_otype():
                    request.object_type = Envariable.get_task_otype()
                request.object_path = lists
                response = client.refresh_object_caches_with_options(request, runtime)
                response_data = response.body.to_map()
            elif Envariable.get_task_type() == 'push':
                taskID = 'PushTaskId'
                request = cdn_20180510_models.PushObjectCacheRequest()
                if Envariable.get_task_area():
                    request.area = Envariable.get_task_area()
                request.object_path = lists
                response = client.push_object_cache_with_options(request, runtime)
                response_data = response.body.to_map()

            if response_data and taskID:
                print(response_data)

                timeout = 0
                while True:
                    count = 0
                    # SDK を使用してタスクのステータスをクエリします。
                    taskreq = cdn_20180510_models.DescribeRefreshTasksRequest()
                    taskreq.task_id = response_data[taskID]
                    taskresp = client.describe_refresh_tasks_with_options(taskreq, runtime)
                    taskresp_data = taskresp.body.to_map()
                    print(f"[{response_data[taskID]}] is doing... ...")
                    
                    for t in taskresp_data['Tasks']['CDNTask']:
                        if t['Status'] != 'Complete':
                            count += 1
                    if count == 0:
                        logging.info(f"[{response_data[taskID]}] is finish")
                        break
                    elif timeout > 5:  # 最大 50 秒 (5 × 10 秒) 待機します。
                        logging.info(f"[{response_data[taskID]}] timeout after 50 seconds")
                        break
                    else:
                        timeout += 1
                        time.sleep(10)  # 10 秒ごとにステータスを確認します。
                        continue
        except Exception as e:
            logging.error(f"\n[error]: {e}")
            sys.exit(1)


class Refresh(object):
    def main(self, argv):
        if len(argv) < 1:
            sys.exit(f"\n[usage]: {sys.argv[0]} -h ")
        try:
            opts, args = getopt.getopt(argv, "hr:t:a:o:")
        except getopt.GetoptError as e:
            sys.exit(f"\n[usage]: {sys.argv[0]} -h ")

        for opt, arg in opts:
            if opt == '-h':
                self.help()
                sys.exit()
            elif opt == '-r':
                Envariable.set_fd(arg)
            elif opt == '-t':
                Envariable.set_task_type(arg)
            elif opt == '-a':
                Envariable.set_task_area(arg)
            elif opt == '-o':
                Envariable.set_task_otype(arg)
            else:
                sys.exit(f"\n[usage]: {sys.argv[0]} -h ")
        
        # ヘルプコマンドでない場合にのみ初期化ステータスを確認します。
        if not _initialization_success:
            sys.exit("\n[error]: Failed to initialize credentials and client. Please check environment variables.\n")

        try:
            if not (Envariable.get_ak() and Envariable.get_sk() and Envariable.get_fd() and Envariable.get_task_type()):
                sys.exit("\n[error]: Must set environment variables ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET, and parameters '-r', '-t'\n")
            if Envariable.get_task_type() not in {"push", "clear"}:
                sys.exit("\n[error]: taskType Error, '-t' option in 'push' or 'clear'\n")
            if Envariable.get_task_area() and Envariable.get_task_otype():
                sys.exit("\n[error]: -a and -o cannot exist at same time\n")
            if Envariable.get_task_area():
                if Envariable.get_task_area() not in {"domestic", "overseas"}:
                    sys.exit("\n[error]: Area value Error, '-a' option in 'domestic' or 'overseas'\n")
            if Envariable.get_task_otype():
                if Envariable.get_task_otype() not in {"File", "Directory"}:
                    sys.exit("\n[error]: ObjectType value Error, '-a' options in 'File' or 'Directory'\n")
                if Envariable.get_task_type() == 'push':
                    sys.exit("\n[error]: -t must be clear and 'push' -a use together\n")
        except Exception as e:
            logging.error(f"\n[error]: Parameter {e} error\n")
            sys.exit(1)

        handler = BaseCheck()
        if handler.urlFormat() and handler.printQuota():
            for g in doTask.doProd():
                doTask.doRefresh(''.join(g))
                time.sleep(1)

    def help(self):
        print("\nscript options explain: \
                    \n\t -r <filename>                   # ファイルパスとファイル名。スクリプトの実行後、ファイルから URL を読み取ります。各行には 1 つの URL を含める必要があります。特殊文字を含む URL は URL エンコードする必要があります。各 URL は http または https で始まる必要があります。 \
                    \n\t -t <taskType>                   # タスクタイプ。`clear`: 更新。`push`: プリフェッチ。 \
                    \n\t -a [String,<domestic|overseas>] # オプション。プリフェッチの範囲。このパラメーターを設定しない場合、リソースはグローバルにプリフェッチされます。\
                    \n\t    domestic                     # 中国本土のみ。 \
                    \n\t    overseas                     # グローバル (中国本土を除く)。 \
                    \n\t -o [String,<File|Directory>]    # オプション。更新するコンテンツのタイプ。 \
                    \n\t    File                         # ファイル (デフォルト)。 \
                    \n\t    Directory                    # ディレクトリ。")


if __name__ == '__main__':
    fun = Refresh()
    fun.main(sys.argv[1:])

コード実行フロー

gop パラメーター (100) の値に基づいてファイルをバッチに分割します。
各バッチの URL を順次処理します。
現在のバッチが完了した後、次のバッチに進みます。

説明

各バッチのサイズは、gop 変数の値を変更することで変更できます。

ヘルプ情報の表示

スクリプトを作成した後、コマンドプロンプト、PowerShell、またはターミナルなどのコマンドラインインターフェイス (CLI) で python $script -h を実行して、スクリプトのヘルプ情報を表示できます。

説明

$script は Python スクリプトのファイル名のプレースホルダーです。たとえば、スクリプト名が Refresh.py の場合、python Refresh.py -h を実行できます。

コマンドプロンプト、PowerShell、またはターミナルなどの CLI で次のコマンドを実行します。このコマンドは、スクリプトとそのパラメーターに関するヘルプ情報を表示します。

python Refresh.py -h

次の出力が返されます。

script options explain:
              -r <filename>                    //ファイルパスとファイル名。スクリプトの実行後、ファイルから URL を読み取ります。各行には 1 つの URL を含める必要があります。特殊文字を含む URL は URL エンコードする必要があります。各 URL は http または https で始まる必要があります。
              -t <taskType>                    //タスクタイプ。`clear`: 更新。`push`: プリフェッチ。
              -a [String,<domestic|overseas>   //オプション。プリフェッチの範囲。このパラメーターを設定しない場合、リソースはグローバルにプリフェッチされます。            
                   domestic                    //中国本土のみ。             
                   overseas                    //グローバル (中国本土を除く)。             
              -o [String,<File|Directory>]     //オプション。更新するコンテンツのタイプ。             
                   File                        //ファイル (デフォルト)。             
                   Directory                   //ディレクトリ。

ステップ 4: 環境変数に Alibaba Cloud AccessKey を設定する

AccessKey を使用して API を呼び出す場合、コードに AccessKey を含めないことをお勧めします。ステップ 3 で提供されるスクリプトは、環境変数から AccessKey ID と AccessKey Secret を読み取ります。コードを実行する前に Alibaba Cloud AccessKey を設定する方法の詳細については、「Linux、macOS、および Windows で環境変数を設定する」をご参照ください。

重要

Linux および macOS では、export コマンドで設定された Alibaba Cloud AccessKey は現在のセッションでのみ有効です。セッションが終了すると、AccessKey は無効になります。

AccessKey を永続的にするには、export コマンドをオペレーティングシステムの起動設定ファイルに追加します。

ステップ 5: スクリプトの実行

コマンドプロンプト、PowerShell、またはターミナルなどの CLI で次のコマンドを実行します。

python Refresh.py -r <PathToUrlFile> -t <TaskType>

説明

<PathToUrlFile>: URL のリストを含むファイルへのパス (例: urllist.txt)。

<TaskType>: タスクのタイプ。有効な値: clear (更新) および push (プリフェッチ)。

リソース更新コマンド

URL ファイル名が urllist.txt で、ファイルと Refresh.py スクリプトが同じディレクトリにあり、タスクタイプが clear (更新) の場合、コマンドプロンプト、PowerShell、またはターミナルなどの CLI で次のコマンドを実行します。
```
python Refresh.py -r urllist.txt -t clear
```
ファイルが別のディレクトリ (例: D:\example\filename\urllist.txt) にある場合、コマンドプロンプト、PowerShell、またはターミナルなどの CLI で次のコマンドを実行します。
```
python Refresh.py -r D:\example\filename\urllist.txt -t clear
```

以下はサンプル出力です。

python Refresh.py -r urllist.txt -t clear
{'RequestId': 'C1686DCA-F3B5-5575-ADD1-05F96617D770', 'RefreshTaskId': '18392588710'}
[18392588710] is doing... ...

エラーメッセージ Failed to initialize credentials and client. Please check environment variables. が返された場合は、ステップ 4 の説明に従って環境変数に Alibaba Cloud AccessKey を設定し、同じターミナルウィンドウでコマンドを実行します。

リソースプリフェッチコマンド

URL ファイル名が urllist.txt で、ファイルと Refresh.py スクリプトが同じディレクトリにあり、タスクタイプが push (プリフェッチ) の場合、コマンドプロンプト、PowerShell、またはターミナルなどの CLI で次のコマンドを実行します。
```
python Refresh.py -r urllist.txt -t push
```
ファイルが別のディレクトリ (例: D:\example\filename\urllist.txt) にある場合、コマンドプロンプト、PowerShell、またはターミナルなどの CLI で次のコマンドを実行します。
```
python Refresh.py -r D:\example\filename\urllist.txt -t push
```

以下はサンプル出力です。

python Refresh.py -r urllist.txt -t push
{'RequestId': 'C1686DCA-F3B5-5575-ADD1-05F96617D771', 'RefreshTaskId': '18392588711'}
[18392588710] is doing... ...