Python 手动刷新腾讯云 CDN 缓存（URL/DIR）

摘要

内容分发网络（CDN）通过将您的业务内容发布到离用户最近的边缘节点，使用户能够就近获取所需内容，提高用户访问的响应速度和成功率。解决因地域分布、带宽、服务器能力等因素影响，而导致的访问延迟高、用户体验较差等问题。

缓存配置可以通过配置缓存规则，根据缓存规则来刷新，如果页面内容临时改动，且刷新规则还未到触发时间，则需要手动刷新。

代码

# -*- coding:utf-8 -*-
__author__ = 'wx'
import hashlib
import requests
import hmac
import random
import time
import getopt,sys
import base64
from urllib import  parse
from datetime import datetime, timedelta

class CdnHelper(object):
    SecretId = 'AKID54rSwEDDtOjuy4nM'
    SecretKey = 'qUI1zjF'
    requestHost = 'cdn.api.qcloud.com'
    requestUri = '/v2/index.php?'
    def __init__(self, **args):
        self.type = args['type']
        self.url = args['url']
        self.startDate = args['startDate']
        self.endDate = args['endDate']

    def flushurl_dict(self):
        keydict = {
            'Action': 'RefreshCdnUrl',
            'Timestamp': str(int(time.time())),
            'Nonce': str(int(random.random() * 1000)),
            'SecretId': CdnHelper.SecretId,
            'urls.0': self.url
        }
        sortlist = sorted(zip(keydict.keys(), keydict.values()))
        return sortlist
    def flushdir_dict(self):
        keydict = {
            'Action': 'RefreshCdnDir',
            'Timestamp': str(int(time.time())),
            'Nonce': str(int(random.random() * 1000)),
            'SecretId': CdnHelper.SecretId,
            'dirs.0': self.url
        }
        sortlist = sorted(zip(keydict.keys(), keydict.values()))
        return sortlist

    def get_str_sign(self):
        if self.type == 'url':
            sortlist = self.flushurl_dict()
        if self.type == 'dir':
            sortlist = self.flushdir_dict()
        sign_str_init = ''
        for value in sortlist:
            sign_str_init += str(value[0]) + '=' + str(value[1]) + '&'
        sign_str = 'GET' + CdnHelper.requestHost + CdnHelper.requestUri + sign_str_init[:-1]
        return sign_str, sign_str_init

    def get_result_url(self):
        sign_str, sign_str_init = self.get_str_sign()
        secretkey = CdnHelper.SecretKey
        signature = bytes(sign_str, encoding='utf-8')
        secretkey = bytes(secretkey, encoding='utf-8')
        my_sign = hmac.new(secretkey, signature, hashlib.sha1).digest()
        my_sign = base64.b64encode(my_sign)
        result_sign = parse.quote(my_sign)
        result_url = 'https://' + CdnHelper.requestHost + CdnHelper.requestUri + sign_str_init + '&Signature=' + result_sign
        return result_url
    def flush_url(self):
        result_url = self.get_result_url()
        try:
            r = requests.get(result_url)
            print(r.json()['codeDesc'])
        except Exception as e:
            print(e)

def usage():
    print('usage:  FlushCDN.py [options] arg1\n' \
        '\n' \
        'options:\n' \
        '   -h, --help            show this help message and exit  \n' \
        '   -v, --verbose         make lots of noise [default]\n' \
        '   -u, --url             Flush URL\n' \
        '   -d, --dir             Flush Dir\n' \
        '')
    exit(0)

if __name__ == '__main__':
    tm = datetime.now() + timedelta(hours=-2)
    startDate = endDate = tm.strftime("%Y-%m-%d %H:00:00")
    if len(sys.argv) == 1:
        usage()
        exit(1)
    try:
        options, args = getopt.getopt(sys.argv[1:], "hvu:d:", ['help', 'verbose', "url=", "dir="])
    except getopt.GetoptError as e:
        print(e)
        exit(1)
    flush_url = sys.argv[-1]
    for name, value in options:
        if name in ('-h', '--help'):
            usage()
        elif name in ('-v', '--verbose'):
            print('Version: 1.0. Date: 2018-9-20')
            exit(0)
        elif name in ('-u', '--url'):
            flush_type = 'url'
            print('Flushing ' + flush_type + ': %s' % flush_url)
        elif name in ('-d', '--dir'):
            flush_type = 'dir'
            print('Flushing ' + flush_type + ': %s' % flush_url)

    obj = CdnHelper(type=flush_type, url=flush_url, startDate=startDate, endDate=endDate)
    obj.flush_url()

