ClusterM

TuyaAPI Client on Python

Dec 7th, 2021 (edited)
475
Never
Not a member of Pastebin yet? Sign Up, it unlocks many cool features!
  1. import os
  2. import time
  3. import hashlib
  4. import hmac
  5. import json
  6. import requests
  7. import logging
  8.  
  9. logger = logging.getLogger()
  10.  
  11. class TuyaAPI:
  12.     def __init__(self, api_region, access_id, api_secret, token_cache=None):
  13.         self.api_region = api_region
  14.         self.access_id = access_id
  15.         self.api_secret = api_secret
  16.         self.token = None
  17.         self.token_cache = token_cache
  18.         if self.token_cache and os.path.isfile(self.token_cache):
  19.             with open(self.token_cache, 'r') as f:
  20.                 self.token = json.loads(f.read())
  21.                 logger.info(f"Tuya - loaded cached token from {self.token_cache}")
  22.  
  23.         api_region = api_region.lower()
  24.         if api_region == "ch":
  25.             self.urlhost = "openapi.tuyacn.com"      # China Data Center
  26.         elif api_region == "us":
  27.             self.urlhost = "openapi.tuyaus.com"      # Western America Data Center
  28.         elif api_region == "us-e":
  29.             self.urlhost = "openapi-ueaz.tuyaus.com" # Eastern America Data Center
  30.         elif api_region == "eu":
  31.             self.urlhost = "openapi.tuyaeu.com"      # Central Europe Data Center
  32.         elif api_region == "eu-w":
  33.             self.urlhost = "openapi-weaz.tuyaeu.com" # Western Europe Data Center
  34.         elif api_region == "in":
  35.             self.urlhost = "openapi.tuyain.com"      # India Datacenter
  36.         else:
  37.             raise NotImplementedError(f"Unknown region '{api_region}'")
  38.  
  39.     def request(self, method, uri, body=None, headers=None, no_token=False):
  40.         # Build URL
  41.         while uri.startswith("/"): uri = uri[1:]
  42.         url = "https://%s/%s" % (self.urlhost, uri)
  43.  
  44.         # Debugging stuff
  45.         if logging.DEBUG >= logger.level:
  46.             d = body
  47.             if body and type(body) == dict: d = json.dumps(body, sort_keys=False, indent=4)
  48.             logger.debug(f"Tuya - {method} request to {url}" + (f":\n{d}" if body else ""))
  49.  
  50.         # Check body
  51.         if body and type(body) == dict:
  52.             body = json.dumps(body)
  53.  
  54.         # Build Header
  55.         now = int(time.time() * 1000)
  56.         headers = dict(list(headers.items()) + [('Signature-Headers', ":".join(headers.keys()))]) if headers else {}
  57.         if no_token:
  58.             payload = self.access_id + str(now)
  59.             headers['secret'] = self.api_secret
  60.         else:
  61.             self.refresh_token()
  62.             payload = self.access_id + self.token["access_token"] + str(now)
  63.  
  64.         payload += (method + '\n' +                                                  # HTTPMethod
  65.             hashlib.sha256(bytes((body or "").encode('utf-8'))).hexdigest() + '\n' + # Content-SHA256
  66.             ''.join(['%s:%s\n'%(key, headers[key])                                   # Headers
  67.                      for key in headers.get("Signature-Headers", "").split(":")
  68.                      if key in headers]) + '\n' +
  69.             '/' + url.split('//', 1)[-1].split('/', 1)[-1])
  70.  
  71.         # Sign Payload
  72.         signature = hmac.new(
  73.             self.api_secret.encode('utf-8'),
  74.             msg=payload.encode('utf-8'),
  75.             digestmod=hashlib.sha256
  76.         ).hexdigest().upper()
  77.  
  78.         # Create Header Data
  79.         headers['client_id'] = self.access_id
  80.         headers['sign'] = signature
  81.         headers['t'] = str(now)
  82.         headers['sign_method'] = 'HMAC-SHA256'
  83.  
  84.         if not no_token:
  85.             headers['access_token'] = self.token["access_token"]
  86.         if method == "GET":
  87.             response = requests.get(url, headers=headers)
  88.         elif method == "POST":
  89.             response = requests.post(url, headers=headers, data=body)
  90.         elif method == "PUT":
  91.             response = requests.put(url, headers=headers, data=body)
  92.         elif method == "DELETE":
  93.             response = requests.delete(url, headers=headers)
  94.         else:
  95.             raise NotImplementedError(f"Unknown method '{method}'")
  96.  
  97.         response_dict = response.json()
  98.         # More debugging stuff
  99.         if logging.DEBUG >= logger.level:
  100.             logger.debug(f"Tuya - response:\n{json.dumps(response_dict, sort_keys=False, indent=4)}")
  101.         success = response_dict["success"]
  102.         if not success: raise TuyaError(response_dict["msg"])
  103.         return response_dict["result"]
  104.  
  105.     def save_token(self):
  106.         if self.token_cache:
  107.             with open(self.token_cache, 'w') as f:
  108.                 f.write(json.dumps(self.token))
  109.  
  110.     def refresh_token(self):
  111.         if self.token and time.time() < self.token["token_time"] + self.token["expire_time"]:
  112.             return
  113.         if not self.token:
  114.             logger.info(f"Tuya - token is empty, requesting new token...")
  115.         else:
  116.             logger.info(f"Tuya - token expired, requesting new one...")
  117.         uri = "v1.0/token?grant_type=1"
  118.         self.token = self.request("GET", uri, no_token=True)
  119.         self.token["token_time"] = time.time()
  120.         self.save_token()
  121.         logger.info(f"Tuya - new token received")
  122.  
  123. def TuyaError(Exception):
  124.     pass
  125.  
RAW Paste Data