阿里云CDN為您提供刷新預熱自動化腳本，可以幫助您分批進行刷新或預熱任務，對文件或目錄快速進行刷新和預熱，替代手動分批提交的繁瑣操作。本文介紹Python自動化腳本的使用說明，并以Windows系統示例為您說明。
功能簡介
當您指定刷新或預熱URL列表文件后，腳本按照指定的并發刷新或預熱數量對URL列表文件進行切割，分批進行刷新或預熱。任務運行后會自動進行判斷，等待上一個任務完成，腳本自動進行下一個刷新或預熱任務的操作。具體的操作邏輯如下：
分批處理：假設您的URL列表中有100個URL，同時您設定了每批次最多處理10個URL，那么腳本會將URL列表切割成10個小批次，每個批次包含10個URL。而如果設定的并發數量更大或更小，批次的大小會相應調整。例如設定的并發數量是20，那么腳本會將100個URL分成5個批次，每個批次包含20個URL。
按批次運行任務：腳本在啟動時會按照批次依次提交刷新或預熱請求。每個批次的任務是并發執行的。
等待完成后進行下一批任務：當一個批次的刷新或預熱任務完成后，腳本會繼續執行下一個批次的任務。這個判斷和操作是自動進行的，不需要人工干預。
適用場景
如果您有以下情況，建議您使用刷新預熱自動化腳本：
無開發人員，需手動提交刷新預熱任務，運維成本高。
刷新或預熱URL過多，分批提交導致刷新或預熱效率低。
需要人工或程序判斷刷新預熱任務是否正常進行，費時費力。
使用限制
請確保操作系統的Python版本為3.x版本。您可以通過在命令行輸入python --version或python3 --version來檢查Python版本是否符合要求。
前提條件
由于阿里云賬號（主賬號）擁有資源的所有權限，其AccessKey一旦泄露風險巨大，所以建議您使用RAM用戶的AccessKey。獲取方法請參見創建AccessKey。
給RAM用戶授予操作域名資源的權限。本示例選擇AliyunDomainFullAccess系統策略。
在環境變量中配置AccessKey，具體操作步驟請參見在Linux、macOS和Windows系統配置環境變量。
步驟一：安裝依賴
執行以下命令安裝Python CDN SDK模塊包，目前使用版本為v20180510。
pip install aliyun-python-sdk-cdn執行以下命令安裝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:])代碼執行流程
按
gop指定的數量（100個）將文件拆分成多個批次。順序處理每個批次的URL。
等待當前批次任務完成后，再執行下一個批次。
說明
您可以通過調整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... ...
文檔內容是否對您有幫助？