简单的cloudflare ddns脚本，无须使用ipv6一堆毛病的docker

cloudflare_ddns.py

import requests
import ipaddress

proxy = {
    "http": '',
    "https": ''
}

# 设置 Cloudflare API token 和域名
api_token = 'https://dash.cloudflare.com/profile/api-tokens'  # 替换为你的 Cloudflare API token 
domain = 'github.com'  # 替换为你要更新的根域
record_prefix = 'gist'  # 替换为你要更新的记录前缀

# 以下两项在脚本运行时会显示在标准输出中，将这两项填入脚本可以节省更新解析记录的时间
zone_id = ''   # 替换为你要更新的 Zone ID
record_id = ''   # 替换为你要更新的 Record ID


def get_zone_id(domain):
    """
    获取根域名对应的zone id
    
    参数:
    - domain: 根域名 (如: example.com)
    """
    url = f'https://api.cloudflare.com/client/v4/zones?name={domain}'
    headers = {
        'Authorization': f'Bearer {api_token}',
        'Content-Type': 'application/json'
    }
    response = requests.get(url, headers=headers, proxies=proxy)
    response_data = response.json()
    if response_data['success']:
        # 检查是否有zone id返回
        if len(response_data['result']) > 0:
            return response_data['result'][0]['id']
        else:
            return None
    else:
        error_messages = [error['message'] for error in response_data['errors']]
        raise Exception(f'无法获取 Zone ID, 错误信息: {"; ".join(error_messages)}')


def get_record_id(zone_id, record_name):
    """
    获取对应解析记录的record id
    
    参数:
    - zone_id: 区域ID
    - record_name: 记录名称 (如: www.example.com)
    """
    url = f'https://api.cloudflare.com/client/v4/zones/{zone_id}/dns_records?name={record_name}'
    headers = {
        'Authorization': f'Bearer {api_token}',
        'Content-Type': 'application/json'
    }
    response = requests.get(url, headers=headers, proxies=proxy)
    response_data = response.json()
    
    if response_data['success']:
        # 检查是否有记录返回
        if len(response_data['result']) > 0:
            return response_data['result'][0]['id']
        else:
            return None  # 记录不存在
    else:
        error_messages = [error['message'] for error in response_data['errors']]
        raise Exception(f'无法获取 Record ID, 错误信息: {"; ".join(error_messages)}')


def create_dns_record(zone_id, record_name, record_type, record_content, ttl=1, proxied=False):
    """
    添加DNS记录
    
    参数:
    - zone_id: 区域ID
    - record_name: 记录名称 (如: www.example.com)
    - record_type: 记录类型 (如: A, AAAA, CNAME, MX, TXT等)
    - record_content: 记录内容 (如: IP地址、目标域名等)
    - ttl: 生存时间，默认auto
    - proxied: 是否启用Cloudflare代理，默认False，proxied可能会影响ddns功能。
    """
    url = f'https://api.cloudflare.com/client/v4/zones/{zone_id}/dns_records'
    headers = {
        'Authorization': f'Bearer {api_token}',
        'Content-Type': 'application/json'
    }
    
    data = {
        'type': record_type,
        'name': record_name,
        'content': record_content,
        'ttl': ttl,
        'proxied': proxied
    }
    
    response = requests.post(url, headers=headers, json=data, proxies=proxy)
    response_data = response.json()
    
    if response_data['success']:
        record_id = response_data['result']['id']
        print(f'DNS记录添加成功, 记录ID: {record_id}')
        return record_id
    else:
        error_messages = [error['message'] for error in response_data['errors']]
        raise Exception(f'无法添加DNS记录, 错误信息: {"; ".join(error_messages)}')


