app.py

"""
Plateforme de veille automatisée - Côte d'Ivoire
--------------------------------------------------
Pipeline : recherche web (Tavily) -> extraction (trafilatura) -> synthèse (Gemini) -> cache (SQLite)

Pour lancer :
    pip install -r requirements.txt
    export TAVILY_API_KEY="ta_cle"
    export GEMINI_API_KEY="ta_cle"
    python app.py
"""

import os
import json
import sqlite3
import hashlib
import time
from datetime import datetime, timedelta

import gradio as gr
import trafilatura
from tavily import TavilyClient
import google.generativeai as genai

# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------

TAVILY_API_KEY = os.environ.get("TAVILY_API_KEY", "")
GEMINI_API_KEY = os.environ.get("GEMINI_API_KEY", "")

DB_PATH = "veille_cache.db"
CACHE_TTL_HOURS = 6  # durée de validité du cache pour un même thème
MAX_SOURCES = 6      # nombre de pages à extraire et synthétiser
MAX_CHARS_PER_SOURCE = 4000  # limite de texte extrait par source

PAYS = "Côte d'Ivoire"

# Domaines de confiance à privilégier (optionnel, améliore la pertinence locale)
DOMAINES_PRIORITAIRES = [
    "gouv.ci",
    "aip.ci",
    "fratmat.info",
    "linfodrome.com",
    "abidjan.net",
    "connectionivoirienne.net",
]

if GEMINI_API_KEY:
    genai.configure(api_key=GEMINI_API_KEY)

# ---------------------------------------------------------------------------
# Cache SQLite
# ---------------------------------------------------------------------------

def init_db():
    conn = sqlite3.connect(DB_PATH)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS cache (
            cle TEXT PRIMARY KEY,
            theme TEXT,
            resultat TEXT,
            date_creation TEXT
        )
    """)
    conn.commit()
    conn.close()


def cle_cache(theme: str) -> str:
    return hashlib.sha256(theme.strip().lower().encode()).hexdigest()


def lire_cache(theme: str):
    conn = sqlite3.connect(DB_PATH)
    row = conn.execute(
        "SELECT resultat, date_creation FROM cache WHERE cle = ?",
        (cle_cache(theme),),
    ).fetchone()
    conn.close()
    if not row:
        return None
    resultat, date_creation = row
    date_obj = datetime.fromisoformat(date_creation)
    if datetime.now() - date_obj > timedelta(hours=CACHE_TTL_HOURS):
        return None
    return json.loads(resultat)


def ecrire_cache(theme: str, resultat: dict):
    conn = sqlite3.connect(DB_PATH)
    conn.execute(
        "INSERT OR REPLACE INTO cache (cle, theme, resultat, date_creation) VALUES (?, ?, ?, ?)",
        (cle_cache(theme), theme, json.dumps(resultat, ensure_ascii=False), datetime.now().isoformat()),
    )
    conn.commit()
    conn.close()


# ---------------------------------------------------------------------------
# Étape 1 : Recherche web (Tavily)
# ---------------------------------------------------------------------------

def rechercher(theme: str) -> list:
    if not TAVILY_API_KEY:
        raise RuntimeError("TAVILY_API_KEY manquante. Voir README.md pour l'obtenir gratuitement.")

    client = TavilyClient(api_key=TAVILY_API_KEY)

    requete = f"{theme} {PAYS} actualité récente 2026"

    reponse = client.search(
        query=requete,
        search_depth="advanced",
        max_results=MAX_SOURCES,
        include_answer=False,
        topic="news",
    )

    resultats = reponse.get("results", [])

    # Tri léger : on remonte les domaines prioritaires en tête de liste
    def score_domaine(url):
        return 0 if any(d in url for d in DOMAINES_PRIORITAIRES) else 1

    resultats.sort(key=lambda r: score_domaine(r.get("url", "")))
    return resultats


# ---------------------------------------------------------------------------
# Étape 2 : Extraction du contenu (trafilatura)
# ---------------------------------------------------------------------------

def extraire_contenu(url: str) -> str:
    try:
        html = trafilatura.fetch_url(url, timeout=15)
        if not html:
            return ""
        texte = trafilatura.extract(html, include_comments=False, include_tables=False)
        if not texte:
            return ""
        return texte[:MAX_CHARS_PER_SOURCE]
    except Exception:
        return ""


# ---------------------------------------------------------------------------
# Étape 3 : Synthèse (Gemini)
# ---------------------------------------------------------------------------

def synthetiser(theme: str, sources: list) -> dict:
    if not GEMINI_API_KEY:
        raise RuntimeError("GEMINI_API_KEY manquante. Voir README.md pour l'obtenir gratuitement.")

    # Construction du contexte numéroté pour permettre la citation [1], [2], etc.
    contexte = ""
    for i, s in enumerate(sources, start=1):
        contexte += f"\n\n--- SOURCE [{i}] ---\nTitre: {s['titre']}\nURL: {s['url']}\nContenu:\n{s['contenu']}\n"

    prompt = f"""Tu es un assistant de veille documentaire spécialisé sur la Côte d'Ivoire.

Thème demandé par l'utilisateur : "{theme}"

Voici des extraits de sources récentes sur ce thème :
{contexte}

Consignes :
1. Rédige une synthèse claire et structurée (400-600 mots) des informations les plus récentes et pertinentes sur ce thème en Côte d'Ivoire.
2. Chaque affirmation factuelle doit être suivie du numéro de sa source entre crochets, ex: [1], [2].
3. Structure ta réponse en 3-5 points clés (puces), pas un simple paragraphe.
4. Si les sources se contredisent ou sont incertaines, signale-le.
5. N'invente aucune information absente des sources.
6. Réponds UNIQUEMENT en JSON valide, sans texte autour, avec ce format exact :

