Spaces:
Sleeping
Sleeping
| import os | |
| import json | |
| import time | |
| import requests | |
| from anthropic import Anthropic, RateLimitError, APIError | |
| from openai import OpenAI | |
| import gradio as gr | |
| import pandas as pd | |
| from huggingface_hub import CommitScheduler | |
| from datetime import datetime, timedelta | |
| import uuid | |
| from user_agents import parse as parse_ua | |
| import schedule | |
| import threading | |
| from sentence_transformers import SentenceTransformer | |
| import numpy as np | |
| import faiss | |
| import re | |
| from difflib import SequenceMatcher | |
| # --- Konfiguration --- | |
| CHARGENODE_URL = "https://www.chargenode.eu" | |
| MAX_CHUNK_SIZE = 2000 | |
| CHUNK_OVERLAP = 200 | |
| RETRIEVAL_K = 5 | |
| MAX_CONTEXT_CHARS = 8000 # ~2000 tokens för kontext | |
| # Kontrollera om vi kör i Hugging Face-miljön | |
| IS_HUGGINGFACE = os.environ.get("SPACE_ID") is not None | |
| # OpenAI-klient behålls för bakåtkompatibilitet | |
| OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY") | |
| if not OPENAI_API_KEY: | |
| raise ValueError("OPENAI_API_KEY saknas") | |
| client = OpenAI(api_key=OPENAI_API_KEY) | |
| # Lägg till Anthropic API-nyckel och klient | |
| ANTHROPIC_API_KEY = os.environ.get("ANTHROPIC_API_KEY") | |
| if not ANTHROPIC_API_KEY: | |
| raise ValueError("ANTHROPIC_API_KEY saknas") | |
| anthropic_client = Anthropic(api_key=ANTHROPIC_API_KEY) | |
| log_folder = "logs" | |
| os.makedirs(log_folder, exist_ok=True) | |
| log_file_path = os.path.join(log_folder, "conversation_log_v2.txt") | |
| # Skapa en tom loggfil om den inte finns | |
| if not os.path.exists(log_file_path): | |
| with open(log_file_path, "w", encoding="utf-8") as f: | |
| f.write("") | |
| print(f"Skapade tom loggfil: {log_file_path}") | |
| hf_token = os.environ.get("HF_TOKEN") | |
| if not hf_token: | |
| raise ValueError("HF_TOKEN saknas") | |
| # Minsta möjliga konfiguration som bör fungera | |
| scheduler = CommitScheduler( | |
| repo_id="ChargeNodeEurope/logfiles", | |
| repo_type="dataset", | |
| folder_path=log_folder, | |
| path_in_repo="logs_v2", | |
| every=300, | |
| token=hf_token | |
| ) | |
| # --- Globala variabler --- | |
| last_log = None | |
| # Globala variabler för embeddings | |
| embedder = None | |
| embeddings = None | |
| index = None | |
| chunks = [] | |
| chunk_sources = [] | |
| faq_dict = {} | |
| # --- Hjälpfunktioner --- | |
| def detect_language(text): | |
| """Enkel språkdetektering baserad på vanliga ord.""" | |
| swedish_indicators = ['hur', 'kan', 'jag', 'är', 'det', 'den', 'ett', 'och', 'som', 'på', 'för', 'med', 'av', 'till', 'om'] | |
| english_indicators = ['how', 'can', 'the', 'is', 'are', 'and', 'or', 'for', 'with', 'what', 'where', 'when'] | |
| text_lower = text.lower() | |
| words = text_lower.split() | |
| swedish_count = sum(1 for word in words if word in swedish_indicators) | |
| english_count = sum(1 for word in words if word in english_indicators) | |
| if english_count > swedish_count and english_count > 2: | |
| return 'en' | |
| return 'sv' | |
| def validate_numeric_field(value, field_name): | |
| """Validerar att ett fält är numeriskt.""" | |
| if value and not value.isdigit(): | |
| return f"{field_name} måste vara numerisk." | |
| return None | |
| # --- Förbättrad loggfunktion --- | |
| def safe_append_to_log(log_entry): | |
| """Säker metod för att lägga till loggdata utan att förlora historisk information.""" | |
| try: | |
| with open(log_file_path, "a", encoding="utf-8") as log_file: | |
| log_json = json.dumps(log_entry, ensure_ascii=False) | |
| log_file.write(log_json + "\n") | |
| log_file.flush() | |
| print(f"Loggpost tillagd: {log_entry.get('timestamp', 'okänd tid')}") | |
| return True | |
| except Exception as e: | |
| print(f"Fel vid loggning: {e}") | |
| try: | |
| os.makedirs(os.path.dirname(log_file_path), exist_ok=True) | |
| with open(log_file_path, "a", encoding="utf-8") as log_file: | |
| log_json = json.dumps(log_entry, ensure_ascii=False) | |
| log_file.write(log_json + "\n") | |
| print("Loggpost tillagd efter återhämtning") | |
| return True | |
| except Exception as retry_error: | |
| print(f"Kritiskt fel vid loggning: {retry_error}") | |
| return False | |
| # --- Laddar textkällor --- | |
| def load_local_files(): | |
| """Laddar alla lokala filer och returnerar som en sammanhängande text.""" | |
| uploaded_text = "" | |
| # Definiera obligatoriska filer | |
| required_files = [ | |
| "FAQ_stadat.xlsx", | |
| "Foretagskonto.txt", | |
| "ChargeNode_App.txt", | |
| "ChargeNode_Portal.txt" | |
| ] | |
| allowed = [".txt", ".docx", ".pdf", ".csv", ".xls", ".xlsx"] | |
| excluded = ["requirements.txt", "app.py", "conversation_log.txt", "conversation_log_v2.txt", "secrets", "prompt.txt"] | |
| # Kontrollera att alla obligatoriska filer finns | |
| missing_files = [] | |
| for req_file in required_files: | |
| if not os.path.exists(req_file): | |
| missing_files.append(req_file) | |
| if missing_files: | |
| print(f"⚠️ VARNING: Följande obligatoriska filer saknas: {', '.join(missing_files)}") | |
| for file in os.listdir("."): | |
| if file.lower().endswith(tuple(allowed)) and file not in excluded: | |
| try: | |
| if file.endswith(".txt"): | |
| with open(file, "r", encoding="utf-8") as f: | |
| content = f.read() | |
| elif file.endswith(".docx"): | |
| from docx import Document | |
| content = "\n".join([p.text for p in Document(file).paragraphs]) | |
| elif file.endswith(".pdf"): | |
| import PyPDF2 | |
| with open(file, "rb") as f: | |
| reader = PyPDF2.PdfReader(f) | |
| content = "\n".join([p.extract_text() or "" for p in reader.pages]) | |
| elif file.endswith(".csv"): | |
| content = pd.read_csv(file).to_string() | |
| elif file.endswith((".xls", ".xlsx")): | |
| if file == "FAQ_stadat.xlsx" or file == "FAQ stadat.xlsx": | |
| df = pd.read_excel(file) | |
| rows = [] | |
| for index, row in df.iterrows(): | |
| row_text = f"Fråga: {row['Fråga']}\nSvar: {row['Svar']}" | |
| if 'kategori' in df.columns: | |
| row_text += f"\nKategori: {row['kategori']}" | |
| elif 'Kategori' in df.columns: | |
| row_text += f"\nKategori: {row['Kategori']}" | |
| rows.append(row_text) | |
| content = "\n\n".join(rows) | |
| else: | |
| content = pd.read_excel(file).to_string() | |
| uploaded_text += f"\n\nFIL: {file}\n{content}" | |
| print(f"✅ Laddade fil: {file}") | |
| except Exception as e: | |
| print(f"❌ Fel vid läsning av {file}: {str(e)}") | |
| return uploaded_text.strip() | |
| def load_prompt(): | |
| """Läser in system-prompts från prompt.txt med bättre felhantering.""" | |
| try: | |
| with open("prompt.txt", "r", encoding="utf-8") as f: | |
| prompt_content = f.read().strip() | |
| if not prompt_content: | |
| print("Varning: prompt.txt är tom, använder standardprompt") | |
| return get_default_prompt() | |
| print("✅ Laddade prompt.txt") | |
| return prompt_content | |
| except FileNotFoundError: | |
| print("Varning: prompt.txt hittades inte, använder standardprompt") | |
| return get_default_prompt() | |
| except Exception as e: | |
| print(f"Fel vid inläsning av prompt.txt: {e}, använder standardprompt") | |
| return get_default_prompt() | |
| def get_default_prompt(): | |
| """Returnerar standardprompt om prompt.txt saknas.""" | |
| return """Du är ChargeNode's AI-assistent som hjälper användare med frågor om ChargeNode's produkter och tjänster. | |
| SPRÅK: Svara alltid på samma språk som användaren skriver på (svenska eller engelska). | |
| SVARSSTIL: | |
| - Var vänlig, professionell och hjälpsam | |
| - Ge konkreta, tydliga svar baserat på den tillhandahållna informationen | |
| - Om informationen inte finns i kontexten, säg det tydligt | |
| - När du ger navigationsinstruktioner, använd numrerade steg | |
| - Håll svaren koncisa men kompletta | |
| KONTAKTINFORMATION: | |
| Om användaren behöver ytterligare hjälp, hänvisa till: | |
| - Email: support@chargenode.eu | |
| - Telefon: 010-2051055""" | |
| # --- Förbättrad chunking --- | |
| def prepare_chunks(text_data): | |
| """Delar upp texten i mindre segment för embedding och sökning med särskild hänsyn till FAQ-format.""" | |
| chunks, sources = [], [] | |
| global faq_dict | |
| for source, text in text_data.items(): | |
| paragraphs = [p for p in text.split("\n") if p.strip()] | |
| i = 0 | |
| while i < len(paragraphs): | |
| current_chunk = "" | |
| start_idx = i | |
| if i < len(paragraphs) and paragraphs[i].startswith("Fråga:"): | |
| question = paragraphs[i][7:].strip() | |
| current_chunk = paragraphs[i] | |
| i += 1 | |
| while i < len(paragraphs) and not paragraphs[i].startswith("Fråga:"): | |
| if len(current_chunk) + len(paragraphs[i]) + 1 <= MAX_CHUNK_SIZE: | |
| current_chunk += "\n" + paragraphs[i] | |
| else: | |
| if "Svar:" in current_chunk: | |
| if len(current_chunk) > MAX_CHUNK_SIZE * 1.5: | |
| break | |
| else: | |
| current_chunk += "\n" + paragraphs[i] | |
| else: | |
| break | |
| i += 1 | |
| if "Svar:" in current_chunk: | |
| answer_start = current_chunk.find("Svar:") | |
| answer_text = current_chunk[answer_start + 5:].strip() | |
| # Lägg till betalningsrelaterade variationer | |
| if any(term in question.lower() for term in ["betalsätt", "betalmetod", "betalmedel", "kort", | |
| "betalkort", "betalning", "betala"]): | |
| payment_variations = [ | |
| "hur ändrar jag betalmedel", | |
| "hur byter jag betalsätt", | |
| "hur uppdaterar jag mitt betalkort", | |
| "hur ändrar jag betalmetod", | |
| "hur byter jag betalningsmetod", | |
| "hur ändrar jag betalkort", | |
| "how do i change payment method", | |
| "how to update credit card", | |
| "how to change payment card" | |
| ] | |
| for variation in payment_variations: | |
| faq_dict[variation] = answer_text | |
| faq_dict[question.lower()] = answer_text | |
| else: | |
| while i < len(paragraphs) and len(current_chunk) + len(paragraphs[i]) + 1 <= MAX_CHUNK_SIZE: | |
| if current_chunk: | |
| current_chunk += " " + paragraphs[i] | |
| else: | |
| current_chunk = paragraphs[i] | |
| i += 1 | |
| if current_chunk.strip(): | |
| chunks.append(current_chunk.strip()) | |
| sources.append(source) | |
| if i == start_idx: | |
| i += 1 | |
| overlap_chunks = [] | |
| overlap_sources = [] | |
| for j in range(0, len(chunks)): | |
| overlap_chunks.append(chunks[j]) | |
| overlap_sources.append(sources[j]) | |
| if j < len(chunks) - 1 and chunks[j].endswith(chunks[j+1][:CHUNK_OVERLAP]): | |
| continue | |
| if j < len(chunks) - 1: | |
| space_left = MAX_CHUNK_SIZE - len(chunks[j]) | |
| if space_left >= CHUNK_OVERLAP: | |
| overlap_text = chunks[j] + " " + chunks[j+1][:CHUNK_OVERLAP] | |
| overlap_chunks.append(overlap_text) | |
| overlap_sources.append(sources[j]) | |
| chunks = overlap_chunks | |
| sources = overlap_sources | |
| print(f"✅ Genererade {len(chunks)} chunks med {len(faq_dict)} FAQ-par") | |
| return chunks, sources | |
| def initialize_embeddings(): | |
| """Initierar SentenceTransformer och FAISS-index vid första anrop.""" | |
| global embedder, embeddings, index, chunks, chunk_sources, faq_dict | |
| if embedder is None: | |
| print("🔄 Initierar SentenceTransformer och FAISS-index...") | |
| print("📁 Laddar textdata...") | |
| text_data = {"local_files": load_local_files()} | |
| print("✂️ Förbereder textsegment...") | |
| chunks, chunk_sources = prepare_chunks(text_data) | |
| print(f"✅ {len(chunks)} segment laddade") | |
| print("🧠 Skapar embeddings...") | |
| try: | |
| # Försök först med svensk modell | |
| embedder = SentenceTransformer('KBLab/sentence-bert-swedish-cased') | |
| print("✅ Använder svensk embeddings-modell") | |
| except: | |
| try: | |
| # Fallback till multilingual | |
| embedder = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2') | |
| print("✅ Använder multilingual embeddings-modell") | |
| except: | |
| # Sista fallback | |
| embedder = SentenceTransformer('all-MiniLM-L6-v2') | |
| print("✅ Använder engelsk embeddings-modell") | |
| embeddings = embedder.encode(chunks, convert_to_numpy=True) | |
| embeddings /= np.linalg.norm(embeddings, axis=1, keepdims=True) | |
| index = faiss.IndexFlatIP(embeddings.shape[1]) | |
| index.add(embeddings) | |
| print("✅ FAISS-index klart") | |
| print(f"📚 FAQ Dictionary innehåller {len(faq_dict)} nycklar") | |
| if len(faq_dict) > 0: | |
| payment_keys = [k for k in faq_dict.keys() if any(term in k for term in ["betalsätt", "betalmetod", "betalmedel", "payment"])] | |
| print(f"💳 Betalningsrelaterade FAQ-nycklar: {len(payment_keys)}") | |
| def check_direct_match(query): | |
| """Kontrollerar om frågan matchar någon av våra fördefinierade FAQ-svar med fuzzy matching.""" | |
| query_lower = query.lower().strip('?!.').strip() | |
| # Normalisera vanliga variationer | |
| query_normalized = query_lower | |
| query_normalized = re.sub(r'\bbyt(a|er)\b', 'ändra', query_normalized) | |
| query_normalized = re.sub(r'\buppdat(era|erar)\b', 'ändra', query_normalized) | |
| # Exakt matchning för betalningsfrågor | |
| if any(query_normalized.startswith(prefix) for prefix in ["hur ändrar jag", "hur byter jag", "hur uppdaterar jag", "how do i change", "how to change", "how to update"]) and \ | |
| any(term in query_normalized for term in ["betalsätt", "betalmetod", "betalmedel", "betalkort", "kort", "payment", "credit card", "card"]): | |
| payment_answer = """Så här gör du om du vill byta betalkort: | |
| 1. Gå in i appen. | |
| 2. Tryck på meny och mina betalsätt | |
| 3. Tryck på ersätt kort. | |
| 4. Godkänn våra villkor | |
| 5. Tryck på kortbetalning under "bekräfta för auktorisering" | |
| 6. Lägg in dina nya kort uppgifter | |
| 7. Bekräfta med BankID. | |
| OBS! Se till att kortet har pengar och att det är upplåst för internetbetalningar.""" | |
| return payment_answer | |
| # Exakt matchning i FAQ dictionary | |
| if query_normalized in faq_dict: | |
| return faq_dict[query_normalized] | |
| # Fuzzy matching för liknande frågor | |
| best_match = None | |
| best_score = 0.0 | |
| for key, value in faq_dict.items(): | |
| similarity = SequenceMatcher(None, query_normalized, key).ratio() | |
| # Ge bonus för matchande nyckelord | |
| query_terms = set(query_normalized.split()) | |
| key_terms = set(key.split()) | |
| common_terms = query_terms.intersection(key_terms) | |
| if len(common_terms) >= 2: | |
| similarity += 0.1 | |
| if similarity > best_score and similarity > 0.75: # 75% likhet krävs | |
| best_score = similarity | |
| best_match = value | |
| if best_match: | |
| print(f"🎯 Fuzzy match hittad (score: {best_score:.2f})") | |
| return best_match | |
| return None | |
| def retrieve_context(query, k=RETRIEVAL_K): | |
| """Hämtar relevant kontext för frågor med direkt matchning för vanliga frågor.""" | |
| initialize_embeddings() | |
| direct_match = check_direct_match(query) | |
| if direct_match: | |
| print(f"✅ Direkt matchning hittad för frågan: {query[:50]}...") | |
| return f"Fråga: {query}\nSvar: {direct_match}", ["direct_match"] | |
| query_embedding = embedder.encode([query], convert_to_numpy=True) | |
| query_embedding /= np.linalg.norm(query_embedding) | |
| D, I = index.search(query_embedding, k) | |
| retrieved, sources = [], set() | |
| for idx in I[0]: | |
| if idx < len(chunks): | |
| retrieved.append(chunks[idx]) | |
| sources.add(chunk_sources[idx]) | |
| context = " ".join(retrieved) | |
| # Truncate om för långt | |
| if len(context) > MAX_CONTEXT_CHARS: | |
| print(f"⚠️ Kontext trunkerad från {len(context)} till {MAX_CONTEXT_CHARS} tecken") | |
| context = context[:MAX_CONTEXT_CHARS] + "..." | |
| return context, list(sources) | |
| prompt_template = load_prompt() | |
| def generate_answer(query, chat_history=None): | |
| """Genererar svar baserat på fråga, chatthistorik och retrieval-baserad kontext med Claude Sonnet 4.5.""" | |
| context, sources = retrieve_context(query) | |
| if not context.strip(): | |
| lang = detect_language(query) | |
| if lang == 'en': | |
| return "I couldn't find any relevant information in my sources.\n\nThis is an AI-generated response." | |
| return "Jag hittar ingen relevant information i mina källor.\n\nDetta är ett AI-genererat svar." | |
| # Detektera språk | |
| lang = detect_language(query) | |
| # Förbered system prompt med språkinstruktion | |
| system_prompt = prompt_template | |
| if lang == 'en': | |
| system_prompt += "\n\nIMPORTANT: The user is writing in ENGLISH. Respond in ENGLISH." | |
| else: | |
| system_prompt += "\n\nVIKTIGT: Användaren skriver på SVENSKA. Svara på SVENSKA." | |
| # Strukturerad user message | |
| user_message = f"""Baserat på följande information om ChargeNode, svara på användarens fråga. | |
| === RELEVANT KONTEXT === | |
| {context} | |
| === ANVÄNDARENS FRÅGA === | |
| {query} | |
| === INSTRUKTIONER === | |
| - Svara på samma språk som frågan ({('engelska' if lang == 'en' else 'svenska')}) | |
| - Var koncis men komplett | |
| - Om informationen inte finns i kontexten, säg det tydligt | |
| - Hänvisa till specifika funktioner i appen/portalen när relevant | |
| - Om frågan gäller navigation, ge steg-för-steg instruktioner med numrerade punkter""" | |
| # Bygg messages array med historik | |
| messages = [] | |
| # Lägg till relevanta historiska meddelanden (max senaste 6 meddelanden = 3 par) | |
| if chat_history: | |
| for msg in chat_history[-6:]: | |
| if msg.get('role') in ['user', 'assistant']: | |
| messages.append({ | |
| "role": msg['role'], | |
| "content": msg['content'] | |
| }) | |
| # Lägg till aktuell fråga | |
| messages.append({"role": "user", "content": user_message}) | |
| try: | |
| response = anthropic_client.messages.create( | |
| model="claude-sonnet-4-5-20250929", | |
| max_tokens=3000, # Ökat från 1500 | |
| temperature=0.1, # Sänkt från 0.3 för mer konsekventa svar | |
| system=system_prompt, | |
| messages=messages | |
| ) | |
| answer = response.content[0].text | |
| print("✅ Svar genererat med Claude Sonnet 4.5") | |
| # Språkspecifik footer | |
| if lang == 'en': | |
| return answer + "\n\nAI-generated. Need more help? Contact support@chargenode.eu or call 010-2051055" | |
| return answer + "\n\nAI-genererat. Otillräcklig hjälp? Kontakta support@chargenode.eu eller 010-2051055" | |
| except RateLimitError: | |
| print("⚠️ Rate limit nådd") | |
| if lang == 'en': | |
| return "Too many requests right now. Please try again in a few seconds.\n\nContact support@chargenode.eu or 010-2051055" | |
| return "För många förfrågningar just nu. Försök igen om några sekunder.\n\nKontakta support@chargenode.eu eller 010-2051055" | |
| except APIError as e: | |
| print(f"⚠️ API-fel: {e}") | |
| if lang == 'en': | |
| return "A technical error occurred. Please try again.\n\nContact support@chargenode.eu or 010-2051055" | |
| return "Tekniskt fel uppstod. Vänligen försök igen.\n\nKontakta support@chargenode.eu eller 010-2051055" | |
| except Exception as e: | |
| print(f"❌ Oväntat fel: {e}") | |
| if lang == 'en': | |
| return f"Technical error: {str(e)}\n\nContact support@chargenode.eu or 010-2051055" | |
| return f"Tekniskt fel: {str(e)}\n\nKontakta support@chargenode.eu eller 010-2051055" | |
| # --- Slack Integration --- | |
| def send_to_slack(subject, content, color="#2a9d8f"): | |
| """Basfunktion för att skicka meddelanden till Slack.""" | |
| webhook_url = os.environ.get("SLACK_WEBHOOK_URL") | |
| if not webhook_url: | |
| print("Slack webhook URL saknas") | |
| 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: {type(e).__name__}: {e}") | |
| return False | |
| # --- Feedback & Like-funktion --- | |
| def vote(data: gr.LikeData): | |
| """Hanterar feedback från Gradio's inbyggda like-funktion.""" | |
| feedback_type = "up" if data.liked else "down" | |
| global last_log | |
| log_entry = { | |
| "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), | |
| "feedback": feedback_type, | |
| "bot_reply": data.value if not isinstance(data.value, dict) else data.value.get("value") | |
| } | |
| if last_log: | |
| log_entry.update({ | |
| "session_id": last_log.get("session_id"), | |
| "user_message": last_log.get("user_message"), | |
| }) | |
| safe_append_to_log(log_entry) | |
| try: | |
| if feedback_type == "down": | |
| feedback_message = f""" | |
| *⚠️ Negativ feedback registrerad* | |
| *Fråga:* {last_log.get('user_message', 'Okänd fråga')} | |
| *Svar:* {log_entry.get('bot_reply', 'Okänt svar')[:300]}{'...' if len(log_entry.get('bot_reply', '')) > 300 else ''} | |
| """ | |
| threading.Thread( | |
| target=lambda: send_to_slack("Negativ feedback", feedback_message, "#ff0000"), | |
| daemon=True | |
| ).start() | |
| except Exception as e: | |
| print(f"❌ Kunde inte skicka feedback till Slack: {e}") | |
| return | |
| # --- Rapportering --- | |
| def read_logs(): | |
| """Läs alla loggposter från loggfilen.""" | |
| logs = [] | |
| try: | |
| if os.path.exists(log_file_path): | |
| with open(log_file_path, "r", encoding="utf-8") as file: | |
| line_count = 0 | |
| for line in file: | |
| line_count += 1 | |
| try: | |
| log_entry = json.loads(line.strip()) | |
| logs.append(log_entry) | |
| except json.JSONDecodeError as e: | |
| print(f"⚠️ Varning: Kunde inte tolka rad {line_count}: {e}") | |
| continue | |
| print(f"✅ Läste {len(logs)} av {line_count} loggposter") | |
| else: | |
| print(f"⚠️ Loggfil saknas: {log_file_path}") | |
| except Exception as e: | |
| print(f"❌ Fel vid läsning av loggfil: {e}") | |
| return logs | |
| def get_latest_conversations(logs, limit=50): | |
| """Hämta de senaste frågorna och svaren.""" | |
| conversations = [] | |
| for log in reversed(logs): | |
| if 'user_message' in log and 'bot_reply' in log: | |
| conversations.append({ | |
| 'user_message': log['user_message'], | |
| 'bot_reply': log['bot_reply'], | |
| 'timestamp': log.get('timestamp', '') | |
| }) | |
| if len(conversations) >= limit: | |
| break | |
| return conversations | |
| def get_feedback_stats(logs): | |
| """Sammanfatta feedback (tumme upp/ned).""" | |
| feedback_count = {"up": 0, "down": 0} | |
| negative_feedback_examples = [] | |
| for log in logs: | |
| if 'feedback' in log: | |
| feedback = log.get('feedback') | |
| if feedback in feedback_count: | |
| feedback_count[feedback] += 1 | |
| if feedback == "down" and 'user_message' in log and len(negative_feedback_examples) < 10: | |
| negative_feedback_examples.append({ | |
| 'user_message': log.get('user_message', 'Okänd fråga'), | |
| 'bot_reply': log.get('bot_reply', 'Okänt svar') | |
| }) | |
| return feedback_count, negative_feedback_examples | |
| def generate_monthly_stats(days=30): | |
| """Genererar omfattande statistik över botanvändning för den senaste månaden.""" | |
| print(f"📊 Genererar statistik för de senaste {days} dagarna...") | |
| logs = read_logs() | |
| if not logs: | |
| return {"error": "Inga loggar hittades för den angivna perioden"} | |
| now = datetime.now() | |
| cutoff_date = now - timedelta(days=days) | |
| filtered_logs = [] | |
| for log in logs: | |
| if 'timestamp' in log: | |
| try: | |
| log_date = datetime.strptime(log['timestamp'], "%Y-%m-%d %H:%M:%S") | |
| if log_date >= cutoff_date: | |
| filtered_logs.append(log) | |
| except: | |
| pass | |
| logs = filtered_logs | |
| total_conversations = sum(1 for log in logs if 'user_message' in log) | |
| unique_sessions = len(set(log.get('session_id', 'unknown') for log in logs if 'session_id' in log)) | |
| unique_users = len(set(log.get('user_id', 'unknown') for log in logs if 'user_id' in log)) | |
| feedback_logs = [log for log in logs if 'feedback' in log] | |
| positive_feedback = sum(1 for log in feedback_logs if log.get('feedback') == 'up') | |
| negative_feedback = sum(1 for log in feedback_logs if log.get('feedback') == 'down') | |
| feedback_ratio = (positive_feedback / len(feedback_logs) * 100) if feedback_logs else 0 | |
| response_times = [log.get('response_time', 0) for log in logs if 'response_time' in log] | |
| avg_response_time = sum(response_times) / len(response_times) if response_times else 0 | |
| platforms = {} | |
| browsers = {} | |
| operating_systems = {} | |
| for log in logs: | |
| if 'platform' in log: | |
| platforms[log['platform']] = platforms.get(log['platform'], 0) + 1 | |
| if 'browser' in log: | |
| browsers[log['browser']] = browsers.get(log['browser'], 0) + 1 | |
| if 'os' in log: | |
| operating_systems[log['os']] = operating_systems.get(log['os'], 0) + 1 | |
| report = { | |
| "period": f"Senaste {days} dagarna", | |
| "generated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), | |
| "basic_stats": { | |
| "total_conversations": total_conversations, | |
| "unique_sessions": unique_sessions, | |
| "unique_users": unique_users, | |
| "messages_per_user": round(total_conversations / unique_users, 2) if unique_users else 0 | |
| }, | |
| "feedback": { | |
| "positive": positive_feedback, | |
| "negative": negative_feedback, | |
| "ratio_percent": round(feedback_ratio, 1) | |
| }, | |
| "performance": { | |
| "avg_response_time": round(avg_response_time, 2) | |
| }, | |
| "platform_distribution": platforms, | |
| "browser_distribution": browsers, | |
| "os_distribution": operating_systems | |
| } | |
| return report | |
| def simple_status_report(): | |
| """Skickar en förenklad statusrapport till Slack.""" | |
| print("📊 Genererar statusrapport för Slack...") | |
| try: | |
| stats = generate_monthly_stats(days=7) | |
| now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") | |
| subject = f"ChargeNode AI Bot - Status {now}" | |
| if 'error' in stats: | |
| content = f"*Fel vid generering av statistik:* {stats['error']}" | |
| return send_to_slack(subject, content, "#ff0000") | |
| basic = stats["basic_stats"] | |
| feedback = stats["feedback"] | |
| perf = stats["performance"] | |
| content = f""" | |
| *ChargeNode AI Bot - Statusrapport {now}* | |
| *Basstatistik* (senaste 7 dagarna) | |
| - Totalt antal konversationer: {basic['total_conversations']} | |
| - Unika sessioner: {basic['unique_sessions']} | |
| - Unika användare: {basic['unique_users']} | |
| - Genomsnittlig svarstid: {perf['avg_response_time']} sekunder | |
| *Feedback* | |
| - 👍 Tumme upp: {feedback['positive']} | |
| - 👎 Tumme ned: {feedback['negative']} | |
| - Nöjdhet: {feedback['ratio_percent']}% | |
| """ | |
| logs = read_logs() | |
| conversations = get_latest_conversations(logs, 3) | |
| if conversations: | |
| content += "\n*Senaste konversationer*\n" | |
| for conv in conversations: | |
| content += f""" | |
| > *Tid:* {conv['timestamp']} | |
| > *Fråga:* {conv['user_message'][:100]}{'...' if len(conv['user_message']) > 100 else ''} | |
| > *Svar:* {conv['bot_reply'][:100]}{'...' if len(conv['bot_reply']) > 100 else ''} | |
| """ | |
| return send_to_slack(subject, content, "#2a9d8f") | |
| except Exception as e: | |
| print(f"❌ Fel vid generering av statusrapport: {e}") | |
| error_subject = f"ChargeNode AI Bot - Fel vid statusrapport" | |
| error_content = f"*Fel vid generering av statusrapport:* {str(e)}" | |
| return send_to_slack(error_subject, error_content, "#ff0000") | |
| def send_support_to_slack(områdeskod, uttagsnummer, email, chat_history): | |
| """Skickar en supportförfrågan till Slack.""" | |
| try: | |
| chat_content = "" | |
| for msg in chat_history: | |
| if msg['role'] == 'user': | |
| chat_content += f">*Användare:* {msg['content']}\n\n" | |
| elif msg['role'] == 'assistant': | |
| chat_content += f">*Bot:* {msg['content'][:300]}{'...' if len(msg['content']) > 300 else ''}\n\n" | |
| subject = f"Support förfrågan - {datetime.now().strftime('%Y-%m-%d %H:%M')}" | |
| content = f""" | |
| *Användarinformation* | |
| - *Områdeskod:* {områdeskod or 'Ej angiven'} | |
| - *Uttagsnummer:* {uttagsnummer or 'Ej angiven'} | |
| - *Email:* {email} | |
| - *Tidpunkt:* {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} | |
| *Chatthistorik:* | |
| {chat_content} | |
| """ | |
| return send_to_slack(subject, content, "#e76f51") | |
| except Exception as e: | |
| print(f"❌ Fel vid sändning av support till Slack: {type(e).__name__}: {e}") | |
| return False | |
| # --- Schemaläggning av rapporter --- | |
| def run_scheduler(): | |
| """Kör schemaläggaren i en separat tråd med förenklad statusrapportering.""" | |
| schedule.every().day.at("08:00").do(simple_status_report) | |
| schedule.every().day.at("12:00").do(simple_status_report) | |
| schedule.every().day.at("17:00").do(simple_status_report) | |
| schedule.every().monday.at("09:00").do(lambda: send_to_slack( | |
| "Veckostatistik", | |
| f"*ChargeNode AI Bot - Veckostatistik*\n\n{json.dumps(generate_monthly_stats(7), indent=2)}", | |
| "#3498db" | |
| )) | |
| while True: | |
| schedule.run_pending() | |
| time.sleep(60) | |
| scheduler_thread = threading.Thread(target=run_scheduler, daemon=True) | |
| scheduler_thread.start() | |
| try: | |
| print("📤 Skickar en inledande statusrapport för att verifiera Slack-integrationen...") | |
| except Exception as e: | |
| print(f"ℹ️ Information: Statusrapport kommer att skickas enligt schema: {e}") | |
| # --- Gradio UI --- | |
| initial_chat = [{"role": "assistant", "content": "Detta är ChargeNode's AI bot. Hur kan jag hjälpa dig idag?"}] | |
| custom_css = """ | |
| body {background-color: #f7f7f7; font-family: Arial, sans-serif; margin: 0; padding: 0;} | |
| h1 {font-family: Helvetica, sans-serif; color: #2a9d8f; text-align: center; margin-bottom: 0.5em;} | |
| .gradio-container {max-width: 400px; margin: 0; padding: 10px; position: fixed; bottom: 20px; right: 20px; box-shadow: 0px 0px 10px rgba(0, 0, 0, 0.1); border-radius: 10px; background-color: #fff;} | |
| #chatbot_conversation { max-height: 300px; overflow-y: auto; } | |
| .gr-button {background-color: #2a9d8f; color: #fff; border: none; border-radius: 4px; padding: 8px 16px; margin: 5px;} | |
| .gr-button:hover {background-color: #264653;} | |
| .support-btn {background-color: #000000; color: #ffffff; margin-top: 5px; margin-bottom: 10px;} | |
| .support-btn:hover {background-color: #333333;} | |
| .flex-row {display: flex; flex-direction: row; gap: 5px;} | |
| .gr-form {padding: 10px; border: 1px solid #eee; border-radius: 4px; margin-bottom: 10px;} | |
| .chat-preview {max-height: 150px; overflow-y: auto; border: 1px solid #eee; padding: 8px; margin-top: 10px; font-size: 12px; background-color: #f9f9f9;} | |
| .success-message {font-size: 16px; font-weight: normal; margin-bottom: 15px;} | |
| footer {display: none !important;} | |
| .footer {display: none !important;} | |
| .gr-footer {display: none !important;} | |
| .gradio-footer {display: none !important;} | |
| .gradio-container .footer {display: none !important;} | |
| .gradio-container .gr-footer {display: none !important;} | |
| """ | |
| with gr.Blocks(css=custom_css, title="ChargeNode Kundtjänst") as app: | |
| gr.Markdown("Ställ din fråga om ChargeNodes produkter och tjänster nedan. Om du inte gillar botten, så ring oss gärna på 010 – 205 10 55") | |
| with gr.Group(visible=True) as chat_interface: | |
| chatbot = gr.Chatbot(value=initial_chat, type="messages", elem_id="chatbot_conversation") | |
| chatbot.like(vote, None, None) | |
| with gr.Row(): | |
| msg = gr.Textbox(label="Meddelande", placeholder="Ange din fråga...") | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| clear = gr.Button("Rensa") | |
| with gr.Column(scale=1): | |
| support_btn = gr.Button("Behöver du mer hjälp?", elem_classes="support-btn") | |
| with gr.Group(visible=False) as support_interface: | |
| gr.Markdown("### Vänligen fyll i din områdeskod, uttagsnummer och din email adress") | |
| with gr.Group(elem_classes="gr-form"): | |
| områdeskod = gr.Textbox(label="Områdeskod", placeholder="Områdeskod (valfritt)", info="Numeriskt värde") | |
| uttagsnummer = gr.Textbox(label="Uttagsnummer", placeholder="Uttagsnummer (valfritt)", info="Numeriskt värde") | |
| email = gr.Textbox(label="Din email adress", placeholder="din@email.se", info="Email adress krävs") | |
| gr.Markdown("### Chat som skickas till support:") | |
| chat_preview = gr.Markdown(elem_classes="chat-preview") | |
| with gr.Row(): | |
| back_btn = gr.Button("Tillbaka") | |
| send_support_btn = gr.Button("Skicka") | |
| with gr.Group(visible=False) as success_interface: | |
| gr.Markdown("Tack för att du kontaktar support@chargenode.eu. Vi återkommer inom kort", elem_classes="success-message") | |
| back_to_chat_btn = gr.Button("Tillbaka till chatten") | |
| def respond(message, chat_history, request: gr.Request): | |
| global last_log | |
| start = time.time() | |
| # Skicka chatthistorik till generate_answer | |
| response = generate_answer(message, chat_history) | |
| elapsed = round(time.time() - start, 2) | |
| timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") | |
| session_id = str(uuid.uuid4()) | |
| if last_log and 'session_id' in last_log: | |
| session_id = last_log.get('session_id') | |
| user_id = request.client.host if request else "okänd" | |
| ua_str = request.headers.get("user-agent", "") | |
| ref = request.headers.get("referer", "") | |
| ip = request.headers.get("x-forwarded-for", user_id).split(",")[0] | |
| ua = parse_ua(ua_str) | |
| browser = f"{ua.browser.family} {ua.browser.version_string}" | |
| osys = f"{ua.os.family} {ua.os.version_string}" | |
| platform = "webb" | |
| if "chargenode.eu" in ref: | |
| platform = "chargenode.eu" | |
| elif "localhost" in ref: | |
| platform = "test" | |
| elif "app" in ref: | |
| platform = "app" | |
| # Detektera språk för loggning | |
| lang = detect_language(message) | |
| log_data = { | |
| "timestamp": timestamp, | |
| "user_id": user_id, | |
| "session_id": session_id, | |
| "user_message": message, | |
| "bot_reply": response, | |
| "response_time": elapsed, | |
| "ip": ip, | |
| "browser": browser, | |
| "os": osys, | |
| "platform": platform, | |
| "language": lang | |
| } | |
| safe_append_to_log(log_data) | |
| last_log = log_data | |
| try: | |
| conversation_content = f""" | |
| *Ny konversation {timestamp}* | |
| *Språk:* {('English' if lang == 'en' else 'Svenska')} | |
| *Användare:* {message} | |
| *Bot:* {response[:300]}{'...' if len(response) > 300 else ''} | |
| *Sessionsinfo:* {session_id[:8]}... | {browser} | {platform} | |
| """ | |
| threading.Thread( | |
| target=lambda: send_to_slack(f"Ny konversation", conversation_content), | |
| daemon=True | |
| ).start() | |
| except Exception as e: | |
| print(f"❌ Kunde inte skicka konversation till Slack: {e}") | |
| chat_history.append({"role": "user", "content": message}) | |
| chat_history.append({"role": "assistant", "content": response}) | |
| return "", chat_history | |
| def format_chat_preview(chat_history): | |
| if not chat_history: | |
| return "Ingen chatthistorik att visa." | |
| preview = "" | |
| for msg in chat_history: | |
| sender = "Användare" if msg["role"] == "user" else "Bot" | |
| content = msg["content"] | |
| if len(content) > 100: | |
| content = content[:100] + "..." | |
| preview += f"**{sender}:** {content}\n\n" | |
| return preview | |
| def show_support_form(chat_history): | |
| preview = format_chat_preview(chat_history) | |
| return { | |
| chat_interface: gr.Group(visible=False), | |
| support_interface: gr.Group(visible=True), | |
| success_interface: gr.Group(visible=False), | |
| chat_preview: preview | |
| } | |
| def back_to_chat(): | |
| return { | |
| chat_interface: gr.Group(visible=True), | |
| support_interface: gr.Group(visible=False), | |
| success_interface: gr.Group(visible=False) | |
| } | |
| def submit_support_form(områdeskod, uttagsnummer, email, chat_history): | |
| """Hanterar formulärinskickningen med bättre felhantering.""" | |
| print(f"📝 Support-förfrågan: områdeskod={områdeskod}, uttagsnummer={uttagsnummer}, email={email}") | |
| validation_errors = [] | |
| # Använd validerings-hjälpfunktionen | |
| if error := validate_numeric_field(områdeskod, "Områdeskod"): | |
| validation_errors.append(error) | |
| if error := validate_numeric_field(uttagsnummer, "Uttagsnummer"): | |
| validation_errors.append(error) | |
| if not email: | |
| validation_errors.append("En giltig e-postadress krävs.") | |
| elif '@' not in email or '.' not in email.split('@')[1]: | |
| validation_errors.append("En giltig e-postadress krävs.") | |
| if validation_errors: | |
| print(f"❌ Valideringsfel: {validation_errors}") | |
| return { | |
| chat_interface: gr.Group(visible=False), | |
| support_interface: gr.Group(visible=True), | |
| success_interface: gr.Group(visible=False), | |
| chat_preview: "\n".join(["**Fel:**"] + validation_errors) | |
| } | |
| try: | |
| print("📤 Försöker skicka supportförfrågan till Slack...") | |
| success = send_support_to_slack(områdeskod, uttagsnummer, email, chat_history) | |
| if success: | |
| print("✅ Support-förfrågan skickad till Slack framgångsrikt") | |
| return { | |
| chat_interface: gr.Group(visible=False), | |
| support_interface: gr.Group(visible=False), | |
| success_interface: gr.Group(visible=True) | |
| } | |
| else: | |
| print("❌ Support-förfrågan till Slack misslyckades") | |
| return { | |
| chat_interface: gr.Group(visible=False), | |
| support_interface: gr.Group(visible=True), | |
| success_interface: gr.Group(visible=False), | |
| chat_preview: "**Ett fel uppstod när meddelandet skulle skickas. Vänligen försök igen senare.**" | |
| } | |
| except Exception as e: | |
| print(f"❌ Oväntat fel vid hantering av support-formulär: {e}") | |
| return { | |
| chat_interface: gr.Group(visible=False), | |
| support_interface: gr.Group(visible=True), | |
| success_interface: gr.Group(visible=False), | |
| chat_preview: f"**Ett fel uppstod: {str(e)}**" | |
| } | |
| msg.submit(respond, [msg, chatbot], [msg, chatbot]) | |
| clear.click(lambda: initial_chat.copy(), None, chatbot, queue=False) | |
| support_btn.click(show_support_form, chatbot, [chat_interface, support_interface, success_interface, chat_preview]) | |
| back_btn.click(back_to_chat, None, [chat_interface, support_interface, success_interface]) | |
| back_to_chat_btn.click(back_to_chat, None, [chat_interface, support_interface, success_interface]) | |
| send_support_btn.click( | |
| submit_support_form, | |
| [områdeskod, uttagsnummer, email, chatbot], | |
| [chat_interface, support_interface, success_interface, chat_preview] | |
| ) | |
| print("🚀 Förbereder embedding-modell och index...") | |
| initialize_embeddings() | |
| print("✅ Embedding-modell och index redo!") | |
| if __name__ == "__main__": | |
| app.launch(share=True) |