IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

如何用脚本自动刷新CDN缓存？

wen 实用脚本 2

本文目录导读：

如何用脚本自动刷新CDN缓存？

  1. 阿里云CDN刷新
  2. 腾讯云CDN刷新
  3. Cloudflare CDN刷新
  4. 通用自动化脚本框架
  5. 使用建议

要用脚本自动刷新CDN缓存,通常需要调用CDN服务商提供的API，以下是几种主流CDN的自动化刷新方法（以Shell脚本和Python为例）：

阿里云CDN刷新

Shell脚本示例

#!/bin/bash
# 配置阿里云AccessKey
ACCESS_KEY_ID="your-access-key-id"
ACCESS_KEY_SECRET="your-access-key-secret"
# 要刷新的URL列表
URLS=(
    "https://example.com/index.html"
    "https://example.com/style.css"
)
# 生成时间戳和签名
TIMESTAMP=$(date -u +"%Y-%m-%dT%H:%M:%SZ")
SIGNATURE=$(echo -n "GET&%2F&AccessKeyId%3D${ACCESS_KEY_ID}%26Action%3DPushObjectCache%26Format%3DJSON%26ObjectPath%3D${URLS[0]}%26Timestamp%3D${TIMESTAMP}%26Version%3D2014-11-11" | openssl dgst -sha1 -hmac "${ACCESS_KEY_SECRET}&" | base64)
# 调用API刷新缓存
curl -X GET "https://cdn.aliyuncs.com/?Action=PushObjectCache&ObjectPath=${URLS[0]}&Format=JSON&Version=2014-11-11&AccessKeyId=${ACCESS_KEY_ID}&Timestamp=${TIMESTAMP}&Signature=${SIGNATURE}"

Python脚本示例（推荐）

import json
import requests
from aliyun_sdk import AliyunClient  # 需要安装aliyun-python-sdk-cdn
def refresh_aliyun_cdn(urls):
    client = AliyunClient(
        access_key_id='your-access-key-id',
        access_key_secret='your-access-key-secret',
        region_id='cn-hangzhou'
    )
    # 刷新URL
    request = {
        'Action': 'PushObjectCache',
        'ObjectPath': '\n'.join(urls)
    }
    response = client.do_action_with_exception(request)
    return json.loads(response)
# 使用示例
urls_to_refresh = [
    "https://example.com/index.html",
    "https://example.com/style.css"
]
result = refresh_aliyun_cdn(urls_to_refresh)
print(result)

腾讯云CDN刷新

Shell脚本示例

#!/bin/bash
# 配置腾讯云密钥
SECRET_ID="your-secret-id"
SECRET_KEY="your-secret-key"
# 要刷新的URL
URL="https://example.com/index.html"
# 生成签名
TIMESTAMP=$(date +%s)
NONCE=$(date +%s | md5sum | head -c 8)
SIGNATURE=$(echo -n "GETcdn.tencentcloudapi.com/?Action=PurgeUrlsCache&Nonce=${NONCE}&SecretId=${SECRET_ID}&Timestamp=${TIMESTAMP}&Urls.0=${URL}&Version=2018-06-06" | openssl dgst -sha256 -hmac "${SECRET_KEY}" | base64)
# 调用API
curl -X GET "https://cdn.tencentcloudapi.com/?Action=PurgeUrlsCache&SecretId=${SECRET_ID}&Timestamp=${TIMESTAMP}&Nonce=${NONCE}&Signature=${SIGNATURE}&Urls.0=${URL}&Version=2018-06-06"

Python脚本

import hashlib
import hmac
import json
import requests
import time
def refresh_tencent_cdn(urls):
    secret_id = 'your-secret-id'
    secret_key = 'your-secret-key'
    # 构造请求参数
    params = {
        'Action': 'PurgeUrlsCache',
        'SecretId': secret_id,
        'Timestamp': int(time.time()),
        'Nonce': int(time.time() * 1000),
        'Version': '2018-06-06'
    }
    # 添加URL
    for i, url in enumerate(urls):
        params[f'Urls.{i}'] = url
    # 生成签名
    sorted_params = sorted(params.items())
    sign_str = 'GETcdn.tencentcloudapi.com/?' + '&'.join([f'{k}={v}' for k, v in sorted_params])
    signature = hmac.new(
        secret_key.encode(),
        sign_str.encode(),
        hashlib.sha256
    ).hexdigest()
    params['Signature'] = signature
    # 发送请求
    response = requests.get('https://cdn.tencentcloudapi.com/', params=params)
    return response.json()
# 使用示例
urls = ["https://example.com/index.html"]
result = refresh_tencent_cdn(urls)
print(result)

Cloudflare CDN刷新

Shell脚本

#!/bin/bash
ZONE_ID="your-zone-id"
API_TOKEN="your-api-token"
# 刷新所有缓存
curl -X POST "https://api.cloudflare.com/client/v4/zones/${ZONE_ID}/purge_cache" \
     -H "Authorization: Bearer ${API_TOKEN}" \
     -H "Content-Type: application/json" \
     -d '{"purge_everything":true}'
