Baselined from internal Repository

last_commit:d8c9ae7e52746700a28d583db0776b35dee61dbb

.gitignore

@@ -18,9 +18,13 @@ build/
 .idea/misc.xml
 
 data/temp.py
+data/boats.csv
 *.iml
 
 .idea/misc.xml
 .idea/gve_devnet_meraki_seamless_sea_ssid.iml
+data/temp.json
 data/temp.py
+data/test.csv
+data/webhook.json
 *.log

README.md

@@ -77,7 +77,7 @@ uvicorn app:app --log-level warning
 
 To run with Docker:
 ```script
-With Docker: docker-compose up
+docker-compose up
 ```
 
 # Screenshots
@@ -107,4 +107,4 @@ See our contributing guidelines [here](CONTRIBUTING.md)
 
 #### DISCLAIMER:
 <b>Please note:</b> This script is meant for demo purposes only. All tools/ scripts in this repo are released for use "AS IS" without any warranties of any kind, including, but not limited to their installation, use, or performance. Any use of these scripts and tools is at your own risk. There is no guarantee that they have been through thorough testing in a comparable environment and we are not responsible for any damage or data loss incurred with their use.
-You are responsible for reviewing and testing any scripts you run thoroughly before use in any non-testing environment.
+You are responsible for reviewing and testing any scripts you run thoroughly before use in any non-testing environment.

data/boats.csv

@@ -1,2 +1 @@
-mac_address,ssid_number,vlan
-BOAT_MAC_ADDRESS, BOAT_SSID_NUMBER, BOAT_VLAN_NUMBER
+SystemName,ssid_number,vlan

data/ms_120_device_list.csv

@@ -0,0 +1 @@
+Serial Number,Model,switch_port_connected,SystemName

src/meraki_seamless_sea_ssid/config.py

@@ -1,14 +1,21 @@
 import os
 import json
-from rich.console import Console
-from rich.table import Table
-
+import pathlib
 from dotenv import load_dotenv
+from pydantic import field_validator, model_validator, validator, root_validator
+from pydantic_settings import BaseSettings
+from typing import Optional, ClassVar, List, Dict, Any
+import re
+
+# Adjust the path to load the .env file from the project root.
+env_path = pathlib.Path(__file__).parents[2] / '.env'
+print(f'env path: {env_path}')  # for debugging
+load_dotenv(dotenv_path=env_path)
 
 load_dotenv()
 
 
-class Config:
+class Config(BaseSettings):
     """
     The EnvironmentManager class is responsible for loading and validating the necessary environment variables
     that the app relies on.
@@ -22,57 +29,45 @@ class Config:
     """
 
     # Construct the absolute path for dir
-    dir_path = os.path.dirname(os.path.realpath(__file__))
-
-    # Define mandatory environment variables
-    MANDATORY_ENV_VARS = ['MERAKI_API_KEY', 'MERAKI_BASE_URL']
-
-    # Construct the absolute path for boats.csv
-    CSV_FILE = os.path.join(dir_path, '..', '..', 'data', 'boats.csv')
-
-    MERAKI_API_KEY = os.getenv('MERAKI_API_KEY')
-    MERAKI_BASE_URL = os.getenv('MERAKI_BASE_URL')
-    MERAKI_NETWORK_ID = os.getenv('MERAKI_NETWORK_ID')
+    DIR_PATH: ClassVar[pathlib.Path] = pathlib.Path(__file__).parents[2]
+    ENV_FILE_PATH: ClassVar[str] = str(DIR_PATH / ".env")
+    BOATS_CSV: ClassVar[str] = str(DIR_PATH / "data" / "boats.csv")     # Construct the absolute path for boats.csv
+    CSV_DB: ClassVar[str] = str(DIR_PATH / "data" / "ms_120_device_list.csv")     # Construct the absolute path for boats.csv
+    BOAT_NAMES: ClassVar[str] = str(DIR_PATH / "data" / "test.csv")
+
+    # Meraki Integration settings
+    MERAKI_API_KEY: str
+    MERAKI_BASE_URL: str
+    MERAKI_NETWORK_ID: str
+    PSK: str
+    MERAKI_SSID_NAME: str
+
+    # FastAPI Settings
+    APP_NAME: Optional[str] = 'Update your app name in .env'
+    APP_VERSION: Optional[str] = 'Update your ap'
+    DEV_ENV: Optional[str] = 'development'
+    IS_PRODUCTION: bool = DEV_ENV.lower() == 'production'  # True if FLASK_ENV is "production," otherwise False
+    LOGGER_LEVEL: str = 'DEBUG'
+
+    _instance: ClassVar[Optional['Config']] = None
 
