123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110 |
- # coding=utf-8
- import base64
- import hmac
- import json
- import sys
- import threading
- import time
- import uuid
- from hashlib import sha256
- if (sys.version_info.major == 3):
- import urllib.request as urllib2
- else:
- import urllib2
- import ssl
- # 调用https必须导入
- ssl._create_default_https_context = ssl._create_unverified_context
- host = "https://192.168.64.44:443" # 代理API网关nginx服务器ip端口
- appKey = "25227103" # 秘钥appkey
- appSecret = "LMGm1LNhAE7zF6Alzqdz" # 秘钥appSecret
- class HkSecure():
- def __init__(self, host, appKey, appSecret):
- self.host = host # 代理API网关nginx服务器ip端口
- self.appKey = appKey # 秘钥appkey
- self.appSecret = appSecret # 秘钥appSecret
- self.token = ""
- flushTokenThread = threading.Thread(target=self.update_token)
- flushTokenThread.start()
- while 1: # 等待第一次刷新token完成
- if self.token:
- break
- time.sleep(1)
- def update_token(self):
- while 1:
- try:
- print("update token every 11 hours")
- self.get_token()
- time.sleep(3600 * 11)
- except Exception as e:
- print("update token error: wait for 60 seconds")
- time.sleep(60)
- ###获取token 12小时需要更新一次
- def get_token(self):
- path = "/artemis/api/v1/oauth/token"
- base_headers = {
- "Accept": "*/*",
- "Content-Type": "application/json",
- }
- headers = self.build_signature("POST", path, base_headers)
- print(headers)
- post1 = self.post(self.host + path, "", headers)
- self.token = json.loads(post1)["data"]["access_token"]
- print(self.token)
- ## 基础请求url
- def base_request(self, url, data):
- headers = {
- "access_token": self.token,
- "Content-Type": "application/json"
- }
- post = self.post(self.host + url, data, headers)
- return post
- def post(self, url, postData, headers): #
- if isinstance(postData, dict):
- postData = json.dumps(postData)
- postData = postData.encode("utf-8")
- req = urllib2.Request(url, data=postData,
- headers=headers)
- res = urllib2.urlopen(req, timeout=60).read().decode("utf-8")
- return res
- # 获取13位时间戳
- def get_millisecond(self):
- millis = int(round(time.time() * 1000))
- return millis
- def get_sha256(self, data, key):
- key = key.encode("utf-8")
- data = data.encode("utf-8")
- return base64.b64encode(hmac.new(key, data, digestmod=sha256).digest()).decode("utf-8")
- def build_signature(self, httpType, url, headers):
- text = httpType + "\n"
- for header in sorted(headers):
- value = str(headers[header])
- if "x-ca-" in header:
- value = header + ":" + value
- text += value
- text += "\n"
- text += url
- sha_ = self.get_sha256(text, self.appSecret)
- headers["x-ca-signature"] = sha_
- headers["x-ca-key"] = self.appKey
- headers["x-ca-nonce"] = str(uuid.uuid4())
- return headers
- hksecure = HkSecure(host, appKey, appSecret)
- url = "/artemis/api/acs/v1/door/states"
- data = {"doorIndexCodes": ["f8f6ec5b4030477ebc9c838dc5ac670d"]}
- re = hksecure.base_request(url, data)
- print(re)
|