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