⚡ ChargeNode Data Manager
Inloggad för uppdatering av företagsdata
# Spar knapp istället! import streamlit as st import gspread from google.oauth2.service_account import Credentials import pandas as pd from datetime import datetime import pytz import json import os import time import io from openpyxl import Workbook from openpyxl.styles import Font, PatternFill, Alignment from openpyxl.utils.dataframe import dataframe_to_rows # Konfigurera sidan st.set_page_config( page_title="ChargeNode Data Manager", page_icon="⚡", layout="wide", initial_sidebar_state="collapsed" ) # Google Sheets setup SCOPES = [ 'https://www.googleapis.com/auth/spreadsheets', 'https://www.googleapis.com/auth/drive' ] # Rate limiting configuration MAX_REQUESTS_PER_MINUTE = 60 REQUEST_DELAY = 1.0 # Svensk tidszon SWEDISH_TZ = pytz.timezone('Europe/Stockholm') # Admin password from Hugging Face secret ADMIN_PASSWORD = os.environ.get('ADMIN') # UPPDATERADE KOLUMNER - Exakt enligt Google Sheet # Readonly kolumner (ej editerbara) READONLY_COLUMNS = [ 'Lösenord', 'Filter', 'Områdeskod', 'Account ID', 'Företagsnamn', 'Fastighetsnamn', 'Address', 'Stad', 'Metod', 'Max kW', '# Uttag', '# F IDs', 'Latitud', 'Longitud', 'Hårdvara', 'Tid dek', 'Bil', 'Vecka', 'Indikativt datum', 'Säljare', 'Status', # Readonly - ej editerbar 'Last update', # Automatiskt genererad 'Editor', 'KAM' # Automatiskt genererad ] # Kolumner som är editerbara EDITABLE_COLUMNS = [ 'Namn', 'Email adress', 'Telefon', 'Tillgänglighet', 'Access info', 'Fritext meddelande', 'Main Contact' ] # Kolumner som ska döljas helt från visningen - UPPDATERAD LISTA HIDDEN_COLUMNS = [ 'Lösenord', 'Filter', 'Tid dek', 'Bil', 'Website', 'Områdeskod', 'Account ID', 'Företagsnamn', 'Latitud', 'Longitud', '# F IDs', 'Max kW', 'Länk till fysisk plats', 'Uppdatera Lat, Long', 'KAM' ] # ============================================================================ # MONITORING FUNCTIONS - NIVÅ 1 # ============================================================================ class LogSheets: """Class för att hantera loggning till Google Sheets""" def __init__(self, spreadsheet): """ Initialisera LogSheets med ett spreadsheet-objekt Args: spreadsheet: gspread Spreadsheet objekt för "Omrade_updater" """ self.spreadsheet = spreadsheet self.logs_sheet = None self.perf_sheet = None self._init_sheets() def _init_sheets(self): """Initiera eller skapa logg-sheets""" try: # Försök hitta eller skapa LOGS sheet try: self.logs_sheet = self.spreadsheet.worksheet("Omrade_updater_LOGS") print("[MONITORING] Logs sheet hittades") except gspread.exceptions.WorksheetNotFound: print("[MONITORING] Skapar nytt Logs sheet...") self.logs_sheet = self.spreadsheet.add_worksheet( title="Omrade_updater_LOGS", rows=1000, cols=10 ) # Lägg till headers headers = [ 'Timestamp', 'Event Type', 'User ID', 'Company Name', 'Details', 'Row Number', 'Column', 'Old Value', 'New Value', 'Status' ] self.logs_sheet.append_row(headers) print("[MONITORING] Logs sheet skapat med headers") # Försök hitta eller skapa Performance sheet (valfritt) try: self.perf_sheet = self.spreadsheet.worksheet("Performance_Log") print("[MONITORING] Performance sheet hittades") except gspread.exceptions.WorksheetNotFound: print("[MONITORING] Skapar nytt Performance sheet...") self.perf_sheet = self.spreadsheet.add_worksheet( title="Performance_Log", rows=1000, cols=6 ) # Lägg till headers headers = [ 'Timestamp', 'Operation', 'Duration (s)', 'User ID', 'Details', 'Status' ] self.perf_sheet.append_row(headers) print("[MONITORING] Performance sheet skapat med headers") except Exception as e: print(f"[MONITORING ERROR] Kunde inte initiera sheets: {e}") self.logs_sheet = None self.perf_sheet = None def log_activity(self, event_type, user_id, company_name="", details="", row_number="", column="", old_value="", new_value="", status="SUCCESS"): """ Logga en aktivitet till LOGS sheet Args: event_type: Typ av händelse (LOGIN, LOGOUT, EDIT, DOWNLOAD, UPLOAD) user_id: Användarens ID (Account ID) company_name: Företagsnamn details: Detaljer om händelsen row_number: Radnummer (för editeringar) column: Kolumnnamn (för editeringar) old_value: Gammalt värde (för editeringar) new_value: Nytt värde (för editeringar) status: Status (SUCCESS, ERROR, WARNING) """ if not self.logs_sheet: return False try: timestamp = datetime.now(SWEDISH_TZ).strftime('%Y-%m-%d %H:%M:%S') row_data = [ timestamp, event_type, str(user_id), str(company_name), str(details), str(row_number), str(column), str(old_value)[:100], # Begränsa längd str(new_value)[:100], # Begränsa längd status ] self.logs_sheet.append_row(row_data) print(f"[MONITORING] Loggade: {event_type} för {user_id}") return True except Exception as e: print(f"[MONITORING ERROR] Kunde inte logga aktivitet: {e}") return False def log_performance(self, operation, duration, user_id="", details="", status="SUCCESS"): """ Logga performance-metrics till Performance_Log sheet Args: operation: Operation som mättes (LOAD_DATA, SAVE_CHANGES, etc.) duration: Tid i sekunder user_id: Användarens ID details: Extra detaljer status: Status """ if not self.perf_sheet: return False try: timestamp = datetime.now(SWEDISH_TZ).strftime('%Y-%m-%d %H:%M:%S') row_data = [ timestamp, operation, f"{duration:.2f}", str(user_id), str(details), status ] self.perf_sheet.append_row(row_data) print(f"[MONITORING] Loggade performance: {operation} ({duration:.2f}s)") return True except Exception as e: print(f"[MONITORING ERROR] Kunde inte logga performance: {e}") return False def safe_log_activity(log_sheets, event_type, user_id, **kwargs): """ Säker wrapper för loggning som aldrig kraschar appen Args: log_sheets: LogSheets instans event_type: Typ av händelse user_id: Användarens ID **kwargs: Extra parametrar att skicka till log_activity """ try: if log_sheets: log_sheets.log_activity(event_type, user_id, **kwargs) except Exception as e: print(f"[MONITORING ERROR] Safe log failed: {e}") # Fortsätt utan att krascha def safe_log_performance(log_sheets, operation, duration, **kwargs): """ Säker wrapper för performance-loggning som aldrig kraschar appen Args: log_sheets: LogSheets instans operation: Operation som mättes duration: Tid i sekunder **kwargs: Extra parametrar att skicka till log_performance """ try: if log_sheets: log_sheets.log_performance(operation, duration, **kwargs) except Exception as e: print(f"[MONITORING ERROR] Safe performance log failed: {e}") # Fortsätt utan att krascha # ============================================================================ # ADMIN PANEL FUNCTIONS # ============================================================================ def show_admin_panel(log_sheets): """ Visa admin-panelen med lösenordsskydd och dashboard Args: log_sheets: LogSheets instans """ st.title("🔐 Admin Panel") # Kontrollera om admin är inloggad if 'admin_authenticated' not in st.session_state: st.session_state.admin_authenticated = False if not st.session_state.admin_authenticated: # Visa login-formulär st.markdown("### Inloggning krävs") password = st.text_input("Admin-lösenord", type="password", key="admin_password") if st.button("Logga in som Admin"): if password == ADMIN_PASSWORD: st.session_state.admin_authenticated = True st.success("✅ Admin-inloggning lyckades!") time.sleep(0.5) st.rerun() else: st.error("❌ Felaktigt lösenord") return # Admin är inloggad - visa dashboard st.markdown("---") # Logout knapp col1, col2 = st.columns([4, 1]) with col2: if st.button("🚪 Logga ut"): st.session_state.admin_authenticated = False st.rerun() # Tabs för olika vyer tab1, tab2, tab3 = st.tabs(["📊 Dashboard", "📋 Aktivitetslogg", "⚡ Performance"]) with tab1: show_dashboard(log_sheets) with tab2: show_activity_log(log_sheets) with tab3: show_performance_log(log_sheets) def show_dashboard(log_sheets): """Visa realtidsstatistik dashboard""" st.markdown("### 📊 Realtidsstatistik") if not log_sheets or not log_sheets.logs_sheet: st.warning("⚠️ Ingen loggdata tillgänglig") return try: # Hämta loggdata all_records = log_sheets.logs_sheet.get_all_records() if not all_records: st.info("ℹ️ Ingen aktivitet loggad än") return df = pd.DataFrame(all_records) # Konvertera timestamp till datetime df['Timestamp'] = pd.to_datetime(df['Timestamp'], errors='coerce') # Dagens datum today = datetime.now(SWEDISH_TZ).date() df['Date'] = df['Timestamp'].dt.date # Metrics col1, col2, col3, col4 = st.columns(4) with col1: total_events = len(df) st.metric("📝 Totala händelser", f"{total_events:,}") with col2: today_events = len(df[df['Date'] == today]) st.metric("📅 Händelser idag", f"{today_events:,}") with col3: unique_users = df['User ID'].nunique() st.metric("👥 Unika användare", f"{unique_users:,}") with col4: total_edits = len(df[df['Event Type'] == 'EDIT']) st.metric("✏️ Totala redigeringar", f"{total_edits:,}") st.markdown("---") # Händelser per typ st.markdown("### 📊 Händelser per typ") event_counts = df['Event Type'].value_counts() st.bar_chart(event_counts) st.markdown("---") # Aktivitet över tid (senaste 7 dagarna) st.markdown("### 📈 Aktivitet senaste 7 dagarna") last_7_days = df[df['Date'] >= (today - pd.Timedelta(days=7))] daily_activity = last_7_days.groupby('Date').size() st.line_chart(daily_activity) st.markdown("---") # Mest aktiva användare st.markdown("### 🏆 Mest aktiva användare") user_activity = df.groupby(['User ID', 'Company Name']).size().reset_index(name='Aktiviteter') user_activity = user_activity.sort_values('Aktiviteter', ascending=False).head(10) st.dataframe(user_activity, use_container_width=True, hide_index=True) except Exception as e: st.error(f"❌ Fel vid hämtning av dashboard-data: {e}") def show_activity_log(log_sheets): """Visa fullständig aktivitetslogg med filter""" st.markdown("### 📋 Fullständig aktivitetslogg") if not log_sheets or not log_sheets.logs_sheet: st.warning("⚠️ Ingen loggdata tillgänglig") return try: # Hämta loggdata all_records = log_sheets.logs_sheet.get_all_records() if not all_records: st.info("ℹ️ Ingen aktivitet loggad än") return df = pd.DataFrame(all_records) # Filter col1, col2, col3 = st.columns(3) with col1: event_types = ['Alla'] + sorted(df['Event Type'].unique().tolist()) selected_event = st.selectbox("Händelsetyp", event_types) with col2: users = ['Alla'] + sorted(df['User ID'].unique().tolist()) selected_user = st.selectbox("Användare", users) with col3: # Datumfilter date_filter = st.selectbox("Tidsperiod", ["Alla", "Idag", "Senaste 7 dagarna", "Senaste 30 dagarna"]) # Applicera filter filtered_df = df.copy() if selected_event != 'Alla': filtered_df = filtered_df[filtered_df['Event Type'] == selected_event] if selected_user != 'Alla': filtered_df = filtered_df[filtered_df['User ID'] == selected_user] if date_filter != 'Alla': filtered_df['Timestamp'] = pd.to_datetime(filtered_df['Timestamp'], errors='coerce') today = datetime.now(SWEDISH_TZ) if date_filter == 'Idag': filtered_df = filtered_df[filtered_df['Timestamp'].dt.date == today.date()] elif date_filter == 'Senaste 7 dagarna': filtered_df = filtered_df[filtered_df['Timestamp'] >= (today - pd.Timedelta(days=7))] elif date_filter == 'Senaste 30 dagarna': filtered_df = filtered_df[filtered_df['Timestamp'] >= (today - pd.Timedelta(days=30))] # Visa resultat st.markdown(f"**Visar {len(filtered_df)} händelser**") # Sortera med senaste först filtered_df = filtered_df.sort_values('Timestamp', ascending=False) # Visa tabell st.dataframe( filtered_df, use_container_width=True, hide_index=True, height=600 ) # Export-knapp if st.button("📥 Exportera till CSV"): csv = filtered_df.to_csv(index=False).encode('utf-8') st.download_button( label="Ladda ner CSV", data=csv, file_name=f"activity_log_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv", mime="text/csv" ) except Exception as e: st.error(f"❌ Fel vid hämtning av loggdata: {e}") def show_performance_log(log_sheets): """Visa performance-metrics""" st.markdown("### ⚡ Performance Metrics") if not log_sheets or not log_sheets.perf_sheet: st.warning("⚠️ Ingen performance-data tillgänglig") return try: # Hämta performance-data all_records = log_sheets.perf_sheet.get_all_records() if not all_records: st.info("ℹ️ Ingen performance-data loggad än") return df = pd.DataFrame(all_records) # Konvertera duration till float df['Duration (s)'] = pd.to_numeric(df['Duration (s)'], errors='coerce') # Metrics col1, col2, col3 = st.columns(3) with col1: avg_duration = df['Duration (s)'].mean() st.metric("⏱️ Genomsnittlig tid", f"{avg_duration:.2f}s") with col2: max_duration = df['Duration (s)'].max() st.metric("🐌 Långsammaste operation", f"{max_duration:.2f}s") with col3: min_duration = df['Duration (s)'].min() st.metric("⚡ Snabbaste operation", f"{min_duration:.2f}s") st.markdown("---") # Genomsnittlig tid per operation st.markdown("### 📊 Genomsnittlig tid per operation") avg_by_operation = df.groupby('Operation')['Duration (s)'].mean().sort_values(ascending=False) st.bar_chart(avg_by_operation) st.markdown("---") # Senaste operationer st.markdown("### 🕐 Senaste operationer") recent_ops = df.sort_values('Timestamp', ascending=False).head(20) st.dataframe(recent_ops, use_container_width=True, hide_index=True) except Exception as e: st.error(f"❌ Fel vid hämtning av performance-data: {e}") # ============================================================================ # ORIGINAL FUNCTIONS # ============================================================================ @st.cache_resource def get_google_sheet(): """Anslut till Google Sheets - använder Omrade_updater""" try: if hasattr(st, 'secrets') and 'gcp_service_account' in st.secrets: creds = Credentials.from_service_account_info( st.secrets["gcp_service_account"], scopes=SCOPES ) sheet_id = st.secrets.get("sheet_id", None) elif 'GOOGLE_CREDENTIALS' in os.environ: google_creds = json.loads(os.environ['GOOGLE_CREDENTIALS']) creds = Credentials.from_service_account_info( google_creds, scopes=SCOPES ) sheet_id = os.environ.get('SHEET_ID', None) else: st.error("❌ Inga credentials hittades!") return None, None, None client = gspread.authorize(creds) # Försök öppna sheetet "Omrade_updater" try: spreadsheet = client.open("Omrade_updater") sheet = spreadsheet.sheet1 # Hämta service account email för loggning if 'GOOGLE_CREDENTIALS' in os.environ: google_creds = json.loads(os.environ['GOOGLE_CREDENTIALS']) service_account_email = google_creds.get('client_email', 'Web User') else: service_account_email = st.secrets["gcp_service_account"].get('client_email', 'Web User') return sheet, service_account_email, spreadsheet except gspread.exceptions.SpreadsheetNotFound: st.error("❌ Kunde inte hitta sheetet 'Omrade_updater'. Kontrollera att namnet är korrekt och att service account har tillgång.") return None, None, None except Exception as e: st.error(f"❌ Kunde inte ansluta till Google Sheets: {e}") return None, None, None def get_sheet_data(sheet): """Hämta all data från sheet med korrekt hantering av textformat""" try: # Använd get_all_values() för att få RAW text all_values = sheet.get_all_values() if not all_values or len(all_values) < 2: st.error("❌ Sheetet är tomt eller har ingen data") return None # Första raden är headers headers = all_values[0] data_rows = all_values[1:] # Skapa DataFrame df = pd.DataFrame(data_rows, columns=headers) # Konvertera viktiga kolumner till string if 'Account ID' in df.columns: df['Account ID'] = df['Account ID'].astype(str) if 'Områdeskod' in df.columns: df['Områdeskod'] = df['Områdeskod'].astype(str) # Rensa alla kolumner från eventuella apostrofer (från gamla data) # Nu när vi använder RAW borde inte nya apostrofer skapas for col in df.columns: if df[col].dtype == 'object': # Text-kolumner df[col] = df[col].astype(str).apply( lambda x: x.lstrip("'") if isinstance(x, str) and x.startswith("'") else x ) # Konvertera "nan", "None" och tomma till tom sträng df[col] = df[col].replace(['nan', 'None', ''], '') print(f"[DEBUG] Laddade {len(df)} rader från Google Sheets") return df except Exception as e: error_str = str(e) # Specialhantering för quota-fel if "quota" in error_str.lower() or "429" in error_str: st.error("❌ **API Quota överskred!**") st.warning(""" Google Sheets API har en gräns på antal förfrågningar per minut. **Vänta 1 minut** och ladda sedan om sidan (tryck F5). Om problemet kvarstår, kontakta migration@chargenode.eu """) print(f"[ERROR] QUOTA EXCEEDED: {error_str}") else: st.error(f"❌ Fel vid hämtning av data: {e}") print(f"[ERROR] get_sheet_data fel: {e}") return None def validate_password(df, password): """Validera lösenord och returnera account ID""" if 'Lösenord' not in df.columns: st.error("❌ Lösenordskolumn saknas i sheetet") return None match = df[df['Lösenord'].astype(str) == password] if not match.empty: account_id = match.iloc[0].get('Account ID', None) return str(account_id) if account_id else None return None def get_company_areas(df, account_id): """Hämta alla områden för ett account ID""" return df[df['Account ID'] == account_id].copy() def get_migration_interval(company_areas): """Hämta första och sista migreringsdatum""" if 'Indikativt datum' not in company_areas.columns: return None, None dates = [] for date_val in company_areas['Indikativt datum']: if not date_val or str(date_val).strip() == '' or str(date_val).strip() == '0': continue try: date_str = str(date_val).strip() parsed_date = None # Försök olika format formats_to_try = [ '%Y/%m/%d', '%Y-%m-%d', '%d/%m/%Y', '%d-%m-%Y', '%Y%m%d', '%d.%m.%Y', '%Y.%m.%d' ] for fmt in formats_to_try: try: parsed_date = datetime.strptime(date_str, fmt) break except ValueError: continue if parsed_date: dates.append(parsed_date) except Exception as e: continue if dates: return min(dates), max(dates) return None, None def get_total_outlets(company_areas): """Beräkna totalt antal uttag""" if '# Uttag' not in company_areas.columns: return 0 total = 0 for val in company_areas['# Uttag']: try: if val and str(val).strip() != '': total += int(float(str(val))) except: continue return total def get_hardware_summary(company_areas): """Hämta sammanfattning av hårdvara""" if 'Hårdvara' not in company_areas.columns: return "Information saknas" # Samla alla hårdvaror i en lista hardware_list = [] for val in company_areas['Hårdvara']: if val and str(val).strip() != '' and str(val).strip() != '0': hardware_list.append(str(val).strip()) if not hardware_list: return "Information saknas" # Om alla är samma, visa bara den hårdvaran en gång if len(set(hardware_list)) == 1: return hardware_list[0] # Annars visa alla unika hårdvaror separerade med komma unique_hardware = list(set(hardware_list)) return ", ".join(unique_hardware) def get_main_contact_name(company_areas): """Hämta namnet på huvudkontakten""" if 'Main Contact' not in company_areas.columns or 'Namn' not in company_areas.columns: return "Ej angiven" # Hitta raden där Main Contact = "Huvudkontakt" main_contacts = company_areas[company_areas['Main Contact'] == 'Huvudkontakt'] if not main_contacts.empty: name = main_contacts.iloc[0]['Namn'] if name and str(name).strip() != '': return str(name).strip() return "Ej angiven" def get_migration_summary(company_areas): """Hämta sammanfattning av migrationsmetod""" if 'Metod' not in company_areas.columns: return "Information saknas" # Hämta metoder methods = [] for val in company_areas['Metod']: if val and str(val).strip() != '' and str(val).strip() != '0': methods.append(str(val).strip()) if not methods: return "Information saknas" # Räkna områden per metod from collections import Counter method_counts = Counter(methods) # Räkna totalt antal uttag total_outlets = get_total_outlets(company_areas) # Räkna antal unika orter unique_cities = set() if 'Stad' in company_areas.columns: for val in company_areas['Stad']: if val and str(val).strip() != '' and str(val).strip() != '0': unique_cities.add(str(val).strip()) num_cities = len(unique_cities) if unique_cities else 1 # Skapa formaterad text summary_parts = [] for method, count in method_counts.items(): summary_parts.append(f"{count}st {method} migrationer") methods_text = ", ".join(summary_parts) return f"{methods_text} om totalt {total_outlets} uttag på {num_cities} {'ort' if num_cities == 1 else 'orter'}" def export_to_excel(company_areas, account_id): """Exportera data till Excel med formatering, instruktioner och validering""" try: from openpyxl.worksheet.datavalidation import DataValidation # Ta bort dolda kolumner export_df = company_areas.drop(columns=HIDDEN_COLUMNS, errors='ignore') # Skapa Excel-fil output = io.BytesIO() wb = Workbook() # === DATAFLIK (FLIK 1) === ws = wb.active ws.title = "Dina områden" # Färger för data header_fill = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid") readonly_fill = PatternFill(start_color="E7E6E6", end_color="E7E6E6", fill_type="solid") editable_fill = PatternFill(start_color="FFFFFF", end_color="FFFFFF", fill_type="solid") # Lägg till headers for col_idx, col_name in enumerate(export_df.columns, 1): cell = ws.cell(row=1, column=col_idx, value=col_name) cell.font = Font(bold=True, color="FFFFFF", size=11) cell.fill = header_fill cell.alignment = Alignment(horizontal="center", vertical="center") # Hitta kolumnindex för Tillgänglighet tillganglighet_col_idx = None for col_idx, col_name in enumerate(export_df.columns, 1): if col_name == "Tillgänglighet": tillganglighet_col_idx = col_idx break # Lägg till data for row_idx, row in enumerate(export_df.itertuples(index=False), 2): for col_idx, (col_name, value) in enumerate(zip(export_df.columns, row), 1): cell = ws.cell(row=row_idx, column=col_idx, value=value) cell.font = Font(size=11) cell.alignment = Alignment(horizontal="left", vertical="center", wrap_text=False) # Sätt bakgrundsfärg baserat på om kolumnen är editerbar if col_name in READONLY_COLUMNS: cell.fill = readonly_fill else: cell.fill = editable_fill # Lägg till dropdown-validering för Tillgänglighet if tillganglighet_col_idx: col_letter = ws.cell(row=1, column=tillganglighet_col_idx).column_letter dv = DataValidation( type="list", formula1='"Alltid tillgängligt,Stängt - se Access info"', allow_blank=True, showErrorMessage=True, errorTitle="Ogiltigt val", error="Välj antingen 'Alltid tillgängligt' eller 'Stängt - se Access info'" ) dv.add(f'{col_letter}2:{col_letter}{len(export_df) + 1}') ws.add_data_validation(dv) # Justera kolumnbredd for col_idx, col_name in enumerate(export_df.columns, 1): max_length = max( len(str(col_name)), max([len(str(val)) for val in export_df[col_name]], default=0) ) ws.column_dimensions[ws.cell(row=1, column=col_idx).column_letter].width = min(max_length + 2, 50) # === INSTRUKTIONSFLIK (FLIK 2) === ws_instructions = wb.create_sheet(title="INSTRUKTIONER") # Färger för instruktioner title_fill = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid") warning_fill = PatternFill(start_color="FFF4E6", end_color="FFF4E6", fill_type="solid") tip_fill = PatternFill(start_color="E8F5E9", end_color="E8F5E9", fill_type="solid") # Rubrik ws_instructions['A1'] = "📋 INSTRUKTIONER - ChargeNode Migrering" ws_instructions['A1'].font = Font(bold=True, size=16, color="FFFFFF") ws_instructions['A1'].fill = title_fill ws_instructions['A1'].alignment = Alignment(horizontal="center", vertical="center") ws_instructions.merge_cells('A1:D1') ws_instructions.row_dimensions[1].height = 30 # Instruktioner - med samma tillgänglighetsförklaring som i datainmatningsfönstret instructions = [ ("", ""), ("📝 HUR DU FYLLER I EXCEL-FILEN", ""), ("", ""), ("1. Editera endast de VITA fälten", "Gråa fält är låsta och kan inte ändras"), ("", ""), ("2. Tillgänglighet - Välj från rullistan", ""), ("", "• Välj 'Alltid tillgängligt' eller 'Stängt - se Access info'"), ("", ""), ("3. ⚠️ VIKTIGT om Tillgänglighet", ""), ("", "• Om 'Stängt - se Access info': Du MÅSTE fylla i 'Access info'"), ("", " (kod, nyckel, kontaktperson, etc.)"), ("", ""), ("", "• Om 'Alltid tillgängligt': Ingen 'Access info' behövs - området är ju tillgängligt!"), ("", ""), ("4. Spara filen", "Ladda sedan upp den i systemet"), ("", ""), ("✅ TIPS FÖR BÄSTA RESULTAT", ""), ("", ""), ("• Ändra INTE kolumnordning eller kolumnnamn", ""), ("• Ta INTE bort rader", ""), ("• Använd dropdown-listor där de finns", ""), ("• Fyll i alla obligatoriska fält", ""), ("", ""), ("❌ FILEN SPARAS INTE OM:", ""), ("", ""), ("• Ett område är markerat som 'Stängt - se Access info' utan 'Access info'", ""), ("", ""), ("💡 Behöver du hjälp?", "Kontakta migration@chargenode.eu"), ] current_row = 2 for text1, text2 in instructions: cell_a = ws_instructions[f'A{current_row}'] cell_b = ws_instructions[f'B{current_row}'] cell_a.value = text1 cell_b.value = text2 # Formatering baserat på innehåll if text1.startswith(("📝", "✅", "❌", "💡")): cell_a.font = Font(bold=True, size=13, color="1976D2") ws_instructions.merge_cells(f'A{current_row}:D{current_row}') elif text1.startswith("⚠️"): cell_a.font = Font(bold=True, size=12, color="E65100") cell_b.font = Font(size=11, color="E65100") cell_a.fill = warning_fill cell_b.fill = warning_fill elif text1.startswith(("1.", "2.", "3.", "4.")): cell_a.font = Font(bold=True, size=12) cell_b.font = Font(size=11, italic=True, color="666666") elif text1.startswith("•"): cell_a.font = Font(size=11) cell_b.font = Font(size=11, italic=True, color="666666") else: cell_a.font = Font(size=11) cell_b.font = Font(size=11) cell_a.alignment = Alignment(horizontal="left", vertical="top", wrap_text=True) cell_b.alignment = Alignment(horizontal="left", vertical="top", wrap_text=True) current_row += 1 # Justera kolumnbredder för instruktioner ws_instructions.column_dimensions['A'].width = 50 ws_instructions.column_dimensions['B'].width = 60 wb.save(output) output.seek(0) return output.getvalue() except Exception as e: st.error(f"❌ Fel vid Excel-export: {e}") return None def process_uploaded_excel(uploaded_file, original_df, account_id): """Bearbeta uppladdad Excel-fil och returnera ändringar""" try: # Läs Excel-filen (från fliken "Dina områden") uploaded_df = pd.read_excel(uploaded_file, sheet_name='Dina områden') # Hämta original data för företaget original_company_data = original_df[original_df['Account ID'] == account_id].copy() original_display = original_company_data.drop(columns=HIDDEN_COLUMNS, errors='ignore') # Validera att kolumnerna matchar if not all(col in original_display.columns for col in uploaded_df.columns): st.error("❌ Kolumnerna i Excel-filen matchar inte de förväntade kolumnerna.") return None # === KRITISK VALIDERING: Stängt område måste ha Access info === validation_errors = [] for idx in range(len(uploaded_df)): tillganglighet = str(uploaded_df.iloc[idx].get('Tillgänglighet', '')).strip() access_info = str(uploaded_df.iloc[idx].get('Access info', '')).strip() fastighetsnamn = str(uploaded_df.iloc[idx].get('Fastighetsnamn', f'Rad {idx + 2}')) # Kolla om området är stängt men saknar access info if tillganglighet == 'Stängt - se Access info' and (access_info == '' or access_info == 'nan' or access_info == 'None'): validation_errors.append(f"• **{fastighetsnamn}** (rad {idx + 2})") if validation_errors: st.error(f""" ❌ **Validering misslyckades!** Följande områden är markerade som **'Stängt - se Access info'** men saknar **'Access info'**: {chr(10).join(validation_errors)} **Åtgärd:** Gå tillbaka till Excel-filen och fyll i 'Access info' för dessa områden. Beskriv hur vi kommer åt området (kod, nyckel, kontaktperson, etc.) 💡 **OBS:** Områden med 'Alltid tillgängligt' behöver ingen 'Access info' - dessa är ju tillgängliga. """) return None # Hitta ändringar changes = [] for idx in range(min(len(uploaded_df), len(original_display))): original_row_idx = original_display.index[idx] for col in uploaded_df.columns: # Endast editerbara kolumner if col not in READONLY_COLUMNS: orig_val = str(original_display.iloc[idx][col]) if pd.notna(original_display.iloc[idx][col]) else "" new_val = str(uploaded_df.iloc[idx][col]) if pd.notna(uploaded_df.iloc[idx][col]) else "" if orig_val != new_val: changes.append({ 'row_idx': original_row_idx, 'column': col, 'old_value': orig_val, 'new_value': new_val }) return changes except Exception as e: st.error(f"❌ Fel vid bearbetning av Excel-fil: {e}") return None def export_to_csv(company_areas): """Exportera data till CSV med korrekt encoding""" try: # Ta bort dolda kolumner export_df = company_areas.drop(columns=HIDDEN_COLUMNS, errors='ignore') # Konvertera till CSV med UTF-8 BOM för Excel-kompatibilitet csv_buffer = io.StringIO() export_df.to_csv(csv_buffer, index=False, encoding='utf-8-sig', sep=';') return csv_buffer.getvalue().encode('utf-8-sig') except Exception as e: st.error(f"❌ Fel vid CSV-export: {e}") return None def test_sheet_connection(sheet): """Testa att anslutningen till Google Sheets fungerar""" try: # Försök läsa första raden (headers) headers = sheet.row_values(1) if headers and len(headers) > 0: return True, "Anslutning till Google Sheets fungerar" else: return False, "Kunde inte läsa data från Google Sheets" except Exception as e: return False, f"Anslutningsfel: {str(e)}" def batch_update_cells_with_tracking(sheet, changes, df, editor_email, log_sheets=None, account_id=None): """ Uppdatera flera celler samtidigt med ändringsloggning och omfattande felhantering MONITORING: Loggar alla ändringar till Google Sheets """ if not changes: return True, "Inga ändringar att spara" # Performance tracking start start_time = time.time() # Logga för debugging print(f"[DEBUG] Försöker uppdatera {len(changes)} ändringar") try: # Test 1: Verifiera att sheet är tillgängligt try: headers = sheet.row_values(1) if not headers: error_msg = "❌ Kunde inte läsa headers från Google Sheets. Kontrollera din internetanslutning." print(f"[ERROR] {error_msg}") return False, error_msg except Exception as e: error_msg = f"❌ Anslutningsfel till Google Sheets: {str(e)}" print(f"[ERROR] {error_msg}") return False, error_msg # Test 2: Verifiera att alla kolumner finns for change in changes: col_name = change['column'] if col_name not in headers: error_msg = f"❌ Kolumnen '{col_name}' finns inte i Google Sheets. Kontakta migration@chargenode.eu" print(f"[ERROR] {error_msg}") return False, error_msg # Hitta kolumnindex för Last update och Editor last_update_col_idx = headers.index('Last update') + 1 if 'Last update' in headers else None editor_col_idx = headers.index('Editor') + 1 if 'Editor' in headers else None # Svensk tid för timestamp current_time = datetime.now(SWEDISH_TZ).strftime('%Y-%m-%d %H:%M:%S') updates = [] # Förbered alla uppdateringar for change in changes: row_idx = change['row_idx'] col_name = change['column'] new_value = change['new_value'] # Hitta rätt rad i sheetet try: df_position = df[df.index == row_idx].index[0] sheet_row = df_position + 2 except Exception as e: error_msg = f"❌ Fel vid radpositionering: {str(e)}" print(f"[ERROR] {error_msg}") return False, error_msg # Hitta kolumnindex if col_name in headers: col_idx = headers.index(col_name) + 1 # Speciell hantering för Telefon-kolumnen: ALLTID som ren text value_to_write = new_value if col_name == 'Telefon': # Skicka som ren text, ingen apostrof behövs value_to_write = str(new_value) if new_value else '' print(f"[DEBUG] Telefonnummer sparas som ren text: '{value_to_write}'") # Lägg till huvuduppdateringen updates.append({ 'range': f'{gspread.utils.rowcol_to_a1(sheet_row, col_idx)}', 'values': [[value_to_write]] }) print(f"[DEBUG] Förbereder uppdatering: {col_name} rad {sheet_row} till '{value_to_write}'") if not updates: return False, "❌ Inga giltiga uppdateringar att utföra" print(f"[DEBUG] Totalt {len(updates)} uppdateringar förberedda") # Batch update alla ändringar med retry-logik max_retries = 3 batch_size = 50 update_successful = False last_error = None for attempt in range(max_retries): try: print(f"[DEBUG] Försök {attempt + 1} av {max_retries}") for i in range(0, len(updates), batch_size): batch = updates[i:i + batch_size] print(f"[DEBUG] Uppdaterar batch {i//batch_size + 1}: {len(batch)} celler") # Använd 'RAW' för att behandla all data som ren text (behåller ledande nollor) sheet.batch_update(batch, value_input_option='RAW') time.sleep(REQUEST_DELAY) # Om vi kom hit så lyckades uppdateringen update_successful = True print(f"[SUCCESS] Batch-uppdatering lyckades på försök {attempt + 1}") break except gspread.exceptions.APIError as e: last_error = str(e) print(f"[ERROR] API-fel på försök {attempt + 1}: {last_error}") if attempt < max_retries - 1: wait_time = (attempt + 1) * 2 print(f"[DEBUG] Väntar {wait_time} sekunder innan nästa försök...") time.sleep(wait_time) continue else: error_msg = f"❌ API-fel efter {max_retries} försök: {last_error}. Vänta en minut och försök igen." return False, error_msg except Exception as e: last_error = str(e) print(f"[ERROR] Oväntat fel på försök {attempt + 1}: {last_error}") if attempt < max_retries - 1: time.sleep(2) continue else: error_msg = f"❌ Oväntat fel vid uppdatering: {last_error}" return False, error_msg # KRITISK CHECK: Verifiera att uppdateringen lyckades if not update_successful: error_msg = f"❌ Uppdateringen misslyckades efter {max_retries} försök. Senaste fel: {last_error}" print(f"[ERROR] {error_msg}") return False, error_msg # Uppdatera Last update och Editor för alla påverkade rader affected_rows = set([change['row_idx'] for change in changes]) tracking_updates = [] for row_idx in affected_rows: try: df_position = df[df.index == row_idx].index[0] sheet_row = df_position + 2 if last_update_col_idx: tracking_updates.append({ 'range': f'{gspread.utils.rowcol_to_a1(sheet_row, last_update_col_idx)}', 'values': [[current_time]] }) if editor_col_idx: tracking_updates.append({ 'range': f'{gspread.utils.rowcol_to_a1(sheet_row, editor_col_idx)}', 'values': [[editor_email]] }) except Exception as e: print(f"[WARNING] Kunde inte förbereda tracking update för rad {row_idx}: {str(e)}") if tracking_updates: try: print(f"[DEBUG] Uppdaterar {len(tracking_updates)} tracking-fält") for i in range(0, len(tracking_updates), batch_size): batch = tracking_updates[i:i + batch_size] # Använd 'RAW' för ren text sheet.batch_update(batch, value_input_option='RAW') time.sleep(REQUEST_DELAY) print("[SUCCESS] Tracking-uppdatering lyckades") except Exception as e: # Om timestamp-uppdateringen misslyckas, fortsätt ändå # eftersom huvuddatan är sparad print(f"[WARNING] Tracking-uppdatering misslyckades: {str(e)}") pass # FINAL VERIFICATION: Läs tillbaka en av ändringarna för att verifiera try: verify_change = changes[0] df_position = df[df.index == verify_change['row_idx']].index[0] sheet_row = df_position + 2 col_idx = headers.index(verify_change['column']) + 1 # Läs tillbaka värdet cell_range = gspread.utils.rowcol_to_a1(sheet_row, col_idx) verified_value = sheet.acell(cell_range).value print(f"[DEBUG] Verifiering: Förväntade '{verify_change['new_value']}', fick '{verified_value}'") # Jämför värdena - behandla None och tom sträng som samma sak expected = str(verify_change['new_value']) if verify_change['new_value'] else '' actual = str(verified_value) if verified_value else '' if expected != actual: error_msg = f"⚠️ VARNING: Verifiering misslyckades! Värdet sparades möjligen inte korrekt. Kontakta migration@chargenode.eu" print(f"[ERROR] {error_msg}") return False, error_msg print("[SUCCESS] Verifiering lyckades - data är sparad!") except Exception as e: print(f"[WARNING] Kunde inte verifiera uppdatering: {str(e)}") # Fortsätt ändå eftersom huvuduppdateringen lyckades # MONITORING: Logga alla redigeringar if log_sheets and account_id: company_name = st.session_state.get('company_name', '') for change in changes: safe_log_activity( log_sheets, 'EDIT', account_id, company_name=company_name, details=f"Redigerade {change['column']}", row_number=str(change['row_idx']), column=change['column'], old_value=change['old_value'], new_value=change['new_value'], status='SUCCESS' ) # Performance tracking end duration = time.time() - start_time if log_sheets and account_id: safe_log_performance( log_sheets, 'BATCH_UPDATE', duration, user_id=account_id, details=f"{len(changes)} ändringar" ) success_msg = f"✅ {len(changes)} ändringar har verifierats och sparats i Google Sheets" print(f"[SUCCESS] {success_msg}") return True, success_msg except Exception as e: error_msg = str(e) print(f"[ERROR] Oväntat exception: {error_msg}") # MONITORING: Logga fel if log_sheets and account_id: safe_log_activity( log_sheets, 'EDIT', account_id, details=f"Fel vid batch update: {error_msg}", status='ERROR' ) if "quota" in error_msg.lower() or "rate" in error_msg.lower() or "429" in error_msg: return False, "❌ API Quota överskred! Google Sheets API-gränsen nådd. Vänta 1 minut, ladda sedan om sidan (F5) och försök igen." elif "permission" in error_msg.lower() or "access" in error_msg.lower(): return False, "❌ Åtkomstfel. Kontakta migration@chargenode.eu" else: return False, f"❌ Fel vid sparande: {error_msg}. Kontakta migration@chargenode.eu om problemet kvarstår." def get_column_config(df): """Generera kolumnkonfiguration för data_editor med beskrivningar och dropdowns""" config = {} # Beskrivningar för editerbara kolumner editable_descriptions = { 'Namn': 'Huvudkontaktperson för detta område. Personen vi kontaktar vid frågor eller problem.', 'Email adress': 'Ange individuell e-postadress till kontaktpersonen för detta område.', 'Telefon': 'Telefonnummer till huvudkontaktpersonen. Format: +46XXXXXXXXX eller 07X-XXXXXXX', 'Tillgänglighet': 'Välj i dropdown-menyn om detta är ett område som vi alltid kan komma åt, eller om det är stängt. Om det är stängt var vänlig fyll i fältet Access info.', 'Access info': 'Om området inte alltid är tillgängligt, vänligen ange hur vi kan få tillgång. T.ex. "Nycklar finns i receptionen" eller "Ring Pelle på 07XXXXXXXX när ni är på gång".', 'Fritext meddelande': 'Övrig information eller meddelanden som är viktiga för oss att veta.', 'Main Contact': 'Ange om denna person är huvudansvarig för kontot. Välj "Huvudkontakt" för helst endast en person per företag.' } # Beskrivningar för readonly kolumner readonly_descriptions = { 'Fastighetsnamn': 'Namnet på fastigheten där laddstationerna finns', 'Address': 'Gatuadress för området', 'Stad': 'Ort/stad där området finns', 'Metod': 'Migrationsmetod som används', '# Uttag': 'Antal ladduttag på området', 'Hårdvara': 'Typ av laddstationer', 'Vecka': 'Planerad vecka för migrering', 'Indikativt datum': 'Planerat datum för migrering', 'Status': 'Nuvarande status för området', 'Last update': 'Senaste uppdatering', 'Editor': 'Senaste redigerare' } # Konfigurera editerbara kolumner for col in EDITABLE_COLUMNS: if col in df.columns: if col == 'Tillgänglighet': # Dropdown för tillgänglighet - tillåter tom värde config[col] = st.column_config.SelectboxColumn( col, help=editable_descriptions.get(col, ''), width="medium", options=[ "", # Tomt alternativ för att kunna rensa "Alltid tillgängligt", "Stängt - se Access info" ], required=False ) elif col == 'Main Contact': # Dropdown för huvudkontakt config[col] = st.column_config.SelectboxColumn( col, help=editable_descriptions.get(col, ''), width="medium", options=[ "", "Huvudkontakt" ], required=False ) else: # Vanlig textkolumn config[col] = st.column_config.TextColumn( col, help=editable_descriptions.get(col, ''), width="medium", required=False ) # Konfigurera readonly kolumner med hänglås-emoji for col in READONLY_COLUMNS: if col in df.columns: config[col] = st.column_config.TextColumn( f"🔒 {col}", help=readonly_descriptions.get(col, 'Detta fält kan inte editeras'), disabled=True, width="medium" ) return config # CSS och styling st.markdown(""" """, unsafe_allow_html=True) # Session state initialization if 'authenticated' not in st.session_state: st.session_state.authenticated = False if 'account_id' not in st.session_state: st.session_state.account_id = None if 'df' not in st.session_state: st.session_state.df = None if 'company_name' not in st.session_state: st.session_state.company_name = None if 'original_data' not in st.session_state: st.session_state.original_data = None if 'last_save_time' not in st.session_state: st.session_state.last_save_time = None if 'saving_in_progress' not in st.session_state: st.session_state.saving_in_progress = False print("[DEBUG] Initierade saving_in_progress till False (första gången)") else: print(f"[DEBUG] Session state laddat: saving_in_progress={st.session_state.saving_in_progress}") if 'service_account_email' not in st.session_state: st.session_state.service_account_email = None if 'log_sheets' not in st.session_state: st.session_state.log_sheets = None if 'show_admin' not in st.session_state: st.session_state.show_admin = False def show_save_status_indicator(): """Visa enkel status-indikator för auto-save""" if st.session_state.saving_in_progress: st.markdown("""
Inloggad för uppdatering av företagsdata
🔒 Säker anslutning • Data krypteras med TLS • ChargeNode.se
ALLA VITA RUTOR MÅSTE FYLLAS I!
Kolumner med 🔒 (gult hänglås) kan du inte ändra på.
Övriga kolumner kan du klicka i och redigera.
Klicka på "SPARA ÄNDRINGAR"-knappen under tabellen när du är klar.
Namn: Huvudkontaktperson för detta område. Personen vi kontaktar vid frågor eller problem.
Email adress: Ange individuell e-postadress till kontaktpersonen för detta område.
Telefon: Telefonnummer till huvudkontaktpersonen. Format: +46XXXXXXXXX eller 07X-XXXXXXX
Tillgänglighet: Välj i dropdown-menyn om detta är ett område som vi alltid kan komma åt, eller om det är stängt. Om det är stängt var vänlig fyll i fältet Access info.
Access info: Om området inte alltid är tillgängligt, vänligen ange hur vi kan få tillgång. T.ex. "Nycklar finns i receptionen" eller "Ring Pelle på 07XXXXXXXX när ni är på gång".
Main Contact: Ange om denna person är huvudansvarig för kontot. Välj "Huvudkontakt" för helst endast en person per företag.
Från {first_date.strftime('%Y-%m-%d')} till {last_date.strftime('%Y-%m-%d')}
", unsafe_allow_html=True) with col3: st.metric( label="🔌 Totalt antal uttag", value=total_outlets, help="Summa av alla uttag för dina områden" ) st.markdown("💡 Tips: Du kan ändra flera fält innan du sparar - alla ändringar sparas samtidigt!
🔒 Företagsdata • Inloggad som {company_name} • {len(company_areas)} områden • {total_outlets} uttag