全部產品
Search
文件中心

CDN:使用自動化指令碼重新整理和預熱

更新時間:Sep 13, 2024

阿里雲CDN為您提供重新整理預熱自動化指令碼,可以協助您分批進行重新整理或預熱任務,對檔案或目錄快速進行重新整理和預熱,替代手動分批提交的繁瑣操作。本文介紹Python自動化指令碼的使用說明,並以Windows系統樣本為您說明。

功能簡介

當您指定重新整理或預熱URL列表檔案後,指令碼按照指定的並發重新整理或預熱數量對URL列表檔案進行切割,分批進行重新整理或預熱。任務運行後會自動進行判斷,等待上一個任務完成,指令碼自動進行下一個重新整理或預熱任務的操作。具體的操作邏輯如下:

  1. 分批處理:假設你的URL列表中有100個URL,同時你設定了每批次最多處理10個URL,那麼指令碼會將URL列表切割成10個小批次,每個批次包含10個URL。而如果設定的並發數量更大或更小,批次的大小會相應調整。例如設定的並發數量是20,那麼指令碼會將100個URL分成5個批次,每個批次包含20個URL。

  2. 按批次運行任務:指令碼在啟動時會按照批次依次提交重新整理或預熱請求。每個批次的任務是並發執行的。

  3. 等待完成後進行下一批任務:當一個批次的重新整理或預熱任務完成後,指令碼會繼續執行下一個批次的任務。這個判斷和操作是自動進行的,不需要人工幹預。

適用情境

如果您有以下情況,建議您使用重新整理預熱自動化指令碼:

  • 無開發人員,需手動提交重新整理預熱任務,營運成本高。

  • 重新整理或預熱URL過多,分批提交導致重新整理或預熱效率低。

  • 需要人工或程式判斷重新整理預熱任務是否正常進行,費時費力。

使用限制

請確保作業系統的Python版本為3.x版本。您可以通過在命令列輸入python --versionpython3 --version來檢查Python版本是否符合要求。

前提條件

  1. 由於阿里雲帳號(主帳號)擁有資源的所有許可權,其AccessKey一旦泄露風險巨大,所以建議您使用RAM使用者的AccessKey。擷取方法請參見建立AccessKey

  2. 給RAM使用者授予操作網域名稱資源的許可權。本樣本選擇AliyunDomainFullAccess系統策略。

    1. 使用系統策略。

      • AliyunCDNFullAccess:管理CDN資源的許可權。

    2. 使用自訂許可權。

      關於如何建立自訂許可權,請參見建立自訂權限原則

  3. 在環境變數中配置AccessKey,具體操作步驟請參見在Linux、macOS和Windows系統配置環境變數

步驟一:安裝依賴

  1. 執行以下命令安裝Python CDN SDK模組包,目前使用版本為v20180510。

    pip install aliyun-python-sdk-cdn
  2. 執行以下命令安裝Python阿里雲核心包,目前使用版本為2.6.0。

    pip install aliyun-python-sdk-core

步驟二:準備URL檔案

建立一個包含需要重新整理或預熱的URL列表的檔案。例如:urllist.txt,每行一個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

步驟三:建立指令碼

將如下代碼儲存為自動化指令碼,並命名為Refresh.py。您可以自訂指令碼名稱,此處為舉例說明。

指令碼範例程式碼

#!/usr/bin/env python3
# coding=utf-8
# __author__ = 'aliyun.cdn'
# __date__ = '2021-04-23'

'''Check Package'''
try:
    # 匯入所需庫
    import os, re, sys, getopt, time, json, logging
    from aliyunsdkcore.client import AcsClient
    from aliyunsdkcore.acs_exception.exceptions import ClientException, ServerException
    from aliyunsdkcdn.request.v20180510.RefreshObjectCachesRequest import RefreshObjectCachesRequest
    from aliyunsdkcdn.request.v20180510.PushObjectCacheRequest import PushObjectCacheRequest
    from aliyunsdkcdn.request.v20180510.DescribeRefreshTasksRequest import DescribeRefreshTasksRequest
    from aliyunsdkcdn.request.v20180510.DescribeRefreshQuotaRequest import DescribeRefreshQuotaRequest

# 捕獲匯入異常
except ImportError as e:
    sys.exit("[error] Please pip install aliyun-python-sdk-cdn and aliyun-python-sdk-core. Details: {e}")