# 或者刷新特定URL
curl -X POST "https://api.cloudflare.com/client/v4/zones/${ZONE_ID}/purge_cache" \
     -H "Authorization: Bearer ${API_TOKEN}" \
     -H "Content-Type: application/json" \
     -d '{"files":["https://example.com/index.html","https://example.com/style.css"]}'

Python脚本

import requests
def refresh_cloudflare_cache(zone_id, api_token, urls=None):
    headers = {
        'Authorization': f'Bearer {api_token}',
        'Content-Type': 'application/json'
    }
    if urls:
        # 刷新特定URL
        payload = {'files': urls}
    else:
        # 刷新所有缓存
        payload = {'purge_everything': True}
    response = requests.post(
        f'https://api.cloudflare.com/client/v4/zones/{zone_id}/purge_cache',
        headers=headers,
        json=payload
    )
    return response.json()
# 使用示例
zone_id = 'your-zone-id'
api_token = 'your-api-token'
result = refresh_cloudflare_cache(zone_id, api_token, 
    urls=['https://example.com/index.html'])
print(result)

通用自动化脚本框架

以下是一个通用的自动化刷新脚本框架,支持多种CDN：

#!/usr/bin/env python3
"""
CDN缓存自动化刷新脚本
支持：阿里云、腾讯云、Cloudflare
"""
import os
import sys
import json
import argparse
from typing import List
class CDNRefresher:
    """CDN刷新器基类"""
    def refresh(self, urls: List[str]):
        raise NotImplementedError
    def refresh_all(self):
        raise NotImplementedError
class AliyunCDNRefresher(CDNRefresher):
    def __init__(self):
        self.access_key_id = os.getenv('ALIYUN_ACCESS_KEY_ID')
        self.access_key_secret = os.getenv('ALIYUN_ACCESS_KEY_SECRET')
    def refresh(self, urls: List[str]):
        # 实现阿里云刷新逻辑
        pass
class TencentCDNRefresher(CDNRefresher):
    def __init__(self):
        self.secret_id = os.getenv('TENCENT_SECRET_ID')
        self.secret_key = os.getenv('TENCENT_SECRET_KEY')
    def refresh(self, urls: List[str]):
        # 实现腾讯云刷新逻辑
        pass
class CloudflareCDNRefresher(CDNRefresher):
    def __init__(self):
        self.api_token = os.getenv('CLOUDFLARE_API_TOKEN')
        self.zone_id = os.getenv('CLOUDFLARE_ZONE_ID')
    def refresh(self, urls: List[str]):
        # 实现Cloudflare刷新逻辑
        pass
def get_refresher(cdn_type: str) -> CDNRefresher:
    """根据CDN类型获取刷新器"""
    refreshers = {
        'aliyun': AliyunCDNRefresher,
        'tencent': TencentCDNRefresher,
        'cloudflare': CloudflareCDNRefresher,
    }
    if cdn_type not in refreshers:
        raise ValueError(f"不支持的CDN类型: {cdn_type}")
    return refreshers[cdn_type]()
def main():
    parser = argparse.ArgumentParser(description='CDN缓存自动刷新脚本')
    parser.add_argument('--cdn', required=True, 
                       choices=['aliyun', 'tencent', 'cloudflare'],
                       help='CDN服务商')
    parser.add_argument('--urls', nargs='+', 
                       help='要刷新的URL列表')
    parser.add_argument('--all', action='store_true',
                       help='刷新所有缓存')
    args = parser.parse_args()
    try:
        refresher = get_refresher(args.cdn)
        if args.all:
            result = refresher.refresh_all()
        elif args.urls:
            result = refresher.refresh(args.urls)
        else:
            # 默认从配置文件读取
            with open('urls.txt', 'r') as f:
                urls = [line.strip() for line in f if line.strip()]
            result = refresher.refresh(urls)
        print(json.dumps(result, indent=2))
    except Exception as e:
        print(f"刷新失败: {e}", file=sys.stderr)
        sys.exit(1)
if __name__ == '__main__':
    main()

使用建议

环境变量管理

建议将敏感信息存储在环境变量或配置文件中：

# .env文件
ALIYUN_ACCESS_KEY_ID=your_key_id
ALIYUN_ACCESS_KEY_SECRET=your_key_secret
TENCENT_SECRET_ID=your_secret_id
TENCENT_SECRET_KEY=your_secret_key
CLOUDFLARE_API_TOKEN=your_api_token
CLOUDFLARE_ZONE_ID=your_zone_id

定时任务设置

使用cron设置定时刷新：

# 每天凌晨3点刷新缓存
0 3 * * * /path/to/refresh_cdn.sh
# 每次文件更新后触发刷新
*/5 * * * * /path/to/check_and_refresh.py

注意事项

  1. API调用限制：注意各CDN的API调用频率限制
  2. 错误处理：添加完善的错误处理和重试机制
  3. 日志记录：记录每次刷新的结果和异常
  4. 权限管理：使用最小权限原则配置API密钥
  5. 成本控制：部分CDN对API调用收费，注意控制频率

根据你的具体需求选择合适的CDN服务商和脚本方案。

上一篇实用脚本能自动切换DNS解析吗？

下一篇脚本能自动预热CDN缓存吗？

相关文章

抱歉，评论功能暂时关闭!