静态IP vs 动态IP：万字长文讲透区别、成本、场景与选型，别再花冤枉钱！

#!/usr/bin/env python3
"""
阿里云 DNS DDNS 更新脚本
功能：检测本机公网IP，若发生变化则自动更新阿里云域名解析记录。
"""

import json
import time
import hashlib
import hmac
import base64
import urllib.request
import urllib.parse
from typing import Optional, Dict

# ========== 用户配置区 ==========
ACCESS_KEY_ID = "你的AccessKey ID"
ACCESS_KEY_SECRET = "你的AccessKey Secret"
DOMAIN_NAME = "yourdomain.com"  # 你的主域名
RR = "ddns"  # 子域名前缀，例如 ddns.yourdomain.com
RECORD_TYPE = "A"  # 记录类型，通常为 A (IPv4) 或 AAAA (IPv6)
TTL = 600  # 记录TTL，单位秒，建议不低于600
# ===============================

class AliDNSClient:
    """阿里云 DNS API 简易客户端"""
    ENDPOINT = "https://alidns.aliyuncs.com"
    API_VERSION = "2015-01-09"

    def __init__(self, access_key_id: str, access_key_secret: str):
        self.access_key_id = access_key_id
        self.access_key_secret = access_key_secret

    def _sign(self, parameters: Dict[str, str]) -> str:
        """生成请求签名 (Signature)"""
        sorted_params = sorted(parameters.items())
        canonicalized_query_string = '&'.join(
            [f"{urllib.parse.quote(k, safe='')}={urllib.parse.quote(v, safe='')}" for k, v in sorted_params]
        )
        string_to_sign = f"GET&%2F&{urllib.parse.quote(canonicalized_query_string, safe='')}"
        key = self.access_key_secret + '&'
        digest = hmac.new(key.encode('utf-8'), string_to_sign.encode('utf-8'), hashlib.sha1).digest()
        return base64.b64encode(digest).decode()

    def _call_api(self, action: str, **kwargs) -> Dict:
        """调用阿里云 DNS API"""
        # 构造公共参数
        params = {
            "Format": "JSON",
            "Version": self.API_VERSION,
            "AccessKeyId": self.access_key_id,
            "SignatureMethod": "HMAC-SHA1",
            "Timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
            "SignatureVersion": "1.0",
            "SignatureNonce": str(int(time.time() * 1000)),
            "Action": action,
        }
        # 添加业务参数
        params.update(kwargs)
        # 计算签名
        params["Signature"] = self._sign(params)
        # 发送请求
        url = f"{self.ENDPOINT}/?{urllib.parse.urlencode(params)}"
        req = urllib.request.Request(url)
        try:
            with urllib.request.urlopen(req, timeout=10) as response:
                result = json.loads(response.read().decode())
                if result.get("Code") is not None:
                    raise Exception(f"API Error: {result.get('Code')} - {result.get('Message')}")
                return result
        except urllib.error.URLError as e:
            raise Exception(f"Network error: {e.reason}")
        except json.JSONDecodeError as e:
            raise Exception(f"Failed to parse API response: {e}")

    def get_record_id(self, domain: str, rr: str, record_type: str) -> Optional[str]:
        """获取指定解析记录的 RecordId"""
        resp = self._call_api("DescribeDomainRecords", DomainName=domain)
        for record in resp.get("DomainRecords", {}).get("Record", []):
            if record.get("RR") == rr and record.get("Type") == record_type:
                return record.get("RecordId")
        return None

    def update_record(self, domain: str, rr: str, record_type: str, value: str, ttl: int) -> bool:
        """更新解析记录，如果记录不存在则创建"""
        record_id = self.get_record_id(domain, rr, record_type)
        action = "UpdateDomainRecord" if record_id else "AddDomainRecord"
        params = {
            "DomainName": domain,
            "RR": rr,
            "Type": record_type,
            "Value": value,
            "TTL": ttl,
        }
        if record_id:
            params["RecordId"] = record_id
        resp = self._call_api(action, **params)
        print(f"✅ 记录 {'更新' if record_id else '创建'}成功: {rr}.{domain} -> {value}")
        return True

def get_public_ip() -> Optional[str]:
    """获取本机当前公网 IPv4 地址"""
    services = [
        "https://api.ipify.org",
        "https://icanhazip.com",
        "https://ident.me",
    ]
    for url in services:
        try:
            with urllib.request.urlopen(url, timeout=5) as response:
                ip = response.read().decode().strip()
                if ip and '.' in ip:  # 简单 IPv4 验证
                    return ip
        except:
            continue
    return None

def main():
    """主函数：检查IP变化并更新DNS记录"""
    # 1. 初始化客户端
    try:
        client = AliDNSClient(ACCESS_KEY_ID, ACCESS_KEY_SECRET)
    except Exception as e:
        print(f"❌ 初始化阿里云客户端失败: {e}")
        return

    # 2. 获取当前公网IP
    current_ip = get_public_ip()
    if not current_ip:
        print("❌ 无法获取公网IP地址，请检查网络连接")
        return
    print(f"🌐 当前检测到的公网IP: {current_ip}")

    # 3. 读取上次记录的IP (简易持久化，可替换为文件/数据库)
    last_ip_file = "last_ip.txt"
    last_ip = None
    try:
        with open(last_ip_file, "r") as f:
            last_ip = f.read().strip()
    except FileNotFoundError:
        pass

    # 4. 如果IP发生变化，则更新DNS记录
    if current_ip != last_ip:
        print(f"🔄 检测到IP变化 ({last_ip or '无记录'} -> {current_ip})，正在更新DNS记录...")
        try:
            success = client.update_record(DOMAIN_NAME, RR, RECORD_TYPE, current_ip, TTL)
            if success:
                # 更新本地记录的IP
                with open(last_ip_file, "w") as f:
                    f.write(current_ip)
        except Exception as e:
            print(f"❌ 更新DNS记录失败: {e}")
    else:
        print("✅ IP未发生变化，无需更新。")

if __name__ == "__main__":
    main()