import os import json import time import requests import gspread from google.oauth2.service_account import Credentials import pandas as pd from datetime import datetime, timedelta import pytz import schedule import threading import gradio as gr # ============================================================================ # KONFIGURATION # ============================================================================ SWEDISH_TZ = pytz.timezone('Europe/Stockholm') # Google Sheets setup SCOPES = [ 'https://www.googleapis.com/auth/spreadsheets', 'https://www.googleapis.com/auth/drive' ] # ============================================================================ # GOOGLE SHEETS ANSLUTNING # ============================================================================ def get_google_sheets_client(): """Anslut till Google Sheets med service account credentials.""" try: # Hämta credentials från Hugging Face Secrets google_credentials_json = os.environ.get('GOOGLE_CREDENTIALS') if not google_credentials_json: raise ValueError("GOOGLE_CREDENTIALS saknas i Hugging Face Secrets") # Parse JSON credentials creds_dict = json.loads(google_credentials_json) # Skapa credentials objekt creds = Credentials.from_service_account_info(creds_dict, scopes=SCOPES) # Skapa gspread client gc = gspread.authorize(creds) print("✅ Google Sheets-anslutning etablerad") return gc except Exception as e: print(f"❌ Fel vid anslutning till Google Sheets: {e}") return None def get_sheet_data(gc): """Hämta data från Google Sheet.""" try: # Öppna spreadsheet spreadsheet = gc.open("Omrade_updater") # Hämta data från huvudflik main_sheet = spreadsheet.worksheet("Sheet1") main_data = pd.DataFrame(main_sheet.get_all_records()) # Hämta loggdata try: logs_sheet = spreadsheet.worksheet("Omrade_updater_LOGS") logs_data = pd.DataFrame(logs_sheet.get_all_records()) except gspread.exceptions.WorksheetNotFound: print("⚠️ LOGS sheet hittades inte, skapar tom DataFrame") logs_data = pd.DataFrame() print(f"✅ Hämtade {len(main_data)} rader från huvudsheet") print(f"✅ Hämtade {len(logs_data)} loggrader") return main_data, logs_data, spreadsheet except Exception as e: print(f"❌ Fel vid hämtning av sheet-data: {e}") return None, None, None # ============================================================================ # METRICS-BERÄKNINGAR # ============================================================================ def calculate_login_metrics(logs_df): """Beräkna inloggningsmetrics för olika tidsperioder.""" if logs_df.empty: return { 'last_24h': 0, 'last_3_days': 0, 'last_7_days': 0, 'unique_companies_24h': [], 'unique_companies_3d': [], 'unique_companies_7d': [] } try: # Filtrera bara LOGIN events login_logs = logs_df[logs_df['Event Type'] == 'LOGIN'].copy() if login_logs.empty: return { 'last_24h': 0, 'last_3_days': 0, 'last_7_days': 0, 'unique_companies_24h': [], 'unique_companies_3d': [], 'unique_companies_7d': [] } # Parse timestamp kolumn login_logs['timestamp_parsed'] = pd.to_datetime( login_logs['Timestamp'], format='%Y-%m-%d %H:%M:%S', errors='coerce' ) # Beräkna cutoff-tider now = datetime.now(SWEDISH_TZ) cutoff_24h = now - timedelta(hours=24) cutoff_3d = now - timedelta(days=3) cutoff_7d = now - timedelta(days=7) # Gör timestamps timezone-aware login_logs['timestamp_parsed'] = login_logs['timestamp_parsed'].dt.tz_localize(SWEDISH_TZ, ambiguous='infer') # Filtrera per tidsperiod logins_24h = login_logs[login_logs['timestamp_parsed'] >= cutoff_24h] logins_3d = login_logs[login_logs['timestamp_parsed'] >= cutoff_3d] logins_7d = login_logs[login_logs['timestamp_parsed'] >= cutoff_7d] # Räkna unika användare (User ID = Account ID) unique_24h = logins_24h['User ID'].nunique() unique_3d = logins_3d['User ID'].nunique() unique_7d = logins_7d['User ID'].nunique() # Hämta företagsnamn för de som loggat in companies_24h = logins_24h['Company Name'].dropna().unique().tolist() companies_3d = logins_3d['Company Name'].dropna().unique().tolist() companies_7d = logins_7d['Company Name'].dropna().unique().tolist() return { 'last_24h': unique_24h, 'last_3_days': unique_3d, 'last_7_days': unique_7d, 'unique_companies_24h': companies_24h, 'unique_companies_3d': companies_3d, 'unique_companies_7d': companies_7d } except Exception as e: print(f"❌ Fel vid beräkning av login metrics: {e}") return { 'last_24h': 0, 'last_3_days': 0, 'last_7_days': 0, 'unique_companies_24h': [], 'unique_companies_3d': [], 'unique_companies_7d': [] } def calculate_edit_metrics(logs_df): """Beräkna redigeringsmetrics - Excel upload vs manuella ändringar.""" if logs_df.empty: return { 'excel_uploads_24h': 0, 'excel_uploads_7d': 0, 'manual_edits_24h': 0, 'manual_edits_7d': 0, 'excel_companies': [], 'manual_companies': [] } try: # Filtrera UPLOAD och EDIT events upload_logs = logs_df[logs_df['Event Type'] == 'UPLOAD'].copy() edit_logs = logs_df[logs_df['Event Type'] == 'EDIT'].copy() # Parse timestamps if not upload_logs.empty: upload_logs['timestamp_parsed'] = pd.to_datetime( upload_logs['Timestamp'], format='%Y-%m-%d %H:%M:%S', errors='coerce' ) upload_logs['timestamp_parsed'] = upload_logs['timestamp_parsed'].dt.tz_localize(SWEDISH_TZ, ambiguous='infer') if not edit_logs.empty: edit_logs['timestamp_parsed'] = pd.to_datetime( edit_logs['Timestamp'], format='%Y-%m-%d %H:%M:%S', errors='coerce' ) edit_logs['timestamp_parsed'] = edit_logs['timestamp_parsed'].dt.tz_localize(SWEDISH_TZ, ambiguous='infer') # Beräkna cutoff-tider now = datetime.now(SWEDISH_TZ) cutoff_24h = now - timedelta(hours=24) cutoff_7d = now - timedelta(days=7) # Räkna uploads if not upload_logs.empty: uploads_24h = len(upload_logs[upload_logs['timestamp_parsed'] >= cutoff_24h]) uploads_7d = len(upload_logs[upload_logs['timestamp_parsed'] >= cutoff_7d]) excel_companies = upload_logs[upload_logs['timestamp_parsed'] >= cutoff_7d]['Company Name'].dropna().unique().tolist() else: uploads_24h = 0 uploads_7d = 0 excel_companies = [] # Räkna manuella editeringar if not edit_logs.empty: edits_24h = len(edit_logs[edit_logs['timestamp_parsed'] >= cutoff_24h]) edits_7d = len(edit_logs[edit_logs['timestamp_parsed'] >= cutoff_7d]) manual_companies = edit_logs[edit_logs['timestamp_parsed'] >= cutoff_7d]['Company Name'].dropna().unique().tolist() else: edits_24h = 0 edits_7d = 0 manual_companies = [] return { 'excel_uploads_24h': uploads_24h, 'excel_uploads_7d': uploads_7d, 'manual_edits_24h': edits_24h, 'manual_edits_7d': edits_7d, 'excel_companies': excel_companies, 'manual_companies': manual_companies } except Exception as e: print(f"❌ Fel vid beräkning av edit metrics: {e}") return { 'excel_uploads_24h': 0, 'excel_uploads_7d': 0, 'manual_edits_24h': 0, 'manual_edits_7d': 0, 'excel_companies': [], 'manual_companies': [] } def calculate_completion_status(main_df): """Beräkna hur många företag som fyllt i all data.""" if main_df.empty: return { 'total_companies': 0, 'completed_companies': 0, 'completion_rate': 0, 'missing_fields': {} } try: # Definiera obligatoriska fält required_fields = ['Namn', 'Email adress', 'Telefon', 'Tillgänglighet'] # Räkna unika företag total_companies = main_df['Account ID'].nunique() # Räkna företag som fyllt i alla fält completed = 0 missing_fields = {} for account_id in main_df['Account ID'].unique(): company_data = main_df[main_df['Account ID'] == account_id] # Kolla om alla required fields är ifyllda för alla områden all_complete = True for field in required_fields: if field in company_data.columns: empty_count = company_data[field].isna().sum() + (company_data[field] == '').sum() if empty_count > 0: all_complete = False if field not in missing_fields: missing_fields[field] = 0 missing_fields[field] += 1 if all_complete: completed += 1 completion_rate = (completed / total_companies * 100) if total_companies > 0 else 0 return { 'total_companies': total_companies, 'completed_companies': completed, 'completion_rate': round(completion_rate, 1), 'missing_fields': missing_fields } except Exception as e: print(f"❌ Fel vid beräkning av completion status: {e}") return { 'total_companies': 0, 'completed_companies': 0, 'completion_rate': 0, 'missing_fields': {} } # ============================================================================ # SLACK INTEGRATION # ============================================================================ def send_to_slack(subject, content, color="#2a9d8f"): """Skicka meddelande till Slack via webhook.""" webhook_url = os.environ.get("SLACK_WEBHOOK_URL") if not webhook_url: print("❌ SLACK_WEBHOOK_URL saknas i Hugging Face Secrets") return False try: payload = { "blocks": [ { "type": "header", "text": { "type": "plain_text", "text": subject } }, { "type": "section", "text": { "type": "mrkdwn", "text": content } } ] } response = requests.post( webhook_url, json=payload, headers={"Content-Type": "application/json"} ) if response.status_code == 200: print(f"✅ Slack-meddelande skickat: {subject}") return True else: print(f"❌ Slack-anrop misslyckades: {response.status_code}, {response.text}") return False except Exception as e: print(f"❌ Fel vid sändning till Slack: {e}") return False # ============================================================================ # HUVUDFUNKTION - GENERERA DAGLIG RAPPORT # ============================================================================ def generate_daily_report(): """Generera och skicka daglig rapport till Slack.""" try: print(f"\n{'='*60}") print(f"📊 Startar daglig rapport - {datetime.now(SWEDISH_TZ).strftime('%Y-%m-%d %H:%M:%S')}") print(f"{'='*60}\n") # Anslut till Google Sheets gc = get_google_sheets_client() if not gc: send_to_slack( "⚠️ ChargeNode Migration - Fel", "Kunde inte ansluta till Google Sheets. Kontrollera credentials.", "#ff0000" ) return # Hämta data main_df, logs_df, spreadsheet = get_sheet_data(gc) if main_df is None or logs_df is None: send_to_slack( "⚠️ ChargeNode Migration - Fel", "Kunde inte hämta data från Google Sheets.", "#ff0000" ) return # Beräkna metrics login_metrics = calculate_login_metrics(logs_df) edit_metrics = calculate_edit_metrics(logs_df) completion_metrics = calculate_completion_status(main_df) # Bygg Slack-meddelande now = datetime.now(SWEDISH_TZ) subject = f"📊 ChargeNode Migration - Daglig Rapport {now.strftime('%Y-%m-%d')}" content = f""" *God morgon! Här är dagens migrationsstatistik* ☕ ━━━━━━━━━━━━━━━━━━━━━━━━━━━ *👥 INLOGGNINGAR* ━━━━━━━━━━━━━━━━━━━━━━━━━━━ • *Senaste 24h:* {login_metrics['last_24h']} unika företag • *Senaste 3 dagar:* {login_metrics['last_3_days']} unika företag • *Senaste 7 dagar:* {login_metrics['last_7_days']} unika företag ━━━━━━━━━━━━━━━━━━━━━━━━━━━ *✏️ UPPDATERINGAR* ━━━━━━━━━━━━━━━━━━━━━━━━━━━ *Excel-uppladdningar:* • Senaste 24h: {edit_metrics['excel_uploads_24h']} st • Senaste 7 dagar: {edit_metrics['excel_uploads_7d']} st *Manuella ändringar i matris:* • Senaste 24h: {edit_metrics['manual_edits_24h']} editeringar • Senaste 7 dagar: {edit_metrics['manual_edits_7d']} editeringar ━━━━━━━━━━━━━━━━━━━━━━━━━━━ *📋 KOMPLETTERINGSSTATUS* ━━━━━━━━━━━━━━━━━━━━━━━━━━━ • *Totalt företag:* {completion_metrics['total_companies']} st • *Komplett ifyllda:* {completion_metrics['completed_companies']} st • *Kompletteringsgrad:* {completion_metrics['completion_rate']}% """ # Lägg till företag som loggat in senaste 24h if login_metrics['unique_companies_24h']: content += f"\n*Företag som loggat in senaste 24h:*\n" for company in login_metrics['unique_companies_24h'][:10]: # Max 10 företag content += f"• {company}\n" if len(login_metrics['unique_companies_24h']) > 10: content += f"_...och {len(login_metrics['unique_companies_24h']) - 10} till_\n" # Lägg till företag som laddat upp Excel if edit_metrics['excel_companies']: content += f"\n*Företag som använt Excel-upload (senaste 7d):*\n" for company in edit_metrics['excel_companies'][:5]: content += f"• {company}\n" if len(edit_metrics['excel_companies']) > 5: content += f"_...och {len(edit_metrics['excel_companies']) - 5} till_\n" content += f"\n━━━━━━━━━━━━━━━━━━━━━━━━━━━\n" content += f"_Rapport genererad: {now.strftime('%Y-%m-%d %H:%M:%S')}_" # Skicka till Slack success = send_to_slack(subject, content, "#2a9d8f") if success: print("\n✅ Daglig rapport skickad till Slack framgångsrikt!\n") else: print("\n❌ Kunde inte skicka rapport till Slack\n") return success except Exception as e: print(f"\n❌ FEL vid generering av daglig rapport: {e}\n") send_to_slack( "⚠️ ChargeNode Migration - Kritiskt Fel", f"Ett fel uppstod vid generering av daglig rapport:\n```{str(e)}```", "#ff0000" ) return False # ============================================================================ # SCHEMALÄGGNING # ============================================================================ def run_scheduler(): """Kör schemaläggaren i en separat tråd.""" # Schemalägg daglig rapport kl 09:00 svensk tid schedule.every().day.at("09:00").do(generate_daily_report) print(f"\n⏰ Scheduler startad - Daglig rapport körs kl 09:00 svensk tid") print(f" Nästa körning: {schedule.next_run()}\n") while True: schedule.run_pending() time.sleep(60) # Kolla varje minut # Starta scheduler i bakgrundstråd scheduler_thread = threading.Thread(target=run_scheduler, daemon=True) scheduler_thread.start() # ============================================================================ # GRADIO UI (minimal för att hålla Space:n aktiv) # ============================================================================ def manual_trigger(): """Manuell trigger för att testa rapporten.""" success = generate_daily_report() if success: return "✅ Rapport skickad till Slack!" else: return "❌ Något gick fel. Kolla loggarna." with gr.Blocks(title="ChargeNode Migration Reporter") as app: gr.Markdown(""" # 📊 ChargeNode Migration - Daglig Rapport Denna applikation kör automatiskt varje dag kl **09:00 svensk tid** och skickar en rapport till Slack-kanalen **#ai-chat**. Rapporten innehåller: - 👥 Inloggningsstatistik (24h, 3 dagar, 7 dagar) - ✏️ Uppdateringar via Excel vs manuellt - 📋 Kompletteringsstatus för företagen """) with gr.Row(): trigger_btn = gr.Button("🔄 Kör rapport manuellt (test)", size="lg") output = gr.Textbox(label="Status", lines=3) trigger_btn.click(manual_trigger, outputs=output) gr.Markdown(f""" --- **Status:** 🟢 Aktiv **Nästa schemalagd körning:** {schedule.next_run().strftime('%Y-%m-%d %H:%M:%S') if schedule.next_run() else 'Väntar på initiering...'} **Tidszon:** Europe/Stockholm (CET/CEST) """) # Skicka en initial rapport när appen startar (för test) print("\n🚀 Skickar initial testrapport vid start...\n") time.sleep(5) # Vänta lite för att systemet ska bli klart generate_daily_report() if __name__ == "__main__": app.launch(server_name="0.0.0.0", server_port=7860)