脚本分享：CDN 流量触顶后自动切换至 Cloudflare

1. 目标域名使用国内 DNS（如阿里 DNS）
2. 目标域名配置好境内 CDN
3. 目标域名借助第二域名配置好 Cloudflare for SaaS
4. DNS 设置国内线路 cname 解析值设定为 CDN 提供的域名
5. DNS 设置海外线路 cname 解析值设定为 Cloudflare for SaaS 提供的域名

@startuml
left to right direction

actor 访客 as visitor

package "互联网" {
  note "域名" as site_address #orange

  node "DNS服务器" as dns {
    component "境内 cname 解析" as cname_cn
    component "境外 cname 解析" as cname_intl
  }
}

node "境内CDN" as cdn #lightblue {
  component "CDN 提供的域名" as cdn_address #orange
}


node "Cloudflare" as cf #skyblue {
  component "Cloudflare for SaaS（自定义主机名）提供的域名" as cf_address #orange
}

visitor --> site_address : 访问站点
site_address --> dns : DNS查询

dns --> cname_cn : 境内解析分流
dns --> cname_intl : 境外解析分流

cname_cn --> cdn_address
cname_intl --> cf_address
@enduml

警告：此代码 dogecloud_api 为多吉云官方提供的 SDK 代码，create_client、disable_default_enable_oversea、enable_default_convert_to_oversea 是由阿里云自动生成得来。笔者花费了 27 分钟将这些 SDK 粘在一起。代码中有无法解释的注释/风格混乱是意料之中的情况。

import hmac
import json
import sys
import time
import tomllib
import urllib
from hashlib import sha1
from typing import List

import requests
from alibabacloud_alidns20150109 import models as alidns_20150109_models
from alibabacloud_alidns20150109.client import Client as Alidns20150109Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_tea_util import models as util_models

with open("env.toml", "rb") as f:
    config: dict = tomllib.load(f)
dogecloud_access_key = config["dogecloud_access_key"]
dogecloud_secret_key = config["dogecloud_secret_key"]
aliyun_accesskey_id = config["aliyun_accesskey_id"]
aliyun_accesskey_secret = config["aliyun_accesskey_secret"]
domain = config["domain"]
rr = config["rr"]
if rr == "@":
    full_domain = domain
else:
    full_domain = rr + "." + domain


def dogecloud_api(api_path, data={}, json_mode=False):
    """
    调用多吉云API

    :param api_path:    调用的 API 接口地址，包含 URL 请求参数 QueryString，例如：/console/vfetch/add.json?url=xxx&a=1&b=2
    :param data:        POST 的数据，字典，例如 {'a': 1, 'b': 2}，传递此参数表示不是 GET 请求而是 POST 请求
    :param json_mode:   数据 data 是否以 JSON 格式请求，默认为 false 则使用表单形式（a=1&b=2）

    :type api_path: string
    :type data: dict
    :type json_mode bool

    :return dict: 返回的数据
    """

    with open("env.toml", "rb") as f:
        config = tomllib.load(f)
    access_key = config["dogecloud_access_key"]
    secret_key = config["dogecloud_secret_key"]

    body = ""
    mime = ""
    if json_mode:
        body = json.dumps(data)
        mime = "application/json"
    else:
        body = urllib.parse.urlencode(data)
        mime = "application/x-www-form-urlencoded"
    sign_str = api_path + "\n" + body
    signed_data = hmac.new(secret_key.encode("utf-8"), sign_str.encode("utf-8"), sha1)
    sign = signed_data.digest().hex()
    authorization = "TOKEN " + access_key + ":" + sign
    response = requests.post(
        " + api_path,
        data=body,
        headers={"Authorization": authorization, "Content-Type": mime},
    )
    return response.json()


def create_client() -> Alidns20150109Client:
    """
    使用AK&SK初始化账号Client
    @return: Client
    @throws Exception
    """
    with open("env.toml", "rb") as f:
        config = tomllib.load(f)
    config = open_api_models.Config(
        access_key_id=config["aliyun_accesskey_id"],
        access_key_secret=config["aliyun_accesskey_secret"],
    )
    # Endpoint 请参考 
    config.endpoint = f"alidns.cn-huhehaote.aliyuncs.com"
    return Alidns20150109Client(config)


