Gå til innhold

Clubsite AI Chatbot

API Backend - app.py (FastAPI)

clubsite-ai-chatbot

Kildekode: `app.py` (FastAPI Integrasjon)

Flaskehalsen mellom databasen, klientsiden (Widget) og Google Gemini AI-modellen.

Mål og Intensjon

Endpoints: Tilbyr et /api/v1/chat-endepunkt som tar imot JSON-historikk, ClubId (for databasekontekst) og system-prompts.
Langchain/Gemini: Kobler seg opp mot AI modellen via \google.generativeai\ logikken.
CORS Sikkerhet: Godtar kun de hvitlistede domene-adressesene for Clubsite slik at ikke tredjeparter misbruke API-kosten vår.

Kildekode

import os
import json
import re
import datetime
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from typing import List, Optional
from pydantic import BaseModel
from google import genai
from google.genai import types
from dotenv import load_dotenv

# Laster miljøvariabler (.env)
load_dotenv()
master_api_key = os.getenv("GEMINI_API_KEY")

if not master_api_key:
    raise RuntimeError("Mangler GEMINI_API_KEY i .env!")

# Initialiserer FastAPI-appen
app = FastAPI(title="Clubsite AI Chatbot API", version="1.0.0")

# Setter opp CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

class Message(BaseModel):
    role: str
    content: str

class ChatRequest(BaseModel):
    tenant_id: str
    message: str
    history: Optional[List[Message]] = []

def clean_html(raw_html):
    """Enkel fjerning av HTML-tags for å spare tokens og rydde støy for AI-en."""
    if not isinstance(raw_html, str):
        return ""
    cleanr = re.compile('<.*?>')
    cleantext = re.sub(cleanr, '', raw_html)
    # Fjerner også unødvendig whitespace og newlines
    return " ".join(cleantext.split())

def build_system_prompt(tenant_id: str) -> str:
    """Leser lokal JSON-cache for å bygge systeminstruksen for AI."""
    base_instructions = (
        f"Du er en vennlig, profesjonell og hjelpsom kundeservice-assistent for {tenant_id} Golfklubb. "
        "Svar alltid på en naturlig, imøtekommende og engasjerende måte. Siden mange brukere er på mobil, MÅ du holde svarene konsise, luftige og lettleste. Finn en god balanse mellom korte kulepunkter og korte avsnitt (unngå lange 'stiler' og store tekstblokker). "
        "Du skal svare kun på norsk. Faktainformasjon om klubben (priser, banelengder, utslagssteder, kontaktinfo) MÅ baseres PÅ informasjonen under. "
        "Dersom brukeren spør om golftips eller ber deg tilpasse strategien (f.eks. for 'senior', 'dame', nybegynner), bruk din generelle golfkunnskap for å gi persontilpassede råd. Vær kort, varm og poengtert, samtidig som du inkluderer de spesifikke banefaktaene. "
        "Finner du ikke nøyaktig informasjon om klubbens spesifikke regler i teksten, si det vennlig, men gi gjerne et kort, generelt golf-svar der det passer. "
        "VIKTIG REGEL OM KONTAKTINFO: Hvis noen spør om hvordan de skal kontakte styremedlemmer, ansatte eller klubben generelt, "
        "skal du ALLTID henvise til klubbens offisielle kontaktinformasjon. Referer til kontaktsiden og epostadressen du finner i 'SETUP_BASIC'. "
        f"For eksempel lenken i 'urlContact' (f.eks. https://{tenant_id.lower()}gk.no/... ) eller annen generell klubb-epost. "
        "Gi ALDRI ut private telefonnumre eller private e-postadresser, vi skjermer personvernet strengt. "
        "Aldri finn på priser eller klubbspesifikk informasjon som ikke ligger i datasettet.\n\n"
    )

    cache_path = os.path.join(os.path.dirname(__file__), "cache", f"{tenant_id}_context.json")

    if not os.path.exists(cache_path):
        return base_instructions + "NB: Ingen ekstra informasjon er lagret for denne klubben ennå."

    try:
        with open(cache_path, "r", encoding="utf-8") as f:
            data = json.load(f)

        context_parts = []

        # Itererer gjennom alle datasett lagret i cachen (unntatt metadata)
        for key, rows in data.items():
            if key == "metadata" or not rows:
                continue

            context_parts.append(f"\n--- DATA FRA: {key.upper()} ---")

            for row in rows:
                # Filtrer ut interne IDer, ubrukelige felter, og PRIVAT PII (epost/telefon)
                row_info = []
                for k, v in row.items():
                    k_lower = k.lower()
                    if "id" in k_lower or k_lower in ["date_created", "created_by", "modified_by", "sortnum", "boolarchive"]:
                        continue

                    # Sørger for at private e-poster og telefonnumre ALDRI lastes inn i AI-ens minne (unntatt Setup_Basic som er klubbens offisielle info)
                    if key.lower() != "setup_basic" and any(pii_word in k_lower for pii_word in ["epost", "email", "mail", "telefon", "phone", "mobil", "tlf"]):
                        continue

                    val_str = clean_html(str(v))

                    # Ignorer veldig lange tekster hvis det trengs, men for Pages er de ofte viktige:
                    if len(val_str) > 1500:
                        val_str = val_str[:1500] + "..."

                    row_info.append(f"{k}: {val_str}")

                # Legg til raden i konteksten
                context_parts.append(" | ".join(row_info))

        full_prompt = base_instructions + "\n".join(context_parts)
        return full_prompt
    except Exception as e:
        print(f"Feil ved lesing av cache for {tenant_id}: {e}")
        return base_instructions + "Feil ved lasting av klubbdata."

