本脚本用于将指定子域名的 A 记录同步到 Cloudflare。通过自定义 DNS 服务器获取 IP，与 Cloudflare 现有解析对比后，如有变动则更新记录。 This script synchronizes A records to Cloudflare for specified subdomains. It fetches current IPs from a custom DNS resolver, compares them to Cloudflare’s existing records, and updates Cloudflare DNS entries if there are any changes.

sync_cf_records.py

#!/usr/bin/env python3
import os
import requests
import socket
import dns.resolver

# 你需要在环境变量中配置 Cloudflare API Token
# export CF_API_TOKEN="xxxx"

# go https://dash.cloudflare.com/profile/api-tokens
CF_API_TOKEN = ""
# os.environ.get("CF_API_TOKEN")
if not CF_API_TOKEN:
    print("请先在环境变量中设置 CF_API_TOKEN")
    exit(1)

# 这些域名对应的记录要从自定义 DNS 服务器查询
SUBDOMAINS_TO_QUERY = [
    "p01.ddns.your-domain.net",
    "p02.ddns.your-domain.net.",
    "p03.ddns.your-domain.net.",
    "p04.ddns.your-domain.net.",
    ]

# 目标域名和记录名称
ZONE_NAME = "your-domain.net"         # 在 Cloudflare 上的根域名
RECORD_NAME = "ddns.ytlink.net"   # 要更新的记录名称

# 可自定义 DNS Server，例: ["8.8.8.8"] 或你的私有 DNS
CUSTOM_RESOLVER = ["ns1.he.net"]


def get_zone_id():
    print("[INFO] 正在获取 Zone ID...")
    url = "https://api.cloudflare.com/client/v4/zones"
    params = {"name": ZONE_NAME}
    headers = {
        "Authorization": f"Bearer {CF_API_TOKEN}",
        "Content-Type": "application/json"
    }
    resp = requests.get(url, headers=headers, params=params)
    try:
        resp.raise_for_status()
    except requests.exceptions.HTTPError as e:
        print("[ERROR] 获取 Zone ID 失败，Cloudflare 返回：", resp.text)
        raise e

    data = resp.json()
    if not data["success"] or len(data["result"]) == 0:
        raise Exception(f"获取 Zone ID 失败: {resp.text}")

    zone_id = data["result"][0]["id"]
    print(f"[INFO] 成功获取 Zone ID: {zone_id}")
    return zone_id

def get_current_records(zone_id):
    print("[INFO] 正在获取现有 DNS 记录...")
    url = f"https://api.cloudflare.com/client/v4/zones/{zone_id}/dns_records"
    params = {"type": "A", "name": RECORD_NAME}
    headers = {
        "Authorization": f"Bearer {CF_API_TOKEN}",
        "Content-Type": "application/json"
    }
    resp = requests.get(url, headers=headers, params=params)
    try:
        resp.raise_for_status()
    except requests.exceptions.HTTPError as e:
        print("[ERROR] 获取现有记录失败，Cloudflare 返回：", resp.text)
        raise e

    data = resp.json()
    if not data["success"]:
        raise Exception(f"获取现有记录失败: {resp.text}")

    records = {r["content"]: r["id"] for r in data["result"]}
    print("[INFO] 获取到现有记录: ", records)
    return records

def delete_record(zone_id, record_id, ip):
    print(f"[INFO] 准备删除 IP={ip} 对应的记录 (record_id={record_id})")
    url = f"https://api.cloudflare.com/client/v4/zones/{zone_id}/dns_records/{record_id}"
    headers = {
        "Authorization": f"Bearer {CF_API_TOKEN}",
        "Content-Type": "application/json"
    }
    resp = requests.delete(url, headers=headers)
    try:
        resp.raise_for_status()
    except requests.exceptions.HTTPError as e:
        print("[ERROR] 删除记录失败，Cloudflare 返回：", resp.text)
        raise e

    data = resp.json()
    if not data["success"]:
        raise Exception(f"删除记录失败: {resp.text}")
    print(f"[INFO] 删除成功: IP={ip}")

def create_record(zone_id, ip):
    print(f"[INFO] 开始创建记录: {ip}")
    url = f"https://api.cloudflare.com/client/v4/zones/{zone_id}/dns_records"
    headers = {
        "Authorization": f"Bearer {CF_API_TOKEN}",
        "Content-Type": "application/json"
    }
    payload = {
        "type": "A",
        "name": RECORD_NAME,
        "content": ip,
        "ttl": 120,
        "proxied": False
    }
    resp = requests.post(url, headers=headers, json=payload)
    try:
        resp.raise_for_status()
    except requests.exceptions.HTTPError as e:
        print("[ERROR] 创建记录失败，Cloudflare 返回：", resp.text)
        raise e

    data = resp.json()
    if not data["success"]:
        raise Exception(f"创建记录失败: {resp.text}")
    print(f"[INFO] 创建成功: {ip}")

def resolve_nameserver_list(server_list):
    """
    支持在 CUSTOM_RESOLVER 中放域名或 IP。
    这个函数会把域名解析成 IP，再返回最终的 IP 列表。
    """
    resolved_ips = []
    for server in server_list:
        try:
            # 如果能成功 inet_aton，就说明是 IP
            socket.inet_aton(server)
            resolved_ips.append(server)
        except socket.error:
            # 如果不是 IP，就当作域名去解析
            print(f"[INFO] 解析自定义解析器域名: {server}")
            answer = dns.resolver.resolve(server, "A")
            for rdata in answer:
                resolved_ips.append(rdata.address)
    print("[INFO] 自定义 DNS 服务器最终解析为: ", resolved_ips)
    return resolved_ips

def resolve_subdomains():
    print("[INFO] 正在解析子域名... ")
    resolver = dns.resolver.Resolver()
    ns_ips = resolve_nameserver_list(CUSTOM_RESOLVER)
    resolver.nameservers = ns_ips

    new_ips = []
    for sub in SUBDOMAINS_TO_QUERY:
        print(f"[INFO] 解析 {sub} ...")
        answer = resolver.resolve(sub, "A")
        for rdata in answer:
            new_ips.append(rdata.address)
    print("[INFO] 子域名解析结果: ", new_ips)
    return new_ips

def main():
    print("[INFO] ================== 开始脚本执行 ==================")
    zone_id = get_zone_id()
    existing_records = get_current_records(zone_id)
    existing_ips = set(existing_records.keys())

    print("[INFO] 开始通过自定义 DNS 解析子域名...")
    new_ips = resolve_subdomains()
    new_ips_set = set(new_ips)

    print("[INFO] 对比旧记录与新记录...")
    ips_to_remove = [ip for ip in existing_ips if ip not in new_ips_set]
    ips_to_add = [ip for ip in new_ips_set if ip not in existing_ips]
    print("[INFO] 需要删除的 IP: ", ips_to_remove)
    print("[INFO] 需要添加的 IP: ", ips_to_add)

    for ip in ips_to_remove:
        record_id = existing_records[ip]
        delete_record(zone_id, record_id, ip)

    for ip in ips_to_add:
        create_record(zone_id, ip)

    print("[INFO] 记录更新完成。最终 IP 列表: ", new_ips)
    print("[INFO] ================== 脚本执行结束 ==================")

if __name__ == "__main__":
    main()