本文目录：

前几天，我的服务器遭受了2次DDoS，然后博客直接挂了1天，Cuckoo主题交流群的一位群友也接着被打掉了，搜索引擎还好没怎么掉排名，这也提醒了我要搞好备份和备用服务器的切换，claw这个厂商如果你被DDoS的次数太多就可能会被删鸡，我昨天也买了个1.5GB网页和数据库空间的带防护的虚拟主机作备份
然后今天就给服务器写了个定时备份并FTP同步到备用虚拟主机的Python脚本，再写好自动监测服务器在线并自动切换dns的脚本
自动切换DNS记录的Python脚本原理是这样的：
2025-05-25T05:24:40.png
那个监测的WebUpTimeAPI是我自己用的监测，文章后面发出来的脚本那一段会更改为用request去请求你服务器上搭建的网站（有重试机制）
定时备份上传的脚本原理大致是这样：
2025-05-25T05:43:20.png

py脚本

监测服务器在线、自动切换dns

依赖库安装命令

pip install requests paramiko cloudflare smtplib retrying

import time
import requests
import paramiko
import smtplib
from email.mime.text import MIMEText
from CloudFlare import CloudFlare
from retrying import retry

TARGET_URL = "https://example.com"
REQUEST_TIMEOUT = 10  # 单次请求超时时间（秒）

SSH_HOST = "ssh_host"
SSH_PORT = 22
SSH_USER = "ssh_username"
SSH_PASS = "ssh_password"

CLOUDFLARE_API_TOKEN = "cloudflare_api_token"
ZONE_NAME = "example.com"
RECORD_NAME = "www.example.com"  # 要修改的 DNS 主机名
NEW_IP = "127.0.0.1"  # 替换为要指向的新 IP 地址

SMTP_SERVER = "smtp.qiye.aliyun.com"
SMTP_PORT = 465
SMTP_USER = "you@yourcompany.com"
SMTP_PASSWORD = "your_email_password_or_auth_code"
EMAIL_TO = "3775610318@qq.com"

@retry(stop_max_attempt_number=3, wait_fixed=5000)
def check_website():
    try:
        print(f"正在请求 {TARGET_URL}")
        response = requests.get(TARGET_URL, timeout=REQUEST_TIMEOUT)
        if response.status_code == 200:
            print("网站正常")
            return True
        else:
            print(f"返回状态码: {response.status_code}")
            raise Exception("网站返回非200状态码")
    except Exception as e:
        print(f"请求失败: {e}")
        return False

def ssh_connect(retries=2):
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    for i in range(retries):
        try:
            print(f"正在尝试连接... 第 {i + 1} 次")
            ssh.connect(SSH_HOST, port=SSH_PORT, username=SSH_USER, password=SSH_PASS, timeout=10)
            print("连接成功")
            ssh.close()
            return True
        except Exception as e:
            print(f"连接失败: {e}")
            time.sleep(5)
    return False

def update_dns_record():
    try:
        cf = CloudFlare(token=CLOUDFLARE_API_TOKEN)
        zones = cf.zones.get(params={"name": ZONE_NAME})
        zone_id = zones[0]["id"]
        records = cf.zones.dns_records.get(zone_id, params={"name": RECORD_NAME})
        record_id = records[0]["id"]

        update_data = {
            "type": "A",
            "name": RECORD_NAME,
            "content": NEW_IP,
            "ttl": 120,
            "proxied": False
        }

        cf.zones.dns_records.put(zone_id, record_id, data=update_data)
        print("DNS记录已更新")
        return True
    except Exception as e:
        print(f"更新失败: {e}")
        return False

def send_email(subject, body):
    msg = MIMEText(body, 'plain', 'utf-8')
    msg['Subject'] = subject
    msg['From'] = SMTP_USER
    msg['To'] = EMAIL_TO

    try:
        server = smtplib.SMTP_SSL(SMTP_SERVER, SMTP_PORT)
        server.login(SMTP_USER, SMTP_PASSWORD)
        server.sendmail(SMTP_USER, [EMAIL_TO], msg.as_string())
        server.quit()
        print("通知邮件已发送")
    except Exception as e:
        print(f"发送失败: {e}")

