python-fastapi-mm-zhcs-yj-api/utils/hk_video_api.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import requests
import json
import time
import uuid
import hmac
import hashlib
import base64
from requests.packages.urllib3.exceptions import InsecureRequestWarning
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)

def get_video_url(id: str, protocol: str):
    video_url = ""
    # api_url = "http://10.181.7.217:8081/previewURLs"
    api_url = "http://10.181.7.236:8081/previewURLs"
    params = {
        "cameraIndexCode": id,
        "protocol": protocol
    }
    print('param', params)
    response = requests.post(url=api_url, params=params, timeout=15)
    print(response.text)
    if response.status_code == 200:
        result = response.json()
        if result['errcode'] == 0:
            video_url = result['data']
    return video_url

def indexCode(id: str):
    api_url = "http://10.181.7.236:8081/indexCode"
    params = {
        "cameraIndexCode": id
    }
    # print('param', params)
    response = requests.post(url=api_url, params=params, timeout=15)
    # print(response.text)
    if response.status_code == 200:
        result = response.json()
        if result['errcode'] == 0:
            data = result['data']
            if isinstance(data,str):
                data = json.loads(data)
                return data

def controlling(id: str, action: int, command: str, speed: int, presetIndex: str):
    api_url = "http://10.181.7.236:8081/controlling"
    params = {
        "cameraIndexCode": id,
        "action": action,
        "command": command,
        "speed": speed,
        "presetIndex": presetIndex
    }
    print('param', params)
    response = requests.post(url=api_url, params=params, timeout=15)
    print(response.text)
    if response.status_code == 200:
        result = response.json()
        if result['errcode'] == 0:
            data = result['data']
    return data


class HikvisionAPI:
    def __init__(self):
        self.app_key = "20034027"
        self.app_secret = "tfa3PNeYeN2MOrbRj3wS"
        self.base_url = f"https://19.155.243.66/artemis"
        # "/artemis/api/resource/v1/cameras"
    def _generate_signature(self, path):
        # 设置请求头
        timestamp = str(int(time.time() * 1000))
        nonce = str(uuid.uuid4())
        headers = {
            "Accept": "*/*",
            "Content-Type": "application/json",
            "x-ca-key": self.app_key,
            "x-ca-nonce": nonce,
            "x-ca-timestamp": timestamp,
            "x-ca-signature-headers": "x-ca-key,x-ca-nonce,x-ca-timestamp"
        }
        # 拼接签名字符串
        nonce = headers.get("x-ca-nonce")
        timestamp = headers.get("x-ca-timestamp")
        sign_str = "POST\n*/*\napplication/json" + "\nx-ca-key:" + self.app_key + "\nx-ca-nonce:" + \
           nonce + "\nx-ca-timestamp:" + timestamp + "\n" + \
           "/artemis"+path

        # 生成签名
        signature = hmac.new(self.app_secret.encode(), sign_str.encode(), hashlib.sha256).digest()
        headers["x-ca-signature"]=base64.b64encode(signature).decode()
        return headers

    def get_cameras(self,pageNo=1,pageSize=1):
        path = "/api/resource/v1/cameras"
        url = f"{self.base_url}{path}"
        body =  {
            "pageNo": pageNo,"pageSize":pageSize
        }
        headers =self._generate_signature(path)
        response = requests.post(url, headers=headers, json=body, verify=False)
        if response.status_code == 200:
            return response.json()
        else:
            raise Exception(f"Failed to get preview URL: {response.text}")
    def get_regions(self,pageNo=1,pageSize=1):
        path = "/api/resource/v1/regions"
        url = f"{self.base_url}{path}"
        body =  {
            "pageNo": pageNo,"pageSize":pageSize,
            "treeCode": "0"
        }
        headers =self._generate_signature(path)
        response = requests.post(url, headers=headers, json=body, verify=False)
        if response.status_code == 200:
            return response.json()
        else:
            raise Exception(f"Failed to get preview URL: {response.text}")