Fixed Hyundai Device Processing

scripts/artifacts/hyundai_callHistory.py

@@ -1,35 +1,76 @@
 __artifacts_v2__ = {
     "hyundaiCallHistory": {
         "name": "Hyundai - Call History",
-        "description": "Bluetooth call history from a Hyundai infotainment CH_*.db.",
-        "author": "Nixy Camacho",
-        "version": "0.2",
+        "description": "Bluetooth call history per connected phone from Hyundai/Kia infotainment CH_{mac}.db databases.",
+        "author": "Nixy Camacho, @pmpulkownik",
+        "version": "0.3",
         "creation_date": "2023-06-09",
-        "last_update_date": "2026-06-29",
+        "last_update_date": "2026-09-02",
         "requirements": "none",
         "category": "Hyundai Vehicles",
-        "notes": "Rewritten as a single query (the original ran nine separate SELECTs and crashed on "
-                 "os.path.splittext). date / date_sort are interpreted as Unix epochs and "
-                 "normalized to UTC.",
+        "notes": "Extracts call logs per device and derives device Bluetooth MAC directly from the database filename (CH_{mac}.db).",
         "paths": ('*/bluetooth/DB_BMS/CH_*.db*',),
         "output_types": "standard",
         "artifact_icon": "phone-call",
     }
 }
 
+import os
+import re
+import sqlite3
 from datetime import datetime, timezone
 
-from scripts.ilapfuncs import artifact_processor, convert_unix_ts_to_utc, open_sqlite_db_readonly
+from scripts.ilapfuncs import (artifact_processor, convert_unix_ts_to_utc, logfunc,
+                               open_sqlite_db_readonly)
+
+# Ordered as queried. A column missing from a phone's database is reported empty
+# rather than costing that phone every row -- see _present_columns.
+_CALL_COLUMNS = ('date', 'date_sort', '_id', 'given_name', 'family_name', 'phone_number',
+                 'calltype', 'duration', 'numberType')
+
+
+def _format_mac_from_filename(filename):
+    """
+    Extracts and formats MAC address from filename like CH_AABBCCDDEEFF.db -> AA:BB:CC:DD:EE:FF
+    """
+    base = os.path.basename(filename)
+    stem = base.rsplit('.', 1)[0]
+    raw_mac = stem[3:] if stem.startswith('CH_') else stem
+
+    clean_hex = re.sub(r'[^A-Fa-f0-9]', '', raw_mac)
+    if len(clean_hex) == 12:
+        return ':'.join(clean_hex[i:i + 2] for i in range(0, 12, 2)).upper()
+    return raw_mac.upper()
+
+
+def _present_columns(cursor, table, wanted):
+    try:
+        cursor.execute(f'PRAGMA table_info({table})')
+        present = {row[1] for row in cursor.fetchall()}
+    except sqlite3.Error:
+        return []
+    return [column for column in wanted if column in present]
+
+
+def _value(record, column):
+    """The stored value, or '' when the column is absent or NULL."""
+    value = record.get(column)
+    return '' if value is None else value
 
 
 def _ts(value):
-    if value is None or value == '':
+    if value is None or value == '' or value == 0:
         return ''
     if isinstance(value, (int, float)):
+        if value > 1e11:
+            value = value / 1000.0
         return convert_unix_ts_to_utc(value)
     text = str(value).strip()
     if text.isdigit():
-        return convert_unix_ts_to_utc(int(text))
+        val = int(text)
+        if val > 1e11:
+            val = val / 1000.0
+        return convert_unix_ts_to_utc(val)
     try:
         dt = datetime.fromisoformat(text.replace('Z', '+00:00'))
         return dt.replace(tzinfo=timezone.utc) if dt.tzinfo is None else dt.astimezone(timezone.utc)
@@ -40,24 +81,83 @@ def _ts(value):
 @artifact_processor
 def hyundaiCallHistory(context):
     data_list = []
-    source_path = ''
+    source_paths = []
+
     for file_found in context.get_files_found():
         file_found = str(file_found)
-        if not file_found.endswith('.db'):
+        if os.path.isdir(file_found) or not file_found.endswith('.db'):
             continue
-        source_path = file_found
+
+        device_mac = _format_mac_from_filename(file_found)
+        db_filename = os.path.basename(file_found)
+
         db = open_sqlite_db_readonly(file_found)
