大概半年之前发现了家里的网络具有公网IP，于是就有了在家里部署一台服务器的想法。
最近由于学校还没有开学，只能居家“学习”，又正好之前自己淘汰了一个笔记本电脑，于是立马把笔记本重装成了 Ubuntu22，又在路由器上设置内网的固定IP，设置DMZ主机为我的笔记本电脑。这样，我在笔记本上部署的所有服务就都可以在公网访问了。
但是，由于家里的公网IP是动态的，不是固定的，每次光猫或者路由器断电，重新拨号就会导致IP变更，即使你能保证家里的路由器和光猫不重启，运营商也会每隔一段时间刷新一次的。所以为了能让域名动态的解析到家里的笔记本电脑。就需要DDNS服务了，即动态域名解析。
我的性格自然不会去买花钱的服务的😏，于是想到了利用 DNS 服务商提供的 API 去设置域名解析，在本地设置好定时任务，就可以实现动态域名解析了。
由于 Cloudflare 已经不再对 freenom 上的免费域名提供 API 支持了，所以我这里使用阿里云的 DNS API。
再设置一个企业微信的推送。就完美了✨

About half a year ago I discovered that my home network has a public IP, so I got the idea of deploying a server at home.
Recently, since school hasn’t started yet, I could only study at home. It just so happened that I had retired an old laptop, so I immediately reinstalled Ubuntu22 on it, set a fixed internal IP on the router, and set the DMZ host to my laptop. In this way, all the services deployed on my laptop can be accessed from the public network.
However, since the public IP at home is dynamic, not fixed, every time the optical modem or router loses power and redials, the IP changes. Even if you can guarantee that your router and optical modem never restart, the ISP will still refresh it every once in a while. So in order to dynamically resolve the domain name to my home laptop, I needed a DDNS service — dynamic domain name resolution.
Given my personality, I naturally wouldn’t buy a paid service 😏, so I thought of using the API provided by the DNS provider to set up domain resolution, and set up a scheduled task locally — that way I could achieve dynamic domain name resolution.
Since Cloudflare no longer provides API support for free domains on freenom, I used Alibaba Cloud’s DNS API here.
Plus a WeCom (WeChat Work) push notification. Perfect ✨

完整代码如下：

The complete code is as follows:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
from aliyunsdkcore.client import AcsClient
from aliyunsdkalidns.request.v20150109.DescribeSubDomainRecordsRequest import DescribeSubDomainRecordsRequest
from aliyunsdkalidns.request.v20150109.DeleteSubDomainRecordsRequest import DeleteSubDomainRecordsRequest
from urllib.request import urlopen

import json,requests,socket,base64,psutil,os,time
from datetime import datetime


# 企业微信推送相关 具体可参考 https://github.com/easychen/wecomchan
qiye_id = ''
AgentId = ''
Secret = ''

def send_to_wecom(text,wecom_cid,wecom_aid,wecom_secret,wecom_touid='@all'):
    get_token_url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={wecom_cid}&corpsecret={wecom_secret}"
    response = requests.get(get_token_url).content
    access_token = json.loads(response).get('access_token')
    if access_token and len(access_token) > 0:
        send_msg_url = f'https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={access_token}'
        data = {
            "touser":wecom_touid,
            "agentid":wecom_aid,
            "msgtype":"text",
            "text":{
                "content":text
            },
            "duplicate_check_interval":600
        }
        response = requests.post(send_msg_url,data=json.dumps(data)).content
        return response
    else:
        return False

def send_to_wecom_image(base64_content,wecom_cid,wecom_aid,wecom_secret,wecom_touid='@all'):
    get_token_url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={wecom_cid}&corpsecret={wecom_secret}"
    response = requests.get(get_token_url).content
    access_token = json.loads(response).get('access_token')
    if access_token and len(access_token) > 0:
        upload_url = f'https://qyapi.weixin.qq.com/cgi-bin/media/upload?access_token={access_token}&type=image'
        upload_response = requests.post(upload_url, files={
            "picture": base64.b64decode(base64_content)
        }).json()
        if "media_id" in upload_response:
            media_id = upload_response['media_id']
        else:
            return False
        send_msg_url = f'https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={access_token}'
        data = {
            "touser":wecom_touid,
            "agentid":wecom_aid,
            "msgtype":"image",
            "image":{
                "media_id": media_id
            },
            "duplicate_check_interval":600
        }
        response = requests.post(send_msg_url,data=json.dumps(data)).content
        return response
    else:
        return False

