"""Reading tools for AsusRouter.""" from __future__ import annotations import ast import json import logging import re from typing import Any from asusrouter.const import ContentType from asusrouter.tools.converters import clean_input, safe_float, safe_float_nn from asusrouter.tools.types import ARCallableType from asusrouter.tools.units import ( DataRateUnitConverter, UnitConverterBase, UnitOfDataRate, ) _LOGGER = logging.getLogger(__name__) # Random symbols to avoid json errors RANDOM_SYMBOLS: list[str] = [ "\u0000", "\u0001", "\u0002", "\u0003", "\u0004", "\u0005", "\u0006", "\u0007", "\u0008", "\u0009", ] def is_non_negative(value: Any) -> bool: """Check if the value is non-negative.""" return safe_float_nn(value) >= 0 def merge_dicts( data: dict[Any, Any], merge_data: dict[Any, Any] ) -> dict[Any, Any]: """Merge two nested dicts into a single one. Keep all the existing values. """ # Create a new dictionary to store the merged data merged_data = data.copy() # Go through the merge data and fill the merged data for key, value in merge_data.items(): # Check if the value is a dict if isinstance(value, dict): # Check if the key is in the merged data if key not in merged_data or merged_data[key] is None: # Add the key to the merged data merged_data[key] = {} # Merge the dicts merged_data[key] = merge_dicts(merged_data[key], value) # Continue to the next value continue # Check if the key is in the merged data if key not in merged_data: # Add the key to the merged data merged_data[key] = value # If the key is already in the merged data, compare the values elif merged_data[key] is None: # If the value in the merged data is None, take the value # from the merge data merged_data[key] = value elif value is None: # If the value in the merge data is not None # and the value in the merge data is None, keep the value # in the merged data pass else: # If both values are not None, keep the value from the merged data pass # Return the merged data return merged_data def read_as_snake_case(data: str) -> str: """Convert a string to snake case.""" string = ( re.sub(r"(?<=[a-z])(?=[A-Z])|[^a-zA-Z]", " ", data) .strip() .replace(" ", "_") ) result = "".join(string.lower()) while "__" in result: result = result.replace("__", "_") return result def read_content_type(headers: dict[str, str]) -> ContentType: """Get the content type from the headers.""" # Get the content type from the headers content_type = headers.get("content-type", "").split(";")[0].strip() # Find the content type in ContentType enum and return correct # ContentType enum for content_type_enum in ContentType: if content_type_enum.value == content_type: return content_type_enum # If the content type is not found, return the content type as text return ContentType.UNKNOWN @clean_input def read_js_variables(content: str, **kwargs: Any) -> dict[str, Any]: """Get all the JS variables from the content.""" # Create a dict to store the data js_variables: dict[str, Any] = {} # regex = re.compile(r"(\w+)\s*=\s*(.*?);$", re.DOTALL | re.MULTILINE) regex = re.compile( r"""(?m) # multiline mode \b # word boundary before variable name (\w+)\s*=\s* # variable name and assignment ( (?: "(?:[^"\\]|\\.)*" # double-quoted string | '(?:[^'\\]|\\.)*' # single-quoted string | [^;]*? # or anything up to semicolon ) ) \s*; # semicolon ends the assignment """, re.VERBOSE, ) for match in regex.finditer(content): key, value = match.groups() # Clean value from the array indexes value = re.sub(r"\[\d+\]", "", value) # Try JSON try: js_variables[key] = json.loads(value.encode().decode("utf-8-sig")) continue except json.JSONDecodeError: pass try: # Try python literal eval js_variables[key] = ast.literal_eval(value) continue except (ValueError, SyntaxError): pass # Clean the value of quotes if it starts and ends with them if (value.startswith("'") and value.endswith("'")) or ( value.startswith('"') and value.endswith('"') ): value = value[1:-1] js_variables[key] = value # Return the JS variables return js_variables @clean_input def read_json_content(content: str | None, **kwargs: Any) -> dict[str, Any]: """Get the json content.""" if not content: return {} # Random control characters to avoid json errors for symbol in RANDOM_SYMBOLS: content = content.replace(symbol, "") # Handle missing values in JSON content = re.sub(r"\s*,\s*,", ", ", content) content = re.sub(r"^\s*{\s*,", "{", content) content = re.sub(r",\s*}\s*$", "}", content) # Handle keys without values content = re.sub(r":\s*(,|\})", ": null\\1", content) # Return the json content try: json_data = json.loads(content.encode().decode("utf-8-sig")) if isinstance(json_data, dict): return json_data return {} except json.JSONDecodeError as ex: _LOGGER.error( "Unable to decode json content with exception `%s`.\ Please, copy this and fill in a bug report: %s", ex, content, ) return {} @clean_input def readable_mac(raw: str | None) -> bool: """Check if string is MAC address.""" return bool( isinstance(raw, str) and re.search( re.compile("^([0-9A-Fa-f]{2}[:-]){5}([0-9A-Fa-f]{2})$"), raw ) ) def read_units_as_base( converter: type[UnitConverterBase], units: Any, check_calls: ARCallableType | list[ARCallableType] | None = None, fallback_value: float = 0.0, ) -> ARCallableType: """Create a reader from the units to the base.""" if not isinstance(converter, type) or not issubclass( converter, UnitConverterBase ): raise TypeError("Converter must be a subclass of UnitConverterBase") # Convert single check call to a list if check_calls is not None and not isinstance(check_calls, list): check_calls = [check_calls] def reader(value: Any) -> float: """Read the value as a base unit.""" # Use the converter with explicit None on unsupported types fval = safe_float(value) # Convert to base if all the checks passed if isinstance(fval, float) and ( check_calls is None or all(call(fval) for call in check_calls) ): return converter.convert_to_base(fval, units) return fallback_value return reader def read_units_data_rate(units: UnitOfDataRate) -> ARCallableType: """Read data rate values from the specified unit type.""" return read_units_as_base( DataRateUnitConverter, units, is_non_negative, 0.0 )