import time import math import requests import datetime import json import threading 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() 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("Hik_Cloud update token every 6 days") self.get_token() time.sleep(3600 * 24 * 6) except Exception as e: print("update token error: wait for 60 seconds") time.sleep(60) ###获取token 6天需要更新一次 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,data,headers) if response["code"] == 200: print("%s,海康云眸权限组绑定人员成功" % (datetime_now())) else: print(datetime_now(), employeeNos, response["message"])