Add pandas to sqlite conversion

app.py

@@ -1,36 +1,114 @@
+#####################################################
+####                 Imports                     ####
+#####################################################
+import os
 import tempfile
+from datetime import datetime
 
 import streamlit as st
 
 from config.log_definitions import log_definitions
 from utils.log2pandas import LogParser
+from utils.pandas2sql import Pandas2SQL
 
-st.title("Log Analyzer")
+#####################################################
+####              Interface Setup               ####
+#####################################################
 
-# Upload area by drag and drop
-uploaded_file = st.file_uploader("Drop your log file here")
+st.title("ShadowLog - Log File Analyzer")
+st.write("Upload a log file to analyze and/or convert it to SQLite")
 
-# Dropdown menu to choose the log type
+# File upload widget
+uploaded_file = st.file_uploader("Choose a log file")
 
-# Extract log types from the configuration file
+# Get available log types from log_definitions
 log_types = list(log_definitions.keys())
+# Set default log type if not already in session state
+if "log_type" not in st.session_state:
+    st.session_state.log_type = log_types[0]  # Default to first log type
 
-log_type = st.selectbox("Select log type", options=log_types)
-
-# Analyze button
-if st.button("Analyze"):
-    if uploaded_file is not None:
-        # Temporarily save the uploaded file
-        with tempfile.NamedTemporaryFile(delete=False, suffix=".log") as tmp_file:
-            tmp_file.write(uploaded_file.getbuffer())
-            tmp_file_path = tmp_file.name
-
-        # Create an instance of LogParser with the temporary path and log type
-        parser = LogParser(tmp_file_path, log_type)
-        # Parse the file and get the DataFrame
-        parsed_df = parser.parse_file()
-        # Display the first rows of the resulting DataFrame
-        st.write("Resulting DataFrame:")
-        st.dataframe(parsed_df)
-    else:
-        st.error("Please upload a log file.")
+st.session_state.log_type = st.selectbox(
+    "Select log type", log_types, index=log_types.index(st.session_state.log_type)
+)
+
+# Store the parsed dataframe in the session state
+if "parsed_df" not in st.session_state:
+    st.session_state.parsed_df = None
+
+if uploaded_file is not None:
+    # Create two columns for the buttons
+    col1, col2 = st.columns(2)
+
+    with col1:
+        # Button to parse the log file
+        if st.button("Parse the log file"):
+            with st.spinner("Analyzing the file..."):
+                # Create a temporary file
+                with tempfile.NamedTemporaryFile(
+                    delete=False, suffix=".log"
+                ) as tmp_file:
+                    tmp_file.write(uploaded_file.getvalue())
+                    tmp_path = tmp_file.name
+
+                try:
+                    # Parse the log file
+                    parser = LogParser(tmp_path, st.session_state.log_type)
+                    st.session_state.parsed_df = parser.parse_file()
+
+                    # Display a success message and the dataframe
+                    st.success("Log file successfully analyzed!")
+                    # st.dataframe(st.session_state.parsed_df)
+                except Exception as e:
+                    st.error(f"Error analyzing the file: {e}")
+                finally:
+                    # Clean up the temporary file
+                    os.unlink(tmp_path)
+
+    with col2:
+        # Button to convert to SQLite and download
+        if st.button("Convert to SQLite"):
+            if st.session_state.parsed_df is not None:
+                with st.spinner("Converting to SQLite..."):
+                    try:
+                        # Create a temporary SQLite file
+                        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+                        sqlite_path = os.path.join(
+                            tempfile.gettempdir(), f"log_data_{timestamp}.sqlite"
+                        )
+
+                        # Create the SQL converter
+                        sql_converter = Pandas2SQL(sqlite_path)
+
+                        # Convert the dataframe to SQLite
+                        sql_converter.create_table(
+                            st.session_state.parsed_df, st.session_state.log_type
+                        )
+
+                        # Read the SQLite file for download
+                        with open(sqlite_path, "rb") as file:
+                            sqlite_data = file.read()
+
+                        # Success message and immediate download
+                        st.success("SQLite file created successfully!")
+
+                        # Download button
+                        st.download_button(
+                            label="Download SQLite file",
+                            data=sqlite_data,
+                            file_name=f"log_file_{st.session_state.log_type}_{timestamp}.sqlite",
+                            mime="application/octet-stream",
+                            key="auto_download",
+                        )
+                    except Exception as e:
+                        st.error(f"Error converting to SQLite: {e}")
+                    finally:
+                        # Clean up the temporary file
+                        if os.path.exists(sqlite_path):
+                            os.unlink(sqlite_path)
+            else:
+                st.warning("Please parse the log file first.")
+
+# Display the dataframe if available
+if st.session_state.parsed_df is not None:
+    st.subheader("Analyzed log data")
+    st.dataframe(st.session_state.parsed_df)

