82 lines
3.5 KiB
Python
82 lines
3.5 KiB
Python
import os
|
|
import gradio
|
|
import fastapi
|
|
import uvicorn
|
|
|
|
from dns import AliyunDNS
|
|
|
|
# Aliyun DNS 相关配置
|
|
BASE_URL = str(os.getenv("BASE_URL", "http://alidns.aliyuncs.com"))
|
|
ACCESS_KEY_ID = str(os.getenv("ACCESS_KEY_ID", ""))
|
|
ACCESS_KEY_SECRET = str(os.getenv("ACCESS_KEY_SECRET", ""))
|
|
|
|
# gradio 管理页面 相关配置
|
|
USERNAME = str(os.getenv("USERNAME", "admin"))
|
|
PASSWORD = str(os.getenv("PASSWORD", "password"))
|
|
SERVER_HOST = str(os.getenv("SERVER_HOST", "0.0.0.0"))
|
|
SERVER_PORT = int(os.getenv("SERVER_PORT", "60000"))
|
|
|
|
|
|
class Server:
|
|
def __init__(self):
|
|
self.dns = AliyunDNS(BASE_URL, ACCESS_KEY_ID, ACCESS_KEY_SECRET)
|
|
domains = self.dns.DescribeDomains()
|
|
domains = [domain["DomainName"] for domain in domains["Domains"]["Domain"]]
|
|
self.resolutions = {domain: list() for domain in domains}
|
|
|
|
def domain(self, domain):
|
|
self.resolutions[domain].clear()
|
|
records = self.dns.DescribeDomainRecords(DomainName=domain)
|
|
for record in records["DomainRecords"]["Record"]:
|
|
self.resolutions[domain].append(
|
|
[record["RR"], record["Value"], record["RecordId"]])
|
|
self.resolutions[domain] = sorted(self.resolutions[domain], key=lambda x: x[0])
|
|
|
|
def webui(self, app, path):
|
|
with gradio.Blocks() as webui:
|
|
domain = gradio.Dropdown(label="主域名", choices=list(self.resolutions.keys()), value=None)
|
|
editor = gradio.DataFrame(label="编辑解析信息", headers=["子域名", "解析地址", "记录编码"],
|
|
row_count=100, col_count=(3, "fixed"), datatype=["str", "str", "str"], type="pandas")
|
|
update = gradio.Button(value="更新解析信息")
|
|
|
|
@domain.select(inputs=[domain], outputs=[editor])
|
|
def _(domain):
|
|
domain = str(domain)
|
|
self.domain(domain=domain) # 更新解析信息
|
|
return self.resolutions[domain]
|
|
|
|
@update.click(inputs=[domain, editor], outputs=[editor])
|
|
def _(domain, editor):
|
|
domain = str(domain)
|
|
editor = [line for line in editor.values.tolist() if "".join(line) != ""] # 去除空行
|
|
|
|
last = {tuple(line) for line in self.resolutions[domain]}
|
|
now = {tuple(line) for line in editor}
|
|
|
|
for rr, ip, id in (now - last):
|
|
if rr == "" or ip == "": continue # 子域名或解析地址为空则不做解析
|
|
print(f"创建解析：{domain} {rr} {ip}")
|
|
self.dns.AddDomainRecord(DomainName=domain, RR=rr, Type="A", Value=ip, Line="default", TTL=600)
|
|
|
|
for rr, ip, id in (last - now):
|
|
print(f"删除解析：{domain} {rr} {ip} {id}")
|
|
self.dns.DeleteDomainRecord(RecordId=id)
|
|
|
|
self.domain(domain=domain) # 更新解析信息
|
|
return self.resolutions[domain]
|
|
|
|
gradio.mount_gradio_app(app, webui, path=path, auth=(USERNAME, PASSWORD))
|
|
|
|
def run(self, SERVER_HOST, SERVER_PORT):
|
|
app = fastapi.FastAPI()
|
|
self.webui(app=app, path="/")
|
|
uvicorn.run(app, host=SERVER_HOST, port=int(SERVER_PORT))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
server = Server()
|
|
server.run(
|
|
SERVER_HOST=SERVER_HOST,
|
|
SERVER_PORT=SERVER_PORT
|
|
)
|