def send_to_wecom_markdown(text,wecom_cid,wecom_aid,wecom_secret,wecom_touid='@all'):
    get_token_url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={wecom_cid}&corpsecret={wecom_secret}"
    response = requests.get(get_token_url).content
    access_token = json.loads(response).get('access_token')
    if access_token and len(access_token) > 0:
        send_msg_url = f'https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={access_token}'
        data = {
            "touser":wecom_touid,
            "agentid":wecom_aid,
            "msgtype":"markdown",
            "markdown":{
                "content":text
            },
            "duplicate_check_interval":600
        }
        response = requests.post(send_msg_url,data=json.dumps(data)).content
        return response
    else:
        return False


def judgeprocess(processname):
    pl = psutil.pids()
    for pid in pl:
        if psutil.Process(pid).name() == processname:
            print(processname + "正在运行，pid=" + str(pid))
            return True
    else:
        return False

def getip():
    """
    查询本机ip地址
    :return: ip
    """
    res = requests.get("https://ifconfig.me/ip")
    return res.text


class Dnscontroller:
    access_key_id = ""
    #阿里云的新建账户id
    access_key_secret = ""
    region = "cn-hangzhou"
    #时区
    record_type = "A"

    domain = "oh-my.ga"
    #上面是你的一级域名
    name_ipv4 = ["@","*"]
    # 我解析的根域名 和 一个泛解析域名

    def __init__(self):
        self.client = AcsClient(
            self.access_key_id,
            self.access_key_secret,
            self.region
        )

    # 添加新的域名解析记录
    def add(self, DomainName, RR, Type, Value):
        from aliyunsdkalidns.request.v20150109.AddDomainRecordRequest import AddDomainRecordRequest
        request = AddDomainRecordRequest()
        request.set_accept_format('json')
        request.set_DomainName(DomainName)
        request.set_RR(RR)
        request.set_Type(Type)
        request.set_Value(Value)
        response = self.client.do_action_with_exception(request)

    # 修改域名解析记录
    def update(self, RecordId, RR, Type, Value):
        from aliyunsdkalidns.request.v20150109.UpdateDomainRecordRequest import UpdateDomainRecordRequest
        request = UpdateDomainRecordRequest()
        request.set_accept_format('json')
        request.set_RecordId(RecordId)
        request.set_RR(RR)
        request.set_Type(Type)
        request.set_Value(Value)
        response = self.client.do_action_with_exception(request)

    # 获取公网IP并进行修改
    def Get_IPv4(self):
        # 获取 IP
        ipv4 = getip()
        print("获取到IPv4地址：%s" % ipv4)

        request = DescribeSubDomainRecordsRequest()
        request.set_accept_format('json')
        request.set_DomainName(self.domain)

        for item in self.name_ipv4:
            request.set_SubDomain(item + '.' + self.domain)
            response = self.client.do_action_with_exception(request)
            domain_list = json.loads(response)

            if domain_list['TotalCount'] == 0:
                self.add(self.domain, item, self.record_type, ipv4)
                print("新建域名：" + item + ",解析成功")

            elif domain_list['TotalCount'] == 1:
                if domain_list['DomainRecords']['Record'][0]['Value'].strip() != ipv4.strip():
                    self.update(domain_list['DomainRecords']['Record'][0]['RecordId'], item, self.record_type, ipv4)
                    ret = send_to_wecom("宏碁笔记本新IP: " + ipv4, qiye_id, AgentId, Secret)
                    print(ret)
                    print("修改域名：" + item + ",解析成功")
                else:
                    print("IPv4地址没变")

            elif domain_list['TotalCount'] > 1:
                # 删除所有解析过的记录，并重新添加新的解析记录
                request = DeleteSubDomainRecordsRequest()
                request.set_accept_format('json')
                request.set_DomainName(self.domain)
                request.set_RR(item)
                response = self.client.do_action_with_exception(request)
                self.add(self.domain, item, self.record_type, ipv4)
                ret = send_to_wecom("宏碁笔记本新IP: " + ipv4, qiye_id, AgentId, Secret)
                print(ret)
                print("修改域名解析成功")

if __name__ == "__main__":
    print(datetime.now())
    # 检测服务器上的代理服务是否开启，如果开启应该先将它关闭，否则会影响到自身 IP 的获取
    # 我这里为了翻墙装了个 v2raya 
    if judgeprocess('v2raya'):
        os.system('systemctl stop v2raya')
        print('关闭v2raya ---')
        time.sleep(3)
        Dnscontroller().Get_IPv4()
        os.system('systemctl start v2raya')
        print('启动v2raya ---')
    else:
        Dnscontroller().Get_IPv4()
    print()