migrate video_nodeQuery.sh to python

Video/video.sh

@@ -254,7 +254,7 @@ else
     if [[ "$session_id" != "null" && "$session_id" != "" && "$session_id" != "reserved" && "$recording_started" = "false" ]]; then
       echo "$(date -u +"${ts_format}") [${process_name}] - Session: $session_id is created"
       session_capabilities="$(jq -r "${JQ_SESSION_CAPABILITIES_QUERY}" "/tmp/status.json")"
-      return_list=($(bash "${VIDEO_CONFIG_DIRECTORY}/video_nodeQuery.sh" "${session_id}" "${session_capabilities}"))
+      return_list=($(python3 "${VIDEO_CONFIG_DIRECTORY}/video_nodeQuery.py" "${session_id}" "${session_capabilities}"))
       caps_se_video_record="${return_list[0]}"
       video_file_name="${return_list[1]}.mp4"
       if [[ "$caps_se_video_record" = "true" ]]; then

Video/video_nodeQuery.py

@@ -0,0 +1,111 @@
+#!/usr/bin/env python3
+
+import json
+import os
+import re
+import sys
+from typing import List, Tuple
+
+
+def main() -> None:
+    """
+    Process video recording configuration based on session capabilities.
+
+    Args:
+        sys.argv[1]: SESSION_ID
+        sys.argv[2]: SESSION_CAPABILITIES (JSON string)
+
+    Outputs:
+        Space-separated values: RECORD_VIDEO TEST_NAME
+    """
+    # Define parameters
+    session_id = sys.argv[1] if len(sys.argv) > 1 else ""
+    session_capabilities = sys.argv[2] if len(sys.argv) > 2 else ""
+
+    # Environment variables with defaults
+    video_cap_name = os.environ.get("VIDEO_CAP_NAME", "se:recordVideo")
+    test_name_cap = os.environ.get("TEST_NAME_CAP", "se:name")
+    video_name_cap = os.environ.get("VIDEO_NAME_CAP", "se:videoName")
+    video_file_name_trim = os.environ.get("SE_VIDEO_FILE_NAME_TRIM_REGEX", "[:alnum:]-_")
+    video_file_name_suffix = os.environ.get("SE_VIDEO_FILE_NAME_SUFFIX", "true")
+
+    # Initialize variables
+    record_video = None
+    test_name = None
+    video_name = None
+
+    # Extract values from session capabilities if provided
+    if session_capabilities:
+        try:
+            capabilities = json.loads(session_capabilities)
+            record_video = capabilities.get(video_cap_name)
+            test_name = capabilities.get(test_name_cap)
+            video_name = capabilities.get(video_name_cap)
+        except (json.JSONDecodeError, AttributeError):
+            # If JSON parsing fails, continue with None values
+            pass
+
+    # Check if enabling to record video
+    if (isinstance(record_video, str) and record_video.lower() == "false") or record_video is False:
+        record_video = "false"
+    else:
+        record_video = "true"
+
+    # Check if video file name is set via capabilities
+    if video_name and video_name != "null":
+        test_name = video_name
+    elif test_name and test_name != "null":
+        test_name = test_name
+    else:
+        test_name = ""
+
+    # Check if append session ID to the video file name suffix
+    if not test_name:
+        test_name = session_id
+    elif video_file_name_suffix.lower() == "true":
+        test_name = f"{test_name}_{session_id}"
+
+    # Normalize the video file name
+    test_name = normalize_filename(test_name, video_file_name_trim)
+
+    # Output the values for other scripts consuming
+    print(f"{record_video} {test_name}")
+
+
+def normalize_filename(filename: str, trim_pattern: str) -> str:
+    """
+    Normalize the filename by replacing spaces with underscores,
+    keeping only allowed characters, and truncating to 251 characters.
+
+    Args:
+        filename: The original filename
+        trim_pattern: Pattern defining allowed characters (e.g., "[:alnum:]-_")
+
+    Returns:
+        Normalized filename
+    """
+    if not filename:
+        return ""
+
+    # Replace spaces with underscores
+    normalized = filename.replace(" ", "_")
+
+    # Convert trim pattern to regex
+    # Handle character classes like [:alnum:]
+    if "[:alnum:]" in trim_pattern:
+        # Create regex pattern for alphanumeric characters plus other allowed chars
+        allowed_chars = trim_pattern.replace("[:alnum:]", "a-zA-Z0-9")
+        pattern = f"[^{allowed_chars}]"
+    else:
+        # Direct character set
+        pattern = f"[^{re.escape(trim_pattern)}]"
+
+    # Remove disallowed characters
+    normalized = re.sub(pattern, "", normalized)
+
+    # Truncate to 251 characters
+    return normalized[:251]
+
+
+if __name__ == "__main__":
+    main()