+        if db is None:
+            continue
         cursor = db.cursor()
-        cursor.execute('''
-            SELECT _id, given_name, family_name, phone_number, calltype, date, date_sort,
-                   duration, numberType
-            FROM bluetooth_callhistory
-        ''')
-        for row in cursor.fetchall():
-            data_list.append((row[0], row[1], row[2], row[3], row[4], _ts(row[5]), _ts(row[6]),
-                              row[7], row[8]))
+
+        columns = _present_columns(cursor, 'bluetooth_callhistory', _CALL_COLUMNS)
+        if not columns:
+            logfunc(f"{db_filename}: no bluetooth_callhistory table, skipped")
+            db.close()
+            continue
+
+        statement = f'SELECT {", ".join(columns)} FROM bluetooth_callhistory'
+        if 'date' in columns:
+            statement += ' ORDER BY date DESC'
+
+        try:
+            cursor.execute(statement)
+            rows = cursor.fetchall()
+        except sqlite3.Error as ex:
+            logfunc(f"{db_filename}: reading bluetooth_callhistory failed, skipped")
+            logfunc(f" - {str(ex)}")
+            db.close()
+            continue
+
         db.close()
+        source_paths.append(file_found)
+
+        missing = [column for column in _CALL_COLUMNS if column not in columns]
+        if missing:
+            logfunc(f"{db_filename}: reported without {', '.join(missing)}")
+
+        for row in rows:
+            record = dict(zip(columns, row))
+            given_name = _value(record, 'given_name')
+            family_name = _value(record, 'family_name')
+
+            name_parts = [str(p).strip() for p in (given_name, family_name) if p]
+            full_name = ' '.join(name_parts) if name_parts else ''
+
+            data_list.append((
+                _ts(record.get('date')),
+                _ts(record.get('date_sort')),
+                device_mac,
+                _value(record, '_id'),
+                full_name,
+                given_name,
+                family_name,
+                _value(record, 'phone_number'),
+                _value(record, 'calltype'),
+                _value(record, 'duration'),
+                _value(record, 'numberType'),
+                db_filename
+            ))
+
+    data_headers = (
+        ('Date (UTC)', 'datetime'),
+        ('Date Sort (UTC)', 'datetime'),
+        'Device MAC',
+        'Record ID',
+        'Full Name',
+        'First Name',
+        'Last Name',
+        ('Phone Number', 'phonenumber'),
+        'Call Type (as stored)',
+        'Duration (as stored)',
+        'Number Type (as stored)',
+        'Source Database'
+    )
 
-    data_headers = ('id', 'given_name', 'family_name', ('phone_number', 'phonenumber'), 'calltype',
-                    ('date', 'datetime'), ('date_sort', 'datetime'), 'duration', 'numberType')
-    return data_headers, data_list, context.get_relative_path(source_path)
+    return data_headers, data_list, '\n'.join(source_paths)

scripts/artifacts/hyundai_contacts.py

@@ -1,39 +1,133 @@
 __artifacts_v2__ = {
     "hyundaiContacts": {
-        "name": "Hyundai - Contacts",
-        "description": "Bluetooth contacts from a Hyundai infotainment MC_*.db.",
-        "author": "Nixy Camacho",
-        "version": "0.2",
+        "name": "Hyundai - Bluetooth Contacts",
+        "description": "Bluetooth contacts per connected phone from Hyundai/Kia infotainment MC_{mac}.db databases.",
+        "author": "Nixy Camacho, @pmpulkownik",
+        "version": "0.3",
         "creation_date": "2023-06-09",
-        "last_update_date": "2026-06-29",
+        "last_update_date": "2026-09-02",
         "requirements": "none",
         "category": "Hyundai Vehicles",
-        "notes": "Rewritten as a single query (the original ran separate SELECTs and crashed on "
-                 "os.path.splittext).",
+        "notes": "Extracts contacts per device and derives device Bluetooth MAC address directly from the database filename (MC_{mac}.db).",
         "paths": ('*/bluetooth/DB_BMS/MC_*.db*',),
         "output_types": "standard",
-        "artifact_icon": "user",
+        "artifact_icon": "users",
     }
 }
 
