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(
        "https://api.dogecloud.com" + 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 请参考 https://api.aliyun.com/product/Alidns
    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 次 / 日