def disable_default_enable_oversea(args: List[str]) -> None | int:
    """
    禁用默认CNAME记录，将海外记录设为默认线路
    """
    client = create_client()
    add_custom_line_request = alidns_20150109_models.DescribeDomainRecordsRequest(
        lang="zh", domain_name=domain, rrkey_word=rr
    )
    try:
        response = client.describe_domain_records(add_custom_line_request)
        if response.status_code != 200:
            return 500

        records = response.body.domain_records.record
        default_cname_record_id = None
        oversea_record_id = None
        oversea_record_value = None

        for record in records:
            if (
                record.domain_name == domain
                and record.line == "default"
                and record.rr == rr
                and record.type == "CNAME"
            ):
                default_cname_record_id = record.record_id
            if record.line == "oversea":
                oversea_record_id = record.record_id
                oversea_record_value = record.value
    except Exception as error:
        # # 错误 message
        # print(error.message)
        # # 诊断地址
        # print(error.data.get("Recommend"))
        # from alibabacloud_tea_util.client import Client as UtilClient
        # UtilClient.assert_as_string(error.message)
        return 500

    set_domain_record_status_request = (
        alidns_20150109_models.SetDomainRecordStatusRequest(
            record_id=default_cname_record_id, status="Disable"
        )
    )
    try:
        response = client.set_domain_record_status_with_options(
            set_domain_record_status_request, util_models.RuntimeOptions()
        )
        if response.status_code != 200:
            return 500
    except Exception as error:
        return 500

    update_domain_record_request = alidns_20150109_models.UpdateDomainRecordRequest(
        record_id=oversea_record_id,
        rr=rr,
        type="CNAME",
        line="default",
        value=oversea_record_value,
    )
    try:
        response = client.update_domain_record_with_options(
            update_domain_record_request, util_models.RuntimeOptions()
        )
        if response.status_code != 200:
            return 500
    except Exception as error:
        return 500


def enable_default_convert_to_oversea(args: List[str]) -> None | int:
    """
    启用默认CNAME记录，将当前默认记录转为海外线路
    """
    client = create_client()
    add_custom_line_request = alidns_20150109_models.DescribeDomainRecordsRequest(
        lang="zh", domain_name=domain, rrkey_word=rr
    )
    try:
        response = client.describe_domain_records(add_custom_line_request)
        if response.status_code != 200:
            return 500

        records = response.body.domain_records.record
        default_cname_record_id = None
        oversea_record_id = None
        oversea_record_value = None

        for record in records:
            # print(record.status)
            if (
                record.domain_name == domain
                and record.line == "default"
                and record.rr == rr
                and record.type == "CNAME"
                and record.status == "DISABLE"
            ):
                default_cname_record_id = record.record_id
            if (
                record.domain_name == domain
                and record.line == "default"
                and record.rr == rr
                and record.type == "CNAME"
                and record.status == "ENABLE"
            ):
                oversea_record_id = record.record_id
                oversea_record_value = record.value
    except Exception as error:
        return 500
    set_domain_record_status_request = (
        alidns_20150109_models.SetDomainRecordStatusRequest(
            record_id=default_cname_record_id, status="Enable"
        )
    )
    try:
        response = client.set_domain_record_status_with_options(
            set_domain_record_status_request, util_models.RuntimeOptions()
        )
        if response.status_code != 200:
            return 500
    except Exception as error:
        return 500

    update_domain_record_request = alidns_20150109_models.UpdateDomainRecordRequest(
        record_id=oversea_record_id,
        rr=rr,
        type="CNAME",
        line="oversea",
        value=oversea_record_value,
    )
    try:
        response = client.update_domain_record_with_options(
            update_domain_record_request, util_models.RuntimeOptions()
        )
        if response.status_code != 200:
            return 500
    except Exception as error:
        return 500


def cdn_offline_monitor(domain_info) -> str:
    if domain_info["status"] not in ("offline"):
        print("CDN 状态: 在线")
        return "offline"
    else:
        print("检测到 CDN 已离线")
        if disable_default_enable_oversea(sys.argv[1:]) is not None:
            print("DNS 修改失败，暂停 1s 后重试")
            time.sleep(1)
        print("开始监控域名上线状态...")
        return "online"


def cdn_online_monitor(domain_info) -> str:
    if domain_info["status"] not in ("online"):
        print("CDN 状态: 离线")
        return "online"
    else:
        print("检测到 CDN 已上线")
        if enable_default_convert_to_oversea(sys.argv[1:]) is not None:
            print("DNS 修改失败，暂停 1s 后重试")
            time.sleep(1)
        print("开始监控域名离线状态...")
        return "offline"


def domain_info_handler(domain_info, monitoring_mode) -> str:
    # 检测离线模式
    if monitoring_mode == "offline":
        return cdn_offline_monitor(domain_info)
    # 检测上线模式
    elif monitoring_mode == "online":
        return cdn_online_monitor(domain_info)


if __name__ == "__main__":
    # 定义当前监控模式：'offline' 监控是否离线，'online' 监控是否上线
    monitoring_mode = "offline"

    while True:
        for domain_info in dogecloud_api("/cdn/domain/list.json")["data"]["domains"]:
            if domain_info["name"] != full_domain:
                continue

            monitoring_mode = domain_info_handler(domain_info, monitoring_mode)

        time.sleep(3)  # 多吉云 API 限制 30000 次 / 日