hksecure.py 3.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110
  1. # coding=utf-8
  2. import base64
  3. import hmac
  4. import json
  5. import sys
  6. import threading
  7. import time
  8. import uuid
  9. from hashlib import sha256
  10. if (sys.version_info.major == 3):
  11. import urllib.request as urllib2
  12. else:
  13. import urllib2
  14. import ssl
  15. # 调用https必须导入
  16. ssl._create_default_https_context = ssl._create_unverified_context
  17. host = "https://192.168.64.44:443" # 代理API网关nginx服务器ip端口
  18. appKey = "25227103" # 秘钥appkey
  19. appSecret = "LMGm1LNhAE7zF6Alzqdz" # 秘钥appSecret
  20. class HkSecure():
  21. def __init__(self, host, appKey, appSecret):
  22. self.host = host # 代理API网关nginx服务器ip端口
  23. self.appKey = appKey # 秘钥appkey
  24. self.appSecret = appSecret # 秘钥appSecret
  25. self.token = ""
  26. flushTokenThread = threading.Thread(target=self.update_token)
  27. flushTokenThread.start()
  28. while 1: # 等待第一次刷新token完成
  29. if self.token:
  30. break
  31. time.sleep(1)
  32. def update_token(self):
  33. while 1:
  34. try:
  35. print("update token every 11 hours")
  36. self.get_token()
  37. time.sleep(3600 * 11)
  38. except Exception as e:
  39. print("update token error: wait for 60 seconds")
  40. time.sleep(60)
  41. ###获取token 12小时需要更新一次
  42. def get_token(self):
  43. path = "/artemis/api/v1/oauth/token"
  44. base_headers = {
  45. "Accept": "*/*",
  46. "Content-Type": "application/json",
  47. }
  48. headers = self.build_signature("POST", path, base_headers)
  49. print(headers)
  50. post1 = self.post(self.host + path, "", headers)
  51. self.token = json.loads(post1)["data"]["access_token"]
  52. print(self.token)
  53. ## 基础请求url
  54. def base_request(self, url, data):
  55. headers = {
  56. "access_token": self.token,
  57. "Content-Type": "application/json"
  58. }
  59. post = self.post(self.host + url, data, headers)
  60. return post
  61. def post(self, url, postData, headers): #
  62. if isinstance(postData, dict):
  63. postData = json.dumps(postData)
  64. postData = postData.encode("utf-8")
  65. req = urllib2.Request(url, data=postData,
  66. headers=headers)
  67. res = urllib2.urlopen(req, timeout=60).read().decode("utf-8")
  68. return res
  69. # 获取13位时间戳
  70. def get_millisecond(self):
  71. millis = int(round(time.time() * 1000))
  72. return millis
  73. def get_sha256(self, data, key):
  74. key = key.encode("utf-8")
  75. data = data.encode("utf-8")
  76. return base64.b64encode(hmac.new(key, data, digestmod=sha256).digest()).decode("utf-8")
  77. def build_signature(self, httpType, url, headers):
  78. text = httpType + "\n"
  79. for header in sorted(headers):
  80. value = str(headers[header])
  81. if "x-ca-" in header:
  82. value = header + ":" + value
  83. text += value
  84. text += "\n"
  85. text += url
  86. sha_ = self.get_sha256(text, self.appSecret)
  87. headers["x-ca-signature"] = sha_
  88. headers["x-ca-key"] = self.appKey
  89. headers["x-ca-nonce"] = str(uuid.uuid4())
  90. return headers
  91. hksecure = HkSecure(host, appKey, appSecret)
  92. url = "/artemis/api/acs/v1/door/states"
  93. data = {"doorIndexCodes": ["f8f6ec5b4030477ebc9c838dc5ac670d"]}
  94. re = hksecure.base_request(url, data)
  95. print(re)