gestionnaire_historique.py — persistance historique SQLite Phase 1 (18/10/2026)

Par /

–– coding: utf-8 ––

"""
MIRA — Gestionnaire de l’historique des conversations
Fichier : gestionnaire_historique.py
Base : /mnt/sda1/MIRA_Project/data/historique.db
Phase : 1 — Persistance

Décisions retenues :

  • SQLite, deux bases séparées (historique / utilisateur)
  • Horodatage ISO 8601 sur chaque échange
  • Compression différée : récents en clair, anciens résumés
  • Seuil de compression : 100 échanges OU 7 jours
  • Seuil géré automatiquement par MIRA à terme (Niveau 2)
    """

import sqlite3
import os
from datetime import datetime, timezone, timedelta

———————————————————-

CHEMIN — à centraliser dans config/chemins.json en v2

———————————————————-

CHEMIN_BASE = "/mnt/sda1/MIRA_Project/data/historique.db"

———————————————————-

SEUILS DE COMPRESSION

———————————————————-

SEUIL_ECHANGES = 100 # nombre d’échanges avant compression
SEUIL_JOURS = 7 # jours avant compression des anciens

==========================================================

INITIALISATION

==========================================================

def initialiser():
"""
Crée la base et les tables si elles n’existent pas.
Peut être appelée à chaque démarrage sans risque
(idempotent).
"""
dossier = os.path.dirname(CHEMIN_BASE)
if dossier and not os.path.exists(dossier):
os.makedirs(dossier)

connexion = sqlite3.connect(CHEMIN_BASE)
curseur   = connexion.cursor()

curseur.execute("""
    CREATE TABLE IF NOT EXISTS echanges (
        id              INTEGER PRIMARY KEY AUTOINCREMENT,
        tour            INTEGER NOT NULL,
        horodatage      TEXT    NOT NULL,
        question        TEXT    NOT NULL,
        reponse         TEXT    NOT NULL,
        source          TEXT    NOT NULL DEFAULT 'phi4',
        compresse       INTEGER NOT NULL DEFAULT 0,
        resume          TEXT
    )
""")

curseur.execute("""
    CREATE TABLE IF NOT EXISTS questions_en_attente (
        id          INTEGER PRIMARY KEY AUTOINCREMENT,
        horodatage  TEXT    NOT NULL,
        question    TEXT    NOT NULL,
        statut      TEXT    NOT NULL DEFAULT 'en_attente'
    )
""")

connexion.commit()
connexion.close()

==========================================================

ÉCRITURE

==========================================================

def ajouter_echange(tour, question, reponse, source="phi4"):
horodatage = datetime.now(timezone.utc).isoformat()
connexion = sqlite3.connect(CHEMIN_BASE)
curseur = connexion.cursor()

curseur.execute("""
    INSERT INTO echanges (tour, horodatage, question, reponse, source)
    VALUES (?, ?, ?, ?, ?)
""", (tour, horodatage, question, reponse, source))

connexion.commit()
connexion.close()

def ajouter_question_en_attente(question):
horodatage = datetime.now(timezone.utc).isoformat()
connexion = sqlite3.connect(CHEMIN_BASE)
curseur = connexion.cursor()

curseur.execute("""
    INSERT INTO questions_en_attente (horodatage, question)
    VALUES (?, ?)
""", (horodatage, question))

connexion.commit()
connexion.close()

==========================================================

LECTURE

==========================================================

def lire_historique_recent(limite=20):
connexion = sqlite3.connect(CHEMIN_BASE)
curseur = connexion.cursor()

curseur.execute("""
    SELECT tour, horodatage, question, reponse, source
    FROM echanges
    WHERE compresse = 0
    ORDER BY id DESC
    LIMIT ?
""", (limite,))

lignes = curseur.fetchall()
connexion.close()

lignes.reverse()
return [
    {
        "tour"       : ligne[0],
        "horodatage" : ligne[1],
        "question"   : ligne[2],
        "reponse"    : ligne[3],
        "source"     : ligne[4]
    }
    for ligne in lignes
]

def lire_questions_en_attente():
connexion = sqlite3.connect(CHEMIN_BASE)
curseur = connexion.cursor()

curseur.execute("""
    SELECT id, horodatage, question
    FROM questions_en_attente
    WHERE statut = 'en_attente'
    ORDER BY id ASC
""")