# 初始化日誌記錄
logging.basicConfig(level=logging.DEBUG, filename='./RefreshAndPredload.log')

# 定義全域變數類,儲存AK、SK、FD等資訊
class Envariable(object):
    LISTS = []
    REGION = 'cn-zhangzhou'
    AK = None
    SK = None
    FD = None
    CLI = None
    TASK_TYPE = None
    TASK_AREA = None
    TASK_OTYPE = None

    # 設定AK
    @staticmethod
    def set_ak(ak):
        Envariable.AK = ak

    # 擷取AK
    @staticmethod
    def get_ak():
        return Envariable.AK

    # 設定SK
    @staticmethod
    def set_sk(sk):
        Envariable.SK = sk

    # 擷取SK
    @staticmethod
    def get_sk():
        return Envariable.SK

    # 設定FD
    @staticmethod
    def set_fd(fd):
        Envariable.FD = 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

    # 建立AcsClient
    @staticmethod
    def set_acs_client():
        Envariable.CLI = AcsClient(Envariable.get_ak(), Envariable.get_sk(), Envariable.REGION)

    # 擷取AcsClient
    @staticmethod
    def get_acs_client():
        return Envariable.CLI


class InitHandler(object):
    def __init__(self, ak, sk, region):
        try:
            self.client = AcsClient(ak, sk, region)
        except Exception:
            logging.info("[error]: initial AcsClient failed")
            exit(1)


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

    # 檢查配額
    def printQuota(self):
        try:
            if Envariable.get_acs_client():
                client = Envariable.get_acs_client()
            else:
                Envariable.set_acs_client()
                client = Envariable.get_acs_client()
            quotas = DescribeRefreshQuotaRequest()
            quotaResp = json.loads(Envariable.get_acs_client().do_action_with_exception(quotas))
        except Exception as e:
            logging.info("\n[error]: initial AcsClient failed\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):
        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

# 批量處理類,將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
        with open(Envariable.get_fd(), "r") as f:
            for line in f.readlines():
                line = doTask.urlencode_pl(line.strip()) + "\n"
                Envariable.LISTS.append(line)
                if mins >= maxs:
                    yield Envariable.LISTS
                    Envariable.LISTS = []
                    mins = 1
                else:
                    mins += 1
        if Envariable.LISTS:
            yield Envariable.LISTS

    # 執行重新整理或預熱任務
    @staticmethod
    def doRefresh(lists):
        try:
            if Envariable.get_acs_client():
                client = Envariable.get_acs_client()
            else:
                Envariable.set_acs_client()
                client = Envariable.get_acs_client()

            if Envariable.get_task_type() == 'clear':
                taskID = 'RefreshTaskId'
                request = RefreshObjectCachesRequest()
                if Envariable.get_task_otype():
                    request.set_ObjectType(Envariable.get_task_otype())
            elif Envariable.get_task_type() == 'push':
                taskID = 'PushTaskId'
                request = PushObjectCacheRequest()
                if Envariable.get_task_area():
                    request.set_Area(Envariable.get_task_area())

            taskreq = DescribeRefreshTasksRequest()
            request.set_accept_format('json')
            request.set_ObjectPath(lists)
            response = json.loads(client.do_action_with_exception(request))
            print(response)

            timeout = 0
            while True:
                count = 0
                taskreq.set_accept_format('json')
                taskreq.set_TaskId(response[taskID])
                taskresp = json.loads(client.do_action_with_exception(taskreq))
                print(f"[{response[taskID]}] is doing... ...")
                for t in taskresp['Tasks']['CDNTask']:
                    if t['Status'] != 'Complete':
                        count += 1
                if count == 0:
                    logging.info(f"[{response[taskID]}] is finish")
                    break
                elif timeout > 5:
                    logging.info(f"[{response[taskID]}] timeout")
                    break
                else:
                    timeout += 1
                    time.sleep(5)
                    continue
        except Exception as e:
            logging.info(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, "hi:k:n:r: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 == '-i':
                Envariable.set_ak(arg)
            elif opt == '-k':
                Envariable.set_sk(arg)
            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 ")

        try:
            if not (Envariable.get_ak() and Envariable.get_sk() and Envariable.get_fd() and Envariable.get_task_type()):
                sys.exit("\n[error]: Must be by parameter '-i', '-k', '-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.info(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 -i <AccessKey>                  訪問阿里雲憑證,存取控制台上可以獲得; \
                    \n\t -k <AccessKeySecret>            訪問阿里雲密鑰,存取控制台上可以獲得; \
                    \n\t -r <filename>                   filename指“檔案所在的路徑+檔案名稱”,自動化指令碼運行後將會讀取檔案內記錄的URL;檔案內的URL記錄方式為每行一條URL,有特殊字元先做URLencode,以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:])

代碼執行流程

  1. gop指定的數量(100個)將檔案拆分成多個批次。

  2. 順序處理每個批次的URL。

  3. 等待當前批次任務完成後,再執行下一個批次。

說明

您可以通過調整gop變數調整每個批次的大小。

查看協助資訊

指令碼建立完成後,您可以在命令列(CMD,PowerShell或終端)中運行python $script -h,用於請求並顯示Python指令碼的命令列協助資訊。

說明

$script通常是指一個變數,這個變數是Python指令碼的檔案名稱。例如,如果您的指令檔名是Refresh.py,您可以運行python Refresh.py -h

在命令列(CMD,PowerShell或終端)運行以下命令,指令碼會顯示協助資訊,告訴您如何正確使用該指令碼及其所有參數。

python Refresh.py -h

運行命令後可能會輸出以下內容:

script options explain:
              -i <AccessKey>                   //訪問阿里雲憑證,存取控制台獲得;
              -k <AccessKeySecret>             //訪問阿里雲密匙,存取控制台獲得;
              -r <filename>                    //filename指“檔案所在的路徑+檔案名稱”,自動化指令碼運行後將會讀取檔案內記錄的URL;檔案內的URL記錄方式為每行一條URL,有特殊字元先做URLencode,以http或https開頭;
              -t <taskType>                    //任務類型,clear:重新整理,push:預熱;
              -a [String,<domestic|overseas>   //可選項,預熱範圍,不傳預設是全球;            
                   domestic                    //僅中國內地;             
                   overseas                    //全球(不包含中國內地);             
              -o [String,<File|Directory>]     //可選項,重新整理的類型;             
                   File                        //檔案重新整理(預設值);             
                   Directory                   //目錄重新整理;

步驟四:運行指令碼

在命令列(CMD,PowerShell或終端)使用以下命令列運行指令碼:

python Refresh.py -i <YourAccessKey> -k <YourAccessKeySecret> -r <PathToUrlFile> -t <TaskType>
    說明

    <YourAccessKey>:您的阿里雲AccessKey ID。

    <YourAccessKeySecret>:您的阿里雲AccessKey Secret。

    <PathToUrlFile>:包含URL列表的檔案路徑,如urllist.txt

    <TaskType>:任務類型,clear(重新整理)或push(預熱)。

樣本命令

  • 假設AccessKey為yourAccessKey,AccessKeySecret為yourAccessKeySecret,URL檔案為urllist.txt,且檔案和Refresh.py指令碼在相同目錄下,任務類型為clear(重新整理),在命令列(CMD,PowerShell或終端)執行以下命令。

    python Refresh.py -i yourAccessKey -k yourAccessKeySecret -r urllist.txt -t clear
  • 如果檔案在不同目錄,例如D:\example\filename\urllist.txt,在命令列(CMD,PowerShell或終端)執行以下命令。

    python Refresh.py -i yourAccessKey -k yourAccessKeySecret -r D:\example\filename\urllist.txt -t clear

運行樣本如下:

python Refresh.py -i yourAccessKey -k yourAccessKeySecret -r urllist.txt -t clear
{'RequestId': 'C1686DCA-F3B5-5575-ADD1-05F96617D770', 'RefreshTaskId': '18392588710'}
[18392588710] is doing... ...
{'RequestId': '5BEAD371-9D82-5DA5-BE60-58EC2C915E82', 'RefreshTaskId': '18392588804'}
[18392588804] is doing... ...
{'RequestId': 'BD0B3D22-66CF-5B1D-A995-D912A5EA8E2F', 'RefreshTaskId': '18392588804'}
[18392588804] is doing... ...
[18392588804] is doing... ...
[18392588804] is doing... ...