123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205 |
- # 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:
- print(self.token)
- break
- time.sleep(1)
- def update_token(self):
- while 1:
- try:
- print("Hik_iSecure 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)
|