-    LOGGER_LEVEL = os.getenv('LOGGER_LEVEL', '').upper() or 'DEBUG'
-    MERAKI_SSID_NAME = os.getenv('MERAKI_SSID_NAME')
-    MY_PSK = os.getenv('MY_PSK')
-
-    APP_NAME = os.environ.get('APP_NAME') or "Update your APP_NAME in .env"
+    @classmethod
+    def get_instance(cls):
+        if cls._instance is None:
+            cls._instance = cls()
+        return cls._instance
 
-    # For Handling SOME_INTEGER with a default value and error check (for integer type).
-    # try:
-    #     SOME_INTEGER = int(os.getenv('SOME_INTEGER', '100'))  # Default to 100 seconds if left blank
-    # except ValueError:
-    #     TIMESPAN_IN_SECONDS = 100  # Default to 100 seconds if the provided value is invalid, so it won't break program.
+    @field_validator('MERAKI_BASE_URL', mode='before')
+    def validate_meraki_base_url(cls, v):
+        if not re.match(r'https://api.meraki\.com/api/v1', v):
+            raise ValueError('WEBEX_BASE_URL must be: https://api.meraki.com/api/v1/')
+        return v
 
+    @field_validator('MERAKI_API_KEY', mode='before')
+    def validate_api_key(cls, v):
+        if not v:
+            raise ValueError('MREAKI_API_KEY must not be empty')
+        return v
 
-    @classmethod
-    def handle_error(cls, error_message):
-        """Handles errors by printing an error message and exiting the program."""
-        console = Console()
-        console.print(f"[bold red]Error:[/bold red] {error_message}", highlight=False)
-        exit(1)
 
-    @classmethod
-    def validate_env_variables(cls):
-        missing_vars = []
-        console = Console()  # Instantiate a console object for rich
-
-        table = Table(title="Environment Variables")
-        table.add_column("Variable", justify="left", style="bright_white", width=30)
-        table.add_column("Value", style="bright_white", width=60)
-
-        for var_name, var_value in cls.__dict__.items():
-            if "os" in var_name or "__" in var_name or isinstance(var_value, classmethod) or var_name == 'MANDATORY_ENV_VARS':
-                continue
-            # Check if mandatory variables are set. if var_name in cls.MANDATORY_ENV_VARS and not var_value: Add variable to the table
-            table.add_row(var_name, str(var_value) if var_value not in [None, ""] else "Not Set")
-            if var_name in cls.MANDATORY_ENV_VARS and var_value in [None, ""]:
-                missing_vars.append(var_name)
-
-        # Display the table
-        console.print(table)
-
-        if missing_vars:
-            cls.handle_error(f"The following environment variables are not set: {', '.join(missing_vars)}")
+config = Config.get_instance()

src/meraki_seamless_sea_ssid/funcs.py

@@ -1,37 +1,115 @@
 import csv
 from rich.console import Console
 from rich.panel import Panel
-from config import Config
+from config import config
 import sys
+import json
+from logrr import lm
 
-# Constants; set in .env
-CSV_FILE = Config.CSV_FILE
 
-# Rich console
-console = Console()
+def read_csv_data(csv_file_path):
+    with open(csv_file_path, mode='r', newline='') as file:
+        reader = csv.DictReader(file)
+        return list(reader), reader.fieldnames
 
 
-def signal_handler(sig, frame):
-    # Implement any clean-up tasks that may be necessary, i.e. closing files or database connections.
-    print_exit_panel()
-    sys.exit(0)  # Exit the app gracefully
+def write_to_csv(csv_file_path, updated_rows, fieldnames):
+    """
+    Write updated data to the CSV file.
 
+    :param csv_file_path: Path to the CSV file.
+    :param updated_rows: List of updated rows to write.
+    :param fieldnames: Fieldnames for the CSV file.
+    """
+    with open(csv_file_path, mode='w', newline='') as file:
+        writer = csv.DictWriter(file, fieldnames=fieldnames)
+        writer.writeheader()
+        writer.writerows(updated_rows)
 