def log_interaction(tenant_id: str, question: str, answer: str):
    """Logger spørsmål og svar til en JSONL-fil per tenant for statistikk og FAQ-bygging"""
    logs_dir = os.path.join(os.path.dirname(__file__), "logs")
    os.makedirs(logs_dir, exist_ok=True)

    log_path = os.path.join(logs_dir, f"{tenant_id}_chatlog.jsonl")

    log_entry = {
        "timestamp": datetime.datetime.now().isoformat(),
        "tenant_id": tenant_id,
        "question": question,
        "answer": answer
    }

    try:
        with open(log_path, "a", encoding="utf-8") as f:
            f.write(json.dumps(log_entry, ensure_ascii=False) + "\n")
    except Exception as e:
        print(f"Feil ved logging av chat: {e}")

@app.post("/api/chat")
async def chat(request: ChatRequest):
    tenant_id = request.tenant_id
    user_message = request.message
    history = request.history or []

    current_api_key = master_api_key

    try:
        # Henter System Prompt fra lokal disk (Cache Worker-en)
        system_instructions = build_system_prompt(tenant_id)

        client = genai.Client(api_key=current_api_key)

        # Konfigurerer requesten og sender inn instruksen til modellen slik at den forstår "hvem den er"
        config = types.GenerateContentConfig(
            system_instruction=system_instructions,
            temperature=0.4 # Litt varmere temperatur = mer beskrivende, naturlig språk og tilpasset tone, men fortsatt faktabasert
        )

        # Bygger samtalekontekst iterativt
        contents = []
        for msg in history:
            role = "model" if msg.role in ["bot", "model", "assistant"] else "user"
            contents.append(types.Content(role=role, parts=[types.Part.from_text(text=msg.content)]))

        # Legger til selve det nye spørsmålet
        contents.append(types.Content(role="user", parts=[types.Part.from_text(text=user_message)]))

        response = client.models.generate_content(
            model='gemini-2.5-flash',
            contents=contents,
            config=config
        )

        # Logg samtalen for statistikk/FAQ-bygging
        log_interaction(tenant_id, user_message, response.text)

        return {
            "tenant_id": tenant_id,
            "response": response.text,
            "status": "success"
        }

    except Exception as e:
        print(f"Feil for tenant {tenant_id}: {str(e)}")
        raise HTTPException(status_code=500, detail="Kunne ikke koble til AI-motoren.")

@app.get("/health")
def health_check():
    return {"status": "ok", "message": "AI Chatbot API er oppe og kjører!"}

if __name__ == "__main__":
    import uvicorn
    # Veldig greit for lokal testing. Kjør med: python app.py
    uvicorn.run("app:app", host="127.0.0.1", port=8000, reload=True)