Here is the code that the manual is based on:
``
python
import os
import sys
# Create a logger
log = {
'version': 1,
}
def log_command(*args, **kwargs):
sys.stdout.write('\033[36m{}\033[0m' .format(log['version']))
if args and kwargs:
for arg in args:
sys.stdout.write(' {}: '.format(arg))
for kwarg in kwargs.values():
sys.stdout.write('{!r}\033[0m'.format(kwarg))
sys.stdout.write('\n')
def handle_command(message):
print(log_command("Running command: {}".format(message)))
# Start the HTTP server
log_command("Starting HTTP server...")
import json, urllib.request
port = 8080
with open("/tmp/dyndns.json") as f:
data = json.load(f)
web_base_url = "http://127.0.0.1:{}/".format(port)
api_key = os.environ.get("DYNDNS_API_KEY")
api_secret = os.environ.get("DYNDNS_API_SECRET")
def send_msg(to, msg):
if api_key and api_secret:
auth = f"{api_key}:{api_secret}"
headers = {"Authorization": "Digest: {}".format(auth)}
# Send message to target
response = urllib.request.urlopen(web_base_url + "/targets/{}".format(len(data)), method="POST", data={"json": json.dumps({"url": web_base_url, "headers": headers})}, params={"Content-Type": "application/json"}, timeout=10)
if response.status_code < 400:
print(msg)
return
# Send message to Telegram chat
try:
request = urllib.request.Request(to, data=json.dumps({"text": msg}), method="POST", headers={"Content-Type": "application/json"})
request.add_header("Authorization", "Bearer {}".format(api_key))
response = urllib.request.urlopen(request)
if response.status_code < 400:
print(msg + "\n")
else:
# Handle error
print("[Error] {}.\n".format(response.reason))
except (urllib.error.URLError, json.JSONDecodeError) as e:
print(e)
send_msg('1003980922525', "Hello! This is a test!")
# Make HTTP requests to DynDNS API
def get_data():
headers = {"Authorization": "Digest: {}".format(api_key)}
response = urllib.request.urlopen(web_base_url + "/api/index.json", method="GET", data={"headers": headers}, timeout=10)
if response.status_code < 400:
print(response.read())
else:
# Handle error
print("[Error] {}.\n".format(response.reason))
get_data()
handle_command(200)
# Set environment variables to access local DynDNS API
os.environ["DYNDNS_API_KEY"] = "your_api_key"
os.environ["DYNDNS_API_SECRET"] = "your_api_secret"
# Start the DynDNS client server on 8081
os.environ["LOCAL_DYNDNS_PORT"] = str(port + 1)
handle_command(200)
time.sleep(3600) # wait for one hour
# Send a new update to all clients with status: Ok
headers = {"Authorization": "Digest: {}".format(api_key)}
web_base_url += "/status/{}".format(len(data))
response = urllib.request.urlopen(web_base_url, header=headers, method="PUT")
if response.status_code < 400:
log_command("Successfully updated DynDNS.")
else:
# Handle error
print("[Error] {}.\n".format(response.reason))
send_msg('1003980922525', "DynDNS Updated: New status is Ok.")
# Set up the HTTP server to keep listening for new commands and handle them immediately.
log_command(0)
if port > 8081:
log_command("Starting new DynDNS client server on {}".format(port + 1))
os.environ["LOCAL_DYNDNS_PORT"] = str(port + 1)
import threading
threading.Thread(target=handle_command, args=(200,))
``
Please note that this code does not test if DynDNS's API returns the expected response for each of its URLs. However, since there is no error checking after sending a request to DynDNS, you may need to add additional error handling code.