100

101

102

103

104

105

106

107

108

109

110

111

# -*- coding:utf-8 -*-

__author__ = 'wx'

import hashlib

import requests

import hmac

import random

import time

import getopt,sys

import base64

from urllib import parse

from datetime import datetime, timedelta

class CdnHelper(object):

SecretId = 'AKID54rSwEDDtOjuy4nM'

SecretKey = 'qUI1zjF'

requestHost = 'cdn.api.qcloud.com'

requestUri = '/v2/index.php?'

def __init__(self, **args):

self.type = args['type']

self.url = args['url']

self.startDate = args['startDate']

self.endDate = args['endDate']

def flushurl_dict(self):

keydict = {

'Action': 'RefreshCdnUrl',

'Timestamp': str(int(time.time())),

'Nonce': str(int(random.random() * 1000)),

'SecretId': CdnHelper.SecretId,

'urls.0': self.url

}

sortlist = sorted(zip(keydict.keys(), keydict.values()))

return sortlist

def flushdir_dict(self):

keydict = {

'Action': 'RefreshCdnDir',

'Timestamp': str(int(time.time())),

'Nonce': str(int(random.random() * 1000)),

'SecretId': CdnHelper.SecretId,

'dirs.0': self.url

}

sortlist = sorted(zip(keydict.keys(), keydict.values()))

return sortlist

def get_str_sign(self):

if self.type == 'url':

sortlist = self.flushurl_dict()

if self.type == 'dir':

sortlist = self.flushdir_dict()

sign_str_init = ''

for value in sortlist:

sign_str_init += str(value[0]) + '=' + str(value[1]) + '&'

sign_str = 'GET' + CdnHelper.requestHost + CdnHelper.requestUri + sign_str_init[:-1]

return sign_str, sign_str_init

def get_result_url(self):

sign_str, sign_str_init = self.get_str_sign()

secretkey = CdnHelper.SecretKey

signature = bytes(sign_str, encoding='utf-8')

secretkey = bytes(secretkey, encoding='utf-8')

my_sign = hmac.new(secretkey, signature, hashlib.sha1).digest()

my_sign = base64.b64encode(my_sign)

result_sign = parse.quote(my_sign)

result_url = 'https://' + CdnHelper.requestHost + CdnHelper.requestUri + sign_str_init + '&Signature=' + result_sign

return result_url

def flush_url(self):

result_url = self.get_result_url()

try:

r = requests.get(result_url)

print(r.json()['codeDesc'])

except Exception as e:

print(e)

def usage():

print('usage: FlushCDN.py [options] arg1\n' \

'\n' \

'options:\n' \

' -h, --help show this help message and exit \n' \

' -v, --verbose make lots of noise [default]\n' \

' -u, --url Flush URL\n' \

' -d, --dir Flush Dir\n' \

'')

exit(0)

if __name__ == '__main__':

tm = datetime.now() + timedelta(hours=-2)

startDate = endDate = tm.strftime("%Y-%m-%d %H:00:00")

if len(sys.argv) == 1:

usage()

exit(1)

try:

options, args = getopt.getopt(sys.argv[1:], "hvu:d:", ['help', 'verbose', "url=", "dir="])

except getopt.GetoptError as e:

print(e)

exit(1)

flush_url = sys.argv[-1]

for name, value in options:

if name in ('-h', '--help'):

usage()

elif name in ('-v', '--verbose'):

print('Version: 1.0. Date: 2018-9-20')

exit(0)

elif name in ('-u', '--url'):

flush_type = 'url'

print('Flushing ' + flush_type + ': %s' % flush_url)

elif name in ('-d', '--dir'):

flush_type = 'dir'

print('Flushing ' + flush_type + ': %s' % flush_url)

obj = CdnHelper(type=flush_type, url=flush_url, startDate=startDate, endDate=endDate)

obj.flush_url()

运行方式

[root@showdoc data]# python3 FLushCDN.py --help
usage:  FlushCDN.py [options] arg1

options:
   -h, --help            show this help message and exit  
   -v, --verbose         make lots of noise [default]
   -u, --url             Flush URL
   -d, --dir             Flush Dir

[root@showdoc data]# python3 FLushCDN.py --help

usage: FlushCDN.py [options] arg1

options:

-h, --help show this help message and exit

-v, --verbose make lots of noise [default]

-u, --url Flush URL

-d, --dir Flush Dir

运行效果

发表于：Python

2018-09-20

# python # 腾讯云

复制链接