#!/usr/bin/env python3
# coding=utf-8
# __author__ = 'aliyun.cdn'
# __date__ = '2025-08-15'
# SDK安裝命令 pip install alibabacloud_cdn20180510
'''Check Package'''
# 匯入所需庫
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')
# 定義全域變數類，儲存AK、SK、FD等資訊
class Envariable(object):
LISTS = []
# Endpoint 請參考 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
# 設定AK
@staticmethod
def set_ak():
Envariable.AK = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID')
# 擷取AK
@staticmethod
def get_ak():
return Envariable.AK
# 設定SK
@staticmethod
def set_sk():
Envariable.SK = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET')
# 擷取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
# 建立新的Client
@staticmethod
def set_acs_client():
try:
# 使用AK 初始化Credentials Client。
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)
# 配置雲產品服務接入地址（endpoint）。
cdnConfig.endpoint = Envariable.ENDPOINT
# 初始化cdn Client。
Envariable.CLI = Cdn20180510Client(cdnConfig)
except Exception as e:
logging.error(f"Failed to create client: {e}")
raise
# 擷取Client
@staticmethod
def get_acs_client():
return Envariable.CLI
# 模組層級別的初始化函數
def initialize_credentials_and_client():
"""在模組載入時初始化AK、SK和Client"""
try:
# 從環境變數初始化AK和SK
Envariable.set_ak()
Envariable.set_sk()
# 檢查AK和SK是否成功擷取
if not Envariable.get_ak() or not Envariable.get_sk():
logging.warning("AK or SK not found in environment variables")
return False
# 初始化Client
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> 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:])