-from scripts.ilapfuncs import artifact_processor, open_sqlite_db_readonly
+import os
+import re
+import sqlite3
+
+from scripts.ilapfuncs import artifact_processor, logfunc, open_sqlite_db_readonly
+
+# Ordered as reported. A column missing from a phone's database is reported empty
+# rather than costing that phone every row -- see _present_columns.
+_CONTACT_COLUMNS = ('_id', 'given_name', 'family_name', 'phone_number', 'phone_type')
+
+
+def _format_mac_from_filename(filename):
+    """
+    Extracts and formats MAC address from filename like MC_AABBCCDDEEFF.db -> AA:BB:CC:DD:EE:FF
+    """
+    base = os.path.basename(filename)
+    stem = base.rsplit('.', 1)[0]
+    raw_mac = stem[3:] if stem.startswith('MC_') else stem
+
+    clean_hex = re.sub(r'[^A-Fa-f0-9]', '', raw_mac)
+    if len(clean_hex) == 12:
+        return ':'.join(clean_hex[i:i + 2] for i in range(0, 12, 2)).upper()
+    return raw_mac.upper()
+
+
+def _present_columns(cursor, table, wanted):
+    try:
+        cursor.execute(f'PRAGMA table_info({table})')
+        present = {row[1] for row in cursor.fetchall()}
+    except sqlite3.Error:
+        return []
+    return [column for column in wanted if column in present]
+
+
+def _value(record, column):
+    """The stored value, or '' when the column is absent or NULL."""
+    value = record.get(column)
+    return '' if value is None else value
 
 
 @artifact_processor
 def hyundaiContacts(context):
     data_list = []
-    source_path = ''
+    source_paths = []
+
     for file_found in context.get_files_found():
         file_found = str(file_found)
-        if not file_found.endswith('.db'):
+        if os.path.isdir(file_found) or not file_found.endswith('.db'):
             continue
-        source_path = file_found
+
+        device_mac = _format_mac_from_filename(file_found)
+        db_filename = os.path.basename(file_found)
+
         db = open_sqlite_db_readonly(file_found)
+        if db is None:
+            continue
         cursor = db.cursor()
-        cursor.execute('SELECT _id, given_name, family_name, phone_number FROM bluetooth_contacts')
-        for row in cursor.fetchall():
-            data_list.append((row[0], row[1], row[2], row[3]))
+
+        columns = _present_columns(cursor, 'bluetooth_contacts', _CONTACT_COLUMNS)
+        if not columns:
+            logfunc(f"{db_filename}: no bluetooth_contacts table, skipped")
+            db.close()
+            continue
+
+        statement = f'SELECT {", ".join(columns)} FROM bluetooth_contacts'
+        if '_id' in columns:
+            statement += ' ORDER BY _id ASC'
+
+        try:
+            cursor.execute(statement)
+            rows = cursor.fetchall()
+        except sqlite3.Error as ex:
+            logfunc(f"{db_filename}: reading bluetooth_contacts failed, skipped")
+            logfunc(f" - {str(ex)}")
+            db.close()
+            continue
+
         db.close()
+        source_paths.append(file_found)
+
+        missing = [column for column in _CONTACT_COLUMNS if column not in columns]
+        if missing:
+            logfunc(f"{db_filename}: reported without {', '.join(missing)}")
+
+        for row in rows:
+            record = dict(zip(columns, row))
+            given_name = _value(record, 'given_name')
+            family_name = _value(record, 'family_name')
+
+            # Build full name cleanly if names are present
+            name_parts = [str(p).strip() for p in (given_name, family_name) if p]
+            full_name = ' '.join(name_parts) if name_parts else ''
+
+            data_list.append((
+                device_mac,
+                _value(record, '_id'),
+                full_name,
+                given_name,
+                family_name,
+                _value(record, 'phone_number'),
+                _value(record, 'phone_type'),
+                db_filename
+            ))
+
+    data_headers = (
+        'Device MAC',
+        'Record ID',
+        'Full Name',
+        'First Name',
+        'Last Name',
+        ('Phone Number', 'phonenumber'),
+        'Phone Type (as stored)',
+        'Source Database'
+    )
 
-    data_headers = ('ID', 'given_name', 'family_name', ('phone_number', 'phonenumber'))
-    return data_headers, data_list, context.get_relative_path(source_path)
+    return data_headers, data_list, '\n'.join(source_paths)