123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147 |
- # 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")
|