96 lines
3.5 KiB
Python
96 lines
3.5 KiB
Python
"""
|
|
CameraSetNTPAction — configures NTP on a Hikvision camera via ISAPI.
|
|
|
|
Reads current NTP config, patches the server address, then PUTs it back.
|
|
"""
|
|
|
|
import xml.etree.ElementTree as ET
|
|
|
|
from app.models.device import Device
|
|
from app.models.object import Object
|
|
from app.plugins.base import ActionResult, BaseAction, Emitter
|
|
from app.services import camera as camera_service
|
|
|
|
|
|
class CameraSetNTPAction(BaseAction):
|
|
name = "camera_set_ntp"
|
|
display_name = "Camera: Set NTP Server"
|
|
compatible_categories = ["camera"]
|
|
params_schema = [
|
|
{
|
|
"name": "server",
|
|
"type": "string",
|
|
"label": "NTP Server",
|
|
"required": True,
|
|
"placeholder": "pool.ntp.org",
|
|
},
|
|
]
|
|
|
|
async def execute(
|
|
self, device: Device, obj: Object, params: dict, emit: Emitter
|
|
) -> ActionResult:
|
|
server = params.get("server", "").strip()
|
|
if not server:
|
|
return ActionResult(success=False, error="No NTP server specified")
|
|
|
|
# Step 1: read current NTP config
|
|
await emit(f"[{device.ip}] GET /ISAPI/System/time/ntpServers", "info")
|
|
get_result = await camera_service.isapi_get(device, obj, "System/time/ntpServers")
|
|
|
|
if not get_result.success:
|
|
# Fall back to simple XML PUT if read fails
|
|
await emit(f"[{device.ip}] Read failed ({get_result.error}), using minimal XML", "warn")
|
|
xml_body = (
|
|
'<?xml version="1.0" encoding="UTF-8"?>'
|
|
"<NTPServerList>"
|
|
"<NTPServer>"
|
|
"<id>1</id>"
|
|
f"<hostName>{server}</hostName>"
|
|
"<ipAddress></ipAddress>"
|
|
"<portNo>123</portNo>"
|
|
"<synchronizeInterval>60</synchronizeInterval>"
|
|
"</NTPServer>"
|
|
"</NTPServerList>"
|
|
)
|
|
else:
|
|
# Patch the first NTP server's hostName in the returned XML
|
|
try:
|
|
root = ET.fromstring(get_result.body)
|
|
ns = root.tag.split("}")[0].strip("{") if "}" in root.tag else ""
|
|
ns_prefix = f"{{{ns}}}" if ns else ""
|
|
|
|
# Find first NTPServer entry
|
|
entry = root.find(f".//{ns_prefix}NTPServer")
|
|
if entry is not None:
|
|
hn = entry.find(f"{ns_prefix}hostName")
|
|
if hn is not None:
|
|
hn.text = server
|
|
ip_el = entry.find(f"{ns_prefix}ipAddress")
|
|
if ip_el is not None:
|
|
ip_el.text = ""
|
|
xml_body = ET.tostring(root, encoding="unicode", xml_declaration=True)
|
|
except ET.ParseError:
|
|
xml_body = get_result.body # send back unchanged, let camera deal with it
|
|
|
|
# Step 2: PUT updated config
|
|
await emit(f"[{device.ip}] PUT /ISAPI/System/time/ntpServers server={server}", "info")
|
|
put_result = await camera_service.isapi_put(
|
|
device, obj, "System/time/ntpServers", xml_body
|
|
)
|
|
|
|
if not put_result.success:
|
|
msg = put_result.error or f"HTTP {put_result.status_code}"
|
|
await emit(f"[{device.ip}] PUT failed: {msg}", "error")
|
|
return ActionResult(
|
|
success=False,
|
|
stdout=put_result.body,
|
|
error=msg,
|
|
)
|
|
|
|
await emit(f"[{device.ip}] NTP server set to {server}", "info")
|
|
return ActionResult(
|
|
success=True,
|
|
stdout=put_result.body,
|
|
exit_code=0,
|
|
data={"server": server},
|
|
)
|