-def print_start_panel(app_name=Config.APP_NAME):
-    console.print(Panel.fit(f"[bold bright_green]{app_name}[/bold bright_green]"))
 
+def save_devices_to_csv(serial_numbers, ms_list):
+    """
+    Save selected device information to a CSV file using the write_to_csv function.
 
-def print_exit_panel():
-    console.print("\n")
-    console.print(Panel.fit("Shutting down...", title="[bright_red]Exit[/bright_red]"))
+    :param serial_numbers: A list of serial numbers of the selected devices.
+    :param ms_list: A list of all devices, each as a dictionary with "serial" and "model" keys.
+    """
+    headers = ["Serial Number", "Model", "switch_port_connected", "SystemName"]
+    csv_file_path = config.CSV_DB
 
+    # Prepare data to be written to CSV
+    updated_rows = []
+    for serial in serial_numbers:
+        for device in ms_list:
+            if device["serial"] == serial:
+                updated_rows.append({
+                    "Serial Number": device["serial"],
+                    "Model": device["model"],
+                    "switch_port_connected": False,  # Default to False
+                    "SystemName": "N/A"  # Default to N/A
+                })
 
-def find_boat_in_csv(mac_address):
-    with open(CSV_FILE, mode='r') as file:
-        reader = csv.DictReader(file)
-        for row in reader:
-            if row['mac_address'] == mac_address:
-                return row['ssid_number'], row['vlan']
-    return None, None
+    # Use write_to_csv function to write data
+    write_to_csv(csv_file_path, updated_rows, headers)
+    lm.tsp(f"Devices saved to {csv_file_path}")
+
+
+def check_serial_in_csv(device_serial):
+    """
+    Check if a given device serial number exists in the CSV file.
+
+    :param csv_file_path: Path to the CSV file.
+    :param device_serial: The device serial number to search for.
+    :return: True if the serial number is found in the CSV, False otherwise.
+    """
+    csv_file_path = config.CSV_DB
+    try:
+        with open(csv_file_path, mode="r", newline="") as file:
+            reader = csv.DictReader(file)
+            for row in reader:
+                if row["Serial Number"] == device_serial:
+                    return True
+    except FileNotFoundError:
+        print(f"CSV file not found at {csv_file_path}")
+
+    return False
 
 
+def update_csv_row(updated_rows, device_serial, switch_port_connected, system_name):
+    for row in updated_rows:
+        if row["Serial Number"] == device_serial:
+            row["switch_port_connected"] = switch_port_connected
+            row["SystemName"] = system_name
+            break
+
+
+def extract_system_name(response, port_number):
+    """
+    Extract the system name from the LLDP response for a specific port number.
+
+    :param response: The LLDP/CDP response data.
+    :param port_number: The port number to extract the system name from.
+    :return: The system name if found, otherwise "N/A".
+    """
+    ports_data = response.get("ports", {})
+    port_data = ports_data.get(str(port_number))
+
+    if port_data and "lldp" in port_data:
+        system_name = port_data["lldp"].get("systemName")
+        if system_name:
+            print(f"System name for port {port_number}: {system_name}")
+            return system_name
+    return "N/A"
+
+
+def check_boat_in_csv(boats_csv_path, system_name):
+    try:
+        with open(boats_csv_path, mode="r", newline="") as file:
+            reader = csv.DictReader(file)
+            for row in reader:
+                if row["SystemName"] == system_name:
+                    lm.tsp("System Name found in boats.csv", system_name)
+                    return row["ssid_number"], row["vlan"]
+    except FileNotFoundError:
+        print(f"Boats CSV file not found at {boats_csv_path}")
+    return None, None