--- name: hik-connect-team Skills description: | Hik-Connect for Teams (HCT) Developer Skills. Integrates a series of skills for managing and controlling HCT devices, including resource management, access control, device capture, video streaming, and alarm push. Use when: Need to perform batch management, remote control, real-time monitoring, media resource acquisition, or alarm push configuration for devices under Hik-Connect for Teams mode. ⚠️ Global Requirement: All sub-modules require configuration of environment variables: - Hik-Connect Team OpenAPI AppKey - Hik-Connect Team OpenAPI SecretKey - Hik-Connect Team OpenAPI Domain (auto-obtained from token response) --- # Hik-Connect Team Skills ## 1. Introduction `Hik-Connect_Team_Skills` is a full-featured integration Skills designed specifically for **Hik-Connect for Teams (HCT)** developers. Based on the **HCTOpen OpenAPI** system, it encapsulates core capabilities from basic resource management to advanced alarm push through Python scripts. This Skills adopts a modular design with built-in automated **Token maintenance mechanisms**, **dynamic path searching**, and **standardized error handling**, aiming to help developers quickly build HCT-based automated O&M, security monitoring, and business integration systems. --- ## 2. Core Modules Deep Dive This Skills consists of five core sub-modules, each providing deep support for specific business scenarios: | Module Name | Core Functions | Core Scripts | Applicable Scenarios | |:---------------------------------------------------------------------------|:----------------------------------------------------------|:-----------------------------------------------------------------------------------|:----------------------------------------------------------------------------------------------------------------------------------------------| | [**📦 Resource Management**](./modules/Hik-Connect_Team_Resource/SKILL.md) | Device discovery, detail acquisition, channel enumeration | `list_devices.py`
`device_detail.py`
`device_channels.py`
`list_doors.py` | Asset inventory, obtaining device serial numbers and channel IDs, access control resources, synchronizing organizational structure resources. | | [**🚪 Access Control (ACS)**](./modules/Hik-Connect_Team_ACS/SKILL.md) | Remote open/close, normally open/normally closed control | `acs_control.py` | Remote office collaboration, unattended entrance management, access control linkage in emergencies. | | [**📸 Device Capture**](./modules/Hik-Connect_Team_Capture/SKILL.md) | Real-time trigger capture, obtain image URL | `capture_pic.py` | Anomaly verification, real-time screen preview, manual secondary verification of AI recognition results. | | [**🎥 Video Streaming**](./modules/Hik-Connect_Team_Video/SKILL.md) | Obtain real-time video stream | `get_video_url.py` | Real-time monitoring embedding, remote video inspection, third-party monitoring large screen integration. | | [**🔔 Alarm Push (Alarm)**](./modules/Hik-Connect_Team_Alarm/SKILL.md) | Webhook subscription, fine-grained event management | `webhook_manager.py`
`event_manager.py` | Real-time alarm notification, third-party system integration (e.g., Feishu/DingTalk robots). | --- ## 3. Environment Preparation and Global Configuration ### 3.1 Credential Configuration Before using any module, credentials must be configured. The system supports two methods: #### Method A: Environment Variables (Recommended) ```bash # Required: Obtain from Hik-Connect HCT Developer Platform export HIK_CONNECT_TEAM_OPENAPI_APP_KEY="Your Hik-Connect Team OpenAPI AppKey" export HIK_CONNECT_TEAM_OPENAPI_SECRET_KEY="Your Hik-Connect Team OpenAPI SecretKey" # Note: API domain is automatically obtained from token response (no longer required) # Optional: Token cache configuration (enabled by default to reduce API call frequency) export HIK_CONNECT_TEAM_TOKEN_CACHE="1" # 1=Enabled, 0=Disabled ``` #### Method B: OpenClaw Config Files (Fallback) If environment variables are not set, the system will automatically search for credentials in OpenClaw config files: ``` Config search order (first found wins): 1. ~/.openclaw/config.json 2. ~/.openclaw/gateway/config.json 3. ~/.openclaw/channels.json ``` Config format: ```json { "channels": { "hik_connect_team_openapi": { "appKey": "Your Hik-Connect Team OpenAPI AppKey", "secretKey": "Your Hik-Connect Team OpenAPI SecretKey", "enabled": true } } } ``` **Recommended: Save to `~/.openclaw/channels.json`** — This is the dedicated file for channel credentials. **⚠️ Security Note**: Storing credentials in config files is convenient but introduces some risk. Environment variables are recommended for better security. ### 3.2 Dependency Installation This Skills is developed based on Python 3.8+. It is recommended to install necessary dependencies using the following command: ```bash pip3 install requests tabulate pycryptodome Pillow ``` --- ## 🔒 Config File Reading Details **Credential Priority** (Highest to Lowest): ``` ┌─────────────────────────────────────────────────────────────┐ │ 1. Environment Variables (Highest Priority - Recommended) │ │ ├─ HIK_CONNECT_TEAM_OPENAPI_APP_KEY │ │ └─ HIK_CONNECT_TEAM_OPENAPI_SECRET_KEY │ │ ✅ Advantage: No config file reading, fully isolated │ ├─────────────────────────────────────────────────────────────┤ │ 2. OpenClaw Config Files (Only when env vars not set) │ │ ├─ ~/.openclaw/config.json │ │ ├─ ~/.openclaw/gateway/config.json │ │ └─ ~/.openclaw/channels.json │ │ ⚠️ Note: Only reads channels.hik_connect_team_openapi field │ ├─────────────────────────────────────────────────────────────┤ ``` ## 4. Directory Structure Description ```text Hik-Connect_Team_Skills/ ├── SKILL.md # This guide file (Full-featured integration guide) ├── lib/ # Core library │ └── token_manager.py # Encapsulates HCTOpenClient base class, handles Token refresh, request retries, and path searching └── modules/ # Functional sub-modules ├── Hik-Connect_Team_Resource/ # Resource Management: Devices, channels, details ├── Hik-Connect_Team_ACS/ # Access Control: Open/close, normally open/normally closed ├── Hik-Connect_Team_Capture/ # Device Capture: Real-time trigger, URL acquisition ├── Hik-Connect_Team_Video/ # Video Streaming: Real-time preview address acquisition └── Hik-Connect_Team_Alarm/ # Alarm Push: Webhook management, event subscription ``` --- ## 5. Security and Best Practices 1. **Token Security**: The Skills automatically caches Tokens locally. Please ensure the security of the running environment to prevent unauthorized reading of cache files in the `lib/` directory. 2. **HTTPS Mandatory Requirement**: All Webhook callbacks from the HCT platform must use HTTPS. It is recommended to use `ngrok` or `cpolar` with SSL certificates for secure access. 3. **Signature Verification**: In the Alarm module, be sure to configure `signSecret` and implement HMAC-SHA256 signature verification on your receiving end to prevent forged alarm pushes. 4. **Error Handling**: All scripts return standard JSON format. If `success` is `false`, please check the `message` field for detailed error reasons. --- FILE:README.md # Hik-Connect Team (HCT) Skills Welcome to the **Hik-Connect Team (HCT) Skills**. This is a comprehensive developer skill set designed for **Hik-Connect for Teams (HCT)**, providing a modular and efficient way to manage and control HCT devices through the **HCTOpen OpenAPI** system. ## 🌟 Overview The HCT Skills empowers developers to integrate professional security and management features into their own applications or automated workflows. It handles the complexities of authentication, token management, and standardized communication with Hikvision's cloud services. ### Key Features - **Resource Management**: Discover devices, get details, and enumerate channels. - **Access Control (ACS)**: Remotely open/close doors and manage access states. - **Real-time Capture**: Trigger and retrieve live snapshots from cameras. - **Video Streaming**: Generate secure, time-limited URLs for live video previews. - **Alarm Management**: Subscribe to events and receive real-time notifications via Webhooks. --- ## 🛠 Modules & Capabilities The Skills is divided into specialized modules, each with its own dedicated scripts and documentation: | Module | Description | Key Scripts | |:----------------------------------------------------------------|:----------------------------|:-------------------------------------------------------| | [**📦 Resource**](./modules/Hik-Connect_Team_Resource/SKILL.md) | Manage your asset inventory | `list_devices.py`, `device_detail.py`, `list_doors.py` | | [**🚪 ACS**](./modules/Hik-Connect_Team_ACS/SKILL.md) | Remote door control | `acs_control.py` | | [**📸 Capture**](./modules/Hik-Connect_Team_Capture/SKILL.md) | Instant image snapshots | `capture_pic.py` | | [**🎥 Video**](./modules/Hik-Connect_Team_Video/SKILL.md) | Live stream URL generation | `get_video_url.py` | | [**🔔 Alarm**](./modules/Hik-Connect_Team_Alarm/SKILL.md) | Webhook & Event management | `webhook_manager.py`, `event_manager.py` | --- ## 🚀 Getting Started ### 1. Prerequisites - **Python 3.8+** - **Node.js** (Required only for the Alarm/Webhook service) - **HCT Developer Credentials**: You must have `HIK_CONNECT_TEAM_OPENAPI_APP_KEY` and `HIK_CONNECT_TEAM_OPENAPI_SECRET_KEY` from the Hik-Connect HCT Developer Platform. The API domain will be automatically obtained from the token response. ### 2. Installation Install the required Python dependencies: ```bash pip3 install requests tabulate pycryptodome Pillow ``` ### 3. Configuration **Credentials only need to be configured ONCE. The system will automatically find and use them.** #### Method A: Environment Variables (Recommended) Set in your shell profile or before running scripts: ```bash export HIK_CONNECT_TEAM_OPENAPI_APP_KEY="Your Hik-Connect Team OpenAPI AppKey" export HIK_CONNECT_TEAM_OPENAPI_SECRET_KEY="Your Hik-Connect Team OpenAPI SecretKey" ``` #### Method B: OpenClaw Config Files (Fallback) If environment variables are not set, the system will automatically search for credentials in OpenClaw config files: ``` Config search order (first found wins): 1. ~/.openclaw/config.json 2. ~/.openclaw/gateway/config.json 3. ~/.openclaw/channels.json ⭐ Recommended ``` Config format: ```json { "channels": { "hik_connect_team_openapi": { "appKey": "Your Hik-Connect Team OpenAPI AppKey", "secretKey": "Your Hik-Connect Team OpenAPI SecretKey", "enabled": true } } } ``` **Note**: API domain is automatically obtained from token response. --- ## 🔒 Credential Priority **The skill obtains credentials in the following order (highest to lowest priority):** ``` ┌─────────────────────────────────────────────────────────────┐ │ 1. Environment Variables (Highest Priority - Recommended) │ │ ├─ HIK_CONNECT_TEAM_OPENAPI_APP_KEY │ │ └─ HIK_CONNECT_TEAM_OPENAPI_SECRET_KEY │ │ ✅ Advantage: No config file reading, fully isolated │ ├─────────────────────────────────────────────────────────────┤ │ 2. OpenClaw Config Files (Only when env vars not set) │ │ ├─ ~/.openclaw/config.json │ │ ├─ ~/.openclaw/gateway/config.json │ │ └─ ~/.openclaw/channels.json │ │ ⚠️ Note: Only reads channels.hik_connect_team_openapi field │ ├─────────────────────────────────────────────────────────────┤ │ 3. Error Handling (When no valid credentials) │ │ Program exits with error message │ └─────────────────────────────────────────────────────────────┘ ``` --- ## 💡 Usage Examples ### Example 1: List All Devices ```bash cd "Hik-Connect Team Skills/modules/Hik-Connect_Team_Resource/scripts" python list_devices.py ``` ### Example 2: Remote Door Opening ```bash cd "Hik-Connect Team Skills/modules/Hik-Connect_Team_ACS/scripts" python acs_control.py --action-type 1 --element-list "your_door_resource_id" ``` ### Example 3: Capture Device Image ```bash cd "Hik-Connect Team Skills/modules/Hik-Connect_Team_Capture/scripts" python capture_pic.py DEVICE_SERIAL ``` ### Example 4: Get a Live Video Stream ```bash cd "Hik-Connect Team Skills/modules/Hik-Connect_Team_Video/scripts" python get_video_url.py --device-serial "SERIAL123" --resource-id "RES_ID_456" ``` ### Example 5: Setting Up Alarms The Alarm module requires a **public HTTPS URL** to receive webhook pushes from HCT platform. #### Option A — Same Server as OpenClaw (Simplest) 1. Configure reverse proxy to route `/hikvision/webhook` to `127.0.0.1:3090` 2. Start Webhook server: `node modules/Hik-Connect_Team_Alarm/scripts/server.js` 3. Register URL: `python modules/Hik-Connect_Team_Alarm/scripts/webhook_manager.py save --url "https://your-domain.com/hikvision/webhook" --secret "your_secret"` 4. Subscribe: `python modules/Hik-Connect_Team_Alarm/scripts/event_manager.py subscribe` #### Option B — Use a Tunnel Tool (ngrok/cpolar) 1. Run `ngrok http 3090` on OpenClaw server 2. Copy the tunnel URL 3. Start Webhook server and register the tunnel URL > **Note**: Tunnel URLs change on restart for free tiers — you must re-register the Webhook after each restart. #### Option C — Different Server with Public URL If you have a separate public server and OpenClaw's port 3090 is reachable from it: 1. On your server, configure a reverse proxy to forward `/hikvision/webhook` to `:3090` 2. Start the Webhook server on OpenClaw server: `node modules/Hik-Connect_Team_Alarm/scripts/server.js` 3. Register your public URL: `python modules/Hik-Connect_Team_Alarm/scripts/webhook_manager.py save --url "https://your-domain.com/hikvision/webhook" --secret "your_secret"` 4. Subscribe to events: `python modules/Hik-Connect_Team_Alarm/scripts/event_manager.py subscribe` > **⚠️ Third-party webhook receiver services (Pipedream, AWS Lambda URL, etc.) are NOT recommended** — they only receive requests, they cannot forward to your internal OpenClaw server. ### About Alarm Message Format When alarm messages are pushed to OpenClaw, the AI agent may inherently attempt to translate, summarize, or reformat the raw data. This behavior is difficult to completely avoid. **If you need a specific alarm message format:** - Explicitly instruct the AI agent: "Do not process/modify/summarize the alarm data, return it as-is" - If the format is still not ideal, directly tell the AI your preferred format (e.g., "Show alarm messages in a table", "Use the raw JSON format", etc.) The raw alarm data from HCT platform contains complete information — the AI's processing is optional and can be overridden by your instructions. --- ## 🔒 Security Recommendations ### 1. Use Minimal Permission Credentials - Create dedicated `HIK_CONNECT_TEAM_OPENAPI_APP_KEY`/`HIK_CONNECT_TEAM_OPENAPI_SECRET_KEY` with only necessary API permissions - Do not use main account credentials - Rotate credentials regularly (recommended every 90 days) ### 2. Environment Variable Security ```bash # Recommended: Use .env file (do not commit to version control) echo "HIK_CONNECT_TEAM_OPENAPI_APP_KEY=your_key" >> .env echo "HIK_CONNECT_TEAM_OPENAPI_SECRET_KEY=your_secret" >> .env chmod 600 .env # Load environment variables source .env ``` ### 3. Disable Token Caching (High Security) ```bash export HIK_CONNECT_TEAM_TOKEN_CACHE=0 python3 scripts/xxx.py ... ``` ### 4. Regular Cache Cleanup ```bash # Clear all cached Tokens rm -rf /tmp/hctopen_global_token_cache/ ``` ### 5. Config File Scanning The skill reads Hikvision configuration from (only when env vars not set): ``` ~/.openclaw/config.json ~/.openclaw/gateway/config.json ~/.openclaw/channels.json ``` **Config Format**: ```json { "channels": { "hik_connect_team_openapi": { "appKey": "Your Hik-Connect Team OpenAPI AppKey", "secretKey": "Your Hik-Connect Team OpenAPI SecretKey", "enabled": true } } } ``` **Security Recommendations**: - ✅ Use dedicated Hikvision credentials, do not share with other services - ✅ Set environment variables to override config file scanning if needed - ✅ Regularly review credential permissions in config files - ❌ Do not store main account credentials in config files --- ## ✅ Security Audit Checklist ### Pre-Installation Checks - [ ] **Review Code** — Read `lib/token_manager.py` and module scripts - [ ] **Verify API Domain** — Confirm domain is Hikvision official endpoint - [ ] **Prepare Test Credentials** — Create dedicated app with only necessary permissions - [ ] **Check Config Files** — Review `~/.openclaw/*.json` for sensitive credentials - [ ] **Confirm Cache Location** — Ensure `/tmp/hctopen_global_token_cache/` is acceptable ### Installation Configuration - [ ] **Use Environment Variables** — Prefer `HIK_CONNECT_TEAM_OPENAPI_APP_KEY` etc. - [ ] **Disable Caching** (Optional) — Set `HIK_CONNECT_TEAM_TOKEN_CACHE=0` for high security - [ ] **Minimal Permission Credentials** — Do not use main account credentials - [ ] **Isolated Environment** (Optional) — Run in container/VM ### Post-Installation Verification - [ ] **Verify Cache Permissions** — Confirm cache file permissions are 600 - [ ] **Test Functionality** — Verify with test device - [ ] **Monitor Logs** — Check API calls are normal - [ ] **Secure Credential Storage** — Use key manager ### Ongoing Maintenance - [ ] **Rotate Credentials** — Recommended every 90 days - [ ] **Review Dependencies** — Check `requests` etc. for security updates - [ ] **Clear Cache** — Clear cache in high security environments - [ ] **Monitor for Anomalies** — Watch for unusual API calls or errors --- ## 🔒 Security & Best Practices - **Least Privilege**: Use credentials with only the permissions necessary for your specific task. - **Token Caching**: Skills automatically caches access tokens in system temp directory (600 permissions) to minimize API calls. - **HTTPS**: All Webhook endpoints **must** use HTTPS. - **Stream Encryption**: If devices have "Stream Encryption" enabled, you must manually decrypt in HCT platform or app. --- ## 📂 Project Structure ```text Hik-Connect_Team_Skills/ ├── README.md # This overview document ├── SKILL.md # Technical integration guide ├── lib/ # Shared libraries │ ├── token_manager.py # Token management & base client │ └── README_TOKEN_MANAGER.md # Token manager documentation └── modules/ # Functional sub-modules ├── Hik-Connect_Team_Resource/ ├── Hik-Connect_Team_ACS/ ├── Hik-Connect_Team_Capture/ ├── Hik-Connect_Team_Video/ └── Hik-Connect_Team_Alarm/ ``` For detailed information on each module, please refer to the `SKILL.md` file within each module's directory. --- FILE:modules/Hik-Connect_Team_Video/SKILL.md --- name: hctopen-video description: | HCTOpen device video stream skill. Supports getting real-time video stream address for specified device channel. Use when: Need to get device real-time video stream URL. Before calling this Skill's script, please check if user provided optional parameters. If user didn't provide video-duration, please clearly inform user in reply: 'Currently using default stream duration (duration 10 minutes), if you need to adjust, please let me know'. After getting confirmation or ignoring, continue execution. ⚠️ Security Requirement: Must set environment variables: Hik-Connect Team OpenAPI AppKey,Hik-Connect Team OpenAPI SecretKey. (API Domain is automatically obtained from token response) parameters: - name: device-serial type: string description: "Device serial number" required: true - name: resource-id type: string description: "Channel/monitoring point resource ID" required: true - name: video-duration type: integer description: "Video stream duration (seconds), default 600s, if user didn't specify duration, please inform user default value will be used, and ask if adjustment is needed." default: 600 output_format: - "⚠️ Important: After getting video stream, must return in Markdown link format: `[url]({url})`, do not return raw URL only!" - "Example: [https://example.com/stream.m3u8]({https://example.com/stream.m3u8})" troubleshooting: scope: on-demand-only trigger: "Only activate when user explicitly reports: 'video won't play', 'stream fails', 'cannot open', or similar playback errors." mandatory_checks: - "Step 1: Verify Stream Encryption is Disabled via device_detail.py" - "Step 2: Verify video encoding format is H264 (ask user to check in HCT platform)" metadata: openclaw: emoji: "🎥" requires: env: ["HIK_CONNECT_TEAM_OPENAPI_APP_KEY", "HIK_CONNECT_TEAM_OPENAPI_SECRET_KEY"] pip: ["requests"] primaryEnv: "HIK_CONNECT_TEAM_OPENAPI_APP_KEY" warnings: - "May read ~/.openclaw/*.json for credentials (env vars have priority)" config: configFileRead: paths: - "~/.openclaw/config.json" - "~/.openclaw/gateway/config.json" - "~/.openclaw/channels.json" priority: "lower than environment variables" description: "Reads Hik-Connect Team credentials from OpenClaw config files as fallback" --- # HCTOpen Video HCT is short for Hik-Connect for Teams, meaning Hik-Connect Team mode. HCTOpen is short for Hik-Connect for Teams OpenAPI. This Skill provides device real-time video stream address acquisition functionality, can be accessed directly through link. --- ## ⚠️ Security Warning (Read Before Use) | # | Check Item | Status | Description | |---|---------------------------|-------------|----------------------------------------------------------------------------------------------------| | 1 | **Credential Permission** | ⚠️ Required | Please use credentials with **video stream permission**, avoid using super admin credentials | | 2 | **Traffic Consumption** | ⚠️ Note | Real-time video stream will consume large bandwidth, please close player in time when not in use | | 3 | **Token Cache** | ✅ Encrypted | Token cached in system temp directory, only current user can read (600 permission) | | 4 | **API Domain** | ✅ Auto | API domain is automatically obtained from token response (no longer requires manual configuration) | --- ## 🚀 Quick Start ### Run Video Stream Script ```bash # Scenario 1: Get video stream for specified device and channel (default 600s) python scripts/get_video_url.py --device-serial J10137390 --resource-id 6a447d3f9cfe4c8e8394c19f8fbcd3ba # Scenario 2: Get video stream for specified duration (60s) python scripts/get_video_url.py --device-serial D72821502 --resource-id 661543ed4b35465a9767081ae0a8bf45 --video-duration 600 ``` > ⚠️ **Important**: The `--resource-id` must be the **camera resource ID** obtained from `device_channels.py`! --- ## 🛠 Workflow ```mermaid graph TD A[Start Script] --> B{Check Environment Variables} B -- Missing --> C[Report Error and Exit] B -- Pass --> D[Get AccessToken] D --> E{Is Token Valid?} E -- Cache Valid --> F[Use Cache Directly] E -- Expired/No Cache --> G[Call API to Get New Token] G --> H[Save to Local Cache] F --> I[Send Video Stream Request] H --> I I --> J{Parse Return Result} J -- Success --> K[Print Video Stream URL and Expiration Time] J -- Failed --> L[Print Error Message] K --> M[Output JSON Result] L --> M M --> N[End] ``` --- ## 📋 API Parameter Details ### 1. Device Video Stream Request Parameters **Endpoint**: `POST /api/hccgw/video/v1/live/address/get` | Parameter Name | Type | Description | Required | Default | Notes | |----------------|---------|--------------------------------------|----------|---------|----------------------------| | `deviceSerial` | String | Device serial number | **Yes** | - | Device unique identifier | | `resourceId` | String | Channel/monitoring point resource ID | **Yes** | - | Channel unique identifier | | `expireTime` | Integer | Preview duration (seconds) | No | 600 | Default 600 seconds | | `protocol` | Integer | Stream protocol | No | 2 | Fixed: 2 (HLS format only) | ### 2. API Return Data Description | Field Name | Type | Description | Notes | |--------------|---------|-------------------------------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------| | `url` | String | Video stream address | Directly accessible video stream URL | | `expireTime` | String | Expiration time in `yyyy-mm-dd hh:mm:ss` format | Local timezone. **IMPORTANT: This value is the authoritative source. Do NOT parse expiration time from URL query parameters (e.g., Expires, expire).** | | `playable` | Boolean | Whether the Video Stream URL is playable | If `false`, check field for reason. | --- ## 📝 Output Example ### Video Stream Success Example: ```text [2026-04-23 18:12:02] Requesting video stream: Device=J10137390, Resource=6a447d3f9cfe4c8e8394c19f8fbcd3ba [SUCCESS] Video stream successful: https://isgpopen.ezvizlife.com/v3/openlive/J10137390_1_1.m3u8?expire=1776939724&id=967488042038833152&c=c3a53f2806&t=4a33a0fa618fec303534c5bb856693aef55b488353c0f56a6edbc6dba8e54079&ev=100 [INFO] Stream URL expiration time: 2026-04-23 18:22:04 [JSON Output] { "success": true, "url": "https://isgpopen.ezvizlife.com/v3/openlive/J10137390_1_1.m3u8?expire=1776939724&id=967488042038833152&c=c3a53f2806&t=4a33a0fa618fec303534c5bb856693aef55b488353c0f56a6edbc6dba8e54079&ev=100", "expireTime": "2026-04-23 18:22:04", "playable": true, "error": null } ====================================================================== Done ====================================================================== ``` ### Video Stream Failed Example( video encoding format is H265,Not Supported): ```text [2026-04-24 13:51:42] Requesting video stream: Device=D72821502, Resource=661543ed4b35465a9767081ae0a8bf45 [SUCCESS] Got stream URL: https://vtmucyn.ezvizlife.com:8883/v3/openlive/D72821502_1_1.m3u8?expire=1777010504&id=967784913892188160&c=caf588fab7&t=837d2555567061dfa6095842439eafaf8536cf660f0f5aa5ee87c3c327916972&ev=100&u=d00f8fbf53ce42c1aaa8731f4ccacd68 [INFO] Stream URL expiration time: 2026-04-24 14:01:44 [ERROR] Stream URL is not playable, the error type is : H265_NOT_SUPPORTED [JSON Output] { "success": false, "url": "https://vtmucyn.ezvizlife.com:8883/v3/openlive/D72821502_1_1.m3u8?expire=1777010504&id=967784913892188160&c=caf588fab7&t=837d2555567061dfa6095842439eafaf8536cf660f0f5aa5ee87c3c327916972&ev=100&u=d00f8fbf53ce42c1aaa8731f4ccacd68", "expireTime": "2026-04-24 14:01:44", "playable": false } ====================================================================== Done ====================================================================== ``` --- ## 📂 File Structure ```text ├── scripts/ │ └── get_video_url.py # Device video stream core execution script └── SKILL.md # Skill usage documentation ``` --- ## ❓ FAQ - **Q: Why is video stream loading slowly?** - A: Video stream quality is affected by network bandwidth, please ensure stable network environment. - **Q: What if "Resource ID error" is shown?** - A: Please first get correct channel `resourceId` through resource management module. - **Q: What is the validity period of video stream address?** - A: **Equals your configured stream duration**, which is the value of the `video-duration` parameter. For example, setting `--video-duration 1080` (18 minutes) means the address validity is exactly 18 minutes. - **Q: What if video stream address is expired?** - A: Video stream address has time limit, please re-run script to get after expiration. - **Q: Can video stream address be opened and played directly?** - A: Yes. - **Q: Video stream address fails to load?** - A: **Must check in this order:** 1. **Stream encryption**: Run `device_detail.py ` — `Stream Encryption` must be `Disabled` 2. **Video encoding format**: Check in HCT platform — must be **H264** (H265 may fail in browser) --- --- **Error Codes**: | Return Code | Return Message | Description | |-------------|-----------------------|---------------------------------------------------------------------------------------------| | EVZ60019 | Encryption is enabled | Stream encryption not disabled, you MUST disable it in HCT platform before stream will work | --- FILE:modules/Hik-Connect_Team_Video/scripts/get_video_url.py #!/usr/bin/env python # -*- coding: utf-8 -*- """ HCTOpen Device Video Stream """ import sys import os import json import argparse from datetime import datetime, timezone try: import requests except ImportError: requests = None # Robust lib directory import logic: search upward until lib directory is found def setup_lib_path(): current_dir = os.path.dirname(os.path.abspath(__file__)) # Search upward 3 levels for root directory containing lib root_dir = current_dir for _ in range(3): root_dir = os.path.dirname(root_dir) potential_lib = os.path.join(root_dir, "lib") if os.path.exists(potential_lib): if potential_lib not in sys.path: sys.path.insert(0, potential_lib) return True return False if not setup_lib_path(): print("[ERROR] Cannot find lib directory, please ensure script is located in Hik-Connect_Team Skills directory structure") sys.exit(1) from token_manager import HCTOpenClient def verify_stream(url): """ Verify stream is playable by fetching m3u8 and checking for error patterns. Returns: (is_valid, error_type) - H265 error pattern: m3u8 contains "ErrCode/9053" """ if not requests: print("[WARN] requests library not installed, skipping stream verification") return True, None try: resp = requests.get(url, timeout=5, headers={"User-Agent": "HCTOpen/1.0"}) if resp.status_code != 200: return True, None # Don't block on HTTP errors, let player handle content = resp.text # Check for H265 error indicator: ErrCode/9053 in playlist if "ErrCode/9053" in content or "9053_0.ts" in content: return False, "H265_NOT_SUPPORTED" # Check if playlist immediately ends (no valid segments) lines = content.split("\n") segment_count = sum(1 for line in lines if line.endswith(".ts")) if segment_count == 0 and "#EXT-X-ENDLIST" in content: return False, "NO_VALID_SEGMENTS" return True, None except Exception as e: print(f"[WARN] Stream verification failed: {e}") return True, None # Don't block on network errors def format_expire_time(exp_time_ms): """Convert millisecond timestamp to yyyy-mm-dd hh:mm:ss in local timezone""" if not exp_time_ms: return None dt = datetime.fromtimestamp(exp_time_ms / 1000, tz=timezone.utc).astimezone() return dt.strftime("%Y-%m-%d %H:%M:%S") class VideoClient(HCTOpenClient): """Device video stream client""" def get_url(self, device_serial: str, resource_id: str, video_duration: int = 600): """Get video stream address""" print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] Requesting video stream: Device={device_serial}, Resource={resource_id}") endpoint = "/api/hccgw/video/v1/live/address/get" payload = { "resourceId": resource_id, "deviceSerial": device_serial, "protocol": 2, #HLS format: Stream retrieval supports only this format; no other formats are supported. "expireTime": video_duration } # Video module API uses "Token" as Header Key result = self.request("POST", endpoint, json_data=payload, token_header_key="Token") if result.get("errorCode") == "0": data = result.get("data", {}) stream_url = data.get("url") exp_time_ms = data.get("expireTime") if stream_url: print(f"[SUCCESS] Got stream URL: {stream_url}") # Verify stream is playable (check for H265 errors) is_valid, error_type = verify_stream(stream_url) # Format expire time as yyyy-mm-dd hh:mm:ss expire_time_str = format_expire_time(exp_time_ms) print(f"[INFO] Stream URL expiration time: {expire_time_str}") if not is_valid: print(f"[ERROR] Stream URL is not playable, the error type is : {error_type}") self.exit_with_json({ "success": False, "url": stream_url, "expireTime": expire_time_str, "playable": False }) self.exit_with_json({ "success": True, "url": stream_url, "expireTime": expire_time_str, "playable": True }) else: self.exit_with_json({ "success": False, "url": None, "expireTime": None, "playable": False }) else: # Use unified message field print(f"[ERROR] Video stream failed: {result.get('message', 'Unknown error')}") self.exit_with_json({ "success": False, "url": None, "expireTime": None, "playable": False }) def main(): parser = argparse.ArgumentParser(description="HCTOpen Device Video Stream") parser.add_argument("--device-serial", required=True, help="Device serial number") parser.add_argument("--resource-id", required=True, help="Resource ID (Channel ID)") parser.add_argument("--video-duration", type=int, default=600, help="Valid duration (seconds)") args = parser.parse_args() client = VideoClient() client.get_url(args.device_serial, args.resource_id, args.video_duration) if __name__ == "__main__": main() FILE:modules/Hik-Connect_Team_Resource/SKILL.md --- name: hctopen-resource-manager description: | HCTOpen resource management skill. Supports viewing device list and specific device details, all channel details under specific device. Use when: Need to view available devices, get specific device detailed information, get all channel information under specific device, etc. Before calling this Skill's script, please check if user provided device serial number. If user didn't provide device serial number, please clearly inform user in reply: 'Currently using default parameters (such as viewing device list), if you need to view specific device information, please provide device serial number'. After getting confirmation or ignoring, continue execution. ⚠️ Security Requirement: Must set environment variables: Hik-Connect Team OpenAPI AppKey, Hik-Connect Team OpenAPI SecretKey. (API Domain is automatically obtained from token response) parameters: - name: device-serial type: string description: "Device serial number" required: false - name: page type: integer description: "Page number, default is 1" default: 1 - name: page-size type: integer description: "Page size, default is 10" default: 10 - name: device-category type: string description: "Device category filter. Options: encodingDevice, accessControllerDevice, alarmDevice, videoIntercomDevice, mobileDevice, businessDisplayDevice" required: false - name: match-key type: string description: "Fuzzy match key for device name or serial number. Only effective when device-category is specified." required: false responses: - success: true template: "Device information retrieved for you:" media: "list_card" metadata: openclaw: emoji: "📦" requires: env: ["HIK_CONNECT_TEAM_OPENAPI_APP_KEY", "HIK_CONNECT_TEAM_OPENAPI_SECRET_KEY"] pip: ["requests"] primaryEnv: "HIK_CONNECT_TEAM_OPENAPI_APP_KEY" warnings: - "May read ~/.openclaw/*.json for credentials (env vars have priority)" config: configFileRead: paths: - "~/.openclaw/config.json" - "~/.openclaw/gateway/config.json" - "~/.openclaw/channels.json" priority: "lower than environment variables" description: "Reads Hik-Connect Team credentials from OpenClaw config files as fallback" --- # HCTOpen Resource Manager HCT is short for Hik-Connect for Teams, meaning Hik-Connect Team mode. HCTOpen is short for Hik-Connect for Teams OpenAPI. This Skill supports three core functions: view device list, query device details, and channel details under device. --- ## ⚠️ Security Warning (Read Before Use) | # | Check Item | Status | Description | |---|---------------------------|-------------|----------------------------------------------------------------------------------------------------| | 1 | **Credential Permission** | ⚠️ Required | Please use credentials with **resource query permission**, avoid using super admin credentials | | 2 | **Token Cache** | ✅ Encrypted | Token cached in system temp directory, only current user can read (600 permission) | | 3 | **API Domain** | ✅ Auto | API domain is automatically obtained from token response (no longer requires manual configuration) | --- ## 🚀 Quick Start ### Run Resource Management Scripts ```bash # Scenario 1: View device list (default pagination) python scripts/list_devices.py # Scenario 1a: Filter by device category (encodingDevice) python scripts/list_devices.py --device-category encodingDevice # Scenario 1b: Filter by device category with fuzzy match on name/serial python scripts/list_devices.py --device-category encodingDevice --match-key D728215 # Scenario 2: Query single device details (by serial number) python scripts/device_detail.py L33721705 # Scenario 3: View specific device channel list python scripts/device_channels.py J10137390 # Scenario 4: View door access resource list (specified serial number) python scripts/list_doors.py L33721705 ``` --- ## 🛠 Workflow ```mermaid graph TD A[Start Script] --> B{Check Environment Variables} B -- Missing --> C[Report Error and Exit] B -- Pass --> D[Get AccessToken] D --> E{Is Token Valid?} E -- Cache Valid --> F[Use Cache Directly] E -- Expired/No Cache --> G[Call API to Get New Token] G --> H[Save to Local Cache] F --> I[Send Resource Query Request] H --> I I --> J{Parse Return Result} J -- Success --> K[Print Resource List Table] J -- Failed --> L[Print Error Message] K --> M[Output JSON Result] L --> M M --> N[End] ``` --- ## 📋 API Parameter Details ### 1. Device List Request Parameters **Endpoint**: `POST /api/hccgw/resource/v1/devices/get` | Parameter Name | Type | Description | Required | Default | Notes | |-------------------|---------|---------------------------------------|----------|---------|---------------------------------------------------------------------------------------------------------------| | `page` | Integer | Page number | No | 1 | Starts from 1 | | `pageSize` | Integer | Page size | No | 10 | Max 100 | | `deviceCategory` | String | Device category filter | No | - | encodingDevice, accessControllerDevice, alarmDevice, videoIntercomDevice, mobileDevice, businessDisplayDevice | | `filter.matchKey` | String | Fuzzy match for device name or serial | No | - | Only effective when deviceCategory is specified | #### deviceCategory Options | deviceCategory Value | Description | |--------------------------|----------------------------| | `encodingDevice` | `Encoding Device / Camera` | | `accessControllerDevice` | `Access Controller Device` | | `alarmDevice` | `Alarm Device` | | `videoIntercomDevice` | `Video Intercom Device` | | `mobileDevice` | `Mobile Device` | | `businessDisplayDevice` | `Business Display Device` | ### Device List Output Field Description | Field Name | Type | Description | |--------------------------|---------|------------------------------------------------------| | `success` | Boolean | Whether request was successful | | `total` | Integer | Total number of devices | | `pageIndex` | Integer | Current page number | | `pageSize` | Integer | Page size | | `devices` | Array | Device list, each element is a device object | | `devices[].id` | String | Device ID | | `devices[].name` | String | Device name | | `devices[].category` | String | Device type | | `devices[].type` | String | Device model | | `devices[].serialNo` | String | Device serial number | | `devices[].version` | String | Firmware version | | `devices[].onlineStatus` | Integer | Network status: 0 (offline), 1 (online), 2 (unknown) | | `devices[].addTime` | String | Added time | ### Device List Success Example: ```text [2026-04-09 15:44:01] Getting device list (page 1, 10 items per page)... ====================================================================== HCTOpen Device List (Total: 2, Current page count: 2) ====================================================================== No. Device ID Device Serial Number Device Name Model Version Device Type Added Time Status --------------------------------------------------------------------------------------------------------------------------------------- 1 2604f502e63247d393e83c07f58705b9 D72821502 Small Cup DS-2CV2026G0-IDW V5.5.110 build 200819 encodingDevice 2026-03-30 01:30:55 Online 2 39a2f72cf2d8404b9067d35cfe2d3501 J10137390 Test Room DS-2TD2637-10/P V5.5.64 build 230207 encodingDevice 2026-04-01 05:57:00 Online ====================================================================== [JSON Output] { "success": true, "totalCount": 2, "pageIndex": 1, "pageSize": 10, "devices": [ { "id": "2604f502e63247d393e83c07f58705b9", "serialNo": "D72821502", "name": "Small Cup", "type": "DS-2CV2026G0-IDW", "version": "V5.5.110 build 200819", "onlineStatus": 1, "category": "encodingDevice", "addTime": "2026-03-30 01:30:55" }, { "id": "39a2f72cf2d8404b9067d35cfe2d3501", "serialNo": "J10137390", "name": "Test Room", "type": "DS-2TD2637-10/P", "version": "V5.5.64 build 230207", "onlineStatus": 1, "category": "encodingDevice", "addTime": "2026-04-01 05:57:00" } ] } ====================================================================== Done ====================================================================== ``` ### Device List Failed Example: ```text [2026-04-22 19:05:43] Getting device list (page 1, 10 items per page)... [WARNING] match-key is only effective when device-category is specified.. {'pageIndex': 1, 'pageSize': 10, 'filter': {'matchKey': 'D728215'}} [ERROR] Failed to get device list: Device category is request{OPEN000010} [JSON Output] { "success": false, "error": "Device category is request{OPEN000010}", "errorCode": "OPEN000010" } ====================================================================== Done ====================================================================== ``` ### 2. Device Detail Request Parameters **Endpoint**: `POST /api/hccgw/resource/v1/devicedetail/get` | Parameter Name | Type | Description | Required | Default | Notes | |------------------|--------|----------------------|----------|---------|--------------------------| | `deviceSerialNo` | String | Device serial number | **Yes** | - | Device unique identifier | ### Device Detail Output Field Description | Field Name | Type | Description | |--------------------------------------------|---------|-------------------------------------------------| | `success` | Boolean | Whether request was successful | | `data` | Object | Device detail data object | | `data.device` | Object | Device detailed information | | `data.device.baseInfo` | Object | Device basic information | | `data.device.baseInfo.id` | String | Device ID | | `data.device.baseInfo.name` | String | Device name | | `data.device.baseInfo.category` | String | Device type | | `data.device.baseInfo.serialNo` | String | Device serial number | | `data.device.baseInfo.version` | String | Firmware version | | `data.device.baseInfo.type` | String | Device model | | `data.device.baseInfo.streamEncryptEnable` | String | Stream encryption enable, 1-enabled, 0-disabled | | `data.device.onlineStatus` | Integer | Device online status: 1-online, 0-offline | ### Device Detail Success Example: ```text ====================================================================== HCTOpen Device Detail ====================================================================== [Time] 2026-04-07 10:00:00 [INFO] Querying device details: F68147103 Device Name Device Serial Number Model Version Status ---------------- -------------- ---------------- -------------------- -------- F68147103 F68147103 DS-9664NI-I8 V4.40.220 build 210125 Online ====================================================================== [JSON Output] { "success": true, "data": { "device": { "baseInfo": { "id": "5c263e4293c84eae81720e9e481e33ad", "name": "F68147103", "category": "encodingDevice", "serialNo": "F68147103", "version": "V4.40.220 build 210125", "type": "DS-9664NI-I8", "streamEncryptEnable": "1", } "onlineStatus": 1, } } } ====================================================================== Done ====================================================================== ``` ### 3. Device Channel List Request Parameters **Endpoint**: `POST /api/hccgw/resource/v1/areas/cameras/get` | Parameter Name | Type | Description | Required | Default | Notes | |----------------|---------|----------------------|----------|---------|--------------------------| | `deviceSerial` | String | Device serial number | **Yes** | - | Device unique identifier | | `page` | Integer | Page number | No | 1 | Starts from 1 | | `pageSize` | Integer | Page size | No | 10 | Max 100 | ### Device Channel List Output Field Description | Field Name | Type | Description | |---------------------------|---------|-------------------------------------------------------| | `success` | Boolean | Whether request was successful | | `data` | Object | Device channel list data object | | `data.totalCount` | Integer | Total channel count | | `data.pageIndex` | Integer | Current page number | | `data.pageSize` | Integer | Page size | | `data.camera` | Array | Camera channel list, each element is a channel object | | `data.camera[].id` | String | Camera ID | | `data.camera[].name` | String | Camera name | | `data.camera[].online` | String | Online status: "1"-online, "0"-offline | | `data.camera[].channelNo` | String | Channel number | ### Device Channel List Success Example: ```text [2026-04-09 17:11:21] Querying device channels: J10137390 ====================================================================== HCTOpen Device Channel List (Current page count: 2) ====================================================================== No. Resource ID Channel Name Status Area Channel No. -------------------------------------------------------------- 1 6a447d3f9cfe4c8e8394c19f8fbcd3ba Test Room_1 Offline OpenClaw 1 2 84b70e3ced36474fb2b8e6d02b9f8efc Test Room_2 Offline OpenClaw 2 ====================================================================== [JSON Output] { "success": true, "pageIndex": 1, "pageSize": 50, "total": 2, "channels": [ { "id": "6a447d3f9cfe4c8e8394c19f8fbcd3ba", "name": "Test Room_1", "online": "1", "channelNo": "1" }, { "id": "84b70e3ced36474fb2b8e6d02b9f8efc", "name": "Test Room_2", "online": "1", "channelNo": "2" } ] } ====================================================================== Done ====================================================================== ``` ### 4. Door Access Resource List Request Parameters **Endpoint**: `POST /api/hccgw/resource/v1/areas/doors/get` | Parameter Name | Type | Description | Required | Default | Notes | |----------------|--------|----------------------|----------|---------|---------------------------------------------------| | `deviceSerial` | String | Device serial number | Yes | - | Filter door access resources for specified device | ### Door Access Resource List Output Field Description | Field Name | Type | Description | |----------------------|---------|----------------------------------------| | `success` | Boolean | Whether request was successful | | `total` | Integer | Total door access resources | | `doors` | Array | Door access list | | `doors[].resourceId` | String | Door Resource ID | | `doors[].name` | String | Door Access name | | `doors[].online` | String | Online status: "1"-online, "0"-offline | ### Door Access Resource List Success Example: ```text [2026-04-10 09:49:51] Getting door access resource list (Device serial number: L33721705)... ====================================================================== HCTOpen Door Access Resource List (Count: 1) ====================================================================== No. Door Resource ID Door Access Name Status --------------------------------------------------- 1 2aabf37ad9804f66acc4ad4fb7bd4698 L33721705 Online ====================================================================== [JSON Output] { "success": true, "total": 1, "doors": [ { "resourceId": "2aabf37ad9804f66acc4ad4fb7bd4698", "name": "L33721705", "online": "1" } ] } ====================================================================== Done ====================================================================== ``` --- ## 📂 File Structure ```text ├── scripts/ │ ├── list_devices.py # Device list query script │ ├── device_detail.py # Device detail query script │ ├── device_channels.py # Device channel query script │ └── list_doors.py # Device door access resource query script └── SKILL.md # Skill usage documentation ``` --- ## ❓ FAQ - **Q: Why can't I find my device?** - A: Please ensure Hik-Connect Team OpenAPI AppKey has permission to access the device, and check if serial number is entered correctly. - **Q: What do status codes 1 and 0 mean?** - A: 1 means online, 0 means offline. - **Q: How to get all devices?** - A: Script supports pagination, if there are many devices, please adjust `--page-size` parameter or loop request. --- --- #### deviceCategory Options **Error Codes**: | Return Code | Return Message | Description | |-------------|-----------------------------|---------------------------------------------------------------------------| | OPEN000010 | Device category is request | `match-key` is only effective when `device-category` is specified. | | OPEN000010 | Device category not support | Please ensure `device-category` is valid and within the supported options | --- FILE:modules/Hik-Connect_Team_Resource/scripts/device_channels.py #!/usr/bin/env python # -*- coding: utf-8 -*- """ HCTOpen Device Channel List """ import sys import os import json import argparse from datetime import datetime # Robust lib directory import logic: search upward until lib directory is found def setup_lib_path(): current_dir = os.path.dirname(os.path.abspath(__file__)) # Search upward 3 levels for root directory containing lib root_dir = current_dir for _ in range(3): root_dir = os.path.dirname(root_dir) potential_lib = os.path.join(root_dir, "lib") if os.path.exists(potential_lib): if potential_lib not in sys.path: sys.path.insert(0, potential_lib) return True return False if not setup_lib_path(): print("[ERROR] Cannot find lib directory, please ensure script is located in Hik-Connect_Team Skills directory structure") sys.exit(1) from token_manager import HCTOpenClient class DeviceChannelsClient(HCTOpenClient): """Device channel query client""" def get_channels(self, device_serial: str, page: int = 1, page_size: int = 50): """Get and print device channel list""" print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] Querying device channels: {device_serial}") endpoint = "/api/hccgw/resource/v1/areas/cameras/get" payload = { "pageIndex": page, "pageSize": page_size, "filter": {"deviceSerialNo": device_serial} } # Resource module API uses "Token" as Header Key result = self.request("POST", endpoint, json_data=payload, token_header_key="Token") if result.get("errorCode") == "0": data = result.get("data", {}) channels = data.get("camera", []) total = len(channels) headers = ["No.", "Resource ID", "Channel Name", "Status", "Area", "Channel No."] rows = [] for i, ch in enumerate(channels, 1): status = "Online" if ch.get("online") == "1" else "Offline" area_name = ch.get("area", {}).get("name", "Unknown") channel_no = ch.get("device", {}).get("channelInfo", {}).get("no", "-") rows.append([ i, ch.get("id"), ch.get("name", "Unknown"), status, area_name, channel_no ]) self.print_table(f"HCTOpen Device Channel List (Current page count: {total})", headers, rows) # Maintain output format consistent with original script self.exit_with_json({ "success": True, "pageIndex": page, "pageSize": page_size, "total": total, "channels": [ { "id": c.get("id"), "name": c.get("name"), # Convert to "1" or "0" "online": c.get("online"), # Map to root-level channelNo "channelNo": c.get("device", {}).get("channelInfo", {}).get("no") } for c in channels ] }) else: # Use unified message field print(f"[ERROR] Failed to get channel list: {result.get('message', 'Unknown error')}") self.exit_with_json({ "success": False, "error": result.get("message", "Unknown error"), "errorCode": result.get("errorCode") }) def main(): parser = argparse.ArgumentParser(description="HCTOpen Get Device Channel List") parser.add_argument("device_serial", help="Device serial number") parser.add_argument("--page", type=int, default=1, help="Page number") parser.add_argument("--page-size", type=int, default=50, help="Page size") args = parser.parse_args() client = DeviceChannelsClient() client.get_channels(args.device_serial, page=args.page, page_size=args.page_size) if __name__ == "__main__": main() FILE:modules/Hik-Connect_Team_Resource/scripts/device_detail.py #!/usr/bin/env python # -*- coding: utf-8 -*- """ HCTOpen Device Detail """ import sys import os import json import argparse from datetime import datetime # Robust lib directory import logic: search upward until lib directory is found def setup_lib_path(): current_dir = os.path.dirname(os.path.abspath(__file__)) # Search upward 3 levels for root directory containing lib root_dir = current_dir for _ in range(3): root_dir = os.path.dirname(root_dir) potential_lib = os.path.join(root_dir, "lib") if os.path.exists(potential_lib): if potential_lib not in sys.path: sys.path.insert(0, potential_lib) return True return False if not setup_lib_path(): print("[ERROR] Cannot find lib directory, please ensure script is located in Hik-Connect_Team Skills directory structure") sys.exit(1) from token_manager import HCTOpenClient class DeviceDetailClient(HCTOpenClient): """Device detail query client""" def get_detail(self, device_serial: str): """Get and print device details""" print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] Querying device details: {device_serial}") endpoint = "/api/hccgw/resource/v1/devicedetail/get" payload = {"deviceSerialNo": device_serial} # Resource module API uses "Token" as Header Key result = self.request("POST", endpoint, json_data=payload, token_header_key="Token") if result.get("errorCode") == "0": data = result.get("data", {}).get("device", {}) base_info = data.get("baseInfo", {}) # 1. Define list of fields to remove exclude_keys = [ "availableCameraChannelNum", "availableAlarmInputChannelNum", "availableAlarmOutputChannelNum", "areaId", "area" ] # 2. Create a simplified base_info for JSON output # Use dict comprehension to filter out unwanted keys filtered_base_info = {k: v for k, v in base_info.items() if k not in exclude_keys} headers = ["Device ID", "Device Name", "Device Serial Number", "Device Type", "Model", "Status", "Version", "Stream Encryption"] status = "Online" if data.get("onlineStatus") == 1 else "Offline" rows = [[ base_info.get("id"), base_info.get("name", "Unknown"), base_info.get("serialNo", "Unknown"), base_info.get("category", "Unknown"), base_info.get("type", "Unknown"), status, base_info.get("version", "Unknown"), "Enabled" if base_info.get("streamEncryptEnable", "0") == "1" else "Disabled", ]] self.print_table("HCTOpen Device Detail", headers, rows) # Maintain output format self.exit_with_json({ "success": True, "total": 1, "devices": [{ "base_info": filtered_base_info, "onlineStatus": data.get("onlineStatus") }] }) else: # Use unified message field print(f"[ERROR] Failed to get device details: {result.get('message', 'Unknown error')}") self.exit_with_json({ "success": False, "error": result.get("message", "Unknown error"), "errorCode": result.get("errorCode") }) def main(): parser = argparse.ArgumentParser(description="HCTOpen Get Device Detail") parser.add_argument("device_serial", help="Device serial number") args = parser.parse_args() client = DeviceDetailClient() client.get_detail(args.device_serial) if __name__ == "__main__": main() FILE:modules/Hik-Connect_Team_Resource/scripts/list_devices.py #!/usr/bin/env python # -*- coding: utf-8 -*- """ HCTOpen Device List """ import sys import os import argparse import json from datetime import datetime # Robust lib directory import logic: search upward until lib directory is found def setup_lib_path(): current_dir = os.path.dirname(os.path.abspath(__file__)) # Search upward 3 levels for root directory containing lib root_dir = current_dir for _ in range(3): root_dir = os.path.dirname(root_dir) potential_lib = os.path.join(root_dir, "lib") if os.path.exists(potential_lib): if potential_lib not in sys.path: sys.path.insert(0, potential_lib) return True return False if not setup_lib_path(): print("[ERROR] Cannot find lib directory, please ensure script is located in Hik-Connect_Team Skills directory structure") sys.exit(1) from token_manager import HCTOpenClient class DeviceListClient(HCTOpenClient): """Device list query client""" def fetch_devices(self, page: int = 1, page_size: int = 10, device_category: str = None, match_key: str = None): """Get device list and print""" print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] Getting device list (page {page}, {page_size} items per page)...") # Validate match_key requirement if match_key and not device_category: print("[WARNING] match-key is only effective when device-category is specified.") endpoint = "/api/hccgw/resource/v1/devices/get" payload = {"pageIndex": page, "pageSize": page_size} # Add device category filter if specified if device_category: payload["deviceCategory"] = device_category if match_key: payload["filter"] = {"matchKey": match_key} print(payload) # Resource module API uses "Token" as Header Key result = self.request("POST", endpoint, json_data=payload, token_header_key="Token") if result.get("errorCode") == "0": data = result.get("data", {}) devices = data.get("device", []) total = len(devices) headers = ["No.", "Device ID", "Device Serial Number", "Device Name", "Model", "Version", "Device Type", "Added Time", "Status"] rows = [] for i, dev in enumerate(devices, 1): status = "Online" if dev.get("onlineStatus") == 1 else "Offline" rows.append([ i, dev.get("id"), dev.get("serialNo", "Unknown"), dev.get("name", "Unknown"), dev.get("type", "Unknown"), dev.get("version", "-"), dev.get("category", "Unknown"), dev.get("addTime", "Unknown"), status ]) self.print_table(f"HCTOpen Device List (Current page count: {total})", headers, rows) self.exit_with_json({ "success": True, "total": total, "devices": [ { "id": d.get("id"), "deviceName": d.get("name"), "serialNo": d.get("serialNo"), "type": d.get("type"), "onlineStatus": d.get("onlineStatus"), "category": d.get("category"), "addTime": d.get("addTime"), } for d in devices ] }) else: # Use unified message field print(f"[ERROR] Failed to get device list: {result.get('message', 'Unknown error')}") self.exit_with_json({ "success": False, "error": result.get("message", "Unknown error"), "errorCode": result.get("errorCode") }) def main(): parser = argparse.ArgumentParser(description="HCTOpen Get Device List") parser.add_argument("--page", type=int, default=1, help="Page number (default: 1)") parser.add_argument("--page-size", type=int, default=10, help="Page size (default: 10)") parser.add_argument("--device-category", type=str, default=None, help="Device category filter (encodingDevice, accessControllerDevice, alarmDevice, videoIntercomDevice, mobileDevice, businessDisplayDevice)") parser.add_argument("--match-key", type=str, default=None, help="Fuzzy match key for device name or serial number. Only effective when device-category is specified.") args = parser.parse_args() client = DeviceListClient() client.fetch_devices(page=args.page, page_size=args.page_size, device_category=args.device_category, match_key=args.match_key) if __name__ == "__main__": main() FILE:modules/Hik-Connect_Team_Resource/scripts/list_doors.py #!/usr/bin/env python # -*- coding: utf-8 -*- """ HCTOpen Door List """ import sys import os import argparse import json from datetime import datetime # Robust lib directory import logic: search upward until lib directory is found def setup_lib_path(): current_dir = os.path.dirname(os.path.abspath(__file__)) # Search upward 3 levels for root directory containing lib root_dir = current_dir for _ in range(3): root_dir = os.path.dirname(root_dir) potential_lib = os.path.join(root_dir, "lib") if os.path.exists(potential_lib): if potential_lib not in sys.path: sys.path.insert(0, potential_lib) return True return False if not setup_lib_path(): print("[ERROR] Cannot find lib directory, please ensure script is located in Hik-Connect_Team Skills directory structure") sys.exit(1) from token_manager import HCTOpenClient class DoorListClient(HCTOpenClient): """Door access resource list query client""" def fetch_doors(self, device_serial: str): """Get door access resource list and print""" print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] Getting door access resource list (Device serial number: {device_serial if device_serial else 'All'})...") endpoint = "/api/hccgw/resource/v1/areas/doors/get" # pageSize=100, pageIndex=1, includeSubArea=1 are fixed values payload = { "pageIndex": 1, "pageSize": 100, "filter": { "includeSubArea": "1", "deviceSerialNo": device_serial } } # Resource module API uses "Token" as Header Key result = self.request("POST", endpoint, json_data=payload, token_header_key="Token") if result.get("errorCode") == "0": data = result.get("data", {}) doors = data.get("door", []) total = len(doors) headers = ["No.", "Door Resource ID", "Door Access Name", "Status"] rows = [] simplified_doors = [] for i, door in enumerate(doors, 1): status = "Online" if door.get("online") == "1" else "Offline" rows.append([ i, door.get("id"), door.get("name", "Unknown"), status ]) # Only keep id, name, online status simplified_doors.append({ "resourceId": door.get("id"), "name": door.get("name"), "online": door.get("online") }) self.print_table(f"HCTOpen Door Access Resource List (Count: {total})", headers, rows) self.exit_with_json({ "success": True, "total": total, "doors": simplified_doors }) else: # Use unified message field print(f"[ERROR] Failed to get door access resource list: {result.get('message', 'Unknown error')}") self.exit_with_json({ "success": False, "error": result.get("message", "Unknown error"), "errorCode": result.get("errorCode") }) def main(): parser = argparse.ArgumentParser(description="HCTOpen Get Door Access Resource List") parser.add_argument("device_serial", help="Device serial number (optional)") args = parser.parse_args() client = DoorListClient() client.fetch_doors(device_serial=args.device_serial) if __name__ == "__main__": main() FILE:modules/Hik-Connect_Team_Capture/SKILL.md --- name: hctopen-capture description: | HCTOpen device capture and decryption skill. Supports capture for specified device channel, and provides encrypted image decryption functionality. The returned capture address is cloud address instead of local address, can be accessed directly. Use when: Need to get device real-time image or decrypt encrypted device image. ⚠️ Security Requirement: Must set environment variables: Hik-Connect Team OpenAPI AppKey, Hik-Connect Team OpenAPI SecretKey. (API Domain is automatically obtained from token response) parameters: - name: device-serial type: string description: "Device serial number" required: true - name: channel-no type: string description: "Channel number, default is 1" default: "1" responses: - success: true template: "Preview image generated for you, click link below to view:" media: "image_card" metadata: openclaw: emoji: "📸" requires: env: ["HIK_CONNECT_TEAM_OPENAPI_APP_KEY", "HIK_CONNECT_TEAM_OPENAPI_SECRET_KEY"] pip: ["requests", "pycryptodome", "Pillow"] primaryEnv: "HIK_CONNECT_TEAM_OPENAPI_APP_KEY" warnings: - "May read ~/.openclaw/*.json for credentials (env vars have priority)" config: configFileRead: paths: - "~/.openclaw/config.json" - "~/.openclaw/gateway/config.json" - "~/.openclaw/channels.json" priority: "lower than environment variables" description: "Reads Hik-Connect Team credentials from OpenClaw config files as fallback" --- # HCTOpen Capture HCT is short for Hik-Connect for Teams, meaning Hik-Connect Team mode. HCTOpen is short for Hik-Connect for Teams OpenAPI. This Skill provides device real-time capture functionality, suitable for anomaly verification, real-time screen preview and other scenarios. > **Note!!!**: This skill only provides capture capability. If device has stream encryption enabled causing image not viewable, user needs to manually decrypt in HCT!!! Skill has no decryption capability. > **Important Pre-check Information**: > - **Check device status before capturing**: Use the device detail function in the resource management module to verify if stream encryption is enabled > - **Example command**: `python scripts/device_detail.py {device_serial}` > - If `Stream Encryption` shows `Enabled`, you must disable it first before capture --- ## ⚠️ Security Warning (Read Before Use) | # | Check Item | Status | Description | |---|---------------------------|-------------|--------------------------------------------------------------------------------------------------------------------------| | 1 | **Credential Permission** | ⚠️ Required | Please use credentials with **capture permission**, avoid using super admin credentials | | 2 | **Image Encryption** | ⚠️ Note | If device has image encryption enabled, returned URL may not be directly viewable, user needs to manually decrypt in HCT | | 3 | **Token Cache** | ✅ Encrypted | Token cached in system temp directory, only current user can read (600 permission) | | 4 | **API Domain** | ✅ Auto | API domain is automatically obtained from token response (no longer requires manual configuration) | --- ## 🚀 Quick Start ```bash # Scenario 1: Capture image for specified device serial number (channel number defaults to 1) python scripts/capture_pic.py L33721705 # Scenario 2: Capture image for specified device serial number and channel number python scripts/capture_pic.py D72821502,2 ``` --- ## 🛠 Workflow ```mermaid graph TD A[Start Script] --> B{Check Environment Variables} B -- Missing --> C[Report Error and Exit] B -- Pass --> D[Get AccessToken] D --> E{Is Token Valid?} E -- Cache Valid --> F[Use Cache Directly] E -- Expired/No Cache --> G[Call API to Get New Token] G --> H[Save to Local Cache] F --> I[Send Capture Request] H --> I I --> J{Parse Return Result} J -- Success --> K[Print Capture URL and Encryption Status] J -- Failed --> L[Print Error Message] K --> M[Output JSON Result] L --> M M --> N[End] ``` --- ## 📋 API Parameter Details ### 1. Device Capture Request Parameters **Endpoint**: `POST /api/hccgw/resource/v1/device/capturePic` | Parameter Name | Type | Description | Required | Default | Notes | |----------------|--------|----------------------|----------|---------|--------------------------| | `deviceSerial` | String | Device serial number | **Yes** | - | Device unique identifier | | `channelNo` | String | Channel number | No | "1" | Default is 1 | ### 2. API Return Data Description | Field Name | Type | Description | Notes | |---------------|---------|-------------------------|--------------------------------------------------| | `captureUrl` | String | Capture preview address | Directly accessible image URL (if not encrypted) | | `isEncrypted` | Integer | Is encrypted | 0-not encrypted, 1-encrypted | --- ## 📝 Output Example ### Capture Success Example: ```text [2026-04-25 22:25:18] Requesting capture: Device=D72821502, Channel=1 [SUCCESS] Capture successful: https://hpc-sgp-prod-s3-hccvis.oss-ap-southeast-1.aliyuncs.com/hccopen/capture/2026-04-25/D72821502/1/c4d29884-5d0c-47d9-8db7-3ccccd6eaf3b.jpeg?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Date=20260425T142521Z&X-Amz-SignedHeaders=host&X-Amz-Expires=900&X-Amz-Credential=LTAI5tQckMpJxMb4qoHXJySP%2F20260425%2Foss-ap-southeast-1%2Fs3%2Faws4_request&X-Amz-Signature=6dbe52fb30120e3fb65a9e5bed420e5e1dea07c5a78a15eca47f105809babb69 [JSON Output] { "success": true, "captureUrl": "https://hpc-sgp-prod-s3-hccvis.oss-ap-southeast-1.aliyuncs.com/hccopen/capture/2026-04-25/D72821502/1/c4d29884-5d0c-47d9-8db7-3ccccd6eaf3b.jpeg?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Date=20260425T142521Z&X-Amz-SignedHeaders=host&X-Amz-Expires=900&X-Amz-Credential=LTAI5tQckMpJxMb4qoHXJySP%2F20260425%2Foss-ap-southeast-1%2Fs3%2Faws4_request&X-Amz-Signature=6dbe52fb30120e3fb65a9e5bed420e5e1dea07c5a78a15eca47f105809babb69", "isEncrypted": 0 } ====================================================================== Done ====================================================================== ``` --- ## 📂 File Structure ```text ├── scripts/ │ └── capture_pic.py # Device capture core execution script └── SKILL.md # Skill usage documentation ``` --- ## ❓ FAQ - **Q: Why can't the image be opened?** - A: **There are two main possible reasons:** 1. **Device has stream encryption enabled**: First check using device detail script (`python scripts/device_detail.py {device_serial}`). If it shows `Stream Encryption: Enabled`, you must disable it in HCT platform first 2. **The returned image's `isEncrypted` field is 1**: This means the captured image is encrypted, same solution - disable stream encryption and retry - **Q: How long is capture URL valid?** - A: Valid for 15 minutes, please view or download as soon as possible. - **Q: What if "Device offline" is shown?** - A: Capture function requires device to be online, please first confirm device status through resource management module. - **Q: Returned image is a URL address?** - A: If user didn't explicitly mention needing URL address, default to returning image to user. --- --- **Error Codes**: | Return Code | Return Message | Description | |-------------|-------------------------|-------------------------------------------------------------| | OPEN000554 | Device Offline | Device is offline, please check device online status | | OPEN000555 | Device Response Timeout | Device response timeout, please check device network status | | OPEN000556 | Device Capture Failed | Device capture failed | --- FILE:modules/Hik-Connect_Team_Capture/scripts/capture_pic.py #!/usr/bin/env python # -*- coding: utf-8 -*- """ HCTOpen Device Picture Capture """ import sys import os import json import argparse from datetime import datetime # Robust lib directory import logic: search upward until lib directory is found def setup_lib_path(): current_dir = os.path.dirname(os.path.abspath(__file__)) # Search upward 3 levels for root directory containing lib root_dir = current_dir for _ in range(3): root_dir = os.path.dirname(root_dir) potential_lib = os.path.join(root_dir, "lib") if os.path.exists(potential_lib): if potential_lib not in sys.path: sys.path.insert(0, potential_lib) return True return False if not setup_lib_path(): print("[ERROR] Cannot find lib directory, please ensure script is located in Hik-Connect_Team Skills directory structure") sys.exit(1) from token_manager import HCTOpenClient class CaptureClient(HCTOpenClient): """Device capture client""" def capture(self, device_serial: str, channel_no: int = 1): """Execute capture operation""" print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] Requesting capture: Device={device_serial}, Channel={channel_no}") endpoint = "/api/hccgw/resource/v1/device/capturePic" payload = { "deviceSerial": device_serial, "channelNo": str(channel_no) } # Capture module API uses "Token" as Header Key result = self.request("POST", endpoint, json_data=payload, token_header_key="Token") if result.get("errorCode") == "0": data = result.get("data", {}) capture_url = data.get("captureUrl") is_encrypted = data.get("isEncrypted") if capture_url: print(f"[SUCCESS] Capture successful: {capture_url}") if is_encrypted == 1: print("[INFO] Note: Image is encrypted, need to use key to decrypt before viewing") self.exit_with_json({ "success": True, "captureUrl": capture_url, "isEncrypted": is_encrypted }) else: print("[ERROR] Response does not contain capture URL") self.exit_with_json({"success": False, "error": "Capture URL not found"}) else: # Use unified message field print(f"[ERROR] Capture failed: {result.get('message', 'Unknown error')}") self.exit_with_json({ "success": False, "error": result.get("message", "Unknown error"), "errorCode": result.get("errorCode") }) def main(): parser = argparse.ArgumentParser(description="HCTOpen Device Capture") parser.add_argument("device_info", help="Device serial number, optional comma-separated channel number (e.g. D72821502,1)") args = parser.parse_args() parts = args.device_info.split(",") device_serial = parts[0].strip() channel_no = 1 if len(parts) > 1: try: channel_no = int(parts[1].strip()) except ValueError: print("[ERROR] Channel number must be integer") sys.exit(1) client = CaptureClient() client.capture(device_serial, channel_no) if __name__ == "__main__": main() FILE:modules/Hik-Connect_Team_Alarm/EVENT_CODES.md # HCT Alarm Event Codes & Descriptions This document lists the alarm event codes supported by HCT platform and their corresponding detailed descriptions for reference when subscribing. ## 1. Video Intercom | Event Code | Description | |:------------|:-------------------------------------| | `Msg140001` | Messages about video intercom events | ## 2. On-Board Monitoring | Event Code | Description | |:------------|:----------------------------------| | `Msg330001` | GPS Data Report | | `Msg330101` | Alarm Triggered by Panic Button | | `Msg330102` | Alarm Input | | `Msg330201` | Forward Collision Warning | | `Msg330202` | Headway Monitoring Warning | | `Msg330203` | Lane Deviation Warning | | `Msg330204` | Pedestrian Collision Warning | | `Msg330205` | Speed Limit Warning | | `Msg330301` | Blind Spot Warning | | `Msg330401` | Sharp Turn | | `Msg330402` | Sudden Brake | | `Msg330403` | Sudden Acceleration | | `Msg330404` | Rollover | | `Msg330405` | Speeding | | `Msg330406` | Collision | | `Msg330407` | ACC ON | | `Msg330408` | ACC OFF | | `Msg330501` | Smoking | | `Msg330502` | Using Mobile Phone | | `Msg330503` | Fatigue Driving | | `Msg330504` | Distraction | | `Msg330505` | Seatbelt Unbuckled | | `Msg330506` | Video Tampering | | `Msg330507` | Yawning | | `Msg330508` | Wearing IR Interrupted Sunglasses | | `Msg330509` | Absence | | `Msg330510` | Front Passenger Detection | | `Msg335000` | Person and Vehicle Match | | `Msg335001` | Person and Vehicle Mismatch | ## 3. Authentication Event | Event Code | Description | |:------------|:-------------------------------------------------| | `Msg110001` | Access Granted by Card and Fingerprint | | `Msg110002` | Access Granted by Card, Fingerprint, and PIN | | `Msg110003` | Access Granted by Card | | `Msg110004` | Access Granted by Card and PIN | | `Msg110005` | Access Granted by Fingerprint | | `Msg110006` | Access Granted by Fingerprint and PIN | | `Msg110007` | Duress Alarm | | `Msg110008` | Access Granted by Face and Fingerprint | | `Msg110009` | Access Granted by Face and PIN | | `Msg110010` | Access Granted by Face and Card | | `Msg110011` | Access Granted by Face, PIN, and Fingerprint | | `Msg110012` | Access Granted by Face, Card, and Fingerprint | | `Msg110013` | Access Granted by Face | | `Msg110018` | Access Granted via Combined Authentication Modes | | `Msg110019` | Skin-Surface Temperature Measured | | `Msg110020` | Password Authenticated | | `Msg110022` | Access Granted by Bluetooth | | `Msg110023` | Access Granted via QR Code | | `Msg110024` | Access Granted via Keyfob | | `Msg110501` | Verifying Card Encryption Failed | | `Msg110502` | Max. Card Access Failed Attempts | | `Msg110505` | Card No. Expired | | `Msg110506` | Access Timed Out by Card and PIN | | `Msg110507` | Access Denied - Door Remained Locked or Inactive | | `Msg110509` | Access Denied by Card and PIN | | `Msg110510` | Access Timed Out by Card, Fingerprint, and PIN | | `Msg110511` | Access Denied by Card, Fingerprint, and PIN | | `Msg110512` | Access Denied by Card and Fingerprint | | `Msg110513` | Access Timed Out by Card and Fingerprint | | `Msg110514` | No Access Level Assigned | | `Msg110515` | Card No. Does Not Exist | | `Msg110516` | Invalid Time Period | | `Msg110517` | Fingerprint Does Not Exist | | `Msg110518` | Access Denied by Fingerprint | | `Msg110519` | Access Denied by Fingerprint and PIN | | `Msg110520` | Access Timed Out by Fingerprint and PIN | | `Msg110521` | Access Denied by Face and Fingerprint | | `Msg110522` | Access Timed Out by Face and Fingerprint | | `Msg110523` | Access Denied by Face and PIN | | `Msg110524` | Access Timed Out by Face and PIN | | `Msg110525` | Access Denied by Face and Card | | `Msg110526` | Access Timed Out by Face and Card | | `Msg110527` | Access Denied by Face, PIN, and Fingerprint | | `Msg110528` | Access Timed Out by Face, PIN, and Fingerprint | | `Msg110529` | Access Denied by Face, Card, and Fingerprint | | `Msg110530` | Access Timed Out by Face, Card, and Fingerprint | | `Msg110531` | Access Denied by Face | | `Msg110533` | Live Facial Detection Failed | | `Msg110545` | Combined Authentication Timed Out | | `Msg110546` | Access Denied by Invalid M1 Card | | `Msg110547` | Verifying CPU Card Encryption Failed | | `Msg110548` | Access Denied - NFC Card Reading Disabled | | `Msg110549` | EM Card Reading Not Enabled | | `Msg110550` | M1 Card Reading Not Enabled | | `Msg110551` | CPU Card Reading Disabled | | `Msg110552` | Authentication Mode Mismatch | | `Msg110554` | Max. Card and Password Authentication Times | | `Msg110555` | Password Mismatches | | `Msg110556` | Employee ID Does Not Exist | | `Msg110557` | Access Denied: Scheduled Sleep Mode | | `Msg110559` | Verifying Desfire Card Encryption Failed | | `Msg110560` | Absence | | `Msg110561` | Authentication Failed Due to Abnormal Features | | `Msg110564` | Access Denied by Bluetooth | | `Msg110565` | Access Denied by QR Code | | `Msg110566` | Verifying QR Code Secret Key Failed | | `Msg110567` | Access Denied via Keyfob | FILE:modules/Hik-Connect_Team_Alarm/SKILL.md --- name: hctopen-alarm description: | HCTOpen alarm webhook subscription and push management skill. Supports subscribing to alarm events and receiving real-time notifications via Webhook. Use when: Need to configure webhook for receiving HCT alarm pushes, subscribe/unsubscribe to alarm events. ⚠️ Security Requirement: Must set environment variables: Hik-Connect Team OpenAPI AppKey, Hik-Connect Team OpenAPI SecretKey. (API Domain is automatically obtained from token response) ⚠️ Prerequisites: - Requires public HTTPS URL (self-owned server or tunnel like ngrok) to receive webhook pushes from HCT platform. - ⚠️ Key Constraint: The server hosting the public HTTPS URL must be able to reach OpenClaw Gateway's port (dynamically detected from openclaw.json). If OpenClaw is on a different server behind NAT/firewall and unreachable externally, third-party webhook receiver services (e.g., Pipedream, AWS Lambda URL) will NOT work — they only receive and cannot forward to internal OpenClaw. In that case, you must use a tunnel tool (ngrok/cpolar) on the OpenClaw server to create a public entry point instead. metadata: openclaw: emoji: "🔔" requires: env: ["HIK_CONNECT_TEAM_OPENAPI_APP_KEY", "HIK_CONNECT_TEAM_OPENAPI_SECRET_KEY"] npm: ["nodejs"] primaryEnv: "HIK_CONNECT_TEAM_OPENAPI_APP_KEY" warnings: - "May read ~/.openclaw/*.json for credentials (env vars have priority)" config: configFileRead: paths: - "~/.openclaw/config.json" - "~/.openclaw/gateway/config.json" - "~/.openclaw/channels.json" priority: "lower than environment variables" description: "Reads Hik-Connect Team credentials from OpenClaw config files as fallback" --- # Hik-Connect_Team_Alarm (HCT Alarm Push Management) ## 1. Module Introduction `Hik-Connect_Team_Alarm` module is designed to help users implement real-time push of HCT alarm messages through Webhook mechanism. This module integrates a complete closed-loop process of **public network access guidance**, **Webhook receiving service**, **OpenClaw Hooks configuration** and **HCT platform subscription**. This document details how to configure the HCT Open platform Webhook alarm push process. The core idea is to use public network address to receive alarm data pushed by HCT Open platform, forward it to internal network Webhook receiving service, and finally have OpenClaw agent organize and send message notifications. The overall architecture flow is as follows: HCT Open Platform → Public Network Tunnel/Self-owned Server → Webhook Service(:3090) → OpenClaw Hooks → Message Notification > **Core Logic**: HCT Platform pushes messages to public network Webhook address -> Webhook service receives and verifies signature -> Forward to OpenClaw Hooks -> Agent organizes and sends notification. --- ### 1.1 Complete Data Flow and Port Responsibilities The full alarm push data flow spans multiple components. Understanding which port belongs to which process is critical for troubleshooting: ``` HCT Platform (port 443 HTTPS) ↓ sends POST / GET [Public Internet] ↓ Reverse Proxy / Tunnel (server:443) — receives on public HTTPS address ↓ forwards internally Webhook Receiving Service (server.js, port 3090 by default) — validates signature, extracts alarm data ↓ forwards internally OpenClaw Gateway (port shown as gateway.port in openclaw.json, dynamically detected) — receives via /hooks/agent ↓ triggers OpenClaw Agent → formats message → sends to notification channel (Feishu/Telegram/etc.) ``` **Port Responsibilities Table:** | Port | Process | Role | Who Owns It | |------------------------------|-------------------------------------------------|----------------------------------------------------------------------------|---------------------------------| | 443 (HTTPS) | Reverse proxy (Caddy/nginx/etc.) or tunnel tool | Public entry point, receives from HCT platform | User's server or tunnel service | | 3090 (default) | server.js (webhook receiving service) | Receives from reverse proxy, validates HMAC signature, extracts alarm data | OpenClaw server | | gateway port (auto-detected) | OpenClaw Gateway | Receives via `/hooks/agent`, triggers agent processing | OpenClaw server | --- ## 2. Core Workflow (Detailed Version) ### 2.1 Flowchart ```mermaid sequenceDiagram participant User as User participant Agent as Agent (AI Assistant) participant Tool as Alarm Module (Python) participant HCT as HCT Open Platform participant Proxy as Public Network Tunnel (Optional) participant Srv as Webhook Receiving Service (server.js) participant OpenClaw as OpenClaw Hooks participant Notify as OpenClaw Agent Note over User, Agent: **Phase 0: OpenClaw Hooks Readiness Check** Agent->>Agent: Run pre_check.py alt hooks not ready Agent->>User: Ask: "Modify hooks config and restart gateway? (yes/no)" User->>Agent: User confirms end Note over User, Agent: **⚠️ Gate 1: Public URL Plan (NO TUNNEL without explicit Option B)** Agent->>User: Ask: "Who hosts the public HTTPS URL? (A: own server / B: tunnel / C: own URL)" User->>Agent: User confirms plan Note right of Agent: **🚨 ABSOLUTE RULE: Tunnel only if user explicitly chose Option B** Note over User, Agent: **⚠️ Gate 2: Signing Secret** Agent->>User: Ask: "Provide an 8-32 character signing secret" User->>Agent: User provides secret (BLOCK if no answer) Note over User, Srv: **Phase 3: Service Startup** User->>Srv: Start Webhook receiving service User->>Srv: Verify public URL is reachable Note over User, Srv: **Phase 4: Webhook Registration** User->>Tool: Run `webhook_manager.py save --url --secret ` Tool->>HCT: POST `/api/hccgw/webhook/v1/config/save` HCT->>Srv: GET `` (verification request) Srv-->>HCT: Return `200 OK` + signature Header HCT-->>Tool: Return `errorCode: "0"` Tool-->>User: Prompt Webhook registration successful Note over User, Srv: **Phase 5: Event Subscription** Agent->>User: Present event types from EVENT_CODES.md User->>Agent: User confirms which events Agent->>Tool: Run `event_manager.py subscribe --types "chosen_types"` Tool->>HCT: POST `/api/hccgw/rawmsg/v1/mq/subscribe` HCT-->>Tool: Return `errorCode: "0"` Note over User, Srv: **Phase 6: Alarm Push and Message Processing** HCT->>Srv: POST `` (alarm data) Srv-->>HCT: Return `200 OK` Srv->>OpenClaw: POST `/hooks/agent` OpenClaw->>Notify: Trigger Agent processing Notify->>User: Send notification via the configured channel ``` ### 2.2 Stage-by-Stage Operation Guide --- ## ⚠️ 2.2.0 Phase 0: OpenClaw Hooks Readiness Check (ALWAYS RUN FIRST) > **Important**: Before doing ANYTHING else, you MUST verify that OpenClaw hooks is properly configured. This is a hard prerequisite. If hooks is not set up, the alarm push chain will break silently. ### Step 0-1: Run Pre-Check Script ```bash cd /modules/Hik-Connect_Team_Alarm/scripts python pre_check.py ``` The script checks all 6 items automatically. ### Step 0-2: If Hooks Needs Configuration — Ask User FIRST If `hooks.enabled` is not `true` or `hooks.token` is missing: **STOP and ask the user explicitly:** > "OpenClaw hooks is not configured on this server. To receive alarm pushes, I need to: > 1. Add a `hooks` section to `~/.openclaw/openclaw.json` > 2. Restart the OpenClaw Gateway > > This will cause a brief interruption to the OpenClaw service (typically a few seconds). > > Do you want me to proceed? (yes/no)" Only proceed if the user confirms. If confirmed: - Generate a new token: `openssl rand -hex 24` - Add to `~/.openclaw/openclaw.json`: ```json { "hooks": { "enabled": true, "token": "" } } ``` - Restart gateway: `openclaw gateway restart` - Verify: `curl http://127.0.0.1:/hooks/agent` returns 200 or 400 (not 404) > ⚠️ Do NOT add `defaultSessionKey` to hooks config — it causes `Malformed agent session key` errors. Only proceed to Phase 1 after pre-check passes or after hooks is confirmed ready. --- ## ⚠️ 2.2.1 Phase 1: Public URL Plan — MUST Confirm Before Taking Action ### Step 1-1: Query Current Status Show the user their existing webhook and subscription state: ```bash cd /modules/Hik-Connect_Team_Alarm/scripts python webhook_manager.py query python event_manager.py query ``` ### Step 1-2: Ask About Public URL Plan (CONFIRMATION GATE — ABSOLUTE BLOCKER) **Ask the user the following question and WAIT for their answer before proceeding:** > "To receive alarm pushes from HCT, I need a public HTTPS URL that HCT can call. How do you want to handle this?" > > Please choose one of the following options: > > **Option A — Use your own server (most stable)** > You have a public IP (124.222.61.228). If you have a domain name pointing to this IP, I can help you set up a reverse proxy (nginx/Caddy) to route HTTPS traffic to the webhook service. > > **Option B — Use a tunnel tool on this server** > I can set up ngrok, cloudflared, or similar on this server to create a public HTTPS URL. This is free but the tunnel may occasionally disconnect. > > **Option C — You have your own public HTTPS URL** > Provide your own URL and I'll configure the webhook service to use it. > > Which option do you prefer? (A / B / C, or describe your situation)" **🚨 ABSOLUTE RULE — No Tunnel Tool Without Explicit User Request:** > **UNDER NO CIRCUMSTANCES may the Agent install, configure, or start any tunnel tool (ngrok, cloudflared, cpolar, serveo, localtunnel, bore, etc.) unless the user has explicitly and affirmatively chosen Option B or otherwise explicitly asked for a tunnel tool in their own words.** > > This rule is absolute and non-negotiable. Violations include: > - Installing a tunnel tool before the user has chosen Option B > - Starting a tunnel without the user's explicit consent > - Creating a public URL without the user confirming tunnel as their chosen approach > - Using a tunnel as a "temporary" or "quick test" solution without explicit approval > > If the user does not respond to the question, re-ask. If the user is unclear, ask follow-up questions. Do not proceed. **Decision Rules Based on User Response:** | User Response | Agent Action | |:---|:---| | Option A (has domain) | Ask for domain → help configure reverse proxy → proceed | | **Option B (tunnel)** | **Only then** install/configure tunnel tool → proceed | | Option C (own URL) | Ask for the URL → verify it points to this server's 3090 → proceed | | Says nothing / unclear | Ask follow-up question — do NOT proceed until clarified | | Has no domain, no tunnel preference | Recommend Option A if public IP exists, otherwise explain limitation | > ⚠️ **Critical**: Do NOT generate any public URL, do NOT install any tunnel tool, do NOT start any tunnel process until the user has explicitly chosen Option B (or equivalent explicit tunnel request). If the user does not respond, ask again. --- ## ⚠️ 2.2.2 Phase 2: Collect Signing Secret — Must Have Before Service Start **Ask the user:** > "Provide an 8-32 character signing secret for webhook verification. This will be used to verify that alarm pushes are genuinely from Hik-Connect. Please provide a secret now (e.g. yourname2026):" **Rules:** - **Do NOT generate or invent a default secret** — the user MUST provide this. - **BLOCK on this step** — do not proceed to Phase 3 until the user provides a secret. - Record the secret. It will be used in: - `webhook_manager.py save --secret ` - `HIK_SIGN_SECRET` environment variable for server.js --- ## ⚠️ 2.2.3 Phase 3: Start Webhook Service — Only After Phases 1 & 2 Are Complete ### Step 3-1: Install Dependencies ```bash cd /modules/Hik-Connect_Team_Alarm/scripts npm install ``` ### Step 3-2: Get OpenClaw Gateway Port ```bash PORT=$(cat ~/.openclaw/openclaw.json | grep -oP '"port":\s*\K\d+') echo "OpenClaw Gateway port: $PORT" ``` ### Step 3-3: Start Webhook Receiving Service **Ask the user for their Feishu open_id (or target user ID) if not already known.** ```bash cd /modules/Hik-Connect_Team_Alarm/scripts HIK_SIGN_SECRET="" \ OPENCLAW_HOOKS_TOKEN="" \ OPENCLAW_HOOKS_URL="http://127.0.0.1:/hooks/agent" \ OPENCLAW_CHANNEL="feishu" \ OPENCLAW_TO="" \ PORT="3090" \ node server.js ``` ### Step 3-4: Verify Service Is Running ```bash curl http://localhost:3090/health ``` Expected: `{"status":"ok",...}` ### Step 3-5: Verify Public URL Is Reachable ```bash curl -sL -o /dev/null -w "%{http_code}" https:///hikvision/webhook ``` Expected: `200` or `302` (redirect). If `000` or timeout → tunnel/proxy is not working. Only proceed to Phase 4 if both service health and public URL are confirmed working. --- ## ⚠️ 2.2.4 Phase 4: Register Webhook with HCT Platform > "Now I'll register the Webhook URL with HCT. Make sure the service is running and the public URL is accessible from the internet." ```bash cd /modules/Hik-Connect_Team_Alarm/scripts python webhook_manager.py save \ --url "https:///hikvision/webhook" \ --secret "" ``` - **Success**: Tell user "Webhook registered successfully! HCT will now push alarms to your URL." - **Failure**: Tell user the error reason and checklist (service running? URL accessible from internet? secret correct?) --- ## ⚠️ 2.2.5 Phase 5: Event Subscription — Must Have Explicit User Confirmation > **⚠️ MANDATORY: Only subscribe when user explicitly asks.** > Never call `event_manager.py subscribe` without explicit user confirmation. **Ask the user:** > "Webhook registration successful! Now let's subscribe to alarm events. Which events do you want to subscribe to? You can find the full list in `EVENT_CODES.md`. Options: > - 'full' — subscribe to all events > - Or provide specific event codes (e.g. 'Msg110001,Msg110002') > > Which would you like?" **Wait for the user's answer.** Only then run: ```bash # All events: python event_manager.py subscribe # Specific events: python event_manager.py subscribe --types "Msg110001,Msg110002,..." ``` **After execution:** - **Success**: Tell the user "Event subscription complete! `{count}` event types subscribed." - **Failure**: Tell the user "Event subscription failed: `{reason}`" --- ## 6. Signature Verification Mechanism (Security) HCT platform and Webhook service ensure communication security through HMAC-SHA256 algorithm. ### 3.1 Verification Request (GET) When you save Webhook configuration on platform, platform will send verification request: * **HCT Platform Sends Header**: * `X-Hook-Batch-Id`: Batch ID * `X-Hook-Timestamp`: Timestamp (milliseconds) * **Webhook Service Processing**: * Service calculates HMAC-SHA256 signature based on configured `HIK_SIGN_SECRET`, `X-Hook-Timestamp` and `X-Hook-Batch-Id`. * **Signature Algorithm**: `signature = HMAC-SHA256(secret, timestamp.batchId)`, result is `sha256=`. * **Webhook Service Returns**: Carries `X-Hook-Signature: sha256=` in Response Header, status code `200 OK`. ### 3.2 Push Request (POST) When alarm occurs, platform pushes data: * **HCT Platform Sends Header**: * `X-Hook-Signature`: Signature calculated by platform * `X-Hook-Timestamp`: Push timestamp * **Webhook Service Processing**: * Service uses same `HIK_SIGN_SECRET` and `timestamp.batchId` (obtained from request Body) to calculate signature, and compares with `X-Hook-Signature` in Header. * If signature matches and timestamp is within acceptable range, request is considered legitimate and processed further, otherwise request is rejected. --- ## 7. Script and API Parameter Details ### 4.1 Webhook Management (`webhook_manager.py`) This script is used to manage HCT platform's Webhook configuration, including query, save and delete. #### 4.1.1 Running Examples * **Query Current Webhook Configuration**: ```bash python scripts/webhook_manager.py query ``` * **Save/Subscribe Webhook Configuration**: ```bash python scripts/webhook_manager.py save --url "https://your-public-domain.com/hikvision/webhook" --secret "YourSignSecret123" --retries 5 --delay 2000 ``` * **Delete Webhook Configuration**: ```bash python scripts/webhook_manager.py delete ``` #### 4.1.2 Request Parameters | Parameter | Type | Required | Default | Description | |:------------|:--------|:----------------------------|:--------|:--------------------------------------------------------------------------------------------------------| | `command` | String | Yes | - | Operation command, options: `query`, `save`, `delete` | | `--url` | String | Required for `save` command | - | Public HTTPS callback address, must start with `https://`, max length 256 characters. | | `--secret` | String | Optional for `save` command | - | Signing secret, used to verify legitimacy of pushed messages. 8-32 alphanumeric combination. | | `--retries` | Integer | Optional for `save` command | 3 | Number of retries after message push failure. Range `[-1, 5]`, -1 means unlimited retry within 2 hours. | | `--delay` | Integer | Optional for `save` command | 1000 | Retry interval after message push failure, in milliseconds. | #### 4.1.3 Output Field Description | Field | Type | Description | |:-------------------|:--------|:-------------------------------------------------------------------------------| | `success` | Boolean | Whether operation was successful. `true` means success, `false` means failure. | | `message` | String | Operation result or error description information. | | `errorCode` | String | Error code, `0` means success, other values are specific error codes. | | `data` | Object | Webhook configuration details returned on successful `query` command. | | `data.callbackUrl` | String | Webhook callback address. | | `data.retryTimes` | Integer | Webhook retry count. | | `data.retryDelay` | Long | Webhook retry interval (milliseconds). | ### 4.2 Event Subscription Management (`event_manager.py`) This script is used to subscribe, unsubscribe, or query HCT platform alarm event subscription status. #### 4.2.1 Running Examples * **Subscribe to Specific Event Types**: ```bash python scripts/event_manager.py subscribe --types "Msg330001,Msg330002" ``` * **Subscribe to All Event Types**: ```bash python scripts/event_manager.py subscribe ``` * **Unsubscribe from Specific Event Types**: ```bash python scripts/event_manager.py unsubscribe --types "Msg330001" ``` * **Unsubscribe from All Event Types**: ```bash python scripts/event_manager.py unsubscribe ``` * **Query Current Subscription Status**: ```bash python scripts/event_manager.py query ``` #### 4.2.2 Request Parameters | Parameter | Type | Required | Default | Description | |:----------|:-------|:---------|:-----------------------------------------|:--------------------------------------------------------------------------------------------------------------------------------| | `command` | String | Yes | - | Operation command, options: `subscribe`, `unsubscribe`, `query` | | `--types` | String | Optional | Empty (subscribe/unsubscribe all events) | Comma-separated event type list (e.g. `Msg330001,Msg330002`). For specific event types please refer to EVENT_CODES.md document. | > **Important Note**: Even without Webhook configuration, you can still execute subscribe/unsubscribe/query operations. But note that without proper Webhook service configuration and registration to HCT platform, you will not receive any alarm message pushes. #### 4.2.3 Output Field Description **For `subscribe` and `unsubscribe` commands:** | Field | Type | Description | |:------------|:--------|:-------------------------------------------------------------------------------| | `success` | Boolean | Whether operation was successful. `true` means success, `false` means failure. | | `message` | String | Operation result or error description information. | | `errorCode` | String | Error code, `0` means success, other values are specific error codes. | **For `query` command:** | Field | Type | Description | |:-------------------------|:--------|:---------------------------------------------------------------------------| | `success` | Boolean | Whether operation was successful. | | `data.isSubscribe` | Boolean | Whether subscribed. `true` means subscribed, `false` means not subscribed. | | `data.subscribeType` | Integer | Subscription type. `0`: Partial subscription, `1`: Full subscription. | | `data.subscribeTypeText` | String | Text description of subscription type. | | `data.events` | Array | List of event type codes user has subscribed to. | --- ## 8. Message Display Format and Agent Processing ### 5.1 Message Structure Webhook service will format received raw JSON into readable text, and forward through OpenClaw Hooks. Below is formatted message example: ```text 📦 HCT Open Webhook Push (batchId: 1a2b3c4d...) Alarms: 1 --- 🚨 Alarm: Human Detection Device: Small Cup Type: alarmCategoryVideo/alarmSubCategoryPersonDetect Time: 2026-04-16 14:20:00 Level: High Serial Number: D72821502 ``` --- ## 5.2 Agent Message Processing Rules > **⚠️ Critical: Agent Must Forward Messages Exactly As-Received** > When the Agent receives a webhook message via OpenClaw Hooks, it must **forward the message content exactly as-is** to the notification channel. **Required Behavior:** - ✅ **DO**: Forward the received message content verbatim to the user - ✅ **DO**: Send the message to the configured notification channel (Feishu/Telegram/etc.) - ❌ **DO NOT**: Summarize, interpret, or rephrase the message - ❌ **DO NOT**: Translate the message (e.g., from Chinese to English or vice versa) - ❌ **DO NOT**: Add comments, analysis, or extra explanations - ❌ **DO NOT**: Reformat or restructure the message content **Why This Matters:** The `server.js` service already formats the webhook payload into a human-readable format. The Agent's only job is to deliver this formatted message to the user without any further processing. Adding summaries or translations introduces noise and delays, and may strip important technical details that the user needs. **If the Agent fails to follow these rules**, it means the instruction was not clear enough — please report this so the skill documentation can be improved. --- ## 9. Troubleshooting | Symptom | Most Likely Cause | Fix | |---------------------------------------------------------|-------------------------------------------------------|-----------------------------------------------------------------------------------------------| | Webhook registration fails | server.js not running or URL not accessible | `curl http://localhost:3090/health`; verify public URL from outside | | Registration succeeds but no alarms | Third-party service (Pipedream/Lambda) used | ❌ Cannot forward to internal OpenClaw. Use tunnel or same-server setup | | Hooks returns 404 | basePath incorrectly included in `OPENCLAW_HOOKS_URL` | Use `http://127.0.0.1:/hooks/agent` — no basePath | | Hooks returns `[RELAY NETWORK ERROR]` | Wrong port, wrong token, or gateway down | Verify `OPENCLAW_HOOKS_URL` port matches `gateway.port`; token matches `hooks.token` | | User gets no notification | server.js stopped, or Feishu card permission missing | Check `curl localhost:3090/health`; enable "card messages" permission in Feishu Open Platform | | ngrok shows ERR_NGROK_4018 | Missing authtoken | Run `ngrok config add-authtoken ` | | gateway "hooks.token must not match gateway auth.token" | tokens are identical | `openssl rand -hex 24` → update hooks.token in openclaw.json → restart gateway | **Quick verification:** ```bash curl http://localhost:3090/health curl -s -o /dev/null -w "%{http_code}" https:///hikvision/webhook curl -X POST http://127.0.0.1:/hooks/agent -H "Authorization: Bearer " -d '{"test":"ping"}' ``` --- ## 10. File Structure ``` Hik-Connect_Team_Alarm/ ├── SKILL.md # This document ├── EVENT_CODES.md # Event type reference └── scripts/ ├── pre_check.py # Phase 0: OpenClaw hooks pre-check (run FIRST) ├── server.js # Webhook receiving service ├── webhook_manager.py # Webhook configuration management ├── event_manager.py # Event subscription management └── package.json # Node.js dependencies ``` --- **Error Codes**: | Return Code | Return Message | Description | |-------------|------------------------------|-------------------------------------------------------------------------------------------------| | CCF000001 | Webhook configuration failed | Webhook configuration failed. Please check the correctness of the public URL and the signature. | --- FILE:modules/Hik-Connect_Team_Alarm/scripts/event_manager.py #!/usr/bin/env python # -*- coding: utf-8 -*- """ HCTOpen Event Manager """ import sys import os import json import argparse from datetime import datetime # Robust lib directory import logic def setup_lib_path(): current_dir = os.path.dirname(os.path.abspath(__file__)) root_dir = current_dir for _ in range(3): root_dir = os.path.dirname(root_dir) potential_lib = os.path.join(root_dir, "lib") if os.path.exists(potential_lib): if potential_lib not in sys.path: sys.path.insert(0, potential_lib) return True return False if not setup_lib_path(): print("[ERROR] Cannot find lib directory") sys.exit(1) from token_manager import HCTOpenClient class EventManager(HCTOpenClient): """Event subscription management client""" def subscribe(self, msg_types: list = None): """Subscribe to events""" print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] Subscribing to events: {msg_types if msg_types else 'All events'}") endpoint = "/api/hccgw/rawmsg/v1/mq/subscribe" payload = { "subscribeType": 1, "msgType": msg_types if msg_types else [] } result = self.request("POST", endpoint, json_data=payload, token_header_key="Token") if result.get("errorCode") == "0": print("[SUCCESS] Event subscription successful") self.exit_with_json({"success": True, "message": "Event subscription successful"}) else: print(f"[ERROR] Subscription failed: {result.get('message', 'Unknown error')}") self.exit_with_json({"success": False, "message": result.get("message", "Unknown error"), "errorCode": result.get("errorCode")}) def unsubscribe(self, msg_types: list = None): """Unsubscribe from events""" print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] Unsubscribing from events: {msg_types if msg_types else 'All events'}") endpoint = "/api/hccgw/rawmsg/v1/mq/subscribe" payload = { "subscribeType": 0, "msgType": msg_types if msg_types else [] } result = self.request("POST", endpoint, json_data=payload, token_header_key="Token") if result.get("errorCode") == "0": print("[SUCCESS] Event unsubscription successful") self.exit_with_json({"success": True, "message": "Event unsubscription successful"}) else: print(f"[ERROR] Unsubscription failed: {result.get('message', 'Unknown error')}") self.exit_with_json({"success": False, "message": result.get("message", "Unknown error"), "errorCode": result.get("errorCode")}) def query(self): """Query current subscription status""" print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] Querying current subscription status...") endpoint = "/api/hccgw/rawmsg/v1/mq/info/subscribe" result = self.request("GET", endpoint, token_header_key="Token") if result.get("errorCode") == "0": data = result.get("data", {}) is_sub = data.get("isSubscribe", False) sub_type = "Full subscription" if data.get("subscribeType") == 1 else "Partial subscription" events = data.get("events", []) print(f"[SUCCESS] Query successful: {'Subscribed' if is_sub else 'Not subscribed'} ({sub_type})") if events: print(f"Subscribed event list: {', '.join(events)}") self.exit_with_json({ "success": True, "data": { "isSubscribe": is_sub, "subscribeType": data.get("subscribeType"), "subscribeTypeText": sub_type, "events": events } }) else: print(f"[ERROR] Query failed: {result.get('message', 'Unknown error')}") self.exit_with_json({"success": False, "message": result.get("message", "Unknown error"), "errorCode": result.get("errorCode")}) def main(): parser = argparse.ArgumentParser(description="HCTOpen Event Subscription Management") subparsers = parser.add_subparsers(dest="command", help="Operation command") # Subscribe command sub_parser = subparsers.add_parser("subscribe", help="Subscribe to events") sub_parser.add_argument("--types", help="Comma-separated event type list (e.g. Msg330001,Msg330002), leave empty to subscribe to all") # Unsubscribe command unsub_parser = subparsers.add_parser("unsubscribe", help="Unsubscribe from events") unsub_parser.add_argument("--types", help="Comma-separated event type list, leave empty to unsubscribe from all") # Query command subparsers.add_parser("query", help="Query current subscription status") args = parser.parse_args() client = EventManager() if args.command == "subscribe": msg_types = [t.strip() for t in args.types.split(',') if t.strip()] if args.types else [] client.subscribe(msg_types) elif args.command == "unsubscribe": msg_types = [t.strip() for t in args.types.split(',') if t.strip()] if args.types else [] client.unsubscribe(msg_types) elif args.command == "query": client.query() else: parser.print_help() if __name__ == "__main__": main() FILE:modules/Hik-Connect_Team_Alarm/scripts/package.json { "name": "hikvision-webhook", "version": "1.0.0", "description": "HikCentral Connect OpenAPI Webhook receiver with dedup & OpenClaw relay", "type": "module", "scripts": { "start": "node server.js", "dev": "node --watch server.js" }, "dependencies": { "express": "^4.21.0" } } FILE:modules/Hik-Connect_Team_Alarm/scripts/pre_check.py #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ HCT Alarm - OpenClaw Hooks Pre-Check Checks whether OpenClaw hooks are properly configured and reachable. Run this BEFORE any other alarm configuration steps. """ import sys import os import json import urllib.request import urllib.error import argparse from datetime import datetime OPENCLAW_CFG = os.path.expanduser("~/.openclaw/openclaw.json") CHECKS = [] def log(status, msg): symbol = {"OK": "✓", "FAIL": "✗", "SKIP": "⊘", "INFO": "ℹ"}.get(status, "?") print(f" [{status}] {msg}") CHECKS.append({"status": status, "msg": msg}) def check_config_file(): if not os.path.exists(OPENCLAW_CFG): log("FAIL", f"Config file not found: {OPENCLAW_CFG}") return False try: with open(OPENCLAW_CFG, "r") as f: json.load(f) log("OK", "Config file is valid JSON") return True except json.JSONDecodeError as e: log("FAIL", f"Config file is not valid JSON: {e}") return False def check_hooks_enabled(): with open(OPENCLAW_CFG, "r") as f: config = json.load(f) hooks = config.get("hooks", {}) if hooks.get("enabled") is True: log("OK", "hooks.enabled = true") return True log("FAIL", "hooks.enabled is not true (or hooks section missing)") return False def check_hooks_token(): with open(OPENCLAW_CFG, "r") as f: config = json.load(f) hooks = config.get("hooks", {}) token = hooks.get("token", "") if token and isinstance(token, str) and len(token) > 0: log("OK", f"hooks.token is set ({len(token)} chars)") return True, token log("FAIL", "hooks.token is missing or empty") return False, None def check_token_not_same_as_gateway(): with open(OPENCLAW_CFG, "r") as f: config = json.load(f) hooks = config.get("hooks", {}) gateway = config.get("gateway", {}) hooks_token = hooks.get("token", "") gateway_token = gateway.get("auth", {}).get("token", "") if hooks_token and gateway_token and hooks_token == gateway_token: log("FAIL", "hooks.token must be different from gateway.auth.token") return False log("OK", "hooks.token differs from gateway.auth.token") return True def check_gateway_port(): with open(OPENCLAW_CFG, "r") as f: config = json.load(f) gateway = config.get("gateway", {}) port = gateway.get("port", "") if port: log("OK", f"gateway.port = {port}") return True, port log("FAIL", "gateway.port is not set") return False, None def check_hooks_reachable(hooks_token, port): url = f"http://127.0.0.1:{port}/hooks/agent" body = json.dumps({"source": "pre_check"}).encode("utf-8") req = urllib.request.Request( url, data=body, headers={ "Authorization": f"Bearer {hooks_token}", "Content-Type": "application/json", }, method="POST", ) try: with urllib.request.urlopen(req, timeout=5) as resp: code = resp.status data = resp.read().decode("utf-8") log("OK", f"Hooks endpoint reachable (HTTP {code})") return True except urllib.error.HTTPError as e: body = e.read().decode("utf-8", errors="replace") if e.code == 400 and "message required" in body.lower(): log("OK", f"Hooks endpoint reachable (HTTP 400 — endpoint alive, needs message body)") return True log("FAIL", f"HTTP {e.code}: {body[:100]}") return False except urllib.error.URLError as e: log("FAIL", f"Cannot reach OpenClaw gateway: {e.reason}") return False except Exception as e: log("FAIL", f"Unexpected error: {e}") return False def main(): parser = argparse.ArgumentParser(description="OpenClaw Hooks Pre-Check for HCT Alarm") parser.add_argument("--json", action="store_true", help="Output results as JSON") args = parser.parse_args() print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] OpenClaw Hooks Pre-Check") print("=" * 50) all_passed = True if not check_config_file(): all_passed = False if not check_hooks_enabled(): all_passed = False token_ok, hooks_token = check_hooks_token() if not token_ok: all_passed = False if not check_token_not_same_as_gateway(): all_passed = False port_ok, port = check_gateway_port() if not port_ok: all_passed = False if token_ok and port_ok: if not check_hooks_reachable(hooks_token, port): all_passed = False else: log("SKIP", "Skipping reachability check (config not ready)") print("=" * 50) if all_passed: print("[RESULT] ✓ All checks passed. OpenClaw hooks are ready.") else: print("[RESULT] ✗ Some checks failed. Fix the issues above before proceeding.") if args.json: print(json.dumps({"ok": all_passed, "checks": CHECKS}, indent=2, ensure_ascii=False)) return 0 if all_passed else 1 if __name__ == "__main__": sys.exit(main()) FILE:modules/Hik-Connect_Team_Alarm/scripts/server.js /** * HikCentral Connect OpenAPI V2.15 — Webhook Receiving Service * * Features: * 1. Receive HCT Open Webhook pushes (Alarms + Events) * 2. HMAC-SHA256 signature verification (X-Hook-Signature) * 3. Configurable window deduplication (same device, same type) * 4. Forward to OpenClaw hooks endpoint → Notification * 5. Extract capture URLs, Agent sends images directly * 6. Auto-detect OpenClaw Gateway port * 7. Startup connection check */ import crypto from 'crypto'; import express from 'express'; import { readFileSync, existsSync } from 'fs'; import { homedir } from 'os'; // ============ Default Values ============ // DEFAULT_WEBHOOK_PORT const DEFAULT_WEBHOOK_PORT = 3090; // ============ Helper: Detect OpenClaw Gateway Port ============ function detectOpenClawPort() { const configPath = `${homedir()}/.openclaw/openclaw.json`; try { if (existsSync(configPath)) { const content = readFileSync(configPath, 'utf-8'); // Remove comments if any (simple JSON doesn't have comments, but just in case) const json = JSON.parse(content); const port = json?.gateway?.port; if (port && typeof port === 'number') { return port; } } } catch (err) { console.error(`[FATAL] Failed to read gateway port from ${configPath}: ${err.message}`); } throw new Error(`OpenClaw gateway port not found in ${configPath}. Please ensure gateway is configured and hooks are enabled.`); } // ============ Helper: Check OpenClaw Connection ============ async function checkOpenClawConnection(url, token) { console.log(`[CHECK] Testing OpenClaw connection at ${url}...`); try { const res = await fetch(url, { method: 'POST', headers: { 'Content-Type': 'application/json', ...(token ? { 'Authorization': `Bearer ${token}` } : {}), }, body: JSON.stringify({ test: 'ping' }), signal: AbortSignal.timeout(5000), }); // 400 means endpoint exists but missing required field (e.g. "message required"), indicating hooks middleware is registered if (res.ok || res.status === 400) { console.log(`[CHECK] ✓ OpenClaw hooks is reachable (status ${res.status})`); return true; } else { console.error(`[CHECK] ✗ OpenClaw returned status ${res.status}`); return false; } } catch (err) { console.error(`[CHECK] ✗ Cannot reach OpenClaw at ${url}`); console.error(`[CHECK] Error: ${err.message}`); console.error(`[CHECK] Please verify:`); console.error(`[CHECK] 1. OpenClaw Gateway is running: openclaw gateway status`); console.error(`[CHECK] 2. Port is correct (detected: ${detectOpenClawPort()})`); console.error(`[CHECK] 3. Or set OPENCLAW_HOOKS_URL environment variable`); return false; } } // ============ Configuration ============ const detectedPort = detectOpenClawPort(); const defaultOpenClawUrl = `http://127.0.0.1:${detectedPort}/hooks/agent`; // Sign secret: MUST be provided via environment variable const signSecretFromEnv = process.env.HIK_SIGN_SECRET; if (!signSecretFromEnv) { console.error('[FATAL] HIK_SIGN_SECRET environment variable is not set.'); console.error('[FATAL] Please set HIK_SIGN_SECRET before starting the webhook service.'); console.error('[FATAL] Example: HIK_SIGN_SECRET="your-custom-secret" node server.js'); process.exit(1); } const signSecret = signSecretFromEnv; const CONFIG = { port: parseInt(process.env.PORT || String(DEFAULT_WEBHOOK_PORT), 10), // HCT Open Webhook Secret (signSecret specified when registering webhook) signSecret: signSecret, // OpenClaw hooks configuration openclaw: { url: process.env.OPENCLAW_HOOKS_URL || defaultOpenClawUrl, token: process.env.OPENCLAW_HOOKS_TOKEN || '', // Supports all OpenClaw channels: feishu, telegram, discord, slack, whatsapp, signal, etc. channel: process.env.OPENCLAW_CHANNEL || '', to: process.env.OPENCLAW_TO || '', }, // Deduplication window (milliseconds), default 1 minute dedupWindowMs: parseInt(process.env.DEDUP_WINDOW_MS || '60000', 10), // Request timeout (HCT Open requires response within 5 seconds) responseTimeoutMs: 4000, }; // ============ Print Configuration ============ function printConfig() { console.log(''); console.log('═══════════════════════════════════════════════════════════'); console.log(' CONFIGURATION SUMMARY '); console.log('═══════════════════════════════════════════════════════════'); console.log(` Webhook Port: ${CONFIG.port}`); console.log(` Sign Secret: ***configured***`); console.log(` OpenClaw URL: ${CONFIG.openclaw.url}`); console.log(` OpenClaw Token: ${CONFIG.openclaw.token ? '***configured***' : 'NOT SET ⚠️'}`); console.log(` Notify Channel: ${CONFIG.openclaw.channel}`); console.log(` Notify Target: ${CONFIG.openclaw.to || 'NOT SET ⚠️'}`); console.log(` Dedup Window: ${CONFIG.dedupWindowMs / 1000}s`); console.log('═══════════════════════════════════════════════════════════'); if (!CONFIG.openclaw.to || !CONFIG.openclaw.channel) { console.log(''); console.log('╔══════════════════════════════════════════════════════════╗'); console.log('║ [FATAL] Missing required configuration ║'); console.log('╠══════════════════════════════════════════════════════════╣'); if (!CONFIG.openclaw.channel) { console.log('║ OPENCLAW_CHANNEL is not set ║'); console.log('║ Please set: export OPENCLAW_CHANNEL="feishu" ║'); } if (!CONFIG.openclaw.to) { console.log('║ OPENCLAW_TO is not set ║'); console.log('║ Please set: export OPENCLAW_TO="user_open_id" ║'); } console.log('╚══════════════════════════════════════════════════════════╝'); console.log(''); console.log('Without these, notifications cannot be delivered. Exiting.'); process.exit(1); } console.log(''); } // ============ Deduplication Cache ============ const dedupCache = new Map(); function dedupKey(item) { if (item.type === 'alarm') { const src = item.eventSource; return `alarm:${src?.sourceID || ''}:${item.alarmMainCategory || ''}:${item.alarmSubCategory || ''}`; } if (item.type === 'event') { return `event:${item.basicInfo?.eventType || ''}:${item.basicInfo?.device?.id || ''}`; } return `${item.type}:${item.guid || JSON.stringify(item).slice(0, 200)}`; } function isDuplicate(key) { const now = Date.now(); const cached = dedupCache.get(key); if (cached && now - cached < CONFIG.dedupWindowMs) return true; dedupCache.set(key, now); // Periodically clean up expired cache if (dedupCache.size > 1000) { for (const [k, v] of dedupCache) { if (now - v > CONFIG.dedupWindowMs) dedupCache.delete(k); } } return false; } // ============ Signature Verification ============ function verifySignature(headers, batchId) { if (!CONFIG.signSecret) { console.warn('[WARN] HIK_SIGN_SECRET not set, skipping signature verification'); return true; } const signature = headers['x-hook-signature'] || headers['X-Hook-Signature']; const timestamp = headers['x-hook-timestamp']; if (!signature || !timestamp) { console.warn('[WARN] Missing signature headers'); return false; } const tsDiff = Math.abs(Date.now() - parseInt(timestamp, 10)); if (tsDiff > 60 * 1000) { console.warn(`[WARN] Timestamp drift too large: ${tsDiff}ms`); return false; } const message = `${timestamp}.${batchId}`; const mac = crypto.createHmac('sha256', CONFIG.signSecret).update(message).digest('hex'); const expected = `sha256=${mac}`; if (signature !== expected) { console.warn(`[WARN] Signature mismatch: expected=${expected}, got=${signature}`); return false; } return true; } // ============ Format Alarm Messages ============ const LEVEL_MAP = { 1: 'High', 2: 'Medium', 3: 'Low' }; // Event code to description mapping const EVENT_CODE_MAP = { // Video Intercom 'Msg140001': 'Messages about video intercom events', // On-Board Monitoring 'Msg330001': 'GPS Data Report', 'Msg330101': 'Alarm Triggered by Panic Button', 'Msg330102': 'Alarm Input', 'Msg330201': 'Forward Collision Warning', 'Msg330202': 'Headway Monitoring Warning', 'Msg330203': 'Lane Deviation Warning', 'Msg330204': 'Pedestrian Collision Warning', 'Msg330205': 'Speed Limit Warning', 'Msg330301': 'Blind Spot Warning', 'Msg330401': 'Sharp Turn', 'Msg330402': 'Sudden Brake', 'Msg330403': 'Sudden Acceleration', 'Msg330404': 'Rollover', 'Msg330405': 'Speeding', 'Msg330406': 'Collision', 'Msg330407': 'ACC ON', 'Msg330408': 'ACC OFF', 'Msg330501': 'Smoking', 'Msg330502': 'Using Mobile Phone', 'Msg330503': 'Fatigue Driving', 'Msg330504': 'Distraction', 'Msg330505': 'Seatbelt Unbuckled', 'Msg330506': 'Video Tampering', 'Msg330507': 'Yawning', 'Msg330508': 'Wearing IR Interrupted Sunglasses', 'Msg330509': 'Absence', 'Msg330510': 'Front Passenger Detection', 'Msg335000': 'Person and Vehicle Match', 'Msg335001': 'Person and Vehicle Mismatch', // Authentication Event 'Msg110001': 'Access Granted by Card and Fingerprint', 'Msg110002': 'Access Granted by Card, Fingerprint, and PIN', 'Msg110003': 'Access Granted by Card', 'Msg110004': 'Access Granted by Card and PIN', 'Msg110005': 'Access Granted by Fingerprint', 'Msg110006': 'Access Granted by Fingerprint and PIN', 'Msg110007': 'Duress Alarm', 'Msg110008': 'Access Granted by Face and Fingerprint', 'Msg110009': 'Access Granted by Face and PIN', 'Msg110010': 'Access Granted by Face and Card', 'Msg110011': 'Access Granted by Face, PIN, and Fingerprint', 'Msg110012': 'Access Granted by Face, Card, and Fingerprint', 'Msg110013': 'Access Granted by Face', 'Msg110018': 'Access Granted via Combined Authentication Modes', 'Msg110019': 'Skin-Surface Temperature Measured', 'Msg110020': 'Password Authenticated', 'Msg110022': 'Access Granted by Bluetooth', 'Msg110023': 'Access Granted via QR Code', 'Msg110024': 'Access Granted via Keyfob', 'Msg110501': 'Verifying Card Encryption Failed', 'Msg110502': 'Max. Card Access Failed Attempts', 'Msg110505': 'Card No. Expired', 'Msg110506': 'Access Timed Out by Card and PIN', 'Msg110507': 'Access Denied - Door Remained Locked or Inactive', 'Msg110509': 'Access Denied by Card and PIN', 'Msg110510': 'Access Timed Out by Card, Fingerprint, and PIN', 'Msg110511': 'Access Denied by Card, Fingerprint, and PIN', 'Msg110512': 'Access Denied by Card and Fingerprint', 'Msg110513': 'Access Timed Out by Card and Fingerprint', 'Msg110514': 'No Access Level Assigned', 'Msg110515': 'Card No. Does Not Exist', 'Msg110516': 'Invalid Time Period', 'Msg110517': 'Fingerprint Does Not Exist', 'Msg110518': 'Access Denied by Fingerprint', 'Msg110519': 'Access Denied by Fingerprint and PIN', 'Msg110520': 'Access Timed Out by Fingerprint and PIN', 'Msg110521': 'Access Denied by Face and Fingerprint', 'Msg110522': 'Access Timed Out by Face and Fingerprint', 'Msg110523': 'Access Denied by Face and PIN', 'Msg110524': 'Access Timed Out by Face and PIN', 'Msg110525': 'Access Denied by Face and Card', 'Msg110526': 'Access Timed Out by Face and Card', 'Msg110527': 'Access Denied by Face, PIN, and Fingerprint', 'Msg110528': 'Access Timed Out by Face, PIN, and Fingerprint', 'Msg110529': 'Access Denied by Face, Card, and Fingerprint', 'Msg110530': 'Access Timed Out by Face, Card, and Fingerprint', 'Msg110531': 'Access Denied by Face', 'Msg110533': 'Live Facial Detection Failed', 'Msg110545': 'Combined Authentication Timed Out', 'Msg110546': 'Access Denied by Invalid M1 Card', 'Msg110547': 'Verifying CPU Card Encryption Failed', 'Msg110548': 'Access Denied - NFC Card Reading Disabled', 'Msg110549': 'EM Card Reading Not Enabled', 'Msg110550': 'M1 Card Reading Not Enabled', 'Msg110551': 'CPU Card Reading Disabled', 'Msg110552': 'Authentication Mode Mismatch', 'Msg110554': 'Max. Card and Password Authentication Times', 'Msg110555': 'Password Mismatches', 'Msg110556': 'Employee ID Does Not Exist', 'Msg110557': 'Access Denied: Scheduled Sleep Mode', 'Msg110559': 'Verifying Desfire Card Encryption Failed', 'Msg110560': 'Absence', 'Msg110561': 'Authentication Failed Due to Abnormal Features', 'Msg110564': 'Access Denied by Bluetooth', 'Msg110565': 'Access Denied by QR Code', 'Msg110566': 'Verifying QR Code Secret Key Failed', 'Msg110567': 'Access Denied via Keyfob' }; function formatAlarmItem(item) { const src = item.eventSource || {}; const dev = src.deviceInfo || {}; const time = item.timeInfo?.startTime || ''; const rule = item.alarmRule || {}; const priority = item.alarmPriority || {}; // Simplify time format const shortTime = time; return [ `🚨 Alarm: ${rule.name || item.alarmSubCategory || 'Unknown Alarm'}`, `Device: ${src.sourceName || dev.devName || 'Unknown Device'}`, `Type: ${item.alarmMainCategory}/${item.alarmSubCategory}`, `Time: ${shortTime}`, `Level: ${priority.levelName || LEVEL_MAP[priority.level] || 'Level ${priority.level}'}`, ].filter(Boolean).join('\n'); } function formatEventItem(item) { const basic = item.basicInfo || {}; const dev = basic.device || {}; // Get event code and map to description const eventCode = item.basicInfo?.msgType || ''; const eventDescription = EVENT_CODE_MAP[eventCode] || eventCode || 'Unknown'; return [ `📡 Event: ${eventDescription}`, `Device: ${dev.name || 'Unknown'}`, `Time: ${basic.occurrenceTime || ''}`, dev.deviceSerial ? `Serial: ${dev.deviceSerial}` : '', ].filter(Boolean).join('\n'); } function buildPayload(body) { const { batchId, list } = body; const messages = []; let alarmCount = 0; let eventCount = 0; for (const item of list || []) { const key = dedupKey(item); if (isDuplicate(key)) { console.log(`[DEDUP] Skipped duplicate: ${key}`); continue; } if (item.type === 'alarm') { alarmCount++; messages.push(formatAlarmItem(item)); } else if (item.type === 'event') { eventCount++; messages.push(formatEventItem(item)); } } if (messages.length === 0) return null; return [ `📦 HCT Open Webhook Push (batchId: ${batchId?.slice(0, 8)}...)`, alarmCount ? `Alarms: ${alarmCount}` : '', eventCount ? `Events: ${eventCount}` : '', `---`, ...messages, ].filter(Boolean).join('\n\n'); } // ============ Relay to OpenClaw ============ async function relayToOpenClaw(message) { if (!CONFIG.openclaw.url || !CONFIG.openclaw.token) { console.error('[RELAY] OPENCLAW_HOOKS_URL or OPENCLAW_HOOKS_TOKEN is not configured. Please set it before starting the webhook service.'); return; } try { const res = await fetch(CONFIG.openclaw.url, { method: 'POST', headers: { 'Content-Type': 'application/json', 'Authorization': `Bearer ${CONFIG.openclaw.token}`, }, body: JSON.stringify({ message: message, channel: CONFIG.openclaw.channel, to: CONFIG.openclaw.to, }), signal: AbortSignal.timeout(10000), }); const data = await res.json().catch(() => ({})); if (res.ok) { console.log('[RELAY] OpenClaw hooks OK, runId:', data.runId || data.id || 'unknown'); } else { console.error('[RELAY] OpenClaw hooks error:', res.status, JSON.stringify(data)); } } catch (err) { console.error('[RELAY] OpenClaw hooks network error:', err.message); } } // ============ Express App ============ const app = express(); app.use('/hikvision/webhook', express.json({ limit: '10mb' })); // Health Check app.get('/health', (req, res) => { res.json({ status: 'ok', uptime: process.uptime(), dedupCacheSize: dedupCache.size, config: { port: CONFIG.port, openclawUrl: CONFIG.openclaw.url, hasToken: !!CONFIG.openclaw.token, hasTarget: !!CONFIG.openclaw.to, } }); }); // GET Request — HCT Open URL Verification Callback app.get('/hikvision/webhook', (req, res) => { const batchId = req.headers['x-hook-batch-id']; const timestamp = req.headers['x-hook-timestamp']; console.log(`[VERIFY] URL verification request (batchId=${batchId})`); if (!CONFIG.signSecret) { console.error('[VERIFY] Cannot verify: HIK_SIGN_SECRET not configured'); return res.status(500).send('signSecret not configured'); } if (!batchId || !timestamp) { return res.status(400).send('Missing X-Hook-Batch-Id or X-Hook-Timestamp'); } const message = `${timestamp}.${batchId}`; const mac = crypto.createHmac('sha256', CONFIG.signSecret).update(message).digest('hex'); res.setHeader('x-hook-signature', `sha256=${mac}`); res.status(200).send('OK'); }); // POST Request — Receive Alarm/Event Push app.post('/hikvision/webhook', async (req, res) => { const startTime = Date.now(); const batchId = req.body?.batchId; const list = req.body?.list || []; console.log(`[IN] batchId=${batchId}, items=${list.length}`); // 1. Verify Signature if (!verifySignature(req.headers, batchId)) { console.warn('[REJECT] Invalid signature'); return res.status(401).json({ error: 'Invalid signature' }); } // 2. Return 200 immediately (HCT Open requires within 5s) res.json({ received: true, batchId, count: list.length }); console.log(`[ACK] Responded in ${Date.now() - startTime}ms`); // 3. Asynchronous processing try { const message = buildPayload(req.body); if (message) { await relayToOpenClaw(message); } else { console.log('[SKIP] All items were duplicates'); } } catch (err) { console.error(`[ERROR] Processing failed: ${err.message}`); } }); // ============ Startup ============ async function main() { // Print configuration summary (exits if required config missing) printConfig(); // Additional startup validation const missing = []; if (!CONFIG.openclaw.url) missing.push('OPENCLAW_HOOKS_URL'); if (!CONFIG.openclaw.token) missing.push('OPENCLAW_HOOKS_TOKEN'); if (!CONFIG.openclaw.channel) missing.push('OPENCLAW_CHANNEL'); if (!CONFIG.openclaw.to) missing.push('OPENCLAW_TO'); if (missing.length > 0) { console.error('[FATAL] Missing required environment variables:', missing.join(', ')); console.error('[FATAL] Cannot start webhook service. Please set them before running server.js'); process.exit(1); } // Check OpenClaw connection await checkOpenClawConnection(CONFIG.openclaw.url, CONFIG.openclaw.token); // Start server app.listen(CONFIG.port, () => { console.log(''); console.log('╔══════════════════════════════════════════╗'); console.log('║ HCT Open Webhook Receiver Started ║'); console.log('╠══════════════════════════════════════════╣'); console.log(`║ Port: ${String(CONFIG.port).padEnd(29)}║`); console.log('║ Endpoint: POST /hikvision/webhook ║'); console.log('║ Verify: GET /hikvision/webhook ║'); console.log('║ Health: GET /health ║'); console.log('╚══════════════════════════════════════════╝'); console.log(''); console.log('Waiting for HCT Open webhook pushes...'); console.log(''); }); } main().catch(err => { console.error('[FATAL] Startup failed:', err); process.exit(1); }); FILE:modules/Hik-Connect_Team_Alarm/scripts/webhook_manager.py #!/usr/bin/env python # -*- coding: utf-8 -*- """ HCTOpen Webhook Manager """ import sys import os import json import argparse from datetime import datetime # Robust lib directory import logic def setup_lib_path(): current_dir = os.path.dirname(os.path.abspath(__file__)) root_dir = current_dir for _ in range(3): root_dir = os.path.dirname(root_dir) potential_lib = os.path.join(root_dir, "lib") if os.path.exists(potential_lib): if potential_lib not in sys.path: sys.path.insert(0, potential_lib) return True return False if not setup_lib_path(): print("[ERROR] Cannot find lib directory") sys.exit(1) from token_manager import HCTOpenClient class WebhookManager(HCTOpenClient): """Webhook configuration management client""" def query(self): """Query Webhook configuration""" print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] Querying Webhook configuration...") endpoint = "/api/hccgw/webhook/v1/config/query" result = self.request("POST", endpoint, token_header_key="Token") if result.get("errorCode") == "0": data = result.get("data", {}) if data: headers = ["Configuration Item", "Content"] rows = [ ["Callback URL (callbackUrl)", data.get("callbackUrl", "-")], ["Retry Count (retryTimes)", data.get("retryTimes", "-")], ["Retry Interval (retryDelay)", f"{data.get('retryDelay', '-')} ms"] ] self.print_table("HCTOpen Webhook Current Configuration", headers, rows) self.exit_with_json({"success": True, "data": data}) else: print("[INFO] Webhook not currently configured") self.exit_with_json({"success": True, "data": None, "message": "No webhook configuration found"}) else: print(f"[ERROR] Query failed: {result.get('message', 'Unknown error')}") self.exit_with_json({"success": False, "message": result.get("message", "Unknown error"), "errorCode": result.get("errorCode")}) def save(self, callback_url: str, sign_secret: str = None, retry_times: int = 3, retry_delay: int = 1000): """Save/Subscribe Webhook configuration""" print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] Saving Webhook configuration: {callback_url}") if not callback_url.startswith("https://"): print("[ERROR] Callback URL must use HTTPS protocol") self.exit_with_json({"success": False, "message": "Callback URL must use HTTPS protocol"}) endpoint = "/api/hccgw/webhook/v1/config/save" payload = { "callbackUrl": callback_url, "retryTimes": retry_times, "retryDelay": retry_delay } if sign_secret: payload["signSecret"] = sign_secret result = self.request("POST", endpoint, json_data=payload, token_header_key="Token") if result.get("errorCode") == "0": print("[SUCCESS] Webhook configuration saved successfully") self.exit_with_json({"success": True, "message": "Webhook configuration saved successfully"}) else: print(f"[ERROR] Save failed: {result.get('message', 'Unknown error')}") self.exit_with_json({"success": False, "message": result.get("message", "Unknown error"), "errorCode": result.get("errorCode")}) def delete(self): """Delete Webhook configuration""" print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] Deleting Webhook configuration...") endpoint = "/api/hccgw/webhook/v1/config/delete" result = self.request("POST", endpoint, token_header_key="Token") if result.get("errorCode") == "0": print("[SUCCESS] Webhook configuration deleted successfully") self.exit_with_json({"success": True, "message": "Webhook configuration deleted successfully"}) else: print(f"[ERROR] Delete failed: {result.get('message', 'Unknown error')}") self.exit_with_json({"success": False, "message": result.get("message", "Unknown error"), "errorCode": result.get("errorCode")}) def main(): parser = argparse.ArgumentParser(description="HCTOpen Webhook Configuration Management") subparsers = parser.add_subparsers(dest="command", help="Operation command") # Query command subparsers.add_parser("query", help="Query current Webhook configuration") # Save command save_parser = subparsers.add_parser("save", help="Save/Subscribe Webhook configuration") save_parser.add_argument("--url", required=True, help="Callback URL (must be HTTPS)") save_parser.add_argument("--secret", help="Signing secret (optional, 8-32 alphanumeric combination)") save_parser.add_argument("--retries", type=int, default=3, help="Retry count (default: 3)") save_parser.add_argument("--delay", type=int, default=1000, help="Retry interval ms (default: 1000)") # Delete command subparsers.add_parser("delete", help="Delete Webhook configuration") args = parser.parse_args() client = WebhookManager() if args.command == "query": client.query() elif args.command == "save": client.save(args.url, args.secret, args.retries, args.delay) elif args.command == "delete": client.delete() else: parser.print_help() if __name__ == "__main__": main() FILE:modules/Hik-Connect_Team_ACS/SKILL.md --- name: hctopen-acs-control description: | HCTOpen door access remote control skill. Supports remote open door, close door, normally open, normally closed operations for Hikvision HCT Team mode (HCT) door access devices. Use when: Need to remotely control open/close status of one or more door access devices, supports specified device or full operations. Before calling this Skill's script, please check if user provided operation type. If user didn't provide operation type, please clearly inform user in reply: 'Currently using default parameters (operation type is open door), if you need to adjust, please let me know'. After getting confirmation or ignoring, continue execution. ⚠️ Security Requirement: Must set environment variables: Hik-Connect Team OpenAPI AppKey, Hik-Connect Team OpenAPI SecretKey. (API Domain is automatically obtained from token response) parameters: - name: action-type type: integer description: "Operation type: 1-open door, 2-close door, 3-normally open, 4-normally closed" required: true enum: [1, 2, 3, 4] - name: element-list type: string description: "Resource point list, comma-separated door access resource ID list" required: true responses: - success: true template: "Door access control operation executed, result as follows:" media: "list_card" metadata: openclaw: emoji: "🚪" requires: env: ["HIK_CONNECT_TEAM_OPENAPI_APP_KEY", "HIK_CONNECT_TEAM_OPENAPI_SECRET_KEY"] pip: ["requests"] primaryEnv: "HIK_CONNECT_TEAM_OPENAPI_APP_KEY" warnings: - "Need Hik-Connect Team OpenAPI AppKey /Hik-Connect Team OpenAPI SecretKey with door access control permission" - "Token automatically cached in system temp directory, permission 600" - "May read ~/.openclaw/*.json for credentials (env vars have priority)" config: tokenCache: default: true envVar: "HIK_CONNECT_TEAM_TOKEN_CACHE" description: "Enable Token cache (enabled by default). Set to 0 to disable." configFileRead: paths: - "~/.openclaw/config.json" - "~/.openclaw/gateway/config.json" - "~/.openclaw/channels.json" priority: "lower than environment variables" description: "Reads Hik-Connect Team credentials from OpenClaw config files as fallback" --- # HCTOpen ACS Control HCT is short for Hik-Connect for Teams, meaning Hik-Connect Team mode. HCTOpen is short for Hik-Connect for Teams OpenAPI. --- ## ⚠️ Security Warning (Read Before Use) **Before executing door access control, please ensure the following security checks are completed:** | # | Check Item | Status | Description | |---|----------------------------|-------------|----------------------------------------------------------------------------------------------------------------| | 1 | **Credential Permission** | ⚠️ Required | Please use credentials with **minimum control permission**, avoid using super admin credentials | | 2 | **Operation Confirmation** | ⚠️ Note | Remote door open operation has physical security risk, please ensure site safety is confirmed before operation | | 3 | **Token Cache** | ✅ Encrypted | Token cached in system temp directory, only current user can read (600 permission) | | 4 | **API Domain** | ✅ Auto | API domain is automatically obtained from token response (no longer requires manual configuration) | --- ## Quick Start ### Run Control Script Skill supports flexible command line parameters: ```bash # Scenario 1: Execute door open operation for specified door access (actionType=1) python scripts/acs_control.py --action-type 2 --element-list "2aabf37ad9804f66acc4ad4fb7bd4694" # Scenario 2: Execute door close operation for specified door access (actionType=2) python scripts/acs_control.py --action-type 2 --element-list "door_resource_id_1,door_resource_id_2" # Scenario 3: Execute normally open operation for specified door access (actionType=3) python scripts/acs_control.py --action-type 3 --element-list "door_resource_id_1" # Scenario 4: Execute normally closed operation for specified door access (actionType=4) python scripts/acs_control.py --action-type 4 --element-list "door_resource_id_1" ``` --- ## API Parameter Details ### 1. Remote Control Request Parameters **Endpoint**: `POST /api/hccgw/acs/v1/remote/control` | Parameter Name | Type | Description | Required | Default | Notes | |----------------|---------|---------------------|----------|---------|---------------------------------------------------------------| | `actionType` | Integer | Operation type | **Yes** | - | 1-open door, 2-close door, 3-normally open, 4-normally closed | | `elementlist` | Array | Resource point list | No | [] | Door logical resource ID list | | `direction` | Integer | Traffic direction | No | 0 | 0-entry, 1-exit. Mainly for gates with direction distinction. | ### 2. API Return Data Description API returns list of devices that failed execution. If `operationResult` is empty, it means all requested devices operated successfully. | Field Name | Type | Description | Notes | |---------------|--------|--------------------------|----------------------------------------| | `elementId` | String | Door logical resource ID | Identifies which door operation failed | | `elementName` | String | Door name | Human-readable device name | | `areaId` | String | Area ID | Device area identifier | | `areaName` | String | Area name | Device area name | | `errorCode` | String | Error code | Specific reason code for failure | --- ## Environment Variables | Variable Name | Required | Description | |---------------------------------------|----------|-----------------------------------------| | `HIK_CONNECT_TEAM_OPENAPI_APP_KEY` | Yes | Hik-Connect Team OpenAPI AppKey | | `HIK_CONNECT_TEAM_OPENAPI_SECRET_KEY` | Yes | Your Hik-Connect Team OpenAPI SecretKey | | `HIK_CONNECT_TEAM_TOKEN_CACHE` | No | 1=Enable cache (default), 0=Disable | --- ## API Endpoints | Function | Endpoint | |---------------------|-----------------------------------------| | Get Token | `POST /api/hccgw/platform/v1/token/get` | | Door Access Control | `POST /api/hccgw/acs/v1/remote/control` | **Domain**: Automatically obtained from token response (`areaDomain` field) --- ## Workflow ```mermaid graph TD A[Start Script] --> B{Check Environment Variables} B -- Missing --> C[Report Error and Exit] B -- Pass --> D[Get AccessToken] D --> E{Is Token Valid?} E -- Cache Valid --> F[Use Cache Directly] E -- Expired/No Cache --> G[Call API to Get New Token] G --> H[Save to Local Cache] F --> I[Send Remote Control Command] H --> I I --> J{Parse Return Result} J -- Failed Devices Exist --> K[Print Failed List Table] J -- All Successful --> L[Print Success Message] K --> M[Output Complete JSON Result] L --> M M --> N[End] ``` --- ## Output Examples ### Partial Operation Failed Example: ```text [2026-04-22 11:31:34] Executing door access control: Type=1, Count=1 [2026-04-22 11:31:34] Executing door access control: Type=1, Count=1 [WARNING] Some devices operation failed: ====================================================================== Failed Device List ====================================================================== No. Door Resource ID Door Name Area Error Code ------------------------------------------------------------------ 1 2aabf37ad9804f66acc4ad4fb7bd4694 VMS000003 ====================================================================== [JSON Output] { "success": false, "operationResult": [ { "elementId": "2aabf37ad9804f66acc4ad4fb7bd4694", "elementName": "", "areaId": "", "areaName": "", "errorCode": "VMS000003" } ], "error": "Some operations failed" } ====================================================================== Done ====================================================================== ``` ### All Operations Successful Example: ```text [2026-04-22 11:36:15] Executing door access control: Type=2, Count=1 [2026-04-22 11:36:15] Executing door access control: Type=2, Count=1 [SUCCESS] All door access operations executed successfully [JSON Output] { "success": true, "operationResult": [] } ====================================================================== Done ====================================================================== ``` --- ## File Structure ``` ├── scripts/ │ └── acs_control.py # Door access control core execution script └── SKILL.md # Skill usage documentation ``` --- ## FAQ - **Q: Why does it show "Credentials required"?** - A: Please ensure `export` command has been correctly executed to set `HIK_CONNECT_TEAM_OPENAPI_APP_KEY` and other environment variables. - **Q: How long is Token cache valid?** - A: Follows HCT API standard, usually 7 days. Script will auto-refresh 5 minutes before expiration. - **Q: How to operate all door access?** - A: Cannot operate all door access, can only operate specific door access. - **Q: Why did operation fail?** - A: Please check device status, permission configuration and network connection. Failed device information will be listed in detail in output. - **Q: How to get door access logical resource ID?** - A: Must first use door access device serial number to call `modules/Hik-Connect_Team_Resource/scripts/list_doors.py `, get door access resource ID from returned list. - **Q: How to get the correct door resource ID?** - A: Use `list_doors.py` API,Example: ```bash python scripts/list_doors.py L33721705 # Output: door resource ID is in "Door Access ID" column ``` --- ## Security Notes - Use Hik-Connect Team OpenAPI AppKey / Hik-Connect Team OpenAPI SecretKey with minimum permissions - Token cached in system temp directory, enabled by default - Automatic 4-second interval between device requests to avoid rate limiting - All remote operations require permission authentication --- ## Other Notes - If user didn't provide operation type, should first inform user and ask about default configuration - Continue executing request after user confirmation - Door access control operations all have physical security risks, please operate with caution --- --- **Error Codes**: | Return Code | Return Message | Description | |-------------|---------------------------|--------------------------------------------------------------------------| | VMS000003 | Resource operation failed | Resource operation failed: The access control resource ID does not exist | --- FILE:modules/Hik-Connect_Team_ACS/scripts/acs_control.py #!/usr/bin/env python # -*- coding: utf-8 -*- """ HCTOpen ACS Control """ import sys import os import json import argparse from datetime import datetime # Robust lib directory import logic: search upward until lib directory is found def setup_lib_path(): current_dir = os.path.dirname(os.path.abspath(__file__)) # Search upward 3 levels for root directory containing lib root_dir = current_dir for _ in range(3): root_dir = os.path.dirname(root_dir) potential_lib = os.path.join(root_dir, "lib") if os.path.exists(potential_lib): if potential_lib not in sys.path: sys.path.insert(0, potential_lib) return True return False if not setup_lib_path(): print("[ERROR] Cannot find lib directory, please ensure script is located in Hik-Connect_Team Skills directory structure") sys.exit(1) from token_manager import HCTOpenClient class ACSControlClient(HCTOpenClient): """Door access control client""" def control(self, action_type: int, element_list: list): """Execute door access control operation""" print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] Executing door access control: Type={action_type}, Count={len(element_list) if element_list else 'All'}") # Check if element_list is empty if not element_list: print( f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] [ERROR] element_list cannot be empty. Please provide at least one resource ID.") self.exit_with_json({ "success": False, "error": "element_list is required and cannot be empty", "errorCode": "PARAMETER_EMPTY" }) print( f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] Executing door access control: Type={action_type}, Count={len(element_list)}") endpoint = "/api/hccgw/acs/v1/remote/control" payload = { "remoteControl": { "actionType": action_type, "elementlist": element_list } } # ACS module API uses "Token" as Header Key result = self.request("POST", endpoint, json_data=payload, token_header_key="Token") if result.get("errorCode") == "0": op_result = result.get("data", {}).get("operationResult", []) if op_result: print("[WARNING] Some devices operation failed:") headers = ["No.", "Door Resource ID", "Door Name", "Area", "Error Code"] rows = [] for i, res in enumerate(op_result, 1): rows.append([ i, res.get("elementId", "-"), res.get("elementName", "-"), res.get("areaName", "-"), res.get("errorCode", "-") ]) self.print_table("Failed Device List", headers, rows) self.exit_with_json({"success": False, "operationResult": op_result, "error": "Some operations failed"}) else: print("[SUCCESS] All door access operations executed successfully") self.exit_with_json({"success": True, "operationResult": []}) else: # Use unified message field print(f"[ERROR] Door access control failed: {result.get('message', 'Unknown error')}") self.exit_with_json({"success": False, "error": result.get("message", "Unknown error"), "errorCode": result.get("errorCode")}) def main(): parser = argparse.ArgumentParser(description="HCTOpen Door Access Remote Control") parser.add_argument("--action-type", type=int, required=True, choices=[1, 2, 3, 4], help="1-open door, 2-close door, 3-normally open, 4-normally closed") parser.add_argument("--element-list", type=str, default="", help="Comma-separated resource ID list") args = parser.parse_args() elements = [e.strip() for e in args.element_list.split(',') if e.strip()] if args.element_list else [] client = ACSControlClient() client.control(args.action_type, elements) if __name__ == "__main__": main() FILE:lib/README_TOKEN_MANAGER.md # HCT Global Token Manager 🔐 Provides unified Token cache management for all HCT skills. ## 📁 Location ``` /Users/jony/.openclaw/workspace/skills/Hik-Connect Team Skills/lib/token_manager.py ``` ## ✨ Features - **Global Cache**: All skills share the same Token, avoiding repeated acquisition - **Smart Reuse**: Use cache directly during Token validity period, no API calls - **Safe Buffer**: Auto-refresh 5 minutes before expiration to avoid boundary issues - **Multi-Account Support**: Identify different accounts based on md5(appKey:appSecret) - **Atomic Write**: Write to temporary file first then replace, ensuring data safety - **Permission Protection**: Cache file permission set to 600 (owner read/write only) ## 🔐 Credential Configuration **Credentials only need to be configured ONCE. The system will automatically find and use them.** **Priority order (highest to lowest):** ``` ┌─────────────────────────────────────────────────────────────┐ │ 1. Environment Variables (if set) │ │ ├─ HIK_CONNECT_TEAM_OPENAPI_APP_KEY │ │ └─ HIK_CONNECT_TEAM_OPENAPI_SECRET_KEY │ ├─────────────────────────────────────────────────────────────┤ │ 2. OpenClaw Config Files (if env vars not set) │ │ ├─ ~/.openclaw/config.json │ │ ├─ ~/.openclaw/gateway/config.json │ │ └─ ~/.openclaw/channels.json ⭐ Recommended │ └─────────────────────────────────────────────────────────────┘ ``` ### OpenClaw Config File Format Config file format (same for all three files): ```json { "channels": { "hik_connect_team_openapi": { "appKey": "Your Hik-Connect Team OpenAPI AppKey", "secretKey": "Your Hik-Connect Team OpenAPI SecretKey", "enabled": true } } } ``` **Recommended: Save to `~/.openclaw/channels.json`** — This is the dedicated file for channel credentials. **⚠️ Security Note**: Before saving credentials to a config file, ask the user for confirmation. Storing credentials on disk is convenient but introduces some risk. Always inform the user of this option and let them choose. --- ## 🚀 Usage ### Method 1: Import and use in Python skills ```python # Add lib directory to path import os import sys script_dir = os.path.dirname(os.path.abspath(__file__)) workspace_dir = os.path.join(script_dir, "..", "..") lib_dir = os.path.abspath(os.path.join(workspace_dir, "Hik-Connect Team Skills", "lib")) if os.path.exists(lib_dir) and lib_dir not in sys.path: sys.path.insert(0, lib_dir) from token_manager import get_cached_token # Get Token (prefer cache, auto-refresh if expired) token_result = get_cached_token(app_key, app_secret, use_cache=True) if token_result["success"]: access_token = token_result["access_token"] print(f"Token: {access_token}") print(f"From Cache: {token_result['from_cache']}") ``` ### Method 2: Command Line Tool ```bash cd /Users/jony/.openclaw/workspace/skills/Hik-Connect Team Skills # Get Token (use cache) python lib/token_manager.py get --app-key "your_key" --app-secret "your_secret" # Force refresh Token (no cache) python lib/token_manager.py refresh --app-key "your_key" --app-secret "your_secret" # View cache list python lib/token_manager.py list # Clear specific account cache python lib/token_manager.py clear --app-key "your_key" --app-secret "your_secret" # Clear all cache python lib/token_manager.py clear ``` ## 📊 Cache Location ``` /var/folders/xx/xxxx/T/hctopen_global_token_cache/global_token_cache.json ``` Cache file format: ```json { "3aa746c5ea5329ab...": { "cache_key": "3aa746c5ea5329ab...", "access_token": "at.ay4x6ris6kl61uao6a3qcjpa1ww...", "area_domain": "https://ieu-team.hikcentralconnect.com", "expire_time": 1774419637518, "created_at": 1773816338280, "app_key_prefix": "26810f3a..." } } ``` ## 🔄 Workflow ``` Skill Startup ↓ Call get_cached_token(app_key, app_secret) ↓ Check cache file ├─ Cache exists and not expired → Return cached Token directly ✅ └─ Cache doesn't exist or expired → Call API to get new Token ↓ Save to cache file ↓ Return new Token ``` ## 🎯 Integrated Skills | Skill | Status | File | |-------------------------------|--------------|----------------------------| | Device List (device_list) | ✅ Integrated | `scripts/list_devices.py` | | Device Detail (device_detail) | ✅ Integrated | `scripts/device_detail.py` | ## 🧪 Test Examples ```bash # 1. Clear cache python lib/token_manager.py clear # 2. First acquisition (from API) python lib/token_manager.py get --app-key "26810f3acd794862b608b6cfbc32a6b8" --app-secret "3155063e93f09f377eaf5ba9f321f8c2" # Output: From Cache: False # 3. Get again (from cache) python lib/token_manager.py get --app-key "26810f3acd794862b608b6cfbc32a6b8" --app-secret "3155063e93f09f377eaf5ba9f321f8c2" # Output: From Cache: True # 4. View cache python lib/token_manager.py list ``` ## ⚠️ Notes 1. **Token Validity**: 7 days, auto-refresh 5 minutes before expiration 2. **Cache Cleanup**: System temp directory may be periodically cleaned 3. **Multi-Account**: Each appKey:appSecret combination has independent cache 4. **Security**: Cache file permission 600, owner read/write only 5. **Concurrency**: Supports multi-process simultaneous reading, atomic operation during writing ## 📝 API Functions ### get_cached_token(app_key, app_secret, use_cache=True) Get Token, prefer using cache. **Returns**: ```json { "success": True, "access_token": "at.xxx", "area_domain": "https://hpc-sgp-uat-5.hik-partner.com", "expire_time": 1774419637518, "from_cache": True } ``` ### refresh_token(app_key, app_secret, cache_key=None) Force refresh Token, do not use cache. ### clear_token_cache(app_key=None, app_secret=None) Clear cache (can specify account or clear all). ### list_cached_tokens() List all cached Token information. --- **Error Codes**: | Return Code | Return Message | Description | |-------------|-------------------|-------------------------------| | OPEN000001 | AK does not exist | Please check if AK is correct | | OPEN000002 | SK error | SK does not match current AK | FILE:lib/token_manager.py #!/usr/bin/env python # -*- coding: utf-8 -*- """ HCTOpen Global Token Manager & Base Client Provides global Token cache management and base API request encapsulation. """ import os import sys import time import json import hashlib import tempfile import requests from typing import Dict, Any, Optional, List, Union from pathlib import Path def _get_openclaw_config_paths(): """Get list of OpenClaw config file paths to search""" home = Path.home() return [ home / ".openclaw" / "config.json", home / ".openclaw" / "gateway" / "config.json", home / ".openclaw" / "channels.json", ] def _load_openclaw_config(): """Load Hik-Connect Team credentials from OpenClaw config files Searches for config in the following order: 1. ~/.openclaw/config.json 2. ~/.openclaw/gateway/config.json 3. ~/.openclaw/channels.json Config format: { "channels": { "hik_connect_team_openapi": { "appKey": "your_app_key", "secretKey": "your_secret_key", "enabled": true } } } """ for config_path in _get_openclaw_config_paths(): if config_path.exists(): try: with open(config_path, "r") as f: content = f.read().strip() if not content: continue data = json.loads(content) hct_config = data.get("channels", {}).get("hik_connect_team_openapi", {}) if hct_config.get("enabled", False) and hct_config.get("appKey") and hct_config.get("secretKey"): return hct_config.get("appKey"), hct_config.get("secretKey") except (json.JSONDecodeError, OSError): continue return None, None class TokenManager: """Manage HCTOpen AccessToken acquisition and caching""" CACHE_DIR_NAME = "hctopen_global_token_cache" CACHE_FILE_NAME = "global_token_cache.json" TOKEN_BUFFER_TIME = 5 * 60 * 1000 # 5-minute buffer TOKEN_URL = "https://ieu-team.hikcentralconnect.com/api/hccgw/platform/v1/token/get" def __init__(self): self.token_url = self.TOKEN_URL self.cache_file = os.path.join(tempfile.gettempdir(), self.CACHE_DIR_NAME, self.CACHE_FILE_NAME) os.makedirs(os.path.dirname(self.cache_file), exist_ok=True) def _get_cache_key(self, app_key: str, secret_key: str) -> str: return hashlib.md5(f"{app_key}:{secret_key}".encode()).hexdigest() def _load_cache(self) -> Dict[str, Any]: if not os.path.exists(self.cache_file): return {} try: with open(self.cache_file, "r") as f: return json.load(f) except Exception as e: print(f"[WARNING] Failed to load Token cache: {e}", file=sys.stderr) return {} def _save_cache(self, cache_data: Dict[str, Any]): try: temp_file = self.cache_file + ".tmp" with open(temp_file, "w") as f: json.dump(cache_data, f, indent=2) os.replace(temp_file, self.cache_file) # Only apply permission on Unix systems (os.chmod has no effect on Windows) if os.name != 'nt': os.chmod(self.cache_file, 0o600) except Exception as e: print(f"[WARNING] Failed to save Token cache: {e}", file=sys.stderr) def get_token(self, app_key: str, secret_key: str, force_refresh: bool = False) -> Dict[str, Any]: """Get Token, prefer using cache""" use_cache = os.environ.get("HIK_CONNECT_TEAM_TOKEN_CACHE", "1") == "1" and not force_refresh cache_key = self._get_cache_key(app_key, secret_key) if use_cache: cache = self._load_cache() if cache_key in cache: token_data = cache[cache_key] # Handle expire_time in seconds or milliseconds # HCT API returns expireTime in seconds (e.g., 3600), # but cache stores it as-is. Convert to milliseconds for comparison. # If value > 10^11, it's already in milliseconds (e.g., 1774419637518) expire_time = token_data.get("expire_time", 0) if expire_time < 10**11: expire_time *= 1000 if time.time() * 1000 + self.TOKEN_BUFFER_TIME < expire_time: return {"success": True, "access_token": token_data["access_token"], "area_domain": token_data.get("area_domain"), "from_cache": True} # Request new Token try: resp = requests.post(self.token_url, json={"appKey": app_key, "secretKey": secret_key}, timeout=10) result = resp.json() if result.get("errorCode") == "0": data = result.get("data", {}) access_token = data.get("accessToken") expire_time = data.get("expireTime") # API usually returns seconds area_domain = data.get("areaDomain", "").rstrip("/") # Update cache cache = self._load_cache() cache[cache_key] = { "access_token": access_token, "expire_time": expire_time, "area_domain": area_domain, "app_key_prefix": app_key[:8] } self._save_cache(cache) return {"success": True, "access_token": access_token, "area_domain": area_domain, "from_cache": False} # Unify error field as message return {"success": False, "message": result.get("message", "Unknown error"), "errorCode": result.get("errorCode")} except Exception as e: return {"success": False, "message": f"Request exception: {str(e)}"} # Robust lib directory import logic: search upward until lib directory is found def setup_lib_path(): current_dir = os.path.dirname(os.path.abspath(__file__)) # Search upward 3 levels for root directory containing lib root_dir = current_dir for _ in range(3): root_dir = os.path.dirname(root_dir) potential_lib = os.path.join(root_dir, "lib") if os.path.exists(potential_lib): if potential_lib not in sys.path: sys.path.insert(0, potential_lib) return True return False if not setup_lib_path(): print("[ERROR] Cannot find lib directory, please ensure script is located in Hik-Connect_Team Skills directory structure") sys.exit(1) class HCTOpenClient: """HCTOpen API Base Client""" def __init__(self): # Priority 1: Environment variables (highest) self.app_key = os.environ.get("HIK_CONNECT_TEAM_OPENAPI_APP_KEY") self.secret_key = os.environ.get("HIK_CONNECT_TEAM_OPENAPI_SECRET_KEY") self._config_source = "environment variables" # Priority 2: OpenClaw config files (only if env vars not set) if not all([self.app_key, self.secret_key]): config_app_key, config_secret_key = _load_openclaw_config() if config_app_key and config_secret_key: self.app_key = config_app_key self.secret_key = config_secret_key self._config_source = "OpenClaw config file" if not all([self.app_key, self.secret_key]): print("[ERROR] Credentials not found. Please set either:") print(" 1. Environment variables: HIK_CONNECT_TEAM_OPENAPI_APP_KEY and HIK_CONNECT_TEAM_OPENAPI_SECRET_KEY") print(" 2. OpenClaw config file: ~/.openclaw/config.json with channels.hik_connect_team_openapi section") sys.exit(1) print(f"[INFO] Using credentials from: {self._config_source}") self.token_manager = TokenManager() self._access_token = None self._area_domain = None def get_access_token(self, force_refresh: bool = False) -> str: if not self._access_token or force_refresh: res = self.token_manager.get_token(self.app_key, self.secret_key, force_refresh) if res["success"]: self._access_token = res["access_token"] self._area_domain = res.get("area_domain", "") else: # Unify error field as message print(f"[ERROR] Failed to get Token: {res.get('message')}") sys.exit(1) return self._access_token def get_area_domain(self) -> str: """Get the area domain from token response, must call get_access_token first""" if not self._area_domain: self.get_access_token() return self._area_domain def request(self, method: str, endpoint: str, json_data: Optional[Dict] = None, params: Optional[Dict] = None, token_header_key: str = "Token") -> Dict[str, Any]: """Send request with Token, supports auto retry (when Token expired)""" # Use areaDomain from token response as the domain area_domain = self.get_area_domain() if not area_domain: return {"errorCode": "-1", "message": "areaDomain not found in token response"} url = f"{area_domain}{endpoint}" for attempt in range(2): headers = { "Content-Type": "application/json", token_header_key: self.get_access_token(force_refresh=(attempt > 0)) } try: response = requests.request(method, url, headers=headers, json=json_data, params=params, timeout=30) result = response.json() # Token invalid error codes - retry once with fresh token # Common token error codes in Hikvision APIs: 10002 (token expired/invalid), 20004 (token malformed) error_code = result.get("errorCode") if error_code in ["10002", "20004"] and attempt == 0: print("[INFO] Token may be invalid, trying to refresh Token and retry...") continue # Unify error field as message if result.get("errorCode") != "0" and "errorMsg" in result: result["message"] = result.pop("errorMsg") return result except requests.exceptions.RequestException as e: # Unify error field as message return {"errorCode": "-1", "message": f"Request exception: {str(e)}"} except json.JSONDecodeError: # Unify error field as message return {"errorCode": "-1", "message": f"JSON parsing failed: {response.text}"} except Exception as e: # Unify error field as message return {"errorCode": "-1", "message": f"Unknown error: {str(e)}"} # Both attempts failed return {"errorCode": "-1", "message": "Request failed, Token refresh still invalid or other issue encountered"} @staticmethod def print_table(title: str, headers: List[str], rows: List[List[Any]]): """Generic table printing utility""" print("=" * 70) print(title) print("=" * 70) if not rows: print("No data found") return # Calculate max width for each column col_widths = [len(h) for h in headers] for row in rows: for i, val in enumerate(row): # Ensure val is string, avoid len() error col_widths[i] = max(col_widths[i], len(str(val))) # Print header header_line = " ".join(f"{headers[i]:<{col_widths[i]}}" for i in range(len(headers))) print(header_line) print("-" * len(header_line)) # Print rows for row in rows: row_line = " ".join(f"{str(val):<{col_widths[i]}}" for i, val in enumerate(row)) print(f"{row_line}") print("=" * 70) @staticmethod def exit_with_json(data: Dict[str, Any]): """Output in JSON format and exit""" print("\n[JSON Output]") print(json.dumps(data, indent=2, ensure_ascii=False)) print("=" * 70) print("Done") print("=" * 70) sys.exit(0 if data.get("success", True) else 1) # Backward compatibility (if external code calls get_cached_token directly) def get_cached_token(app_key, secret_key, use_cache=True): tm = TokenManager() return tm.get_token(app_key, secret_key, force_refresh=not use_cache) if __name__ == "__main__": # Simple CLI test # Ensure HIK_CONNECT_TEAM_OPENAPI_APP_KEY, HIK_CONNECT_TEAM_OPENAPI_SECRET_KEY environment variables are set try: client = HCTOpenClient() token = client.get_access_token() print(f"Test Token: {token[:10]}...") # Simulate a request test_endpoint = "/api/hccgw/resource/v1/devices/get" # Hypothetical test endpoint test_result = client.request("POST", test_endpoint, json_data={"pageIndex":1, "pageSize":1}, token_header_key="Token") print("Test request result:") print(json.dumps(test_result, indent=2, ensure_ascii=False)) except Exception as e: print(f"Test failed: {e}")

hik-connect-team-skill

Comments (0)