163 lines
5 KiB
Python
163 lines
5 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_snapshot(
|
|
device: Device,
|
|
obj: Object,
|
|
channel: int = 1,
|
|
timeout: int = 10,
|
|
) -> tuple[bytes, str]:
|
|
"""
|
|
Fetch a JPEG snapshot from the camera via ISAPI.
|
|
Returns (image_bytes, content_type). Raises RuntimeError on failure.
|
|
"""
|
|
user, password, port = _resolve_camera_credentials(device, obj)
|
|
|
|
# Try connection_params.frame_url first (custom snapshot URL)
|
|
frame_url: Optional[str] = (device.connection_params or {}).get("frame_url")
|
|
if frame_url:
|
|
url = frame_url
|
|
else:
|
|
# Hikvision ISAPI channel format: channel 1 main = 101
|
|
ch = f"{channel:02d}1"
|
|
url = f"http://{device.ip}:{port}/ISAPI/Streaming/channels/{ch}/picture"
|
|
|
|
try:
|
|
async with httpx.AsyncClient(auth=(user, password), timeout=timeout) as client:
|
|
resp = await client.get(url)
|
|
except httpx.TimeoutException:
|
|
raise RuntimeError("Камера не отвечает (timeout)")
|
|
except Exception as exc:
|
|
raise RuntimeError(str(exc))
|
|
|
|
if resp.status_code == 401:
|
|
raise RuntimeError("Ошибка авторизации (401) — проверьте логин/пароль камеры")
|
|
if resp.status_code >= 400:
|
|
raise RuntimeError(f"Камера вернула HTTP {resp.status_code}")
|
|
|
|
content_type = resp.headers.get("content-type", "image/jpeg")
|
|
return resp.content, content_type
|
|
|
|
|
|
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,
|
|
)
|