def handle_exception():
    print("网站无法访问，开始尝试SSH连接...")
    if not ssh_connect():
        print("SSH连接失败，更改DNS并发送邮件通知")
        if update_dns_record():
            send_email(
                subject="⚠️ 服务器异常切换通知",
                body=f"检测到 {TARGET_URL} 无法访问且无法通过SSH连接，已将DNS记录 {RECORD_NAME} 指向备用服务器IP：{NEW_IP}"
            )
        else:
            print("DNS更新失败，未发送邮件")
    else:
        print("SSH连接正常")

if __name__ == "__main__":
    while True:
        try:
            is_up = check_website()
            if not is_up:
                handle_exception()
        except:
            print("请求过程中发生异常")
            handle_exception()

        time.sleep(600)  # 10分钟

自动备份Mysql/MariaDB数据上传到FTP

依赖库安装命令

pip install schedule requests

import os
import time
import datetime
import subprocess
import schedule
from ftplib import FTP
import requests

db_config = {
    "host": "localhost", # 数据库连接地址
    "user": "root", # 数据库用户名
    "password": "your_password",  # 数据库密码
    "database": "typecho", # 数据库名
}
backup_dir = "/data/dbackup" # 本地sql文件保存地址（临时的，上传到FTP后会自动删除）
ftp_config = {
    "host": "127.0.0.1", # FTP连接地址
    "user": "username", # FTP用户名
    "passwd": "password", #FTP密码
    "backup_folder": "/backupdb" # FTP sql文件保存位置（需提前创建）
}
notification_url = "https://back.xiaorin.com/backupdb/in.php"

def backup_database():
    try:
        # 创建备份目录（如果不存在）
        if not os.path.exists(backup_dir):
            os.makedirs(backup_dir)

        backup_file = os.path.join(backup_dir, f"{db_config['database']}.sql")

        # 构建 mysqldump 命令
        command = (
            f"mysqldump -h {db_config['host']} -u{db_config['user']} "
            f"-p{db_config['password']} {db_config['database']} > {backup_file}"
        )

        # 执行命令
        subprocess.run(command, shell=True, check=True)
        print(f"数据库已备份到: {backup_file}")

        if upload_to_ftp(backup_file):
            os.remove(backup_file)
            print(f"本地备份文件已删除: {backup_file}")
            send_notification()
        else:
            print("上传失败，未删除本地备份文件。")

    except Exception as e:
        print(f"数据库备份失败: {e}")

def upload_to_ftp(local_file_path):
    try:
        with FTP(ftp_config["host"]) as ftp:
            ftp.login(user=ftp_config["user"], passwd=ftp_config["passwd"])

            # 切换到远程备份文件夹
            ftp.cwd(ftp_config["backup_folder"])

            # 读取本地文件并上传
            with open(local_file_path, 'rb') as file:
                remote_filename = os.path.basename(local_file_path)
                ftp.storbinary(f'STOR {remote_filename}', file)
            print(f"文件已上传到FTP: {remote_filename}")
            return True
    except Exception as e:
        print(f"上传到FTP失败: {e}")
        return False

def send_notification():
    try:
        response = requests.get(notification_url, timeout=10)
        if response.status_code == 200:
            print(f"[通知] 已成功访问远程通知地址: {notification_url}")
        else:
            print(f"[警告] 访问通知地址返回状态码: {response.status_code}")
    except Exception as e:
        print(f"[错误] 无法访问通知地址: {e}")

def set_backup_schedule(interval_minutes):
    schedule.every(interval_minutes).minutes.do(backup_database)
    print(f"已设置每 {interval_minutes} 分钟自动备份一次，并上传至FTP。")

set_backup_schedule(60)

# 启动定时任务
if __name__ == "__main__":
    print("启动自动备份服务...")
    while True:
        schedule.run_pending()
        time.sleep(1)