lignes = curseur.fetchall()
connexion.close()

return [
    {
        "id"         : ligne[0],
        "horodatage" : ligne[1],
        "question"   : ligne[2]
    }
    for ligne in lignes
]

def compter_echanges_en_clair():
connexion = sqlite3.connect(CHEMIN_BASE)
curseur = connexion.cursor()
curseur.execute("SELECT COUNT(*) FROM echanges WHERE compresse = 0")
total = curseur.fetchone()[0]
connexion.close()
return total

==========================================================

COMPRESSION DIFFÉRÉE (Option B)

==========================================================

def verifier_seuil_compression():
connexion = sqlite3.connect(CHEMIN_BASE)
curseur = connexion.cursor()

curseur.execute("SELECT COUNT(*) FROM echanges WHERE compresse = 0")
total = curseur.fetchone()[0]

if total >= SEUIL_ECHANGES:
    connexion.close()
    return True

curseur.execute("""
    SELECT horodatage FROM echanges
    WHERE compresse = 0
    ORDER BY id ASC
    LIMIT 1
""")
ligne = curseur.fetchone()
connexion.close()

if ligne:
    horodatage_ancien = datetime.fromisoformat(ligne[0])
    maintenant        = datetime.now(timezone.utc)
    if (maintenant - horodatage_ancien) > timedelta(days=SEUIL_JOURS):
        return True

return False

def lire_echanges_a_comprimer(garder_recents=20):
connexion = sqlite3.connect(CHEMIN_BASE)
curseur = connexion.cursor()

curseur.execute("""
    SELECT id FROM echanges
    WHERE compresse = 0
    ORDER BY id DESC
    LIMIT ?
""", (garder_recents,))
ids_a_garder = {ligne[0] for ligne in curseur.fetchall()}

curseur.execute("""
    SELECT id, tour, horodatage, question, reponse, source
    FROM echanges
    WHERE compresse = 0
    ORDER BY id ASC
""")
lignes = curseur.fetchall()
connexion.close()

return [
    {
        "id"         : ligne[0],
        "tour"       : ligne[1],
        "horodatage" : ligne[2],
        "question"   : ligne[3],
        "reponse"    : ligne[4],
        "source"     : ligne[5]
    }
    for ligne in lignes
    if ligne[0] not in ids_a_garder
]

def enregistrer_compression(id_echange, resume):
connexion = sqlite3.connect(CHEMIN_BASE)
curseur = connexion.cursor()

curseur.execute("""
    UPDATE echanges
    SET compresse = 1, resume = ?
    WHERE id = ?
""", (resume, id_echange))

connexion.commit()
connexion.close()

def marquer_question_traitee(id_question):
connexion = sqlite3.connect(CHEMIN_BASE)
curseur = connexion.cursor()

curseur.execute("""
    UPDATE questions_en_attente
    SET statut = 'traitee'
    WHERE id = ?
""", (id_question,))

connexion.commit()
connexion.close()

==========================================================

VÉRIFICATION D’INTÉGRITÉ

==========================================================

def verifier_integrite():
try:
if not os.path.exists(CHEMIN_BASE):
return True, None

    connexion = sqlite3.connect(CHEMIN_BASE)
    curseur   = connexion.cursor()
    curseur.execute("PRAGMA integrity_check")
    resultat  = curseur.fetchone()
    connexion.close()

    if resultat and resultat[0] == "ok":
        return True, None
    else:
        return False, "Résultat integrity_check : {}".format(resultat)

except Exception as e:
    return False, str(e)

==========================================================

POINT D’ENTRÉE — TEST RAPIDE

==========================================================

if name == "main":
print("Initialisation de historique.db…")
initialiser()

ok, erreur = verifier_integrite()
if ok:
    print("Base historique : OK")
else:
    print("ERREUR base historique : {}".format(erreur))

ajouter_echange(1, "quel jour sommes-nous ?",
                "Nous sommes le 18/10/2026.", "degradee")
print("Échange test ajouté.")

historique = lire_historique_recent(5)
print("Historique récent ({} échange(s)) :".format(len(historique)))
for e in historique:
    print("  Tour {} — {} — Q: {}".format(
        e["tour"], e["horodatage"][:10], e["question"]))

print("Test terminé.")

Laisser un commentaire

Votre adresse e-mail ne sera pas publiée. Les champs obligatoires sont indiqués avec *

Défiler vers le haut