def get_record_ip(zone_id, record_id):
    """
    根据zone id和record id查询对应解析记录的内容
    
    参数:
    - zone_id: 区域ID
    - record_id: 记录ID
    """
    url = f'https://api.cloudflare.com/client/v4/zones/{zone_id}/dns_records/{record_id}'
    headers = {
        'Authorization': f'Bearer {api_token}',
        'Content-Type': 'application/json'
        }
    response = requests.get(url, headers=headers, proxies=proxy)
    response_data = response.json()
    if response_data['success']:
        return response_data['result']['content']
    else:
        error_messages = [error['message'] for error in response_data['errors']]
        raise Exception(f'无法获取 Record IP, 错误信息: {"; ".join(error_messages)}')


def get_ipv6_address():
    """
    从 https://6.ipw.cn 获取本机ipv6地址
    """
    url = 'https://6.ipw.cn'
    response = requests.get(url)
    return response.text.strip()


def get_ipv4_address():
    """
    从 https://4.ipw.cn 获取本机ipv4地址
    """
    url = 'https://4.ipw.cn'
    response = requests.get(url)
    return response.text.strip()


def update_dns_record(zone_id, record_id, record_name, record_type, record_content):
    """
    更新DNS记录
    
    参数:
    - zone_id: 区域ID
    - record_id: 记录ID
    - record_name: 记录名称 (如: www.example.com)
    - record_type: 记录类型 (如: A, AAAA, CNAME, MX, TXT等)
    - record_content: 记录内容 (如: IP地址、目标域名等)
    """
    url = f'https://api.cloudflare.com/client/v4/zones/{zone_id}/dns_records/{record_id}'
    headers = {
        'Authorization': f'Bearer {api_token}',
        'Content-Type': 'application/json'
    }
    data = {
        'type': record_type,
        'name': record_name,
        'content': record_content,
        'ttl': 1,  # 设置 TTL 为 1，表示自动
        'proxied': False  # 是否启用 Cloudflare 的代理
    }
    response = requests.put(url, headers=headers, json=data, proxies=proxy)
    response_data = response.json()
    if response_data['success']:
        print('DNS 记录更新成功')
    else:
        error_messages = [error['message'] for error in response_data['errors']]
        raise Exception(f'无法更新DNS记录, 错误信息: {"; ".join(error_messages)}')


# 主函数
if __name__ == '__main__':
    try:
        # 通过根域和前缀拼接完整的域名
        record_name = f'{record_prefix}.{domain}'

        # 如果没有 Zone ID 或 Record ID，则获取
        if not zone_id:
            zone_id = get_zone_id(domain)

            if zone_id == None:
                raise Exception(f'Cloudflare未托管域名{domain}')
            
            print('zone id: ' + zone_id)
        if not record_id:
            record_id = get_record_id(zone_id, record_name)

        # 解析记录存在时，比较IP地址以减少DNS更新次数，避免风控
        if record_id != None:
            print('record id: ' + record_id)
    
            # 获取dns记录和本地ip地址
            record_address = ipaddress.IPv6Address(get_record_ip(zone_id, record_id))
            print(f'DNS记录IP地址: {record_address}')
            local_address = ipaddress.IPv6Address(get_ipv6_address())
            print(f'本地IP地址: {local_address}')

            # 对两个地址进行比较
            if local_address == record_address:
                # 如果本地 IP 和 Cloudflare 的记录 IP 相同，则不更新 DNS 记录
                print(f'本地IP 和 Cloudflare 的记录 IP 相同，无需更新({local_address})')
            else:
                # 如果本地 IP 和 Cloudflare 的记录 IP 不相同，则更新 DNS 记录
                print(f'本地IP 和 Cloudflare 的记录 IP 不相同，开始更新({record_address} -> {local_address})')
                update_dns_record(zone_id, record_id, record_name, 'AAAA', str(local_address))

        # 解析记录不存在时，创建新的解析记录
        else:
            print(f'{record_name}的解析记录不存在, 创建新的解析记录')

            local_address = ipaddress.IPv6Address(get_ipv6_address())
            print(f'本地IP地址: {local_address}')

            record_id = create_dns_record(zone_id, record_name, 'AAAA', str(local_address))
            print('record id: ' + record_id)
        
    except Exception as e:
        print(f'发生错误: {e}')