Hik.py 4.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147
  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. import ssl
  11. import math
  12. import random
  13. # 调用https必须导入
  14. ssl._create_default_https_context = ssl._create_unverified_context
  15. if (sys.version_info.major == 3):
  16. import urllib.request as urllib2
  17. else:
  18. import urllib2
  19. class Hik_iSecure():
  20. def __init__(self, host, appKey, appSecret):
  21. self.host = host # 代理API网关nginx服务器ip端口
  22. self.appKey = appKey # 秘钥appkey
  23. self.appSecret = appSecret # 秘钥appSecret
  24. self.get_token()
  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("Hik_iSecure 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. def build_signature(self, httpType, url, headers):
  54. text = httpType + "\n"
  55. for header in sorted(headers):
  56. value = str(headers[header])
  57. if "x-ca-" in header:
  58. value = header + ":" + value
  59. text += value
  60. text += "\n"
  61. text += url
  62. sha_ = self.get_sha256(text, self.appSecret)
  63. headers["x-ca-signature"] = sha_
  64. headers["x-ca-key"] = self.appKey
  65. headers["x-ca-nonce"] = str(uuid.uuid4())
  66. return headers
  67. def post(self, url, postData, headers): #
  68. if isinstance(postData, dict):
  69. postData = json.dumps(postData)
  70. postData = postData.encode("utf-8")
  71. req = urllib2.Request(url, data=postData,
  72. headers=headers)
  73. res = urllib2.urlopen(req, timeout=60).read().decode("utf-8")
  74. # res = urllib2.urlopen(req, timeout=60).read()
  75. return res
  76. # 获取13位时间戳
  77. def get_millisecond(self):
  78. millis = int(round(time.time() * 1000))
  79. return millis
  80. def get_sha256(self, data, key):
  81. key = key.encode("utf-8")
  82. data = data.encode("utf-8")
  83. return base64.b64encode(hmac.new(key, data, digestmod=sha256).digest()).decode("utf-8")
  84. def add_user(self,user_name,user_id):
  85. url = "/artemis/api/resource/v1/person/single/add"
  86. headers = {
  87. "access_token": self.token,
  88. "Content-Type": "application/json"
  89. }
  90. certificateNo = ''.join(random.sample("0123456789121234222212312",20))
  91. data = {
  92. "personName": user_name,
  93. "orgIndexCode": "12390179471245",
  94. "personId": user_id,
  95. "certificateNo": certificateNo,
  96. "certificateType": "111"
  97. }
  98. resp = self.post(self.host + url ,data ,headers)
  99. resp = json.loads(resp)["msg"]
  100. if resp == "success":
  101. print("用户 %s 添加成功,user_id为 %s"%(user_name,user_id))
  102. return resp
  103. else:
  104. print(resp)
  105. def create_card(self,startDate,endDate,user_id):
  106. url = "/artemis/api/cis/v1/card/bindings"
  107. headers = {
  108. "access_token": self.token,
  109. "Content-Type": "application/json"
  110. }
  111. cardNo = ''.join(random.sample("0123456789121234222212312", 16))
  112. data = {
  113. "startDate": startDate,
  114. "endDate": endDate,
  115. "cardList": [
  116. {
  117. "cardNo": cardNo,
  118. "personId": user_id
  119. }
  120. ]
  121. }
  122. resp = self.post(self.host + url, data, headers)
  123. resp = json.loads(resp)["msg"]
  124. if resp == "success":
  125. print("用户%s的卡片添加成功"%user_id)
  126. else:
  127. print(resp)
  128. if __name__ == '__main__':
  129. host = "https://192.168.64.44:443"
  130. appKey = "25227103"
  131. appSecret = "LMGm1LNhAE7zF6Alzqdz"
  132. Hik_iSecure = Hik_iSecure(host,appKey,appSecret)
  133. Hik_iSecure.add_user("test11","123321")