# 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 import random # 调用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("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 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 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 add_user(self,user_name,user_id): url = "/artemis/api/resource/v1/person/single/add" headers = { "access_token": self.token, "Content-Type": "application/json" } certificateNo = ''.join(random.sample("0123456789121234222212312",20)) data = { "personName": user_name, "orgIndexCode": "12390179471245", "personId": user_id, "certificateNo": certificateNo, "certificateType": "111" } resp = self.post(self.host + url ,data ,headers) resp = json.loads(resp)["msg"] if resp == "success": print("用户 %s 添加成功,user_id为 %s"%(user_name,user_id)) return resp else: print(resp) def create_card(self,startDate,endDate,user_id): url = "/artemis/api/cis/v1/card/bindings" headers = { "access_token": self.token, "Content-Type": "application/json" } cardNo = ''.join(random.sample("0123456789121234222212312", 16)) data = { "startDate": startDate, "endDate": endDate, "cardList": [ { "cardNo": cardNo, "personId": user_id } ] } resp = self.post(self.host + url, data, headers) resp = json.loads(resp)["msg"] if resp == "success": print("用户%s的卡片添加成功"%user_id) else: print(resp) if __name__ == '__main__': host = "https://192.168.64.44:443" appKey = "25227103" appSecret = "LMGm1LNhAE7zF6Alzqdz" Hik_iSecure = Hik_iSecure(host,appKey,appSecret) Hik_iSecure.add_user("test11","123321")