app.py

from dotenv import load_dotenv
from openai import OpenAI
from huggingface_hub import HfApi, create_repo, hf_hub_download
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse
from pydantic import BaseModel, Field, validator
import json
import logging
import os
import re
import requests
import sys
import threading
from datetime import datetime

load_dotenv(override=True)


class JsonLogFormatter(logging.Formatter):
    """Une ligne JSON par événement (LOG_FORMAT=json sur HF)."""

    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "timestamp": self.formatTime(
                record, datefmt="%Y-%m-%dT%H:%M:%S"
            ),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
        }
        if record.exc_info:
            payload["exception"] = self.formatException(record.exc_info)
        return json.dumps(payload, ensure_ascii=False)


def _configure_logging() -> None:
    level_name = os.getenv("LOG_LEVEL", "INFO").upper()
    level = getattr(logging, level_name, logging.INFO)
    root = logging.getLogger()
    root.setLevel(level)
    if root.handlers:
        return
    handler = logging.StreamHandler(sys.stdout)
    if os.getenv("LOG_FORMAT", "").lower() == "json":
        handler.setFormatter(JsonLogFormatter())
    else:
        handler.setFormatter(logging.Formatter(
            "%(asctime)s | %(levelname)s | %(name)s | %(message)s",
            datefmt="%Y-%m-%dT%H:%M:%S",
        ))
    root.addHandler(handler)


_configure_logging()

log = logging.getLogger("wadagni.hf_logs")
questions_log = logging.getLogger("wadagni.questions")
upload_log = logging.getLogger("wadagni.upload")
security_log = logging.getLogger("wadagni.security")
tools_log = logging.getLogger("wadagni.tools")


# ── Configuration confidentialité ────────────────────────────────────────────

PRIVACY_NOTICE = (
    "Avant l'envoi de votre message, veuillez noter que son contenu peut être enregistré "
    "à des fins de suivi et d'amélioration du service .\n\n"
    "Merci de ne pas inclure de données sensibles (opinions politiques détaillées, "
    "données de santé, informations bancaires) ni de données personnelles non nécessaires "
    "(adresse, téléphone, numéro d'identité).\n\n"
    "En cochant la case de consentement et en envoyant votre message, vous acceptez ces conditions."
)

CONSENT_REQUIRED_MESSAGE = (
    "⚠️ Avant de pouvoir échanger avec l'assistant, veuillez accepter notre "
    "politique de confidentialité en cochant la case prévue à cet effet."
)


# ── Initialisation du dataset de logs ────────────────────────────────────────

def init_logs_dataset():
    token = os.getenv("HF_TOKEN")
    dataset_id = os.getenv("LOGS_DATASET_ID")
    if not token or not dataset_id:
        log.warning(
            "HF_TOKEN ou LOGS_DATASET_ID manquant, dataset non initialisé"
        )
        return
    try:
        create_repo(
            repo_id=dataset_id,
            repo_type="dataset",
            private=True,
            token=token,
            exist_ok=True,
        )
        log.info("Dataset logs prêt : %s", dataset_id)
    except Exception:
        log.exception("Erreur init dataset logs")
    try:
        path = hf_hub_download(
            repo_id=dataset_id,
            repo_type="dataset",
            filename="db/log.txt",
            token=token,
        )
        with open(path, "r", encoding="utf-8") as src:
            content = src.read()
        with open("db/log.txt", "w", encoding="utf-8") as dst:
            dst.write(content)
        log.info(
            "Fichier log.txt récupéré (%s lignes)",
            len(content.splitlines()),
        )
    except Exception:
        log.info("Aucun fichier log.txt sur HF, démarrage à vide")
init_logs_dataset()

# ── Initialisation du dataset Questions ─────────────────────────────────────

def init_questions_dataset():
    token = os.getenv("HF_TOKEN")
    questions_id = os.getenv("QUESTIONS_DATASET_ID")
    if not token or not questions_id:
        questions_log.warning(
            "HF_TOKEN ou QUESTIONS_DATASET_ID manquant, "
            "dataset non initialisé"
        )
        return
    try:
        create_repo(
            repo_id=questions_id,
            repo_type="dataset",
            private=True,
            token=token,
            exist_ok=True,
        )
        questions_log.info("Dataset questions prêt : %s", questions_id)
    except Exception:
        questions_log.exception("Erreur init dataset questions")
    try:
        path = hf_hub_download(
            repo_id=questions_id,
            repo_type="dataset",
            filename="questions.txt",
            token=token,
        )
        with open(path, "r", encoding="utf-8") as src:
            content = src.read()
        with open("db/questions.txt", "w", encoding="utf-8") as dst:
            dst.write(content)
        questions_log.info(
            "Fichier questions.txt récupéré (%s lignes)",
            len(content.splitlines()),
        )
    except Exception:
        questions_log.info(
            "Aucun fichier questions.txt sur HF, démarrage à vide"
        )
