CDN:スクリプトを使用したコンテンツの更新とプリフェッチ

仕組み

このスクリプトは、3 つの段階で URL を処理します。

1. バッチへの分割：スクリプトは URL ファイルを読み取り、URL を最大 20 個 (スクリプト内の gop 変数で制御) のバッチにグループ化します。

2. 送信とポーリング：各バッチに対して、スクリプトは更新またはプリフェッチリクエストを送信し、10 秒ごとにタスクのステータスを確認し、バッチごとに最大 50 秒間待機します。

3. 順次処理：バッチが完了すると、スクリプトは自動的に次のバッチに進みます。

このスクリプトを使用する状況

このスクリプトは、次のような場合に使用します。

• 更新またはプリフェッチする URL の数が多く、手動で送信すると時間がかかりすぎる場合。

• カスタムの統合コードを記述せずに、バッチ送信を自動化したい場合。

• タスクが完了したかどうかを手動で確認する代わりに、自動でステータスをポーリングする必要がある場合。

クォータと制限

Python のバージョン：このスクリプトには Python 3.x が必要です。バージョンを確認するには、次のように確認します。

python --version

または

python3 --version

1 日あたりのクォータ：1 日あたりの更新およびプリフェッチのクォータは、スクリプトによるものを含め、すべての送信に適用されます。大規模なバッチを実行する前に、Alibaba Cloud CDN コンソールの [更新およびプリフェッチ] で残りのクォータを確認してください。スクリプトはタスクを送信する前に利用可能なクォータを自動的に確認し、クォータが不十分な場合はエラーで終了します。

前提条件

開始する前に、以下が準備されていることを確認してください。

• Python 3.x がインストールされていること

• Resource Access Management (RAM) ユーザーが設定された Alibaba Cloud アカウント (詳細については、「AccessKey ペアの作成」をご参照ください)

• RAM ユーザーに AliyunCDNFullAccess システムポリシー、または同等の CDN 権限を持つカスタムポリシーが付与されていること (詳細については、「カスタムポリシーの作成」をご参照ください)

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

Alibaba Cloud CDN の Python 用ソフトウェア開発キット (SDK) をインストールします。

pip install alibabacloud_cdn20180510

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

プレーンテキストファイル (例：urllist.txt) を作成し、1 行に 1 つの URL を記述します。各 URL は http:// または https:// で始まる必要があります。

http://example.com/file1.jpg
http://example.com/file2.jpg
http://example.com/file3.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 シークレット、ファイルパスなどの情報を格納するためのグローバル変数クラスを定義します。
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 シークレットを設定します。
    @staticmethod
    def set_sk():
        Envariable.SK = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET')

    # AccessKey シークレットを取得します。
    @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 シークレットに設定します。
                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 x 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, '-o' options in 'File' or 'Directory'\n")
                if Envariable.get_task_type() == 'push':
                    sys.exit("\n[error]: -o パラメータはタスクタイプ 'push' と一緒に使用できません。\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("\n スクリプトオプションの説明: \
                    \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:])

python Refresh.py -h

スクリプトオプションの説明:
      -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: AccessKey の環境変数への設定

スクリプトは、環境変数 ALIBABA_CLOUD_ACCESS_KEY_ID および ALIBABA_CLOUD_ACCESS_KEY_SECRET から認証情報を読み取ります。設定手順については、「Linux、macOS、Windows で環境変数を設定する」をご参照ください。

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

ターミナル (コマンドプロンプト、PowerShell、またはターミナル) を開き、次を実行します。

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

<PathToUrlFile> を URL ファイルのパスに、<TaskType> を clear (更新) または push (プリフェッチ) に置き換えます。

python Refresh.py -r urllist.txt -t clear

python Refresh.py -r D:\example\filename\urllist.txt -t clear

{'RequestId': 'C1686DCA-F3B5-5575-ADD1-05F96617D770', 'RefreshTaskId': '18392588710'}
[18392588710] is doing... ...

python Refresh.py -r urllist.txt -t push

python Refresh.py -r D:\example\filename\urllist.txt -t push

{'RequestId': 'C1686DCA-F3B5-5575-ADD1-05F96617D771', 'PushTaskId': '18392588711'}
[18392588711] is doing... ...