Hik_Cloud.py 3.9 KB

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