背景需要采集智慧商业地产管理平台海康威视的客流数据1、进入智慧商业地产管理平台系统管理界面安装并且开启API网关系统维护--安装部署---软件包管理---搜索API2.进入状态监控选择API网关点击API管理配置AK/SK3.根据接口文档协议编写连接代码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:0008:00 的 ISO 字符串 tz pytz.timezone(Asia/Shanghai) today_start datetime.now(tz).replace(hour0, minute0, second0, microsecond0) return today_start.isoformat() stat_time_str get_stat_time() warnings.filterwarnings(ignore, categoryInsecureRequestWarning) body { granularity: daily, statTime: stat_time_str } #body_str {granularity: daily, statTime: 2026-03-10T00:00:0008:00 }; # 确保与MD5使用的字符串完全一致 body_str json.dumps(body, ensure_asciiFalse) # ensure_asciiFalse 保留中文如果有 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, headersrequest_headers, databody_str, verifyFalse) # verifyFalse忽略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代码测试是否正常6.接入Zabbix采集需要创建一个监控项键值是调用zabbix中的python脚本添加UserParameter客流有进客量和出客量需要创建两个依赖项分别使用预处理取数使用JSONPath最后接入grafana查看实时数据
Zabbix采集海康卫视客流摄像头数据接入grafana
背景需要采集智慧商业地产管理平台海康威视的客流数据1、进入智慧商业地产管理平台系统管理界面安装并且开启API网关系统维护--安装部署---软件包管理---搜索API2.进入状态监控选择API网关点击API管理配置AK/SK3.根据接口文档协议编写连接代码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:0008:00 的 ISO 字符串 tz pytz.timezone(Asia/Shanghai) today_start datetime.now(tz).replace(hour0, minute0, second0, microsecond0) return today_start.isoformat() stat_time_str get_stat_time() warnings.filterwarnings(ignore, categoryInsecureRequestWarning) body { granularity: daily, statTime: stat_time_str } #body_str {granularity: daily, statTime: 2026-03-10T00:00:0008:00 }; # 确保与MD5使用的字符串完全一致 body_str json.dumps(body, ensure_asciiFalse) # ensure_asciiFalse 保留中文如果有 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, headersrequest_headers, databody_str, verifyFalse) # verifyFalse忽略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代码测试是否正常6.接入Zabbix采集需要创建一个监控项键值是调用zabbix中的python脚本添加UserParameter客流有进客量和出客量需要创建两个依赖项分别使用预处理取数使用JSONPath最后接入grafana查看实时数据
