lmb-fe/functions.py

import json
from typing import Any, Dict, List, Union
import requests
import pandas as pd


def remove_empty_values(data: Any) -> Any:
    """
    Rimuove ricorsivamente tutte le chiavi con valori vuoti (stringa vuota, None, False, liste/dict vuoti).
    
    Args:
        data: Il dato da pulire (può essere dict, list o valore primitivo)
    
    Returns:
        Il dato pulito senza valori vuoti
    """
    if isinstance(data, dict):
        # Filtra il dizionario rimuovendo chiavi con valori vuoti
        cleaned = {}
        for key, value in data.items():
            # Ricorsivamente pulisci il valore
            cleaned_value = remove_empty_values(value)
            
            # Aggiungi solo se il valore non è vuoto
            if cleaned_value not in [None, "", [], {}, False]:
                cleaned[key] = cleaned_value
                
        return cleaned if cleaned else None
    
    elif isinstance(data, list):
        # Pulisci ogni elemento della lista
        cleaned = []
        for item in data:
            cleaned_item = remove_empty_values(item)
            if cleaned_item not in [None, "", [], {}, False]:
                cleaned.append(cleaned_item)
        return cleaned if cleaned else None
    
    else:
        # Per valori primitivi, ritorna il valore se non è vuoto
        if data in [None, "", False]:
            return None
        return data


def transform_labels_to_keys(data: Any) -> Any:
    """
    Trasforma ricorsivamente le 'label' in chiavi del dizionario.
    
    Args:
        data: Il dato da trasformare
    
    Returns:
        Il dato trasformato con le label come chiavi
    """
    if isinstance(data, dict):
        # Se c'è una label, usala come chiave principale
        if 'label' in data:
            label = data['label']
            # Crea una copia del dizionario senza la label
            new_dict = {k: v for k, v in data.items() if k != 'label'}
            
            # Se ci sono subsections, trasformale ricorsivamente
            if 'subsections' in new_dict:
                subsections = new_dict.pop('subsections')
                # Trasforma ogni subsection
                for subsection in subsections:
                    transformed_sub = transform_labels_to_keys(subsection)
                    if isinstance(transformed_sub, dict):
                        new_dict.update(transformed_sub)
            
            # Trasforma ricorsivamente tutti gli altri valori
            for key, value in list(new_dict.items()):
                new_dict[key] = transform_labels_to_keys(value)
            
            # Ritorna con la label come chiave principale
            return {label: new_dict if new_dict else {}}
        
        else:
            # Se non c'è label, trasforma ricorsivamente tutti i valori
            result = {}
            for key, value in data.items():
                if key == 'subsections' and isinstance(value, list):
                    # Gestisci le subsections senza label nel dizionario padre
                    for subsection in value:
                        transformed = transform_labels_to_keys(subsection)
                        if isinstance(transformed, dict):
                            result.update(transformed)
                elif key == 'sections' and isinstance(value, list):
                    # Gestisci le sections trasformandole in un dizionario
                    sections_dict = {}
                    for section in value:
                        transformed = transform_labels_to_keys(section)
                        if isinstance(transformed, dict):
                            sections_dict.update(transformed)
                    result[key] = sections_dict
                else:
                    result[key] = transform_labels_to_keys(value)
            return result
    
    elif isinstance(data, list):
        # Trasforma ogni elemento della lista
        return [transform_labels_to_keys(item) for item in data]
    
    else:
        # Ritorna il valore così com'è
        return data


def clean_and_transform_json(json_data: Union[str, dict], keep_false: bool = False) -> dict:
    """
    Funzione principale che pulisce e trasforma il JSON.
    
    Args:
        json_data: Il JSON come stringa o dizionario
        keep_false: Se True, mantiene i valori False (default: False li rimuove)
    
    Returns:
        Il JSON pulito e trasformato
    """
    # Se è una stringa, parsala
    if isinstance(json_data, str):
        data = json.loads(json_data)
    else:
        data = json_data
    
    # Step 1: Rimuovi i valori vuoti
    if not keep_false:
        cleaned_data = remove_empty_values(data)
    else:
        # Versione alternativa che mantiene False
        def remove_empty_keep_false(d):
            if isinstance(d, dict):
                cleaned = {}
                for key, value in d.items():
                    cleaned_value = remove_empty_keep_false(value)
                    if cleaned_value not in [None, "", [], {}]:
                        cleaned[key] = cleaned_value
                return cleaned if cleaned else None
            elif isinstance(d, list):
                cleaned = []
                for item in d:
                    cleaned_item = remove_empty_keep_false(item)
                    if cleaned_item not in [None, "", [], {}]:
                        cleaned.append(cleaned_item)
                return cleaned if cleaned else None
            else:
                if d in [None, ""]:
                    return None
                return d
        
        cleaned_data = remove_empty_keep_false(data)
    
    # Step 2: Trasforma le label in chiavi
    if cleaned_data:
        transformed_data = transform_labels_to_keys(cleaned_data)
    else:
        transformed_data = {}
    
    return transformed_data


def api_req(query: str) -> Dict[str, Any]:
    url = 'https://api.cosmoguard.it/api/v1/echa/search'
    response = requests.post(url, json={'cas': query})
    data = response.json()
    if data['success'] == True:
        results = data['data']
        if results:
            substance_info = data['data']['substance']
            last_update = data['data']['dossier_info']['lastUpdatedDate']
            tox = data['data']['index']['toxicological_information_link']
            rpt = data['data']['index']['repeated_dose_toxicity_link']
            act = data['data']['index']['acute_toxicity_link']
            toxicological_information = data['data']['toxicological_information']
            repeated_dose_toxicity = data['data']['repeated_dose_toxicity']
            acute_toxicity = data['data']['acute_toxicity']
            return {
                'substance_info': substance_info,
                'last_update': last_update,
                'toxicological_information': toxicological_information,
                'repeated_dose_toxicity': repeated_dose_toxicity,
                'acute_toxicity': acute_toxicity,
                'link': {
                    'toxicological_information': tox,
                    'repeated_dose_toxicity': rpt,
                    'acute_toxicity': act
                }
            }
        else:
            return {'error': 'No results found for the given CAS number.'}
    else:
        return data['error']