DDNS/run.py

#!/usr/bin/env python
# -*- coding:utf-8 -*-
"""
DDNS
@author: New Future
@modified: rufengsuixing
"""

from os import path, environ, name as os_name
from io import TextIOWrapper
from subprocess import check_output
from tempfile import gettempdir
from logging import basicConfig, info, warning, error, debug, DEBUG, NOTSET

import sys

from util import ip
from util.cache import Cache
from util.config import init_config, get_config

__version__ = "${BUILD_VERSION}@${BUILD_DATE}"  # CI 时会被Tag替换
__description__ = "automatically update DNS records to my IP [域名自动指向本机IP]"
__doc__ = """
ddns[%s]
(i) homepage or docs [文档主页]: https://ddns.newfuture.cc/
(?) issues or bugs [问题和帮助]: https://github.com/NewFuture/DDNS/issues
Copyright (c) New Future (MIT License)
""" % (__version__)

environ["DDNS_VERSION"] = "${BUILD_VERSION}"


def is_false(value):
    """
    判断值是否为 False
    字符串 'false', 或者 False, 或者 'none';
    0 不是 False
    """
    if isinstance(value, str):
        return value.strip().lower() in ['false', 'none']
    return value is False


def get_ip(ip_type, index="default"):
    """
    get IP address
    """
    # CN: 捕获异常
    # EN: Catch exceptions
    value = None
    try:
        debug("get_ip(%s, %s)", ip_type, index)
        if is_false(index):  # disabled
            return False
        elif isinstance(index, list):  # 如果获取到的规则是列表，则依次判断列表中每一个规则，直到获取到IP
            for i in index:
                value = get_ip(ip_type, i)
                if value:
                    break
        elif str(index).isdigit():  # 数字 local eth
            value = getattr(ip, "local_v" + ip_type)(index)
        elif index.startswith('cmd:'):  # cmd
            value = str(check_output(index[4:]).strip().decode('utf-8'))
        elif index.startswith('shell:'):  # shell
            value = str(check_output(
                index[6:], shell=True).strip().decode('utf-8'))
        elif index.startswith('url:'):  # 自定义 url
            value = getattr(ip, "public_v" + ip_type)(index[4:])
        elif index.startswith('regex:'):  # 正则 regex
            value = getattr(ip, "regex_v" + ip_type)(index[6:])
        else:
            value = getattr(ip, index + "_v" + ip_type)()
    except Exception as e:
        error("Failed to get %s address: %s", ip_type, e)
    return value


def change_dns_record(dns, proxy_list, **kw):
    for proxy in proxy_list:
        if not proxy or (proxy.upper() in ['DIRECT', 'NONE']):
            dns.Config.PROXY = None
        else:
            dns.Config.PROXY = proxy
        record_type, domain = kw['record_type'], kw['domain']
        info("%s(%s) ==> %s [via %s]", domain, record_type, kw['ip'], proxy)
        try:
            return dns.update_record(domain, kw['ip'], record_type=record_type)
        except Exception as e:
            error("Failed to update %s record for %s: %s", record_type, domain, e)
    return False


def update_ip(ip_type, cache, dns, proxy_list):
    """
    更新IP
    """
    ipname = 'ipv' + ip_type
    domains = get_config(ipname)
    if not domains:
        return None
    if not isinstance(domains, list):
        domains = domains.strip('; ').replace(
            ',', ';').replace(' ', ';').split(';')
    index_rule = get_config('index' + ip_type, "default")  # 从配置中获取index配置
    address = get_ip(ip_type, index_rule)
    if not address:
        error('Fail to get %s address!', ipname)
        return False
    elif cache and (address == cache[ipname]):
        info('%s address not changed, using cache.', ipname)
        return True
    record_type = (ip_type == '4') and 'A' or 'AAAA'
    update_fail = False  # https://github.com/NewFuture/DDNS/issues/16
    for domain in domains:
        domain = domain.lower()  # https://github.com/NewFuture/DDNS/issues/431
        if change_dns_record(dns, proxy_list, domain=domain, ip=address, record_type=record_type):
            update_fail = True
    if cache is not False:
        # 如果更新失败删除缓存
        cache[ipname] = update_fail and address


def main():
    """
    更新
    """
    init_config(__description__, __doc__, __version__)

    log_level = get_config('log.level')
    log_format = get_config('log.format', '%(asctime)s %(levelname)s [%(module)s]: %(message)s')
    # Override log format in debug mode to include filename and line number for detailed debugging
    if (log_level == DEBUG or log_level == NOTSET) and not log_format:
        log_format = '%(asctime)s %(levelname)s [%(filename)s:%(lineno)d]: %(message)s'
    basicConfig(
        level=log_level,
        format=log_format,
        datefmt=get_config('log.datefmt', '%Y-%m-%dT%H:%M:%S'),
        filename=get_config('log.file'),
    )

    info("DDNS[ %s ] run: %s %s", __version__, os_name, sys.platform)

    # Dynamically import the dns module as configuration
    dns_provider = str(get_config('dns', 'dnspod').lower())
    dns = getattr(__import__('dns', fromlist=[dns_provider]), dns_provider)
    dns.Config.ID = get_config('id')
    dns.Config.TOKEN = get_config('token')
    dns.Config.TTL = get_config('ttl')

    if get_config("config"):
        info('loaded Config from: %s', path.abspath(get_config('config')))

    proxy = get_config('proxy') or 'DIRECT'
    proxy_list = proxy if isinstance(
        proxy, list) else proxy.strip(';').replace(',', ';').split(';')

    cache_config = get_config('cache', True)
    if cache_config is False:
        cache = cache_config
    elif cache_config is True:
        cache = Cache(path.join(gettempdir(), 'ddns.cache'))
    else:
        cache = Cache(cache_config)

    if cache is False:
        info('Cache is disabled!')
    elif not get_config('config_modified_time') or get_config('config_modified_time') >= cache.time:
        warning('Cache file is outdated.')
        cache.clear()
    else:
        debug('Cache is empty.')
    update_ip('4', cache, dns, proxy_list)
    update_ip('6', cache, dns, proxy_list)


if __name__ == '__main__':
    encode = sys.stdout.encoding
    if encode is not None and encode.lower() != 'utf-8' and hasattr(sys.stdout, 'buffer'):
        # 兼容windows 和部分ASCII编码的老旧系统
        sys.stdout = TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
        sys.stderr = TextIOWrapper(sys.stderr.buffer, encoding='utf-8')
    main()

# Nuitka Project Configuration
# nuitka-project: --mode=onefile
# nuitka-project: --output-filename=ddns
# nuitka-project: --product-name=DDNS
# nuitka-project: --product-version=0.0.0
# nuitka-project: --onefile-tempdir-spec="{TEMP}/{PRODUCT}_{VERSION}"
# nuitka-project: --no-deployment-flag=self-execution
# nuitka-project: --company-name="New Future"
# nuitka-project: --copyright=https://ddns.newfuture.cc
# nuitka-project: --assume-yes-for-downloads
# nuitka-project: --python-flag=no_site,no_asserts,no_docstrings,isolated,static_hashes
# nuitka-project: --nofollow-import-to=tkinter,unittest,pydoc,doctest,distutils,setuptools,lib2to3,test,idlelib,lzma
# nuitka-project: --noinclude-dlls=liblzma.*