这里使用Google Cloud和Cloudflare来实现，解决海外服务器被误封IP，访问不到的问题。
这段脚本的核心目的，是自动监测你在 Cloudflare 上管理的 VPS 域名是否可达，一旦发现域名无法 Ping 通，就会帮你更换IP：
-
重置该实例的外部 IP（通过 gcloud compute instances delete-access-config +
add-access-config）， -
把新分配到的 IP 同步到 Cloudflare DNS，
-
刷新本地 DNS 缓存，
-
并且全程将操作日志保存到按天分文件的 info_YYYYMMDD.log 中，方便你审计和排查。
—— 具体是这样解决问题的 ——
| 问题场景 | 代码模块 | 解决办法 |
|---|---|---|
| VPS 外网 IP 被 GCP 回收或被 ISP 屏蔽 | is_ping_reachable() | 用系统 ping 命令检测域名连通性，如果返回码≠0，就判定“被屏蔽/不可达”。 |
| 需要把 VPS 的新外网 IP 同步到用户的域名 | update_external_ip() | 调用 gcloud compute instances delete-access-config + add-access-config，强制给实例分配一个新的外网 IP；再用 gcloud compute instances list 把这个 IP 取出来。 |
| 要让域名马上解析到新的 IP | update_dns() | 通过 Cloudflare REST API，查询并更新（或新建）对应 A 记录，把最新 IP 写入 DNS。 |
| 本地可能还在缓存旧的 DNS 记录 | flush_local_dns() | 在 Windows 下执行 ipconfig /flushdns，Linux 下尝试刷新 systemd-resolve/nscd/dnsmasq 的缓存，确保下一次访问能拿到新 IP。 |
| 整个流程要长期自动运行 | while True / time.sleep(300) | 把上面所有逻辑放到一个无限循环里，每轮处理完等待 5 分钟（300 秒）再重新「读配置 → 逐条处理」。 |
| 运维过程中出现任何错误也不影响后续 | try/except + 日志记录 | 每次对单个 VPS 的操作都包在 try…except 中，捕获异常后用 logging.error 把错误写日志，但不抛出，保证后面的 VPS 还能继续被处理。 |
总结：
-
监控（Ping）→ 修复（重置 IP + 更新 DNS）→ 验证（本地 DNS 刷新），
-
全程自动化、可重试、有日志，从根本上解决了 VPS 外网 IP 不可用时，域名无法访问的问题。
import os
import sys
import requests
import json
import platform
import subprocess
import re
import shutil
import time
from datetime import datetime
from pathlib import Path
import logging"""
脚本功能：
1. 轮询 vps.config.json 中的 VPS 列表；
2. 若域名无法 ping 通，则重置实例外网 IP（gcloud）并同步到 Cloudflare；
3. 成功后刷新本地主机 DNS 缓存（Windows / Linux 通用）；
4. 所有日志按日期写入当前目录的 info_YYYYMMDD.log，并保留 print 直观输出；
5. 出现任何异常仅记录日志，不影响继续处理下一条 VPS；
6. 每完成一轮后暂停 5 分钟，重新加载配置再开始下一轮。
"""# === Cloudflare 相关参数 ===
API_TOKEN = "HGccH7pI34n65ZAz3MkBM1QGAQB7xo69_40J1" # 具有 DNS 编辑权限
ZONE_NAME = "askdfjsdd5.xyz" # 顶级域名
PROXIED = False # 关闭橙云（CDN）
TTL_SECONDS = 60 # TTL 最小值 60 秒CF_API = "https://api.cloudflare.com/client/v4"
HEADERS = {"Authorization": f"Bearer {API_TOKEN}","Content-Type": "application/json"
}# ================ 日志配置 ================
log_file = Path(__file__).with_name(f"info_{datetime.now():%Y%m%d}.log")
logging.basicConfig(level=logging.INFO,format="[%(asctime)s] %(message)s",datefmt="%Y-%m-%d %H:%M:%S",handlers=[logging.FileHandler(log_file, encoding="utf-8")]
)# ================ 数据结构 ================
class VPSInfo:"""保存单个 VPS/域名 的相关信息"""def __init__(self, proId: str, area: str, vpsId: str, dn: str):self.proId = proId # GCP 项目 IDself.area = area # GCP 可用区self.vpsId = vpsId # 实例名称self.dn = dn # 对应域名def __repr__(self) -> str:return f"VPSInfo(proId={self.proId}, area={self.area}, vpsId={self.vpsId}, dn={self.dn})"# ================ 工具函数 ================
def load_vps_list(path: str = "vps.config.json"):"""加载配置文件并返回 VPSInfo 列表"""with open(path, "r", encoding="utf-8") as f:data = json.load(f)return [VPSInfo(item["proId"], item["area"], item["vpsId"], item["dn"]) for item in data]def is_ping_reachable(domain: str) -> bool:"""ping 判断域名可达性"""count_param = "-n" if platform.system().lower() == "windows" else "-c"try:res = subprocess.run(["ping", count_param, "1", domain],stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL)return res.returncode == 0except Exception:return Falsedef cf(method: str, path: str, **kw):"""调用 Cloudflare API，成功返回 result"""url = CF_API + pathresp = requests.request(method, url, headers=HEADERS, timeout=30, **kw)resp.raise_for_status()data = resp.json()if not data.get("success"):raise RuntimeError(f"Cloudflare API 调用失败: {data.get('errors')}")return data["result"]def update_dns(record_name: str, new_ip: str):"""创建或更新 A 记录"""zone = cf("GET", f"/zones?name={ZONE_NAME}&status=active&per_page=1")if not zone:raise RuntimeError(f"未找到域 {ZONE_NAME} 对应的 Zone")zone_id = zone[0]["id"]body = {"type": "A", "name": record_name, "content": new_ip,"ttl": TTL_SECONDS, "proxied": PROXIED}recs = cf("GET", f"/zones/{zone_id}/dns_records?type=A&name={record_name}&per_page=1")if recs:rec_id = recs[0]["id"]cf("PUT", f"/zones/{zone_id}/dns_records/{rec_id}", json=body)print(f"已更新 DNS：{record_name} → {new_ip}")logging.info(f"已更新 DNS：{record_name} → {new_ip}")else:rec = cf("POST", f"/zones/{zone_id}/dns_records", json=body)print(f"已创建 DNS：{record_name} → {new_ip} (id={rec['id']})")logging.info(f"已创建 DNS：{record_name} → {new_ip} (id={rec['id']})")def flush_local_dns():"""刷新本机 DNS 缓存（Windows / Linux）"""system = platform.system().lower()cmds = [["ipconfig", "/flushdns"]] if system == "windows" else [["systemd-resolve", "--flush-caches"],["resolvectl", "flush-caches"],["service", "nscd", "restart"],["service", "dnsmasq", "restart"],]for argv in cmds:if shutil.which(argv[0]):res = subprocess.run(argv, stdout=subprocess.PIPE,stderr=subprocess.PIPE, text=True)if res.returncode == 0:print(f"已刷新本地 DNS：{' '.join(argv)}")logging.info(f"已刷新本地 DNS：{' '.join(argv)}")returnprint("刷新本地 DNS 失败：未找到可用命令")logging.warning("刷新本地 DNS 失败：未找到可用命令")def update_external_ip(vps: VPSInfo) -> str:"""重置外网 IP 并返回新的 IP"""gcloud = shutil.which("gcloud") or shutil.which("gcloud.cmd")if not gcloud:raise RuntimeError("找不到 gcloud，可执行文件不在 PATH 中")cmds = [[gcloud, "compute", "instances", "describe", vps.vpsId,f"--project={vps.proId}", f"--zone={vps.area}","--format=get(networkInterfaces[0].accessConfigs[0].name)"],[gcloud, "compute", "instances", "delete-access-config", vps.vpsId,f"--project={vps.proId}", f"--zone={vps.area}","--access-config-name=External NAT"],[gcloud, "compute", "instances", "add-access-config", vps.vpsId,f"--project={vps.proId}", f"--zone={vps.area}","--access-config-name=External NAT"],[gcloud, "compute", "instances", "list",f"--filter=name={vps.vpsId}"]]last_output = ""for idx, argv in enumerate(cmds, 1):proc = subprocess.run(argv, stdout=subprocess.PIPE,stderr=subprocess.PIPE, text=True)if proc.returncode != 0:raise RuntimeError(f"第 {idx} 步命令失败: {' '.join(argv)}\n{proc.stderr.strip()}")last_output = proc.stdoutprint(f"第 {idx} 步命令执行成功")logging.info(f"第 {idx} 步命令执行成功")ips = re.findall(r"\b(?:\d{1,3}\.){3}\d{1,3}\b", last_output)if not ips:raise RuntimeError("未在实例列表输出中找到 IP")new_ip = ips[-1]print(f"获得新外部 IP: {new_ip}")logging.info(f"获得新外部 IP: {new_ip}")return new_ip# ================ 主流程 ================
def process_once():"""处理一轮：读取配置并遍历所有 VPS"""vps_list = load_vps_list()for vps in vps_list:print(f"开始处理: {vps}")logging.info(f"开始处理: {vps}")try:if is_ping_reachable(vps.dn):print(f"{vps.dn} 可 ping，跳过")logging.info(f"{vps.dn} 可 ping，跳过")else:print(f"{vps.dn} 不可 ping，准备重置 IP")logging.info(f"{vps.dn} 不可 ping，准备重置 IP")new_ip = update_external_ip(vps)update_dns(vps.dn, new_ip)flush_local_dns()print("等待 60 秒以确保解析生效")logging.info("等待 60 秒以确保解析生效")time.sleep(60)except Exception as e:print(f"处理 {vps.dn} 时发生异常: {e}")logging.error(f"处理 {vps.dn} 时发生异常: {e}")finally:print(f"完成处理: {vps}\n")logging.info(f"完成处理: {vps}\n")if __name__ == "__main__":while True: # 无限循环，每轮结束休眠 5 分钟try:process_once()except Exception as e:print(f"脚本发生致命错误: {e}")logging.critical(f"脚本发生致命错误: {e}")print("=== 本轮结束，暂停 5 分钟 ===\n")logging.info("=== 本轮结束，暂停 5 分钟 ===\n")time.sleep(300) # 300 秒 = 5 分钟
)