init_questions_dataset()

def _upload_to_hf(path_or_fileobj, path_in_repo, repo_id):
    """Upload en arrière-plan vers HF avec retry."""
    try:
        api = HfApi()
        api.upload_file(
            path_or_fileobj=path_or_fileobj,
            path_in_repo=path_in_repo,
            repo_id=repo_id,
            repo_type="dataset",
            token=os.getenv("HF_TOKEN"),
        )
    except Exception:
        upload_log.exception(
            "Erreur upload HF path_in_repo=%s repo_id=%s",
            path_in_repo,
            repo_id,
        )

def push_question(question):
    """Enregistre chaque question posée par un utilisateur dans le dataset Questions."""
    now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    questions_log.info("%s — %s", now, question)
    questions_path = "db/questions.txt"
    entry = f"[{now}] {question}\n"
    with open(questions_path, "a", encoding="utf-8") as f:
        f.write(entry)
    threading.Thread(
        target=_upload_to_hf,
        args=(questions_path, "questions.txt", os.getenv("QUESTIONS_DATASET_ID")),
        daemon=True,
    ).start()

def push(text):
    log.info("%s", text)
    log_path = "db/log.txt"
    entry = f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {text}\n"
    with open(log_path, "a", encoding="utf-8") as f:
        f.write(entry)
    threading.Thread(
        target=_upload_to_hf,
        args=(log_path, log_path, os.getenv("LOGS_DATASET_ID")),
        daemon=True,
    ).start()


# ── Outils (tools) ───────────────────────────────────────────────────────────

def record_user_details(email, name="Name not provided", notes="not provided"):
    push(f"Recording {name} with email {email} and notes {notes}")
    return {"recorded": "ok"}

def record_unknown_question(question):
    push(f"Recording {question}")
    return {"recorded": "ok"}

record_user_details_json = {
    "name": "record_user_details",
    "description": "Utilise cet outil pour enregistrer qu'un utilisateur souhaite être contacté et a fourni une adresse e-mail",
    "parameters": {
        "type": "object",
        "properties": {
            "email": {
                "type": "string",
                "description": "L'adresse e-mail de cet utilisateur"
            },
            "name": {
                "type": "string",
                "description": "Le nom de l'utilisateur, s'il l'a fourni"
            },
            "notes": {
                "type": "string",
                "description": "Toute information supplémentaire sur la conversation qui mérite d'être enregistrée pour donner du contexte"
            }
        },
        "required": ["email"],
        "additionalProperties": False
    }
}

record_unknown_question_json = {
    "name": "record_unknown_question",
    "description": "Utilise toujours cet outil pour enregistrer toute question à laquelle tu n'as pas pu répondre faute de connaissance",
    "parameters": {
        "type": "object",
        "properties": {
            "question": {
                "type": "string",
                "description": "La question à laquelle il n'a pas été possible de répondre"
            },
        },
        "required": ["question"],
        "additionalProperties": False
    }
}

tools = [
    {"type": "function", "function": record_user_details_json},
    {"type": "function", "function": record_unknown_question_json}
]

# ── Whitelist des outils autorisés (sécurité) ──────────────────────────────

ALLOWED_TOOLS = {
    "record_user_details": record_user_details,
    "record_unknown_question": record_unknown_question,
}


# ── Classe principale ─────────────────────────────────────────────────────────

