Futurebot/newfetch.py

from msal import ConfidentialClientApplication
import requests
import csv
from typing import List, Dict, Any, Union

# Define your Azure AD and API configurations
TENANT_ID = '82360567-02b6-4a45451f-a4z6-910a811b8131'
CLIENT_ID = 'a3d88cc7-8889-42356548-bd3b-f5689372f07084'
CLIENT_SECRET = 'ibh8Q~LVSzZeeewdeWF546456UHNSoUBP.lz8GBmenXGJTlnbdU'
SCOPE = ['api://a3d88cc7-0789-4238-bd3hb-456456f89667372f07084/.default']

# Define API endpoint configurations
data_endpoint_config = {
    'projects': {
        'fields': 'projects?select=id,name,route,imeis,subline,ws10Icon,routeSecondLayer',
        'order': '&order=name.asc&limit=10',
    },
    'latestDeviceData': {
        'fields': 'devices',
        'order': '&order=name.desc',
    },
    'data': {
        'fields': 'data?select=id,version,imei,timestamp,network,data,project,location,latest',
        'order': '&order=timestamp.desc&limit=25',
    },
}

# Function to retrieve access token using MSAL
def get_access_token(tenant_id: str, client_id: str, client_secret: str, scope: List[str]) -> Union[str, None]:
    authority = f"https://login.microsoftonline.com/{tenant_id}"
    app = ConfidentialClientApplication(
        client_id,
        authority=authority,
        client_credential=client_secret,
    )
    result = app.acquire_token_for_client(scopes=scope)
    if 'access_token' in result:
        return result['access_token']
    else:
        print('Failed to retrieve access token:', result.get('error_description', 'Unknown error'))
        return None

# Function to call API with access token
def call_api_data(access_token: str, query: str) -> Union[Dict[str, Any], List[Dict[str, Any]]]:
    headers = {
        'Authorization': f'Bearer {access_token}',
        'Content-Type': 'application/json',
    }
    url = f"https://api.vodafone.dev/iot-db/{query}"

    print(f"Requesting URL: {url}")  # Debug: Print URL
    print(f"Request Headers: {headers}")  # Debug: Print headers

    try:
        response = requests.get(url, headers=headers)
        response.raise_for_status()  # Raise an error for bad status codes

        return response.json()
    except requests.exceptions.HTTPError as http_err:
        print(f"HTTP error occurred: {http_err}")
        print("Response content:", response.text)  # Print full response content
    except Exception as err:
        print(f"Other error occurred: {err}")

    return {}  # Return empty dictionary if request fails

# Function to get device data
def get_device_data(query_name: str) -> Union[List[Dict[str, Any]], Dict[str, Any]]:
    try:
        access_token = get_access_token(TENANT_ID, CLIENT_ID, CLIENT_SECRET, SCOPE)
        if access_token:
            query_config = data_endpoint_config.get(query_name, {})
            query_fields = query_config.get('fields', '')
            query_order = query_config.get('order', '')
            query = f"{query_fields}{query_order}"  # Combine fields and order
            if query:
                return call_api_data(access_token, query)
            else:
                print(f'No query found for query name: {query_name}')
        else:
            print('Failed to retrieve access token.')
    except Exception as e:
        print(f'Error retrieving data: {str(e)}')

    return []

# Function to convert data to CSV
def write_data_to_csv(data: List[Dict[str, Any]], filename: str):
    if not data:
        print("No data to write to CSV.")
        return

    # Extract header from the first data item keys
    headers = data[0].keys()

    try:
        with open(filename, 'w', newline='', encoding='utf-8') as csvfile:
            writer = csv.DictWriter(csvfile, fieldnames=headers)
            writer.writeheader()
            writer.writerows(data)
        print(f"Data successfully written to {filename}")
    except Exception as e:
        print(f"Error writing to CSV: {e}")

# Fetch latest device data and write to CSV
if __name__ == '__main__':
    query_name = 'data'  # Example query name (replace with actual query name)
    device_data = get_device_data(query_name)
    
    # Check if device_data is in the expected format
    if isinstance(device_data, dict):
        # If data is returned as a dictionary, try to get the list of items
        device_data = device_data.get('items', [])

    # Convert to CSV
    write_data_to_csv(device_data, 'device_data.csv')