import json import logging import requests from flask import Flask, request from .settings import * logging.basicConfig(level=logging.INFO) class EagleEyev3(): def __init__(self): self.client_id = None self.client_secret = None self.access_token = None self.refresh_token = None self.redirect_uri = None self._load_vars_from_settings() self.user_base_url = None self.current_user = None self.users = [] self.bridges = [] self.cameras = [] self.users = [] self.accounts = [] self.base_url = None def _load_vars_from_settings(self): """Load variables from the settings module.""" self.client_id = settings.client_id self.client_secret = settings.client_secret self.server_host = settings.server_host self.server_port = settings.server_port # Combine server_protocol, server_host, and server_port to make the redirect_uri # Note: Please see the note in settings.py about trailing slashes and modify this line if needed self.redirect_uri = f"{settings.server_protocol}://{settings.server_host}:{settings.server_port}" def login_tokens(self, code=None, cascade=True): """Obtains login tokens using the authorization code. Args: code (str): The authorization code. cascade (bool): Indicates whether to cascade and get the base URL and current user information. Returns: dict: Dictionary containing the success status, response HTTP status code, data, and current user information. """ baseUrl = "https://auth.eagleeyenetworks.com/oauth2/token" pathUrl = f"?grant_type=authorization_code&scope=vms.all&code={code}&redirect_uri={self.redirect_uri}" # Note the trailing slash, make sure it matches the whitelist url = baseUrl + pathUrl # Send a POST request to obtain login tokens response = requests.post(url, auth=(self.client_id, self.client_secret)) response_json = response.json() logging.info(f"{response.status_code} in login_tokens") if response.status_code == 200: success = True self.access_token = response_json['access_token'] self.refresh_token = response_json['refresh_token'] if cascade: self.get_base_url() else: success = False return { "success": success, "response_http_status": response.status_code, "data": response_json, 'current_user': self.current_user } def get_base_url(self, cascade=True): """Obtains the base URL for the user. Args: cascade (bool): Indicates whether to cascade and get the current user information. Returns: dict: Dictionary containing the success status, response HTTP status code, and data. """ url = "https://api.eagleeyenetworks.com/api/v3.0/clientSettings" headers = { "Authorization": f"Bearer {self.access_token}", "Accept": "application/json" } # Send a GET request to obtain the base URL response = requests.get(url, headers=headers) response_json = response.json() logging.info(f"{response.status_code} in get_base_url") if response.status_code == 200: success = True if 'httpsBaseUrl' in response_json and 'hostname' in response_json['httpsBaseUrl']: self.user_base_url = response_json['httpsBaseUrl']['hostname'] if cascade: self.get_current_user() else: success = False return { "success": success, "response_http_status": response.status_code, "data": response_json } def get_current_user(self): """Obtains the information of the current user. Returns: dict: Dictionary containing the success status, response HTTP status code, and data. """ url = f"https://{self.user_base_url}/api/v3.0/users/self" headers = { "Authorization": f"Bearer {self.access_token}", "Accept": "application/json" } # Send a GET request to obtain the current user information response = requests.get(url, headers=headers) response_json = response.json() logging.info(f"{response.status_code} in get_current_user") if response.status_code == 200: success = True self.current_user = response_json else: success = False return { "success": success, "response_http_status": response.status_code, "data": response_json }