本文提供 Cloudflare DNS 记录批量清理脚本：先获取有限分页快照与 JSON/BIND 备份，默认仅预览，确认目标后逐条删除，遇到失败立即停止。执行前需核对目标 Zone 和备份。
清空 DNS 会同时影响网站、邮件、验证记录等。这里的“全部”指执行时在指定 Zone 中列出的记录快照；运行期间应暂停其他 DNS 修改。不要把删除当作关闭代理或暂停站点的替代操作。
一、准备权限和目标
使用仅有目标 Zone 的 DNS Write 权限、且能读取该 Zone 信息和 DNS 列表的 API Token。凭据通过 CF_API_TOKEN 环境变量提供，不写入脚本、命令历史或备份。zone_id 与账户 ID 不是同一个值，脚本会再读取 Zone 名称并和你输入的完整域名比较。
安装依赖：
python -m pip install requests
脚本使用 per_page=100 作为便于检查的小批次，这不是 API 允许的最大值；参数范围以当前列表文档为准。Cloudflare DNS 列表 API
二、保存脚本
保存为 clear_dns.py。默认只读取和备份；加 --apply 后还要输入域名及记录数，才会删除。写操作失败立即停止，不自动重试，也不会继续打印“全部成功”。
import argparse
import json
import os
from pathlib import Path
from datetime import datetime, timezone
import uuid
import requests
API = 'https://api.cloudflare.com/client/v4'
PER_PAGE = 100
def api(session, method, path, **kwargs):
response = session.request(method, API + path, timeout=(10, 30), **kwargs)
response.raise_for_status()
data = response.json()
if data.get('success') is not True:
raise RuntimeError(f'Cloudflare API 失败：{data.get("errors", [])}')
return data
def snapshot(session, zone_id):
path = f'/zones/{zone_id}/dns_records'
first = api(session, 'GET', path, params={'page': 1, 'per_page': PER_PAGE})
info = first['result_info']
pages = int(info['total_pages'])
expected = int(info['total_count'])
records = list(first['result'])
for page in range(2, pages + 1):
data = api(session, 'GET', path, params={'page': page, 'per_page': PER_PAGE})
if int(data['result_info']['total_count']) != expected:
raise RuntimeError('DNS 数量在分页中改变，请暂停其他修改后重试')
records.extend(data['result'])
ids = [record['id'] for record in records]
if len(ids) != len(set(ids)) or len(ids) != expected:
raise RuntimeError('分页记录数量或唯一性检查失败，未执行删除')
return records
def save_private(path, text):
# 备份包含域名和源站信息；Unix 下限制权限，Windows 需配合目录 ACL。
fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
with os.fdopen(fd, 'w', encoding='utf-8') as handle:
handle.write(text)
def main():
parser = argparse.ArgumentParser()
parser.add_argument('--zone-id', required=True)
parser.add_argument('--zone-name', required=True)
parser.add_argument('--backup', default='dns-backups')
parser.add_argument('--apply', action='store_true')
args = parser.parse_args()
token = os.environ.get('CF_API_TOKEN')
if not token:
raise RuntimeError('请先在当前进程设置 CF_API_TOKEN')
with requests.Session() as session:
session.headers['Authorization'] = f'Bearer {token}'
zone = api(session, 'GET', f'/zones/{args.zone_id}')['result']
name = zone['name']
if name.lower() != args.zone_name.rstrip('.').lower():
raise RuntimeError(f'Zone 名称不匹配：实际为 {name}')
records = snapshot(session, args.zone_id)
export = session.request('GET', API + f'/zones/{args.zone_id}/dns_records/export',
timeout=(10, 30))
export.raise_for_status()
# API 错误不是 BIND 文件，不能把 JSON 错误响应当作可恢复备份。
if 'json' in export.headers.get('Content-Type', '').lower() or not export.text.strip():
raise RuntimeError('BIND 导出未返回有效文本，未执行删除')
stamp = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')
folder = Path(args.backup) / f'{stamp}-{uuid.uuid4().hex[:8]}'
folder.mkdir(parents=True, mode=0o700, exist_ok=False)
save_private(folder / 'records.json', json.dumps(
{'zone_id': args.zone_id, 'zone_name': name, 'records': records},
ensure_ascii=False, indent=2))
save_private(folder / 'records.bind', export.text)
print(f'目标：{name}，快照：{len(records)} 条，备份：{folder}')
for record in records[:10]:
print(record['type'], record['name'], record['content'])
if not args.apply or not records:
print('预览结束，未删除记录。')
return
confirmation = f'{name} {len(records)}'
if input(f'如确认删除，输入 {confirmation!r}：').strip() != confirmation:
raise RuntimeError('确认不匹配，未执行删除')
# 确认等待期间也可能有人修改；快照变动则要求重新预览。
current = snapshot(session, args.zone_id)
if current != records:
raise RuntimeError('记录已变动，请重新预览和备份')
log = folder / 'deleted-ids.txt'
for index, record in enumerate(records, start=1):
result = api(session, 'DELETE',
f'/zones/{args.zone_id}/dns_records/{record["id"]}')['result']
if result.get('id') != record['id']:
raise RuntimeError('删除响应 ID 不匹配，已停止；请检查线上状态')
# 只记录已经被 API 确認删除的 ID，便于中断后核对。
fd = os.open(log, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600)
with os.fdopen(fd, 'a', encoding='utf-8') as handle:
handle.write(record['id'] + '\n')
print(f'已删除 {index}/{len(records)}：{record["type"]} {record["name"]}')
remaining = snapshot(session, args.zone_id)
target_ids = {record['id'] for record in records}
if target_ids & {record['id'] for record in remaining}:
raise RuntimeError('复核仍有目标记录，请检查备份和删除日志')
print(f'快照中的 {len(records)} 条已删除；当前 Zone 还有 {len(remaining)} 条记录。')
if __name__ == '__main__':
main()
三、先预览，再执行
在当前终端交互式设置环境变量后，先运行：
python clear_dns.py --zone-id YOUR_ZONE_ID --zone-name example.com
核对备份中的 A、AAAA、CNAME、MX、TXT、SRV 等记录，再重新运行：
python clear_dns.py --zone-id YOUR_ZONE_ID --zone-name example.com --apply
第二次运行会重新备份和确认。脚本不会删除快照之后新增的记录；分页不是事务，也无法阻止其他人在删除期间修改，实际操作必须安排维护窗口。遇到 429、超时或 API 错误，依据备份、已删 ID 日志和线上列表判断完成到哪一步，再决定是否继续。不要假定超时代表删除没有发生。删除 API、Requests 超时说明
BIND 文件用于恢复 DNS 内容，JSON 用于核对 Cloudflare 字段。导入 BIND 不应被描述为逐字段恢复代理状态、备注和所有账户设置；导入后按 JSON 逐项核查，并测试网站与邮件。Cloudflare 导入与导出说明
批量处理思路参考《使用 SHELL 批量删除 Cloudflare 解析记录》。本文 Python 脚本的字段和调用方式以 Cloudflare 官方 API 文档为依据。
评论区