import requests
import socket
import re
import time
API_TOKEN = "你的API令牌"
DNS_NAME = "你的完整域名" # 如 sub.example.com
def get_ipv6_address():
try:
for info in socket.getaddrinfo(socket.gethostname(), None, socket.AF_INET6):
addr = info[4][0]
if not addr.startswith("fe80"): # 排除 link-local 地址
return addr
except Exception as e:
print("获取IPv6失败：", e)
return None
def get_zone_id(session, headers, zone_name):
url = f"https://api.cloudflare.com/client/v4/zones?name={zone_name}"
resp = session.get(url, headers=headers).json()
if resp.get("success"):
return resp["result"][0]["id"]
print("获取 Zone ID 失败：", resp.get("errors"))
return None
def get_record(session, headers, zone_id, dns_name):
url = f"https://api.cloudflare.com/client/v4/zones/{zone_id}/dns_records?type=AAAA&name={dns_name}"
resp = session.get(url, headers=headers).json()
if resp.get("success") and resp["result"]:
return resp["result"][0]
return None
def update_record(session, headers, zone_id, record_id, dns_name, ipv6, proxied):
url = f"https://api.cloudflare.com/client/v4/zones/{zone_id}/dns_records/{record_id}"
data = {
"type": "AAAA",
"name": dns_name,
"content": ipv6,
"ttl": 1,
"proxied": proxied
}
resp = session.put(url, json=data, headers=headers).json()
return resp.get("success", False)
def create_record(session, headers, zone_id, dns_name, ipv6):
url = f"https://api.cloudflare.com/client/v4/zones/{zone_id}/dns_records"
data = {
"type": "AAAA",
"name": dns_name,
"content": ipv6,
"ttl": 1,
"proxied": False
}
resp = session.post(url, json=data, headers=headers).json()
return resp.get("success", False)
def main():
headers = {
"Authorization": f"Bearer {API_TOKEN}",
"Content-Type": "application/json"
}
zone_name = ".".join(DNS_NAME.split(".")[-2:])
with requests.Session() as session:
ipv6 = get_ipv6_address()
if not ipv6:
print("未找到有效的 IPv6 地址")
return
print(f"当前 IPv6：{ipv6}")
zone_id = get_zone_id(session, headers, zone_name)
if not zone_id:
return
record = get_record(session, headers, zone_id, DNS_NAME)
if record:
if record["content"] == ipv6:
print("IPv6 地址未变，无需更新")
else:
print("IPv6 地址已变，正在更新...")
if update_record(session, headers, zone_id, record["id"], DNS_NAME, ipv6, record["proxied"]):
print("更新成功")
else:
print("更新失败")
else:
print("未找到记录，正在创建...")
if create_record(session, headers, zone_id, DNS_NAME, ipv6):
print("创建成功")
else:
print("创建失败")
if __name__ == "__main__":
main()