utils/pandas2sql.py

@@ -0,0 +1,88 @@
+import sqlite3
+
+import pandas as pd
+
+
+class Pandas2SQL:
+    """
+    Classe pour convertir un DataFrame pandas en table SQLite
+    avec détection automatique des types de colonnes.
+    """
+
+    def __init__(self, db_path=":memory:"):
+        """
+        Initialise la connexion à la base de données SQLite
+
+        Args:
+            db_path (str): Chemin vers le fichier de base de données SQLite
+                           (par défaut utilise une base de données en mémoire)
+        """
+        self.db_path = db_path
+
+    def _get_sqlite_type(self, pandas_dtype):
+        """
+        Convertit un type pandas en type SQLite approprié
+
+        Args:
+            pandas_dtype: Type pandas
+
+        Returns:
+            str: Type SQLite correspondant
+        """
+        if pd.api.types.is_integer_dtype(pandas_dtype):
+            return "INTEGER"
+        elif pd.api.types.is_float_dtype(pandas_dtype):
+            return "REAL"
+        elif pd.api.types.is_bool_dtype(pandas_dtype):
+            return "INTEGER"  # SQLite n'a pas de type booléen, utilise INTEGER (0/1)
+        elif pd.api.types.is_datetime64_dtype(pandas_dtype):
+            return "TIMESTAMP"
+        else:
+            return "TEXT"  # Pour les types object, string, category, etc.
+
+    def create_table(self, df, table_name, if_exists="replace", primary_key=None):
+        """
+        Crée une table SQLite basée sur un DataFrame pandas
+
+        Args:
+            df (pandas.DataFrame): DataFrame à convertir
+            table_name (str): Nom de la table à créer
+            if_exists (str): Action si la table existe ('fail', 'replace', 'append')
+            primary_key (str): Nom de la colonne à définir comme clé primaire (optionnel)
+        """
+        # Création du schéma de table basé sur les types de colonnes
+        columns = []
+        for col_name, dtype in df.dtypes.items():
+            sqlite_type = self._get_sqlite_type(dtype)
+            col_def = f'"{col_name}" {sqlite_type}'
+            if primary_key and col_name == primary_key:
+                col_def += " PRIMARY KEY"
+            columns.append(col_def)
+
+        # Création de la requête SQL
+        create_query = f'CREATE TABLE "{table_name}" ({", ".join(columns)})'
+
+        # Connexion et création de la table
+        conn = sqlite3.connect(self.db_path)
+        cursor = conn.cursor()
+
+        try:
+            if if_exists == "replace":
+                cursor.execute(f'DROP TABLE IF EXISTS "{table_name}"')
+            elif if_exists == "fail":
+                cursor.execute(
+                    f'SELECT name FROM sqlite_master WHERE type="table" AND name="{table_name}"'
+                )
+                if cursor.fetchone():
+                    raise ValueError(f"La table '{table_name}' existe déjà.")
+
+            cursor.execute(create_query)
+
+            # Insertion des données
+            df.to_sql(table_name, conn, if_exists="append", index=False)
+            conn.commit()
+        except Exception as e:
+            conn.rollback()
+            raise e
+        finally:
+            conn.close()