ipcam_ntp_util (wip: only supports hikvision cams for now)

1 files changed, 63 insertions, 20 deletions

diff --git a/bin/ipcam_ntp_util.py b/bin/ipcam_ntp_util.py
index 98639bd..0268a06 100755
--- a/bin/ipcam_ntp_util.py
+++ b/bin/ipcam_ntp_util.py
@@ -4,12 +4,22 @@ import requests
 import hashlib
 import xml.etree.ElementTree as ET
 
+from enum import Enum, auto
 from time import time
+from typing import Optional
 from argparse import ArgumentParser, ArgumentError
 from homekit.util import validate_ipv4, validate_ipv4_or_hostname
 from homekit.camera import IpcamConfig
 
 
+ipcam_config = IpcamConfig()
+
+
+class Action(Enum):
+    GET_NTP = auto()
+    SET_NTP = auto()
+
+
 def xml_to_dict(xml_data: str) -> dict:
     # Parse the XML data
     root = ET.fromstring(xml_data)
@@ -131,11 +141,14 @@ class HikvisionISAPIClient:
         data += '<Time><timeMode>NTP</timeMode><timeZone>CST-3:00:00</timeZone></Time>'
 
         r = requests.put(self.isapi_uri('System/time'), cookies=self.cookies, data=data, headers={
-            'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8'
+            'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8',
+            'X-Requested-With': 'XMLHttpRequest',
         })
         self.isapi_check_put_response(r)
 
-    def set_ntp_server(self, ntp_host: str, ntp_port: int = 123):
+    def set_ntp_server(self,
+                       ntp_host: str,
+                       ntp_port: int = 123):
         format = 'ipaddress' if validate_ipv4(ntp_host) else 'hostname'
 
         data = '<?xml version="1.0" encoding="UTF-8"?>'
@@ -145,7 +158,8 @@ class HikvisionISAPIClient:
                          data=data,
                          cookies=self.cookies,
                          headers={
-                            'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8'
+                            'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8',
+                            'X-Requested-With': 'XMLHttpRequest',
                          })
         self.isapi_check_put_response(r)
 
@@ -153,7 +167,12 @@ class HikvisionISAPIClient:
         return f'http://{self.host}/ISAPI/{path}'
 
     def isapi_check_put_response(self, r):
-        r.raise_for_status()
+        try:
+            r.raise_for_status()
+        except requests.exceptions.HTTPError as e:
+            # print(r.text)
+            raise e
+
         resp = xml_to_dict(r.text)['ResponseStatus']
 
         status_code = int(resp['statusCode'][0])
@@ -163,36 +182,60 @@ class HikvisionISAPIClient:
             raise ResponseError('response status looks bad')
 
 
+def process_hikvision_camera(host: str,
+                             action: Action,
+                             login: str,
+                             password: str,
+                             ntp_server: Optional[str] = None):
+    client = HikvisionISAPIClient(host)
+    try:
+        client.auth(login, password)
+        if action == Action.GET_NTP:
+            print(f'[{host}] {client.get_ntp_server()}')
+            return
+        client.set_ntp_server(ntp_server)
+    except AuthError as e:
+        print(f'[{host}] ({str(e)})')
+    except ResponseError as e:
+        print(f'[{host}] ({str(e)})')
+
+
 def main():
     parser = ArgumentParser()
-    parser.add_argument('--host', type=str, required=True)
+    parser.add_argument('--host', type=str)
+    parser.add_argument('--all', action='store_true')
     parser.add_argument('--get-ntp-server', action='store_true')
     parser.add_argument('--set-ntp-server', type=str)
     parser.add_argument('--username', type=str)
     parser.add_argument('--password', type=str)
     args = parser.parse_args()
 
+    if not args.host and not args.all:
+        raise ArgumentError(None, 'either --all or --host <IP> is required')
+
     if not args.get_ntp_server and not args.set_ntp_server:
         raise ArgumentError(None, 'either --get-ntp-server or --set-ntp-server is required')
 
-    ipcam_config = IpcamConfig()
+    action = Action.GET_NTP if args.get_ntp_server else Action.SET_NTP
     login = args.username if args.username else ipcam_config['web_creds']['login']
     password = args.password if args.password else ipcam_config['web_creds']['password']
 
-    client = HikvisionISAPIClient(args.host)
-    client.auth(args.username, args.password)
-
-    if args.get_ntp_server:
-        print(client.get_ntp_server())
-        return
-
-    if not args.set_ntp_server:
-        raise ArgumentError(None, '--set-ntp-server is required')
-
-    if not validate_ipv4_or_hostname(args.set_ntp_server):
-        raise ArgumentError(None, 'input ntp server is neither ip address nor a valid hostname')
-
-    client.set_ntp_server(args.set_ntp_server)
+    if action == Action.SET_NTP:
+        if not args.set_ntp_server:
+            raise ArgumentError(None, '--set-ntp-server is required')
+        if not validate_ipv4_or_hostname(args.set_ntp_server):
+            raise ArgumentError(None, 'input ntp server is neither ip address nor a valid hostname')
+
+    kwargs = {}
+    if args.set_ntp_server:
+        kwargs['ntp_server'] = args.set_ntp_server
+    if not args.all:
+        process_hikvision_camera(args.host, action, login, password, **kwargs)
+    else:
+        for cam in ipcam_config.get_all_cam_names():
+            if not ipcam_config.get_camera_type(cam).is_hikvision():
+                continue
+            process_hikvision_camera(ipcam_config.get_camera_ip(cam), action, login, password, **kwargs)
 
 
 if __name__ == '__main__':