fixes issue with call log entries for FritzOS 7.90 #122

fritzinfluxdb/classes/fritzbox/service_definitions/telephone_list.py

@@ -18,50 +18,74 @@
 
 read_interval = 60
 
+call_types = {
+    "1": "incoming",
+    "2": "unanswered",
+    "3": "blocked",
+    "4": "outgoing"
+}
 
-def get_hash(data):
-    """
-    compute a MD5 hash and use as ID to track and group log data by uid tag
-    """
 
-    hit = hash_cache.get(data)
-    if hit is not None:
-        return hit
+class CallLogEntry:
 
-    hash_cache[data] = hashlib.md5(data.encode("UTF-8")).hexdigest()
+    def __init__(self, data: str):
 
-    return hash_cache.get(data)
+        # compute a MD5 hash and use as ID to track and group log data by uid tag
+        self._hash = hashlib.md5(data.encode("UTF-8")).hexdigest()
 
+        data_fields = data.split(";")
 
-def get_date(data):
-    return datetime.strptime(data.split(";")[1], '%d.%m.%y %H:%M')
+        self._call_type = call_types.get(data_fields[0], "undefined")
+        self._date_time = datetime.strptime(data_fields[1], '%d.%m.%y %H:%M')
+        self._caller_name = data_fields[2]
+        self._caller_number = data_fields[3]
+        if len(data_fields) == 7:
+            self._extension = data_fields[4]
+            self._number_called = data_fields[5]
+            self._duration = self.get_call_duration(data_fields[6])
+        else:
+            self._extension = data_fields[5]
+            self._number_called = data_fields[6]
+            self._duration = self.get_call_duration(data_fields[7])
 
+    @staticmethod
+    def get_call_duration(field) -> int:
+        """
+        returns call duration in minutes
+        """
 
-def get_call_type(data):
+        # noinspection PyBroadException
+        try:
+            hours, minutes = field.split(":")
+            duration = int(hours) * 60 + int(minutes)
+        except Exception:
+            duration = 0
 
-    call_types = {
-        "1": "incoming",
-        "2": "unanswered",
-        "3": "blocked",
-        "4": "outgoing"
-    }
+        return duration
 
-    return call_types.get(data.split(";")[0], "undefined")
+    def hash(self) -> str:
+        return self._hash
 
+    def type(self) -> str:
+        return self._call_type
 
-def get_call_duration(data):
-    """
-    returns call duration in minutes
-    """
+    def date_time(self) -> datetime:
+        return self._date_time
 
-    # noinspection PyBroadException
-    try:
-        hours, minutes = data.split(";")[6].split(":")
-        duration = int(hours) * 60 + int(minutes)
-    except Exception:
-        return 0
+    def caller_name(self) -> str:
+        return self._caller_name
 
-    return duration
+    def caller_number(self) -> str:
+        return self._caller_number
+
+    def extension(self) -> str:
+        return self._extension
+
+    def number_called(self) -> str:
+        return self._number_called
+
+    def duration(self) -> int:
+        return self._duration
 
 
 def prepare_response_data(response):
@@ -73,10 +97,11 @@ def prepare_response_data(response):
     filter_strings = [
         'sep=;',
         'Typ;Datum;Name;Rufnummer;Nebenstelle;Eigene Rufnummer;Dauer',
+        'Typ;Datum;Name;Rufnummer;Landes-/Ortsnetzbereich;Nebenstelle;Eigene Rufnummer;Dauer',
         ''
     ]
 
-    return [x for x in response.text.split("\n") if x not in filter_strings]
+    return [CallLogEntry(x) for x in response.text.split("\n") if x not in filter_strings]
 
 
 # due to the tracking of measurements multiple short calls from the same number within the same minute
@@ -100,61 +125,61 @@ def prepare_response_data(response):
                 "value_function": lambda data: data,
                 "next": {
                     "type": str,
-                    "tags_function": lambda data: {"uid": get_hash(data)},
-                    "value_function": get_call_type,
-                    "timestamp_function": get_date,
+                    "tags_function": lambda entry: {"uid": entry.hash()},
+                    "value_function": lambda entry: entry.type(),
+                    "timestamp_function": lambda entry: entry.date_time(),
                 }
             },
             "call_list_caller_name": {
                 "type": list,
                 "value_function": lambda data: data,
                 "next": {
                     "type": str,
-                    "tags_function": lambda data: {"uid": get_hash(data)},
-                    "value_function": lambda data: data.split(";")[2],
-                    "timestamp_function": get_date,
+                    "tags_function": lambda entry: {"uid": entry.hash()},
+                    "value_function": lambda entry: entry.caller_name(),
+                    "timestamp_function": lambda entry: entry.date_time(),
                 }
             },
             "call_list_caller_number": {
                 "type": list,
                 "value_function": lambda data: data,
                 "next": {
                     "type": str,
-                    "tags_function": lambda data: {"uid": get_hash(data)},
-                    "value_function": lambda data: data.split(";")[3],
-                    "timestamp_function": get_date,
+                    "tags_function": lambda entry: {"uid": entry.hash()},
+                    "value_function": lambda entry: entry.caller_number(),
+                    "timestamp_function": lambda entry: entry.date_time(),
                 }
             },
             "call_list_extension": {
                 "type": list,
                 "value_function": lambda data: data,
                 "next": {
                     "type": str,
-                    "tags_function": lambda data: {"uid": get_hash(data)},
-                    "value_function": lambda data: data.split(";")[4],
-                    "timestamp_function": get_date,
+                    "tags_function": lambda entry: {"uid": entry.hash()},
+                    "value_function": lambda entry: entry.extension(),
+                    "timestamp_function": lambda entry: entry.date_time(),
                 }
             },
             "call_list_number_called": {
                 "type": list,
                 "value_function": lambda data: data,
                 "next": {
                     "type": str,
-                    "tags_function": lambda data: {"uid": get_hash(data)},
-                    "value_function": lambda data: data.split(";")[5],
-                    "timestamp_function": get_date,
+                    "tags_function": lambda entry: {"uid": entry.hash()},
+                    "value_function": lambda entry: entry.number_called(),
+                    "timestamp_function": lambda entry: entry.date_time(),
                 }
             },
             "call_list_duration": {
                 "type": list,
                 "value_function": lambda data: data,
                 "next": {
                     "type": int,
-                    "tags_function": lambda data: {"uid": get_hash(data)},
-                    "value_function": get_call_duration,
-                    "timestamp_function": get_date,
+                    "tags_function": lambda entry: {"uid": entry.hash()},
+                    "value_function": lambda entry: entry.duration(),
+                    "timestamp_function": lambda entry: entry.date_time(),
                 }
-            },
+            }
         }
     }
 )