CDN：Gunakan skrip untuk menyegarkan dan memuat ulang konten

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

# Perintah instalasi SDK: pip install alibabacloud_cdn20180510

'''Periksa Paket'''
# Impor library yang diperlukan.
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

# Tangkap pengecualian impor.
except ImportError as e:
    sys.exit(f"[error] Harap pip install alibabacloud_cdn20180510. Detail: {e}")

# Inisialisasi logging.
logging.basicConfig(level=logging.DEBUG, filename='./RefreshAndPredload.log')

# Definisikan kelas variabel global untuk menyimpan informasi seperti ID AccessKey, rahasia AccessKey, dan direktori file.
class Envariable(object):
    LISTS = []
    # Untuk Endpoint, lihat 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

    # Tetapkan ID AccessKey.
    @staticmethod
    def set_ak():
        Envariable.AK = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID')

    # Dapatkan ID AccessKey.
    @staticmethod
    def get_ak():
        return Envariable.AK

    # Tetapkan rahasia AccessKey.
    @staticmethod
    def set_sk():
        Envariable.SK = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET')

    # Dapatkan rahasia AccessKey.
    @staticmethod
    def get_sk():
        return Envariable.SK

    # Tetapkan direktori file.
    @staticmethod
    def set_fd(fd):
        Envariable.FD = fd

    # Dapatkan direktori file.
    @staticmethod
    def get_fd():
        return Envariable.FD

    # Tetapkan jenis tugas.
    @staticmethod
    def set_task_type(task_type):
        Envariable.TASK_TYPE = task_type

    # Dapatkan jenis tugas.
    @staticmethod
    def get_task_type():
        return Envariable.TASK_TYPE
    
    # Tetapkan area tugas.
    @staticmethod
    def set_task_area(task_area):
        Envariable.TASK_AREA = task_area

    # Dapatkan area tugas.
    @staticmethod
    def get_task_area():
        return Envariable.TASK_AREA

    # Tetapkan jenis objek tugas.
    @staticmethod
    def set_task_otype(task_otype):
        Envariable.TASK_OTYPE = task_otype

    # Dapatkan jenis objek tugas.
    @staticmethod
    def get_task_otype():
        return Envariable.TASK_OTYPE

    # Buat klien baru.
    @staticmethod
    def set_acs_client():
        try:
            # Gunakan pasangan AccessKey untuk menginisialisasi klien Credentials.
            credentialsConfig = CreConfig(
                # Jenis kredensial.
                type='access_key',
                # Atur ke ID AccessKey.
                access_key_id=Envariable.get_ak(),
                # Atur ke rahasia AccessKey.
                access_key_secret=Envariable.get_sk(),
            )
            credentialClient = CredentialClient(credentialsConfig)

            cdnConfig = Config(credential=credentialClient)
            # Konfigurasikan titik akhir layanan.
            cdnConfig.endpoint = Envariable.ENDPOINT
            # Inisialisasi klien CDN.
            Envariable.CLI = Cdn20180510Client(cdnConfig)
        except Exception as e:
            logging.error(f"Gagal membuat klien: {e}")
            raise

    # Dapatkan klien.
    @staticmethod
    def get_acs_client():
        return Envariable.CLI


# Fungsi inisialisasi modul.
def initialize_credentials_and_client():
    """Inisialisasi pasangan AccessKey dan klien ketika modul dimuat."""
    try:
        # Inisialisasi pasangan AccessKey dari variabel lingkungan.
        Envariable.set_ak()
        Envariable.set_sk()
        
        # Periksa apakah pasangan AccessKey diperoleh.
        if not Envariable.get_ak() or not Envariable.get_sk():
            logging.warning("AK atau SK tidak ditemukan dalam variabel lingkungan")
            return False
            
        # Inisialisasi klien.
        Envariable.set_acs_client()
        logging.info("Kredensial dan klien berhasil diinisialisasi")
        return True
    except Exception as e:
        logging.error(f"Gagal menginisialisasi kredensial dan klien: {e}")
        return False


