# pages/users.py import os import sys import solara import solara import dataclasses from typing import Optional, cast import solara import solara.lab import pandas as pd import json import pandas as pd from psycopg2 import pool # Get the project root directory project_root = os.path.dirname(os.path.dirname(os.path.realpath(__file__))) print(project_root) module_path = os.path.join(project_root, 'public') print(module_path) # Add the module path to the Python path sys.path.append(module_path) from auth import * key_content = os.environ.get('DATABASE_SECRETS') key_dict = json.loads(key_content) connection_pool = pool.SimpleConnectionPool(1, 10, **key_dict) import crud import status import dataviz def get_columns(): conn = connection_pool.getconn() with conn: with conn.cursor() as cur: cur.execute("SELECT column_name FROM information_schema.columns WHERE table_name = 'clientes_test';") columns = [col[0] for col in cur.fetchall()] connection_pool.putconn(conn) return columns def fetch_data(): conn = connection_pool.getconn() with conn: with conn.cursor() as cur: cur.execute("SELECT * FROM clientes_test;") data = cur.fetchall() connection_pool.putconn(conn) df = pd.DataFrame(data, columns=get_columns()) return df # Reactive states for the filter selected_filter = solara.reactive("Choose a filter") filter_value = solara.reactive("") @solara.component def Page(): assert user.value is not None #### Data #### columns = get_columns() inputs = {col: solara.use_state('') for col in columns} selected_row, set_selected_row = solara.use_state(None) data, set_data = solara.use_state(fetch_data()) # List of available filters filters = ['cod_cliente', 'doc_identidad','nombre', 'apellido', 'departamento', 'distrito', 'cta_transaccion', 'feching', 'fechact'] # Initialize filtered_data as an empty DataFrame and in a reactive state filtered_data, set_filtered_data = solara.use_state_or_update(data) # Reactive state for filters selected_filter, set_selected_filter = solara.use_state(filters[0]) # Default to the first filter filter_value, set_filter_value = solara.use_state("") # 0022-3344-5566-7788 1212-1639-5551-4075 4416-3921-4905-2757 def apply_dropdown_filter(): # Start with the original data filtered_data_temp = data # Remove whitespace and convert to lowercase cleaned_filter_value = filter_value.strip().lower() # Filter data based on the selected filter and filter value if cleaned_filter_value: # Convert column values to string and lowercase for comparison filtered_data_temp = filtered_data_temp[filtered_data_temp[selected_filter].astype(str).str.lower() == cleaned_filter_value] filtered_data_temp = filtered_data_temp.reset_index(drop=True) # Update the filtered_data reactive state with the filtered data set_filtered_data(filtered_data_temp) def reset_filter(): set_filtered_data(data) set_filter_value("") #### Functions to interact with AWS database #### def handle_submit(): conn = connection_pool.getconn() with conn: with conn.cursor() as cur: columns_str = ", ".join(columns[1:]) # Exclude 'id' since it's auto-incremented values_str = ", ".join(["%s"] * len(columns[1:])) cur.execute(f"INSERT INTO clientes_test ({columns_str}) VALUES ({values_str});", [inputs[col][0] for col in columns[1:]]) conn.commit() connection_pool.putconn(conn) set_data(fetch_data()) def handle_update(): conn = connection_pool.getconn() with conn: with conn.cursor() as cur: if selected_row is not None: set_str = ", ".join([f"{col} = %s" for col in columns[1:]]) cur.execute(f"UPDATE clientes_test SET {set_str} WHERE id = {selected_row};", [inputs[col][0] for col in columns[1:]]) conn.commit() connection_pool.putconn(conn) set_data(fetch_data()) def handle_delete(): conn = connection_pool.getconn() with conn: with conn.cursor() as cur: if selected_row is not None: cur.execute(f"DELETE FROM clientes_test WHERE id = {selected_row};") conn.commit() connection_pool.putconn(conn) set_data(fetch_data()) def handle_row_selection(row_index): conn = connection_pool.getconn() with conn: with conn.cursor() as cur: cur.execute("SELECT * FROM clientes_test WHERE id = %s;", (row_index,)) row_data = cur.fetchone() for idx, col in enumerate(columns): inputs[col][1](str(row_data[idx])) connection_pool.putconn(conn) #### Statistics for charts #### option, set_option = solara.use_state_or_update("Total") mouseover_data, set_mouseover_data = solara.use_state_or_update(None) mouseout_data, set_mouseout_data = solara.use_state_or_update(None) # Statistics calculations total_customers = len(data) unique_razon_social = data['razon_social'].nunique() cust_by_dept = data.groupby(['departamento']).size().reset_index(name='count') cust_by_district = data.groupby([ 'distrito']).size().reset_index(name='count') cust_by_bank = data['banco'].value_counts() data['date_column_name'] = pd.to_datetime(data['feching']) data['year'] = data['date_column_name'].dt.year data['month'] = data['date_column_name'].dt.month cust_by_year_month = data.groupby(['year', 'month']).size().reset_index(name='count') # Visualization options for ECharts options = { "Total": { "title": {"text": "Numero De Clientes Total"}, "tooltip": {}, "xAxis": {'type':'category', 'data': ['Total']}, "yAxis":{'type': 'value'}, "series": [ { "name": "Total", "type": "bar", 'color': '#61a0a8', "data": [total_customers], "universalTransition": True, } ], }, "Departamento": { "title": {"text": "Numero De Clientes Por Departamento"}, "tooltip": {}, "xAxis": {'type':'category', 'data': cust_by_dept['departamento'].tolist()}, "yAxis":{'type': 'value'}, "series": [ { "name": "Departamento", "type": "bar", "data": cust_by_dept['count'].tolist(), "universalTransition": True, } ], }, "Distrito": { "title": {"text": "Numero De Clientes Por Distrito"}, "tooltip": {}, "xAxis": {'type':'category', 'data': cust_by_district['distrito'].tolist()}, "yAxis":{'type': 'value'}, "series": [ { "name": "Distrito", "type": "bar", "data": cust_by_district['count'].tolist(), "universalTransition": True, } ], } } with solara.Column(): with solara.Card('Statistics', margin= 20): with solara.ToggleButtonsSingle("Total", on_value=set_option): solara.Button("Total") solara.Button("Departamento") solara.Button("Distrito") solara.FigureEcharts(option=options[option], on_mouseover=mouseover_data, on_mouseout=mouseout_data) # Place buttons in a horizontal row below the map with solara.Row(): solara.Button("Submit", on_click=lambda: [handle_submit(), set_data(fetch_data())]) solara.Button("Update", on_click=lambda: [handle_update(), set_data(fetch_data())]) solara.Button("Delete", on_click=lambda: [handle_delete(), set_data(fetch_data())]) solara.Button("Clear", on_click=lambda: [inputs[col][1]('') for col in columns]) # Dropdown Filter UI with solara.Row(): solara.Select(label="Select Filter", value=selected_filter, values=filters, on_value=set_selected_filter) solara.InputText(label="Filter Value", value=filter_value, on_value=set_filter_value) solara.Button("Apply Filter", on_click=apply_dropdown_filter) solara.Button("Reset", on_click=reset_filter) # Input Fields and DataFrame with solara.Columns(widths=[1, 3]): with solara.Column(): # Input fields for data entry for col in columns: solara.InputText(col, value=inputs[col][0], on_value=inputs[col][1]) # Dropdown for row selection if data is not empty if not data.empty: solara.Select(label="Select a row to update", values=[row[0] for row in data.itertuples()], value=selected_row, on_value=handle_row_selection) # DataFrame with filters applied with solara.Row(): solara.CrossFilterDataFrame(filtered_data, items_per_page=50) # Banner image at the top with solara.Row(): image_url = "https://huggingface.co/spaces/romero61/hendata/resolve/main/public/logo.png" solara.Image(image_url, width="100%")