125 lines
3.6 KiB
Python
125 lines
3.6 KiB
Python
"""
|
|
CameraService — HTTP client for Hikvision ISAPI.
|
|
|
|
All endpoints use HTTP Basic auth with the camera's web credentials.
|
|
Credentials are resolved from device SSH overrides (reusing the same
|
|
username/password fields) — camera HTTP auth often matches SSH.
|
|
|
|
ISAPI base: http://<ip>/ISAPI/
|
|
"""
|
|
|
|
import xml.etree.ElementTree as ET
|
|
from dataclasses import dataclass
|
|
from typing import Optional
|
|
|
|
import httpx
|
|
|
|
from app.models.device import Device
|
|
from app.models.object import Object
|
|
from app.services.crypto import decrypt
|
|
|
|
|
|
@dataclass
|
|
class CameraResult:
|
|
success: bool
|
|
status_code: int = 0
|
|
body: str = ""
|
|
data: dict = None
|
|
error: str = ""
|
|
|
|
def __post_init__(self):
|
|
if self.data is None:
|
|
self.data = {}
|
|
|
|
|
|
def _resolve_camera_credentials(device: Device, obj: Object) -> tuple[str, str, int]:
|
|
"""Returns (username, password, http_port) — device overrides take priority."""
|
|
user = device.ssh_user_override or obj.ssh_user or "admin"
|
|
port = device.device_meta.get("http_port", 80) if device.device_meta else 80
|
|
|
|
if device.ssh_pass_override_enc:
|
|
password = decrypt(device.ssh_pass_override_enc)
|
|
elif obj.ssh_pass_enc:
|
|
password = decrypt(obj.ssh_pass_enc)
|
|
else:
|
|
password = ""
|
|
|
|
return user, password, int(port)
|
|
|
|
|
|
def _xml_to_dict(element: ET.Element) -> dict:
|
|
"""Recursively convert an XML element to a dict (simple, no attributes)."""
|
|
result = {}
|
|
for child in element:
|
|
# Strip namespace if present: {ns}tag → tag
|
|
tag = child.tag.split("}")[-1] if "}" in child.tag else child.tag
|
|
if len(child):
|
|
result[tag] = _xml_to_dict(child)
|
|
else:
|
|
result[tag] = child.text
|
|
return result
|
|
|
|
|
|
async def isapi_get(
|
|
device: Device,
|
|
obj: Object,
|
|
path: str,
|
|
timeout: int = 10,
|
|
) -> CameraResult:
|
|
"""Send GET to ISAPI endpoint, return parsed XML body."""
|
|
user, password, port = _resolve_camera_credentials(device, obj)
|
|
url = f"http://{device.ip}:{port}/ISAPI/{path.lstrip('/')}"
|
|
|
|
try:
|
|
async with httpx.AsyncClient(auth=(user, password), timeout=timeout) as client:
|
|
resp = await client.get(url)
|
|
except httpx.TimeoutException:
|
|
return CameraResult(success=False, error="Connection timed out")
|
|
except Exception as exc:
|
|
return CameraResult(success=False, error=str(exc))
|
|
|
|
body = resp.text
|
|
data: dict = {}
|
|
if body.strip().startswith("<"):
|
|
try:
|
|
root = ET.fromstring(body)
|
|
data = _xml_to_dict(root)
|
|
except ET.ParseError:
|
|
pass
|
|
|
|
return CameraResult(
|
|
success=resp.status_code < 400,
|
|
status_code=resp.status_code,
|
|
body=body,
|
|
data=data,
|
|
)
|
|
|
|
|
|
async def isapi_put(
|
|
device: Device,
|
|
obj: Object,
|
|
path: str,
|
|
xml_body: str,
|
|
timeout: int = 10,
|
|
) -> CameraResult:
|
|
"""Send PUT with XML body to ISAPI endpoint."""
|
|
user, password, port = _resolve_camera_credentials(device, obj)
|
|
url = f"http://{device.ip}:{port}/ISAPI/{path.lstrip('/')}"
|
|
|
|
try:
|
|
async with httpx.AsyncClient(auth=(user, password), timeout=timeout) as client:
|
|
resp = await client.put(
|
|
url,
|
|
content=xml_body.encode(),
|
|
headers={"Content-Type": "application/xml"},
|
|
)
|
|
except httpx.TimeoutException:
|
|
return CameraResult(success=False, error="Connection timed out")
|
|
except Exception as exc:
|
|
return CameraResult(success=False, error=str(exc))
|
|
|
|
return CameraResult(
|
|
success=resp.status_code < 400,
|
|
status_code=resp.status_code,
|
|
body=resp.text,
|
|
)
|