# coding=utf-8 import base64 import hmac import json import sys import threading import time import uuid from hashlib import sha256 import ssl import math # 调用https必须导入 ssl._create_default_https_context = ssl._create_unverified_context if (sys.version_info.major == 3): import urllib.request as urllib2 else: import urllib2 class Hik_iSecure(): def __init__(self, host, appKey, appSecret): self.host = host # 代理API网关nginx服务器ip端口 self.appKey = appKey # 秘钥appkey self.appSecret = appSecret # 秘钥appSecret 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("update token every 11 hours") # self.get_token() # time.sleep(3600 * 11) # except Exception as e: # print("update token error: wait for 60 seconds") # time.sleep(60) ###获取token 12小时需要更新一次 def get_token(self): path = "/artemis/api/v1/oauth/token" base_headers = { "Accept": "*/*", "Content-Type": "application/json", } headers = self.build_signature("POST", path, base_headers) # print(headers) post1 = self.post(self.host + path, "", headers) self.token = json.loads(post1)["data"]["access_token"] # print(self.token) def post(self, url, postData, headers): # if isinstance(postData, dict): postData = json.dumps(postData) postData = postData.encode("utf-8") req = urllib2.Request(url, data=postData, headers=headers) # res = urllib2.urlopen(req, timeout=60).read().decode("utf-8") res = urllib2.urlopen(req, timeout=60).read() return res # 获取13位时间戳 def get_millisecond(self): millis = int(round(time.time() * 1000)) return millis def get_sha256(self, data, key): key = key.encode("utf-8") data = data.encode("utf-8") return base64.b64encode(hmac.new(key, data, digestmod=sha256).digest()).decode("utf-8") def build_signature(self, httpType, url, headers): text = httpType + "\n" for header in sorted(headers): value = str(headers[header]) if "x-ca-" in header: value = header + ":" + value text += value text += "\n" text += url sha_ = self.get_sha256(text, self.appSecret) headers["x-ca-signature"] = sha_ headers["x-ca-key"] = self.appKey headers["x-ca-nonce"] = str(uuid.uuid4()) return headers ## 基础请求url def base_request(self, url, data): headers = { "access_token": self.token, "Content-Type": "application/json" } post = self.post(self.host + url, data, headers) return post def person_list(self,person_list,url): headers = { "access_token": self.token, "Content-Type": "application/json" } data = { "pageNo": "1", "pageSize": "999" } resp = self.post(self.host + url, data, headers) resp = json.loads(resp) totalsize = resp["data"]["total"] pageSize = resp["data"]["pageSize"] page = math.ceil(int(totalsize) / 999) data_list = resp["data"]["list"] for i in data_list: if "personPhoto" in i : personId = i["personId"] if str(personId).isdigit() is False: personId = ''.join(filter(lambda x: x.isdigit(), personId)) personName = i["personName"] personPhoto = i["personPhoto"][0] #图片加密 picture_base64 = self.get_picture(personPhoto) person_list[personId] = [personName,picture_base64] return pageSize,page def get_personlist(self): url = "/artemis/api/resource/v2/person/personList" person_list = {} pageSize,page = self.person_list(person_list,url) while pageSize < page: pageSize = pageSize + 1 self.person_list(person_list,url) return person_list def get_picture(self,personPhoto): url = "/artemis/api/resource/v1/person/picture" headers = { "access_token": self.token, "Content-Type": "application/json" } resp = self.post(self.host + url, personPhoto, headers) resp = base64.b64encode(resp).decode() return resp def get_card_post(self,resp,cards,diff_value): data_list = resp["data"]["list"] for i in data_list: if i["personId"] in diff_value: personId = i["personId"] card = i["cardNo"] card = { "cardNo": card, "cardType": "normalCard", "employeeNo": personId } cards.append(card) def get_card(self,diff_value): url = "/artemis/api/resource/v1/card/cardList" headers = { "access_token": self.token, "Content-Type": "application/json" } data = { "pageNo": "1", "pageSize": "999" } resp = self.post(self.host + url, data, headers) resp = json.loads(resp) totalsize = resp["data"]["total"] pageSize = resp["data"]["pageSize"] page = math.ceil(int(totalsize) / 999) cards = [] self.get_card_post(resp,cards,diff_value) while pageSize < page: pageSize = pageSize + 1 resp = self.post(self.host + url,data,headers) resp = json.loads(resp) self.get_card_post(resp, cards,diff_value) return cards # url = "/artemis/api/resource/v2/person/personList" # person_lists = HikiSecure.get_personlist(url) # print(person_lists) # url = "/artemis/api/resource/v1/person/picture" # person_lists = [['1661305020782497', '张勇', [{'serverIndexCode': 'cb5092fd-61fe-4fd9-880f-319483d3949d', 'personPhotoIndexCode': '15a08a88-d9e7-4577-990f-b6bb62822b15', 'picUri': '/pic?0dd100=ac-0ip2e1o06ee74e--fei3617badd92i5b2*=*d3d8i*s1d=i1p4t=pe*m6i13=1964190*79z47bs=091c7*l-0dod024=1i005'}]]] # aa = HikiSecure.get_picture(url,person_lists) # print(aa) # url = "/artemis/api/resource/v2/person/personList" # person_lists = HikiSecure.get_personlist(url) # picture_url = "/artemis/api/resource/v1/person/picture" # HikiSecure.get_picture(picture_url,person_lists) # url = "/artemis/api/resource/v1/card/cardList" # HikiSecure.get_card(url,person_lists)