cat aliyun_ddns.py

import requests
import json
import os
import datetime
# 安裝aliyunsdkcore、aliyunsdkcore，pip install aliyun-python-sdk-core-v3 aliyun-python-sdk-alidns aliyun-fc2
import aliyunsdkcore
import aliyunsdkalidns

# 阿里云配置  --- 修改：修改對應value
ACCESS_KEY_ID = "L******************UX"
ACCESS_KEY_SECRET = "9u****************3"
DOMAIN_NAME = "yuming.com"
# REGION固定即可
REGION = "cn-hangzhou"

# RECORD_NAMES :# 要綁定的 RR 值，域名前綴，例如www.本次使用到v6.yuming.com，v4.yuming.com。 RECORD_TYPES：v6對應 AAAA記錄類型, v4對應A記錄類型
RECORD_NAMES_TYPE = {
    'v4': 'A',
    'v6': 'AAAA'
}

# 日志配置 --- 修改：根據自身需求要修改參數
# 是否記錄日志 yes:記錄日志，no：不記錄日志，只在控制臺輸出。
WHETHER_RECORD_LOG='no'
# 日志文本名稱：位置為腳本所在位置
LOG_DIR="me-test.log"

# ===============================================
def log(message):
    # 帶時間戳的日志記錄
    now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    if WHETHER_RECORD_LOG == 'yes' :
        # 如下代碼將日志追加到文件中
        # 獲取當前腳本所在的目錄
        current_dir = os.path.dirname(os.path.abspath(__file__))
        # 構建日志文件的完整路徑
        log_file_path = os.path.join(current_dir, LOG_DIR)
        # 以追加模式打開文件
        with open(log_file_path, 'a') as file:
            # 向文件中追加內容
            file.write(f"[{now}] {message}\n")
    elif WHETHER_RECORD_LOG == 'no':
        print(f"[{now}] {message}")

def get_public_ip():
    """獲取公網 IP 地址方法一：此只能獲取到ipv4
    try:
        response = requests.get("http://www.3322.org/dyndns/getip", timeout=5)
        response.raise_for_status()
        ip_address = response.text.strip()
        return ip_address
    except requests.exceptions.RequestException as e:
        log(f"獲取公網 IP 失敗: {e}")
        return None
    """
    """獲取公網 IP 地址方法二：獲取ipv4+ipv6公網地址"""
    try:
        response_ipv4 = requests.get("http://ipv4.lookup.test-ipv6.com/ip/", timeout=5)
        response_ipv6 = requests.get("http://ipv6.lookup.test-ipv6.com/ip/", timeout=5)
        ipv4 = json.loads(response_ipv4.text.strip())['ip']
        ipv6 = json.loads(response_ipv6.text.strip())['ip']
        ip_address = {}
        ip_address['v4'] = ipv4
        ip_address['v6'] = ipv6
        return ip_address
    except requests.exceptions.RequestException as e:
        log(f"獲取公網 IP 失敗: {e}")
        return None

def get_aliyun_dns_record_id(record_name, ip_address, record_type):
    """獲取阿里云 DNS 記錄的 RecordId"""
    from aliyunsdkcore.client import AcsClient
    from aliyunsdkalidns.request.v20150109.DescribeDomainRecordsRequest import DescribeDomainRecordsRequest
    client = AcsClient(ACCESS_KEY_ID, ACCESS_KEY_SECRET, REGION)
    request = DescribeDomainRecordsRequest()
    request.set_DomainName(DOMAIN_NAME)
    request.set_RRKeyWord(record_name)  # 使用 RRKeyWord
    request.set_TypeKeyWord(record_type)
    request.set_accept_format('json')
    response = client.do_action_with_exception(request)
    response_json = json.loads(response.decode('utf-8'))
    #
    for record in response_json["DomainRecords"]["Record"]:
        if record["RR"] == record_name and record["Type"] == record_type:  # Double Check
            return record["RecordId"], record["Value"]
    return None, None

def update_aliyun_dns_record(record_id, record_name, ip_address, old_ip, record_type):
    """更新阿里云 DNS 記錄"""
    from aliyunsdkcore.client import AcsClient
    from aliyunsdkalidns.request.v20150109.UpdateDomainRecordRequest import UpdateDomainRecordRequest
    if old_ip == ip_address:
        log(f"域名: {DOMAIN_NAME}, RR: {record_name} IP地址沒有變化，無需更新")
        return
    client = AcsClient(ACCESS_KEY_ID, ACCESS_KEY_SECRET, REGION)
    request = UpdateDomainRecordRequest()
    request.set_RecordId(record_id)
    request.set_RR(record_name)
    request.set_Type(record_type)
    request.set_Value(ip_address)
    request.set_accept_format('json')

    response = client.do_action_with_exception(request)
    log(str(response, encoding='utf-8'))  # 打印完整響應信息

def process_record(record_name, public_ip, record_type):
    """處理單個 DNS 記錄"""
    record_id, old_ip = get_aliyun_dns_record_id(record_name, public_ip, record_type)
    if not record_id:
        log(f"未找到匹配的 DNS 記錄 (域名: {DOMAIN_NAME}, RR: {record_name}, Type: {record_type})。請檢查配置。")
        return

    log(f"找到 DNS 記錄 ID: {record_id}, 域名: {DOMAIN_NAME}, RR: {record_name}, 當前IP地址: {old_ip}")

    update_aliyun_dns_record(record_id, record_name, public_ip, old_ip, record_type)
    log(f"域名: {DOMAIN_NAME}, RR: {record_name} DNS 記錄更新完成。")

def main():
    """主函數"""
    all_public_ip = get_public_ip()
    if not all_public_ip:
        log("未能獲取公網 IP 地址，程序終止。")
        return

    log(f"當前公網 IP 地址: {all_public_ip}")
    for record_name in RECORD_NAMES_TYPE.keys():
        record_type = RECORD_NAMES_TYPE[record_name]
        public_ip = all_public_ip[record_name]
        process_record(record_name, public_ip, record_type)


if __name__ == "__main__":
    # 檢查是否安裝了 aliyun-python-sdk-core 和 aliyun-python-sdk-alidns
    try:
        import aliyunsdkcore
        import aliyunsdkalidns
    except ImportError:
        print("缺少必要的 Python 庫。請先安裝 aliyun-python-sdk-core 和 aliyun-python-sdk-alidns。")
        print(
            "使用命令: pip install aliyun-python-sdk-core-v3 aliyun-python-sdk-alidns aliyun-fc2")
        os._exit(1)
    # 如果記錄日志到文本需要，初始化日志分隔符
    if WHETHER_RECORD_LOG == 'yes':
        # 獲取當前腳本所在的目錄
        current_dir = os.path.dirname(os.path.abspath(__file__))
        # 構建日志文件的完整路徑
        log_file_path = os.path.join(current_dir, LOG_DIR)
        with open(log_file_path, 'a') as file:
            # 向文件中追加內容
            file.write(f"=" * 20 + '\n')
    main()