Zabbix采集海康卫视客流摄像头数据接入grafana

Jerry@1998

122人浏览 · 2026-03-10 14:02:36

Jerry@1998 · 2026-03-10 14:02:36 发布

背景：需要采集智慧商业地产管理平台（海康威视的客流数据）

1、进入智慧商业地产管理平台系统管理界面安装并且开启API网关

（系统维护--安装部署---软件包管理---搜索API）

2.进入状态监控选择API网关，点击API管理，配置AK/SK

3.根据接口文档协议，编写连接代码

4.获取客流数据的python代码（需要安装requests库）

import requests
import hashlib
import base64
import hmac
import uuid
import time
from datetime import datetime, timezone
import pytz
import json
import warnings
from urllib3.exceptions import InsecureRequestWarning


# 配置
app_key = 'xxxx'
app_secret = 'xxx'
url = 'https://10.11.20.1:443/artemis/api/cfas/v2/passengerFlow/allGroup'
method = 'POST'

def get_stat_time():
    """返回当天 00:00:00+08:00 的 ISO 字符串"""
    tz = pytz.timezone('Asia/Shanghai')
    today_start = datetime.now(tz).replace(hour=0, minute=0, second=0, microsecond=0)
    return today_start.isoformat()


stat_time_str = get_stat_time()
warnings.filterwarnings('ignore', category=InsecureRequestWarning)
body = {
    "granularity": "daily",
    "statTime": stat_time_str
}
#body_str = '{"granularity": "daily", "statTime": "2026-03-10T00:00:00+08:00" };'  # 确保与MD5使用的字符串完全一致
body_str = json.dumps(body, ensure_ascii=False)  # ensure_ascii=False 保留中文（如果有）
print(body_str)

# 计算Content-MD5
md5 = hashlib.md5(body_str.encode('utf-8')).digest()
content_md5 = base64.b64encode(md5).decode('utf-8')

# 准备头
accept = 'application/json'
content_type = 'application/json'
timestamp = str(int(time.time() * 1000))  # 毫秒
nonce = str(uuid.uuid4())
date = datetime.now(timezone.utc).strftime('%a, %d %b %Y %H:%M:%S GMT')  # RFC1123

# 参与签名的自定义头
headers_to_sign = {
    'x-ca-key': app_key,
    'x-ca-timestamp': timestamp,
    'x-ca-nonce': nonce,
}
# 排序并拼接Headers字符串
sorted_keys = sorted(headers_to_sign.keys())
headers_str = ''
for k in sorted_keys:
    headers_str += f"{k}:{headers_to_sign[k]}\n"

# 构建签名字符串
# 实际path应为 /artemis/api/cfas/v2/passengerFlow/allGroup
# 更稳妥的方法：from urllib.parse import urlparse; path = urlparse(url).path
path = '/artemis/api/cfas/v2/passengerFlow/allGroup'
string_to_sign = f"{method}\n{accept}\n{content_md5}\n{content_type}\n{date}\n{headers_str}{path}"

# 计算签名
signature = base64.b64encode(
    hmac.new(app_secret.encode('utf-8'), string_to_sign.encode('utf-8'), hashlib.sha256).digest()
).decode('utf-8')

# 构建最终请求头
request_headers = {
    'Accept': accept,
    'Content-MD5': content_md5,
    'Content-Type': content_type,
    'Date': date,
    'X-Ca-Key': app_key,
    'X-Ca-Signature': signature,
    'X-Ca-Signature-Headers': ','.join(sorted_keys),
    'X-Ca-Timestamp': timestamp,
    'X-Ca-Nonce': nonce,
}

# 发送请求
response = requests.post(url, headers=request_headers, data=body_str, verify=False)  # verify=False忽略SSL证书，生产环境建议验证

# 输出结果
#print("Status Code:", response.status_code)
#print("Response:", response.json())
data = response.json()


try:
    # 获取数据逻辑...
    j_region_data = None
    for item in data.get('data', {}).get('list', []):
        if item.get('groupName') == 'J区客流统计':
            j_region_data = item
            break

    if j_region_data:
        result = {
            "flowin": j_region_data.get('flowInNum'),
            "flowout": j_region_data.get('flowOutNum')
        }
        print(json.dumps(result))
    else:
        # 未找到数据时输出空JSON或错误标识，但不能什么都不打印
        print(json.dumps({"error": "not found"}))
        # 或者直接输出一个默认值，如 {"flowin": 0, "flowout": 0}
        sys.exit(1)  # 可选：返回非0退出码让Zabbix感知错误
except Exception as e:
    # 捕获所有异常，避免脚本崩溃无输出
    print(json.dumps({"error": str(e)}))
    sys.exit(1)

5.运行python代码测试是否正常