网站 开发 成本,推广图片发布混淆信息,深圳做棋牌网站建设哪家好,百度推广包做网站吗这里使用Google Cloud和Cloudflare来实现#xff0c;解决海外服务器被误封IP#xff0c;访问不到的问题。
这段脚本的核心目的#xff0c;是自动监测你在 Cloudflare 上管理的 VPS 域名是否可达#xff0c;一旦发现域名无法 Ping 通#xff0c;就会帮你更换IP#xff1a…这里使用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 取出来。要让域名马上解析到新的 IPupdate_dns()通过 Cloudflare REST API查询并更新或新建对应 A 记录把最新 IP 写入 DNS。本地可能还在缓存旧的 DNS 记录flush_local_dns()在 Windows 下执行 ipconfig /flushdnsLinux 下尝试刷新 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 通则重置实例外网 IPgcloud并同步到 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: fBearer {API_TOKEN},Content-Type: application/json
}# 日志配置
log_file Path(__file__).with_name(finfo_{datetime.now():%Y%m%d}.log)
logging.basicConfig(levellogging.INFO,format[%(asctime)s] %(message)s,datefmt%Y-%m-%d %H:%M:%S,handlers[logging.FileHandler(log_file, encodingutf-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 fVPSInfo(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, encodingutf-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 -ctry:res subprocess.run([ping, count_param, 1, domain],stdoutsubprocess.DEVNULL,stderrsubprocess.DEVNULL)return res.returncode 0except Exception:return Falsedef cf(method: str, path: str, **kw):调用 Cloudflare API成功返回 resulturl CF_API pathresp requests.request(method, url, headersHEADERS, timeout30, **kw)resp.raise_for_status()data resp.json()if not data.get(success):raise RuntimeError(fCloudflare 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}statusactiveper_page1)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?typeAname{record_name}per_page1)if recs:rec_id recs[0][id]cf(PUT, f/zones/{zone_id}/dns_records/{rec_id}, jsonbody)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, jsonbody)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 / Linuxsystem 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, stdoutsubprocess.PIPE,stderrsubprocess.PIPE, textTrue)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 并返回新的 IPgcloud 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},--formatget(networkInterfaces[0].accessConfigs[0].name)],[gcloud, compute, instances, delete-access-config, vps.vpsId,f--project{vps.proId}, f--zone{vps.area},--access-config-nameExternal NAT],[gcloud, compute, instances, add-access-config, vps.vpsId,f--project{vps.proId}, f--zone{vps.area},--access-config-nameExternal NAT],[gcloud, compute, instances, list,f--filtername{vps.vpsId}]]last_output for idx, argv in enumerate(cmds, 1):proc subprocess.run(argv, stdoutsubprocess.PIPE,stderrsubprocess.PIPE, textTrue)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():处理一轮读取配置并遍历所有 VPSvps_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 分钟
