Hik_Cloud.py 4.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132
  1. import time
  2. import math
  3. import requests
  4. import datetime
  5. import json
  6. import threading
  7. def datetime_now():
  8. datetimenow = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
  9. return datetimenow
  10. class Hik_Cloud():
  11. def __init__(self,host,client_id,client_secret):
  12. self.host = host
  13. self.client_id = client_id
  14. self.client_secret = client_secret
  15. # self.get_token()
  16. self.token = ""
  17. flushTokenThread = threading.Thread(target=self.update_token)
  18. flushTokenThread.start()
  19. while 1: # 等待第一次刷新token完成
  20. if self.token:
  21. break
  22. time.sleep(1)
  23. def update_token(self):
  24. while 1:
  25. try:
  26. print("Hik_Cloud update token every 6 days")
  27. self.get_token()
  28. time.sleep(3600 * 24 * 6)
  29. except Exception as e:
  30. print("update token error: wait for 60 seconds")
  31. time.sleep(60)
  32. ###获取token 6天需要更新一次
  33. def get_token(self):
  34. path = "/oauth/token"
  35. base_headers = {
  36. "Content-Type": "application/x-www-form-urlencoded"
  37. }
  38. base_data = {
  39. "client_id": self.client_id,
  40. "client_secret": self.client_secret,
  41. "grant_type": "client_credentials"
  42. }
  43. req = requests.post(self.host + path, data=base_data,
  44. headers=base_headers)
  45. res = req.json()
  46. self.token = res["access_token"]
  47. def post(self, url, postData, headers): #
  48. if isinstance(postData, dict):
  49. postData = json.dumps(postData)
  50. postData = postData.encode("utf-8")
  51. req = requests.post(url, data=postData,
  52. headers=headers)
  53. res = req.json()
  54. return res
  55. def hik_adduser(self,employeeNo,personName,faceImageBase64):
  56. path = "/api/v1/open/basic/persons/create?access_token=%s"%self.token
  57. if str(employeeNo).isdigit() is False:
  58. employeeNo = ''.join(filter(lambda x: x.isdigit(), employeeNo))
  59. data = {
  60. "employeeNo": employeeNo,
  61. "personName": personName,
  62. "faceImageBase64": faceImageBase64,
  63. "verifyImage": "false",
  64. }
  65. headers = {"content-type": "application/json"}
  66. response = self.post(self.host + path, data, headers)
  67. if response["code"] == 200:
  68. print("%s,%s(%s)用户添加成功" % (datetime_now(), personName, employeeNo))
  69. else:
  70. print("%s,%s(%s)%s" % (datetime_now(), personName, employeeNo, response["message"]))
  71. time.sleep(2)
  72. #创建卡片
  73. def hik_addcard(self,cards):
  74. path = "/api/v1/open/basic/cards/batchCreate?access_token=%s"%self.token
  75. headers = {"content-type": "application/json"}
  76. data = {
  77. "cards": cards
  78. }
  79. response = self.post(self.host + path, data,headers)
  80. if response["code"] == 200:
  81. print("%s,海康云眸创建卡片成功"%(datetime_now()))
  82. else:
  83. print(datetime_now(),response["message"])
  84. def resp(self, pageNo, person_dict):
  85. path = "/api/v1/open/basic/persons/list?access_token=%s&pageNo=%s&pageSize=999"%(self.token,pageNo)
  86. response = requests.get(self.host + path)
  87. response = response.json()
  88. totalsize = response["data"]["total"]
  89. persons = response["data"]["rows"]
  90. for person in persons:
  91. personId = person["employeeNo"]
  92. personName = person["personName"]
  93. person_dict[personId] = personName
  94. return totalsize, person_dict
  95. def hik_getuserlist(self):
  96. person_dict = {}
  97. pageNo = 1
  98. totalsize, person_dict = self.resp(pageNo,person_dict)
  99. page = math.ceil(int(totalsize) / 999)
  100. while pageNo < page:
  101. pageNo = pageNo + 1
  102. self.resp(pageNo, person_dict)
  103. return person_dict
  104. def hik_permissiongroupspersons(self, groupId, employeeNos):
  105. path = "/api/v1/open/accessControl/permissionGroups/actions/addPersons?access_token=%s"%self.token
  106. headers = {"content-type": "application/json"}
  107. data = {
  108. "groupId": groupId,
  109. "employeeNos": employeeNos,
  110. "autoIssue": "true"
  111. }
  112. response = self.post(self.host + path,data,headers)
  113. if response["code"] == 200:
  114. print("%s,海康云眸权限组绑定人员成功" % (datetime_now()))
  115. else:
  116. print(datetime_now(), employeeNos, response["message"])