class Me:

    def __init__(self):
        self.openai = OpenAI()
        self.name = "Romuald WADAGNI"

        # Chargement de la base vectorielle FAISS
        self.db_index = FAISS.load_local(
            folder_path="db",
            index_name="db_index",
            embeddings=OpenAIEmbeddings(),
            allow_dangerous_deserialization=True,
        )

    def handle_tool_call(self, tool_calls):
        results = []
        for tool_call in tool_calls:
            tool_name = tool_call.function.name
            arguments = json.loads(tool_call.function.arguments)
            tools_log.info("Tool called: %s", tool_name)
            tool = ALLOWED_TOOLS.get(tool_name)
            if not tool:
                security_log.warning(
                    "Tool non autorisé refusé : %s", tool_name
                )
                result = {"error": "tool not allowed"}
            else:
                result = tool(**arguments)
            results.append({
                "role": "tool",
                "content": json.dumps(result),
                "tool_call_id": tool_call.id
            })
        return results

    def system_prompt(self, query):
        # Recherche des passages les plus pertinents dans la base vectorielle
        similar_documents = self.db_index.similarity_search(query, k=3)
        message_content = re.sub(r'\n{2}', ' ', '\n '.join(
            [f'Extrait du document n°{i+1} :\n' + doc.page_content
             for i, doc in enumerate(similar_documents)]))

        prompt = (
            f"You are the communication assistant of {self.name}, elected President of Benin. \
            You answer questions from visitors on his website dedicated to his record in office and his ambitions. \
            Your mission is to represent {self.name} faithfully and to convince visitors to support his actions and projects for the future. \
            Be professional and engaging, as if you were speaking to citizens who care about their country's development. \
            If you do not know the answer to a question, use the record_unknown_question tool to record it. \
            Do not forget: you must highlight {self.name}'s record and initiatives, while encouraging citizens to continue supporting his actions.\
            If a user wishes to be contacted, ask for their email address and record it using the record_user_details tool." 
        )
        prompt += f"\n\n Relevant documents:\n{message_content}\n\n"
        prompt += f"With this context, please chat with the user, always staying in character as the communication assistant of {self.name}."
        return prompt

    def chat(self, message, history):
        push_question(message)
        history_messages = []
        for item in history:
            if isinstance(item, dict):
                history_messages.append(item)
            else:
                history_messages.append({"role": "user", "content": item[0]})
                if item[1]:
                    history_messages.append({"role": "assistant", "content": item[1]})
        messages = (
            [{"role": "system", "content": self.system_prompt(message)}]
            + history_messages
            + [{"role": "user", "content": message}]
        )
        try:
            done = False
            while not done:
                response = self.openai.chat.completions.create(
                    model="gpt-4o-mini",
                    messages=messages,
                    tools=tools,
                    temperature=0.3,
                )
                if response.choices[0].finish_reason == "tool_calls":
                    msg = response.choices[0].message
                    results = self.handle_tool_call(msg.tool_calls)
                    messages.append(msg)
                    messages.extend(results)
                else:
                    done = True
            return response.choices[0].message.content
        except requests.RequestException:
            return "Notre service est momentanément indisponible. Veuillez réessayer dans quelques instants."
        except (KeyError, IndexError, ValueError):
            return "Une erreur inattendue s'est produite lors du traitement de votre demande. Veuillez reformuler votre question ou réessayer."


# ── Application FastAPI ───────────────────────────────────────────────────────

app = FastAPI()
me = Me()


# ── Rate limiting ───────────────────────────────────────────────────

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)


# ── En-têtes de sécurité HTTP ─────────────────────────────────────────

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request

class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        response = await call_next(request)
        response.headers["X-Frame-Options"] = "SAMEORIGIN"
        response.headers["X-Content-Type-Options"] = "nosniff"
        response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
        response.headers["Content-Security-Policy"] = (
            "default-src 'self'; "
            "script-src 'self' 'unsafe-inline' https://cdn.jsdelivr.net; "
            "style-src 'self' 'unsafe-inline'; "
            "img-src 'self' data:; "
            "frame-ancestors 'self' https://*.hf.space https://huggingface.co"
        )
        return response

app.add_middleware(SecurityHeadersMiddleware)


class ChatRequest(BaseModel):
    message: str = Field(..., max_length=2000)
    history: list = Field(default_factory=list)
    consent: bool = False

    @validator("history")
    def limit_history(cls, v):
        if len(v) > 50:
            return v[-50:]
        return v


@app.get("/health")
def health():
    """Sonde de disponibilité pour monitoring / Hugging Face."""
    return {"status": "ok"}


@app.get("/privacy")
def privacy():
    """Retourne la notice de confidentialité pour affichage côté frontend."""
    return {"notice": PRIVACY_NOTICE}


@app.get("/photo")
def get_photo():
    photo_path = "db/photo.jpg"
    if os.path.exists(photo_path):
        return FileResponse(photo_path, media_type="image/jpeg")
    return FileResponse("static/placeholder.png", media_type="image/png")


@app.post("/chat")
@limiter.limit("20/minute")
def chat_endpoint(req: ChatRequest, request: Request):
    # ── Vérification du consentement explicite ───────────────────────────
    if not req.consent:
        return {
            "response": CONSENT_REQUIRED_MESSAGE,
            "consent_required": True
        }

    # ── Traitement normal avec consentement validé ───────────────────────
    response = me.chat(req.message, req.history)
    return {"response": response, "consent_required": False}


# Servir le frontend (doit être monté en dernier)
app.mount("/", StaticFiles(directory="static", html=True), name="static")