123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111 |
- import time
- import math
- import requests
- import datetime
- import json
- def datetime_now():
- datetimenow = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
- return datetimenow
- class Hik_Cloud():
- def __init__(self,host,client_id,client_secret):
- self.host = host
- self.client_id = client_id
- self.client_secret = client_secret
- self.get_token()
- def get_token(self):
- path = "/oauth/token"
- base_headers = {
- "Content-Type": "application/x-www-form-urlencoded"
- }
- base_data = {
- "client_id": self.client_id,
- "client_secret": self.client_secret,
- "grant_type": "client_credentials"
- }
- req = requests.post(self.host + path, data=base_data,
- headers=base_headers)
- res = req.json()
- self.token = res["access_token"]
- def post(self, url, postData, headers): #
- if isinstance(postData, dict):
- postData = json.dumps(postData)
- postData = postData.encode("utf-8")
- req = requests.post(url, data=postData,
- headers=headers)
- res = req.json()
- return res
- def hik_adduser(self,employeeNo,personName,faceImageBase64):
- path = "/api/v1/open/basic/persons/create?access_token=%s"%self.token
- if str(employeeNo).isdigit() is False:
- employeeNo = ''.join(filter(lambda x: x.isdigit(), employeeNo))
- data = {
- "employeeNo": employeeNo,
- "personName": personName,
- "faceImageBase64": faceImageBase64,
- "verifyImage": "false",
- }
- headers = {"content-type": "application/json"}
- response = self.post(self.host + path, data, headers)
- if response["code"] == 200:
- print("%s,%s(%s)用户添加成功" % (datetime_now(), personName, employeeNo))
- else:
- print("%s,%s(%s)%s" % (datetime_now(), personName, employeeNo, response["message"]))
- time.sleep(2)
- #创建卡片
- def hik_addcard(self,cards):
- path = "/api/v1/open/basic/cards/batchCreate?access_token=%s"%self.token
- headers = {"content-type": "application/json"}
- data = {
- "cards": cards
- }
- response = self.post(self.host + path, data,headers)
- if response["code"] == 200:
- print("%s,海康云眸创建卡片成功"%(datetime_now()))
- else:
- print(datetime_now(),response["message"])
- def resp(self, pageNo, person_dict):
- path = "/api/v1/open/basic/persons/list?access_token=%s&pageNo=%s&pageSize=999"%(self.token,pageNo)
- response = requests.get(self.host + path)
- response = response.json()
- totalsize = response["data"]["total"]
- persons = response["data"]["rows"]
- for person in persons:
- personId = person["employeeNo"]
- personName = person["personName"]
- person_dict[personId] = personName
- return totalsize, person_dict
- def hik_getuserlist(self):
- person_dict = {}
- pageNo = 1
- totalsize, person_dict = self.resp(pageNo,person_dict)
- page = math.ceil(int(totalsize) / 999)
- while pageNo < page:
- pageNo = pageNo + 1
- self.resp(pageNo, person_dict)
- return person_dict
- def hik_permissiongroupspersons(self, groupId, employeeNos):
- path = "/api/v1/open/accessControl/permissionGroups/actions/addPersons?access_token=%s"%self.token
- headers = {"content-type": "application/json"}
- data = {
- "groupId": groupId,
- "employeeNos": employeeNos,
- "autoIssue": "true"
- }
- response = self.post(self.host + path,headers,data)
- if response["code"] == 200:
- print("%s,海康云眸权限组绑定人员成功" % (datetime_now()))
- else:
- print(datetime_now(), employeeNos, response["message"])
|