# Jalankan inisialisasi ketika modul dimuat.
_initialization_success = initialize_credentials_and_client()





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

    # Periksa kuota.
    def printQuota(self):
        try:
            client = Envariable.get_acs_client()
            if not client:
                raise Exception("Klien CDN belum diinisialisasi")
            
            # Gunakan SDK untuk melakukan panggilan.
            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]: inisialisasi Cdn20180510Client gagal: {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 tidak cukup {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 tidak cukup {0}".format(quotaResp['UrlRemain']))
                elif Envariable.get_task_otype() == 'Directory' and self.lines > int(quotaResp['DirRemain']):
                    sys.exit("\n[error]:DirRemain tidak cukup {0}".format(quotaResp['DirRemain']))
                else:
                    return True

    # Verifikasi format 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]: Format URL ilegal \n{0}".format(self.invalidurl))
                return True
        except FileNotFoundError:
            sys.exit(f"\n[error]: File tidak ditemukan: {self.urllist}\n")
        except Exception as e:
            sys.exit(f"\n[error]: Gagal membaca file {self.urllist}: {e}\n")

# Kelas pemrosesan batch yang membagi daftar URL menjadi beberapa batch dengan ukuran tertentu.
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

    # Proses URL dalam batch.
    @staticmethod
    def doProd():
        gop = 20  # Menentukan jumlah maksimum URL per batch.
        mins = 1
        maxs = gop
        current_batch = []  # Gunakan variabel lokal alih-alih variabel global.
        
        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 tidak ditemukan: {Envariable.get_fd()}\n")
        except Exception as e:
            sys.exit(f"\n[error]: Gagal membaca file {Envariable.get_fd()}: {e}\n")

    # Jalankan tugas penyegaran atau pemuatan ulang.
    @staticmethod
    def doRefresh(lists):
        try:
            client = Envariable.get_acs_client()
            if not client:
                raise Exception("Klien CDN belum diinisialisasi")

            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
                    # Gunakan SDK untuk memeriksa status tugas.
                    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]}] sedang berjalan... ...")
                    
                    for t in taskresp_data['Tasks']['CDNTask']:
                        if t['Status'] != 'Complete':
                            count += 1
                    if count == 0:
                        logging.info(f"[{response_data[taskID]}] selesai")
                        break
                    elif timeout > 5:  # Tunggu maksimal 50 detik (5 × 10 detik).
                        logging.info(f"[{response_data[taskID]}] timeout setelah 50 detik")
                        break
                    else:
                        timeout += 1
                        time.sleep(10)  # Periksa status setiap 10 detik.
                        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[penggunaan]: {sys.argv[0]} -h ")
        try:
            opts, args = getopt.getopt(argv, "hr:t:a:o:")
        except getopt.GetoptError as e:
            sys.exit(f"\n[penggunaan]: {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[penggunaan]: {sys.argv[0]} -h ")
        
        # Periksa status inisialisasi hanya ketika bukan perintah bantuan.
        if not _initialization_success:
            sys.exit("\n[error]: Gagal menginisialisasi kredensial dan klien. Silakan periksa variabel lingkungan.\n")

        try:
            if not (Envariable.get_ak() and Envariable.get_sk() and Envariable.get_fd() and Envariable.get_task_type()):
                sys.exit("\n[error]: Harus menetapkan variabel lingkungan ALIBABA_CLOUD_ACCESS_KEY_ID dan ALIBABA_CLOUD_ACCESS_KEY_SECRET, dan parameter '-r', '-t'\n")
            if Envariable.get_task_type() not in {"push", "clear"}:
                sys.exit("\n[error]: Jenis tugas salah, opsi '-t' dalam 'push' atau 'clear'\n")
            if Envariable.get_task_area() and Envariable.get_task_otype():
                sys.exit("\n[error]: -a dan -o tidak dapat ada pada saat yang sama\n")
            if Envariable.get_task_area():
                if Envariable.get_task_area() not in {"domestic", "overseas"}:
                    sys.exit("\n[error]: Nilai Area salah, opsi '-a' dalam 'domestic' atau 'overseas'\n")
            if Envariable.get_task_otype():
                if Envariable.get_task_otype() not in {"File", "Directory"}:
                    sys.exit("\n[error]: Nilai ObjectType salah, opsi '-a' dalam 'File' atau 'Directory'\n")
                if Envariable.get_task_type() == 'push':
                    sys.exit("\n[error]: -t harus clear dan 'push' -a digunakan bersama\n")
        except Exception as e:
            loggingerror(f"\n[error]: Parameter {e} salah\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("\npilihan skrip menjelaskan: \
                    \n\t -r <filename>                    //Jalur file dan nama file. Setelah skrip dijalankan, ia membaca URL dari file. Setiap baris harus berisi satu URL. URL dengan karakter khusus harus dienkripsi URL. Setiap URL harus dimulai dengan http atau https. \
                    \n\t -t <taskType>                    //Jenis tugas. `clear`: segarkan. `push`: muat ulang. \
                    \n\t -a [String,<domestic|overseas>   //Opsional. Ruang lingkup pemuatan ulang. Jika Anda tidak menyetel parameter ini, sumber daya dimuat ulang secara global.            
                   domestic                    //Hanya daratan Tiongkok.             
                   overseas                    //Global (tidak termasuk daratan Tiongkok).             
              -o [String,<File|Directory>]     //Opsional. Jenis konten untuk disegarkan.             
                   File                        //File (default).             
                   Directory                   //Direktori.")


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

• Jika file URL bernama urllist.txt, file dan skrip Refresh.py berada dalam direktori yang sama, dan jenis tugas adalah push (muat ulang), jalankan perintah berikut dalam CLI, seperti Command Prompt, PowerShell, atau Terminal.

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

• Jika file berada di direktori berbeda, seperti D:\example\filename\urllist.txt, jalankan perintah berikut dalam CLI, seperti Command Prompt, PowerShell, atau Terminal.

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

Berikut adalah contoh keluaran:

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

Jika pesan kesalahan Gagal menginisialisasi kredensial dan klien. Silakan periksa variabel lingkungan. dikembalikan, tetapkan Alibaba Cloud AccessKey dalam variabel lingkungan seperti yang dijelaskan pada Langkah 4, lalu jalankan perintah di jendela terminal yang sama.