{{
  "titre_synthese": "titre court et clair",
  "points_cles": [
    "point 1 avec citation [1]",
    "point 2 avec citation [2][3]"
  ],
  "synthese_complete": "texte complet de la synthèse avec citations [1][2] intégrées",
  "fiabilite": "haute|moyenne|faible",
  "note_fiabilite": "courte explication si fiabilité moyenne ou faible"
}}
"""

    model = genai.GenerativeModel("gemini-2.0-flash")
    reponse = model.generate_content(prompt)

    texte_brut = reponse.text.strip()
    # Nettoyage si le modèle entoure le JSON de balises markdown
    if texte_brut.startswith("```"):
        texte_brut = texte_brut.split("```")[1]
        if texte_brut.startswith("json"):
            texte_brut = texte_brut[4:]
    texte_brut = texte_brut.strip()

    return json.loads(texte_brut)


# ---------------------------------------------------------------------------
# Pipeline complet
# ---------------------------------------------------------------------------

def pipeline_veille(theme: str, forcer_actualisation: bool = False, progress=gr.Progress()):
    if not theme or not theme.strip():
        return "Merci d'entrer un thème (ex: éducation, santé, agriculture...).", ""

    theme = theme.strip()

    if not forcer_actualisation:
        cache = lire_cache(theme)
        if cache:
            return formater_resultat(cache, depuis_cache=True), formater_sources_html(cache["sources"])

    progress(0.1, desc="Recherche des dernières informations...")
    resultats_bruts = rechercher(theme)

    if not resultats_bruts:
        return f"Aucun résultat trouvé pour '{theme}'. Essaie une formulation plus large.", ""

    progress(0.35, desc="Extraction du contenu des pages...")
    sources = []
    for r in resultats_bruts:
        contenu = extraire_contenu(r.get("url", ""))
        if contenu:
            sources.append({
                "titre": r.get("title", "Sans titre"),
                "url": r.get("url", ""),
                "contenu": contenu,
                "date": r.get("published_date", "Date inconnue"),
            })
        if len(sources) >= MAX_SOURCES:
            break

    if not sources:
        return f"Impossible d'extraire le contenu des sources trouvées pour '{theme}'.", ""

    progress(0.7, desc="Synthèse par l'IA...")
    synthese = synthetiser(theme, sources)
    synthese["sources"] = sources
    synthese["theme"] = theme
    synthese["date_generation"] = datetime.now().strftime("%d/%m/%Y %H:%M")

    ecrire_cache(theme, synthese)

    progress(1.0, desc="Terminé")
    return formater_resultat(synthese, depuis_cache=False), formater_sources_html(sources)


def formater_resultat(resultat: dict, depuis_cache: bool) -> str:
    badge_cache = "\n\n*(Résultat servi depuis le cache — regénère si besoin)*" if depuis_cache else ""
    badge_fiabilite = {
        "haute": "🟢 Fiabilité haute",
        "moyenne": "🟡 Fiabilité moyenne",
        "faible": "🔴 Fiabilité faible",
    }.get(resultat.get("fiabilite", "moyenne"), "🟡 Fiabilité moyenne")

    points = "\n".join(f"- {p}" for p in resultat.get("points_cles", []))

    note = resultat.get("note_fiabilite", "")
    note_bloc = f"\n\n> ⚠️ {note}" if note else ""

    return f"""## {resultat.get('titre_synthese', 'Synthèse')}

**Thème :** {resultat.get('theme', '')} | **Généré le :** {resultat.get('date_generation', '')} | {badge_fiabilite}{note_bloc}

### Points clés
{points}

### Synthèse complète
{resultat.get('synthese_complete', '')}
{badge_cache}
"""


def formater_sources_html(sources: list) -> str:
    if not sources:
        return ""
    html = "<div style='font-family: sans-serif;'><h3>Sources</h3><ol>"
    for s in sources:
        html += f"<li style='margin-bottom:10px;'><a href='{s['url']}' target='_blank'><b>{s['titre']}</b></a><br><span style='color:#666;font-size:0.9em;'>{s['url']}</span></li>"
    html += "</ol></div>"
    return html


# ---------------------------------------------------------------------------
# Interface Gradio
# ---------------------------------------------------------------------------

def construire_interface():
    with gr.Blocks(title="Veille CI - IA") as demo:
        gr.Markdown("""
        # 🔎 Veille automatisée — Côte d'Ivoire
        Entre un thème (ex: *éducation*, *santé*, *agriculture*, *numérique*...) et l'IA cherche,
        lit et synthétise les informations les plus récentes, avec les sources citées.
        """)

        with gr.Row():
            theme_input = gr.Textbox(
                label="Thème",
                placeholder="ex: éducation",
                scale=4,
            )
            forcer = gr.Checkbox(label="Forcer l'actualisation (ignorer le cache)", scale=1)

        bouton = gr.Button("Lancer la veille", variant="primary")

        resultat_md = gr.Markdown()
        sources_html = gr.HTML()

        bouton.click(
            fn=pipeline_veille,
            inputs=[theme_input, forcer],
            outputs=[resultat_md, sources_html],
        )
        theme_input.submit(
            fn=pipeline_veille,
            inputs=[theme_input, forcer],
            outputs=[resultat_md, sources_html],
        )

    return demo


if __name__ == "__main__":
    init_db()
    demo = construire_interface()
    demo.launch()