Move common function to util

frigate/config.py

@@ -4,7 +4,6 @@ import json
 import logging
 import os
 from enum import Enum
-import re
 from typing import Dict, List, Optional, Tuple, Union
 
 import matplotlib.pyplot as plt
@@ -13,8 +12,13 @@ import yaml
 from pydantic import BaseModel, Extra, Field, validator
 from pydantic.fields import PrivateAttr
 
-from frigate.const import BASE_DIR, CACHE_DIR, REGEX_CAMERA_NAME, REGEX_CAMERA_USER_PASS, YAML_EXT
+from frigate.const import (
-from frigate.util import create_mask, deep_merge, load_labels
+    BASE_DIR,
+    CACHE_DIR,
+    REGEX_CAMERA_NAME,
+    YAML_EXT,
+)
+from frigate.util import clean_camera_user_pass, create_mask, deep_merge, load_labels
 
 logger = logging.getLogger(__name__)
 
@@ -690,7 +694,7 @@ class CameraConfig(FrigateBaseModel):
         input_args = (
             input_args if isinstance(input_args, list) else input_args.split(" ")
         )
-        cleaned_input = re.sub(REGEX_CAMERA_USER_PASS, "*:*@", ffmpeg_input.path)
+        cleaned_input = clean_camera_user_pass(ffmpeg_input.path)
 
         cmd = (
             ["ffmpeg"]

frigate/log.py

@@ -12,6 +12,7 @@ from typing import Deque
 from collections import deque
 
 from frigate.const import REGEX_CAMERA_USER_PASS
+from frigate.util import clean_camera_user_pass
 
 
 def listener_configurer() -> None:
@@ -60,7 +61,7 @@ class LogPipe(threading.Thread):
 
     def cleanup_log(self, log: str) -> str:
         """Cleanup the log line to remove sensitive info and string tokens."""
-        log = re.sub(REGEX_CAMERA_USER_PASS, "*:*@", log).strip("\n")
+        log = clean_camera_user_pass(log).strip("\n")
         return log
 
     def fileno(self) -> int:

frigate/util.py

@@ -4,6 +4,7 @@ import hashlib
 import json
 import logging
 import math
+import re
 import signal
 import subprocess as sp
 import threading
@@ -20,6 +21,8 @@ import numpy as np
 import os
 import psutil
 
+from frigate.const import REGEX_CAMERA_USER_PASS
+
 logger = logging.getLogger(__name__)
 
 
@@ -624,6 +627,10 @@ def load_labels(path, encoding="utf-8"):
         else:
             return {index: line.strip() for index, line in enumerate(lines)}
 
+def clean_camera_user_pass(line: str) -> str:
+    """Removes user and password from line."""
+    return re.sub(REGEX_CAMERA_USER_PASS, "*:*@", line)
+
 
 class FrameManager(ABC):
     @abstractmethod