OpenXP 发表于 2022-7-24 19:04

porkbun API 简单封装 + DDNS

我的域名在 porkbun 上,而 prokbun 提供了实用的 API 可以不用登录就操作域名。我对其中一些我用得上的做了 python 的封装

`main` 函数会将配置里的子域名对应的记录改为当前机器的地址,结合 crontab 一类的定时任务就可以实现 DDNS. 这是我的用法。

如果不需要 DDNS, 也可以自己改一改,比如写一个不同的 `main` 函数,根据传入参数,调用相应的 API.

不多说了,直接上代码

```python
#!/usr/local/bin/python3
import requests
import json
import sys
import logging

# https://porkbun.com/api/json/v3/documentation

LOG_FORMAT = "%(asctime)s %(name)s %(levelname)s %(message)s"
DATE_FORMAT = "%Y-%m-%d %H:%M:%S"
logging.basicConfig(level = logging.INFO, format = LOG_FORMAT, datefmt = DATE_FORMAT)

config = json.load(open(sys.argv))
auth = { "secretapikey": config["secretapikey"], "apikey": config["apikey"] }
session = requests.Session()
session.trust_env = False

def ping():
    logger = logging.getLogger("ping")
    result = json.loads(session.post(config["endpoint"] + '/ping/', data = json.dumps(auth)).text)
    if result["status"] == "SUCCESS":
      logger.debug(result["yourIp"])
      return(result["yourIp"])
    else:
      logger.error(result["message"])
      return(None)

def create():
    logger = logging.getLogger("create")
    info = auth.copy()
    info.update({ "name": config["subdomain"], "type": config["type"], "content": config["ip"], "ttl": 600 })
    endpoint = config["endpoint"] + "/dns/create/" + config["domain"]
    result = json.loads(session.post(endpoint, data = json.dumps(info)).text)

    if result["status"] == "SUCCESS":
      hostInfo = config["domain"] + "/" + config["type"] + "/" + config["subdomain"]
      logger.info(hostInfo + " created: " + config["ip"])
      return(True)
    else:
      logger.error(result["message"])
      return(False)

def editByNameType():
    logger = logging.getLogger("edit")
    info = auth.copy()
    info.update({ "content": config["ip"], "ttl": 600 })
    endpoint = config["endpoint"] + "/dns/editByNameType/" + config["hostInfo"]
    result = json.loads(session.post(endpoint, data = json.dumps(info)).text)

    if result["status"] == "SUCCESS":
      if "recordAddr" not in config:
            recordAddr = "Unknown"
      else:
            recordAddr = config["recordAddr"]
      logger.info(config["hostInfo"] + " updated: " + recordAddr + " -> " + config["ip"])
      return(True)
    else:
      logger.error(result["message"])
      return(False)

def deleteByNameType():
    logger = logging.getLogger("delete")
    endpoint = config["endpoint"] + "/dns/deleteByNameType/" + config["hostInfo"]
    result = json.loads(session.post(endpoint, data = json.dumps(auth)).text)

    if result["status"] == "SUCCESS":
      logger.info(config["hostInfo"] + " deleted")
      return(True)
    else:
      logger.error(result["message"])
      return(False)

def retrieveByNameType():
    logger = logging.getLogger("retrieve")
    endpoint = config["endpoint"] + "/dns/retrieveByNameType/" + config["hostInfo"]
    result = json.loads(session.post(endpoint, data = json.dumps(auth)).text)

    if result["status"] == "SUCCESS":
      if "records" not in result or len(result["records"]) == 0:
            logger.error(config["hostInfo"] + " nothing retrieved")
            return(None)
      recordAddr = result["records"]["content"]
      logger.debug(config["hostInfo"] + " " + recordAddr)
      return(recordAddr)
    else:
      logger.error(result["message"])
      return(None)

def main():
    logger = logging.getLogger("main")
    ip = ping()

    if not ip:
      sys.exit(1)
    if "type" not in config:
      config.update({ "type": "A" })

    if "subdomain" not in config or config["subdomain"] == "":
      hostInfo = config["domain"] + "/" + config["type"]
    else:
      hostInfo = config["domain"] + "/" + config["type"] + "/" + config["subdomain"]

    config.update({ "ip": ip, "hostInfo": hostInfo })

    recordAddr = retrieveByNameType()
    if not recordAddr:
      if not create():
            sys.exit(1)
      else:
            sys.exit(0)

    if recordAddr == ip:
      logger.info(hostInfo + " " + ip + " unchanged")
      sys.exit(0)

    config.update({ "recordAddr": recordAddr })
    logger.debug(config)

    if not editByNameType():
      sys.exit(1)

if __name__ == '__main__':
    main()
```

这里的 `config` 将配置文件名作为参数传入,配置文件是一个 json 文件,格式如下

```json
{
"endpoint": "https://porkbun.com/api/json/v3",
"secretapikey": "",
"apikey": "",
"domain": "domain.com",
"subdomain": "subdomain",
"type": "A"
}
```

jhcybb 发表于 2022-7-25 07:58

感觉很厉害

qqdns 发表于 2022-7-25 08:39

值得学习!
页: [1]
查看完整版本: porkbun API 简单封装 + DDNS