import json import os import random import re import sqlite3 import unicodedata from datetime import datetime, timezone from pathlib import Path from typing import Literal from dotenv import load_dotenv from flask import Flask, jsonify, render_template, request, url_for from openai import OpenAI from pydantic import BaseModel, Field BASE_DIR = Path(__file__).resolve().parent load_dotenv(BASE_DIR / ".env") app = Flask(__name__) app.secret_key = os.getenv("SECRET_KEY", "dev-change-me") app.config["MAX_CONTENT_LENGTH"] = 12 * 1024 * 1024 app.config["SESSION_COOKIE_HTTPONLY"] = True app.config["SESSION_COOKIE_SAMESITE"] = "Lax" app.config["SESSION_COOKIE_SECURE"] = os.getenv("COOKIE_SECURE", "1") == "1" INSTANCE_DIR = BASE_DIR / "instance" INSTANCE_DIR.mkdir(parents=True, exist_ok=True) SINGLE_DB_PATH = INSTANCE_DIR / "lao.db" USER_DB_DIR = INSTANCE_DIR / "users" USER_DB_DIR.mkdir(parents=True, exist_ok=True) OPENAI_MODEL = os.getenv("OPENAI_MODEL", "gpt-4.1-mini") OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "") client = OpenAI(api_key=OPENAI_API_KEY) if OPENAI_API_KEY else None LEVELS = { "A1": {"vocab_target": 150, "next": "A2"}, "A2": {"vocab_target": 400, "next": "B1"}, "B1": {"vocab_target": 1000, "next": "B2"}, "B2": {"vocab_target": 2500, "next": "C1"}, "C1": {"vocab_target": 5000, "next": "C2"}, "C2": {"vocab_target": 9000, "next": None}, } LEVEL_ORDER = ["A1", "A2", "B1", "B2", "C1", "C2"] LEVEL_TARGET_SCORE = {"A1": 15, "A2": 30, "B1": 50, "B2": 70, "C1": 85, "C2": 100} def now_iso(): return datetime.now(timezone.utc).isoformat(timespec="seconds") def today_iso(): return datetime.now(timezone.utc).date().isoformat() def clamp(value, low=0, high=100): return max(low, min(high, value)) def user_db_path(user_id=None): """Mode personnel sans connexion : une seule base locale.""" return SINGLE_DB_PATH def connect_user_db(path): conn = sqlite3.connect(path) conn.row_factory = sqlite3.Row conn.execute("PRAGMA foreign_keys=ON") return conn def table_columns(conn, table): return {r["name"] for r in conn.execute(f"PRAGMA table_info({table})").fetchall()} def ensure_column(conn, table, name, definition): if name not in table_columns(conn, table): conn.execute(f"ALTER TABLE {table} ADD COLUMN {name} {definition}") def init_user_db(path): path.parent.mkdir(parents=True, exist_ok=True) with connect_user_db(path) as conn: conn.executescript(""" CREATE TABLE IF NOT EXISTS settings ( id INTEGER PRIMARY KEY CHECK (id = 1), learner_name TEXT NOT NULL DEFAULT '', mode TEXT NOT NULL DEFAULT 'informal', correction_mode INTEGER NOT NULL DEFAULT 1, known_only INTEGER NOT NULL DEFAULT 0, current_level TEXT NOT NULL DEFAULT 'A1', grammar_score REAL NOT NULL DEFAULT 0, comprehension_score REAL NOT NULL DEFAULT 0, pronunciation_score REAL NOT NULL DEFAULT 0, proficiency_score REAL NOT NULL DEFAULT 0, demonstrated_level TEXT NOT NULL DEFAULT 'A1', streak INTEGER NOT NULL DEFAULT 0, last_active_date TEXT ); INSERT OR IGNORE INTO settings (id) VALUES (1); CREATE TABLE IF NOT EXISTS vocabulary ( id INTEGER PRIMARY KEY AUTOINCREMENT, hanzi TEXT NOT NULL DEFAULT '', pinyin TEXT NOT NULL, meaning_fr TEXT NOT NULL, theme TEXT NOT NULL DEFAULT 'général', word_type TEXT NOT NULL DEFAULT 'autre', status TEXT NOT NULL DEFAULT 'new', mastery REAL NOT NULL DEFAULT 10, successes INTEGER NOT NULL DEFAULT 0, errors INTEGER NOT NULL DEFAULT 0, spontaneous_uses INTEGER NOT NULL DEFAULT 0, last_seen TEXT, created_at TEXT NOT NULL, UNIQUE(pinyin, meaning_fr) ); CREATE TABLE IF NOT EXISTS conversations ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL DEFAULT 'Nouvelle conversation', created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS chat_messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, conversation_id INTEGER, role TEXT NOT NULL, content TEXT NOT NULL, created_at TEXT NOT NULL, FOREIGN KEY(conversation_id) REFERENCES conversations(id) ON DELETE CASCADE ); CREATE TABLE IF NOT EXISTS learning_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, event_type TEXT NOT NULL, payload TEXT NOT NULL DEFAULT '{}', created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS character_attempts ( id INTEGER PRIMARY KEY AUTOINCREMENT, hanzi TEXT NOT NULL, self_score INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS progress_snapshots ( snapshot_date TEXT PRIMARY KEY, total_words INTEGER NOT NULL DEFAULT 0, known_words INTEGER NOT NULL DEFAULT 0, mastered_words INTEGER NOT NULL DEFAULT 0, avg_mastery REAL NOT NULL DEFAULT 0, level TEXT NOT NULL DEFAULT 'A1', overall INTEGER NOT NULL DEFAULT 0 ); CREATE TABLE IF NOT EXISTS proficiency_evidence ( id INTEGER PRIMARY KEY AUTOINCREMENT, estimated_level TEXT NOT NULL, performance_score REAL NOT NULL, created_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_vocab_status ON vocabulary(status); CREATE INDEX IF NOT EXISTS idx_vocab_theme ON vocabulary(theme); CREATE INDEX IF NOT EXISTS idx_vocab_type ON vocabulary(word_type); CREATE INDEX IF NOT EXISTS idx_chat_conversation ON chat_messages(conversation_id, id); CREATE INDEX IF NOT EXISTS idx_chat_created ON chat_messages(created_at); """) # Migrations from V1 / legacy databases. ensure_column(conn, "settings", "proficiency_score", "REAL NOT NULL DEFAULT 0") ensure_column(conn, "settings", "demonstrated_level", "TEXT NOT NULL DEFAULT 'A1'") ensure_column(conn, "vocabulary", "word_type", "TEXT NOT NULL DEFAULT 'autre'") ensure_column(conn, "chat_messages", "conversation_id", "INTEGER") # If old messages exist, attach them to one legacy conversation. orphan_count = conn.execute( "SELECT COUNT(*) AS c FROM chat_messages WHERE conversation_id IS NULL" ).fetchone()["c"] if orphan_count: conv = conn.execute( "SELECT id FROM conversations ORDER BY id LIMIT 1" ).fetchone() if conv: conv_id = conv["id"] else: cur = conn.execute( "INSERT INTO conversations(title,created_at,updated_at) VALUES(?,?,?)", ("Conversation précédente", now_iso(), now_iso()) ) conv_id = cur.lastrowid conn.execute( "UPDATE chat_messages SET conversation_id=? WHERE conversation_id IS NULL", (conv_id,) ) def db(): init_user_db(SINGLE_DB_PATH) return connect_user_db(SINGLE_DB_PATH) init_user_db(SINGLE_DB_PATH) STARTER_WORDS = [ ("ສະບາຍດີ", "sabaidee", "bonjour / salut", "salutations", "expression"), ("ຂອບໃຈ", "khop chai", "merci", "salutations", "expression"), ("ຂໍໂທດ", "kho thot", "pardon / désolé", "salutations", "expression"), ("ແມ່ນ", "maen", "oui / c'est exact", "essentiels", "autre"), ("ບໍ່", "bo", "non / ne... pas", "essentiels", "adverbe"), ("ຂ້ອຍ", "khoy", "je / moi", "pronoms", "pronom"), ("ເຈົ້າ", "chao", "tu / toi", "pronoms", "pronom"), ("ລາວ", "lao", "il / elle", "pronoms", "pronom"), ("ພວກເຮົາ", "phuak hao", "nous", "pronoms", "pronom"), ("ໄປ", "pai", "aller", "verbes", "verbe"), ("ມາ", "ma", "venir", "verbes", "verbe"), ("ກິນ", "kin", "manger", "verbes", "verbe"), ("ດື່ມ", "duem", "boire", "verbes", "verbe"), ("ນ້ຳ", "nam", "eau", "nourriture", "nom"), ("ເຂົ້າ", "khao", "riz / repas", "nourriture", "nom"), ("ເຮືອນ", "heuan", "maison", "quotidien", "nom"), ("ໝູ່", "mu", "ami", "quotidien", "nom"), ("ມີ", "mi", "avoir / il y a", "verbes", "verbe"), ("ຢາກ", "yak", "vouloir", "verbes", "verbe"), ("ຮັກ", "hak", "aimer (amour)", "verbes", "verbe"), ("ມັກ", "mak", "aimer / apprécier", "verbes", "verbe"), ("ຮຽນ", "hian", "étudier / apprendre", "verbes", "verbe"), ("ເວົ້າ", "vao", "parler / dire", "verbes", "verbe"), ("ເຂົ້າໃຈ", "khao chai", "comprendre", "verbes", "verbe"), ("ຫຍັງ", "nyang", "quoi", "questions", "pronom"), ("ໃຜ", "phai", "qui", "questions", "pronom"), ("ຢູ່ໃສ", "yu sai", "où", "questions", "expression"), ("ເປັນຫຍັງ", "pen nyang", "pourquoi", "questions", "expression"), ("ມື້ນີ້", "mue ni", "aujourd'hui", "temps", "adverbe"), ("ມື້ວານນີ້", "mue van ni", "hier", "temps", "adverbe"), ("ມື້ອື່ນ", "mue un", "demain", "temps", "adverbe"), ("ແລ້ວ", "laeo", "déjà / marque d'accompli", "temps", "adverbe"), ("ຈະ", "cha", "aller (futur)", "temps", "autre"), ("ດີ", "di", "bien / bon", "adjectifs", "adjectif"), ("ຫຼາຍ", "lai", "beaucoup", "quantités", "adverbe"), ("ອີກ", "ik", "encore / de plus", "quotidien", "adverbe"), ("ບໍ?", "bo?", "particule interrogative", "questions", "autre"), ] with connect_user_db(SINGLE_DB_PATH) as conn: if not conn.execute("SELECT 1 FROM vocabulary LIMIT 1").fetchone(): conn.executemany("""INSERT OR IGNORE INTO vocabulary (hanzi,pinyin,meaning_fr,theme,word_type,status,mastery,created_at) VALUES(?,?,?,?,?,'new',10,?)""", [(*w, now_iso()) for w in STARTER_WORDS]) conn.execute("UPDATE settings SET learner_name=' ' WHERE id=1 AND learner_name='' ") @app.context_processor def inject_user(): return {"current_user": None} def vocab_status(mastery, successes): if mastery >= 90 and successes >= 3: return "mastered" if mastery >= 65: return "known" if mastery >= 25: return "learning" return "new" def get_settings(): with db() as conn: return dict(conn.execute("SELECT * FROM settings WHERE id=1").fetchone()) def mastered_vocab_count(): with db() as conn: return conn.execute( "SELECT COUNT(*) AS c FROM vocabulary WHERE status IN ('known','mastered')" ).fetchone()["c"] def calculate_progress(): s = get_settings() current = s["current_level"] target = LEVELS.get(current, LEVELS["A1"])["vocab_target"] known = mastered_vocab_count() vocab_component = clamp((known / target) * 100 if target else 100) grammar = clamp(float(s["grammar_score"])) comprehension = clamp(float(s["comprehension_score"])) pronunciation = clamp(float(s["pronunciation_score"])) proficiency = clamp(float(s["proficiency_score"])) # The proficiency score allows a genuinely strong learner to move faster, # while vocabulary still matters. overall = round( vocab_component * 0.30 + grammar * 0.25 + comprehension * 0.15 + pronunciation * 0.10 + proficiency * 0.20 ) return { "level": current, "overall": overall, "vocab": round(vocab_component), "grammar": round(grammar), "comprehension": round(comprehension), "pronunciation": round(pronunciation), "proficiency": round(proficiency), "demonstrated_level": s["demonstrated_level"], "known_count": known, "target": target, "next_level": LEVELS.get(current, {}).get("next"), } def record_progress_snapshot(): p = calculate_progress() with db() as conn: row = conn.execute(""" SELECT COUNT(*) total, SUM(CASE WHEN status IN ('known','mastered') THEN 1 ELSE 0 END) known, SUM(CASE WHEN status='mastered' THEN 1 ELSE 0 END) mastered, AVG(mastery) avg_m FROM vocabulary """).fetchone() conn.execute(""" INSERT INTO progress_snapshots( snapshot_date,total_words,known_words,mastered_words,avg_mastery,level,overall ) VALUES(?,?,?,?,?,?,?) ON CONFLICT(snapshot_date) DO UPDATE SET total_words=excluded.total_words, known_words=excluded.known_words, mastered_words=excluded.mastered_words, avg_mastery=excluded.avg_mastery, level=excluded.level, overall=excluded.overall """, ( today_iso(), row["total"] or 0, row["known"] or 0, row["mastered"] or 0, float(row["avg_m"] or 0), p["level"], p["overall"] )) def maybe_level_up(): p = calculate_progress() if p["overall"] < 100: return None nxt = p["next_level"] if not nxt: return None with db() as conn: conn.execute("UPDATE settings SET current_level=? WHERE id=1", (nxt,)) conn.execute( "INSERT INTO learning_events(event_type,payload,created_at) VALUES(?,?,?)", ("level_up", json.dumps({"from": p["level"], "to": nxt}), now_iso()) ) record_progress_snapshot() return nxt def normalize_pinyin(text): text = text.lower().strip() text = unicodedata.normalize("NFD", text) text = "".join(c for c in text if unicodedata.category(c) != "Mn") text = text.replace("ü", "v") text = re.sub(r"[^a-z0-9vàâäéèêëîïôöùûüçœæ\s'-]", " ", text) text = re.sub(r"\s+", " ", text).strip() return text class VocabEvent(BaseModel): pinyin: str hanzi: str = "" meaning_fr: str result: Literal["correct", "partial", "error", "new"] note: str = "" theme: str = "conversation" word_type: Literal["nom", "verbe", "adjectif", "adverbe", "connecteur", "pronom", "classificateur", "expression", "autre"] = "autre" class ChatEvaluation(BaseModel): reply: str correction: str = "" explanation_fr: str = "" vocab_events: list[VocabEvent] = Field(default_factory=list) grammar_delta: float = Field(default=0, ge=-5, le=5) comprehension_delta: float = Field(default=0, ge=-3, le=3) estimated_cefr: Literal["A1", "A2", "B1", "B2", "C1", "C2"] = "A1" performance_score: int = Field(default=50, ge=0, le=100) class GeneratedWord(BaseModel): hanzi: str pinyin: str meaning_fr: str example_pinyin: str example_fr: str word_type: Literal["nom", "verbe", "adjectif", "adverbe", "connecteur", "pronom", "classificateur", "expression", "autre"] = "autre" class VocabPack(BaseModel): words: list[GeneratedWord] class SpeechEvaluation(BaseModel): feedback_fr: str score: int = Field(ge=0, le=100) content_correct: bool def add_learning_event(event_type, payload): with db() as conn: conn.execute( "INSERT INTO learning_events(event_type,payload,created_at) VALUES(?,?,?)", (event_type, json.dumps(payload, ensure_ascii=False), now_iso()) ) def update_vocab_from_event(event): pinyin = event.pinyin.strip() meaning = event.meaning_fr.strip() if not pinyin or not meaning: return with db() as conn: row = conn.execute(""" SELECT * FROM vocabulary WHERE lower(pinyin)=lower(?) OR (hanzi<>'' AND hanzi=?) ORDER BY mastery DESC LIMIT 1 """, (pinyin, event.hanzi)).fetchone() if row: mastery = float(row["mastery"]) successes = int(row["successes"]) errors = int(row["errors"]) spontaneous = int(row["spontaneous_uses"]) if event.result == "correct": mastery += 6 successes += 1 spontaneous += 1 elif event.result == "partial": mastery += 2 successes += 1 elif event.result == "error": mastery -= 5 errors += 1 else: mastery += 0.5 mastery = clamp(mastery) status = vocab_status(mastery, successes) conn.execute(""" UPDATE vocabulary SET mastery=?,successes=?,errors=?,spontaneous_uses=?, status=?,last_seen=?, hanzi=CASE WHEN hanzi='' THEN ? ELSE hanzi END, theme=CASE WHEN theme IN ('','général','conversation') THEN ? ELSE theme END, word_type=CASE WHEN word_type IN ('','autre') THEN ? ELSE word_type END WHERE id=? """, ( mastery, successes, errors, spontaneous, status, now_iso(), event.hanzi, event.theme or "conversation", event.word_type, row["id"] )) else: base = 22 if event.result == "correct" else 12 if event.result == "partial" else 8 successes = 1 if event.result in ("correct", "partial") else 0 errors = 1 if event.result == "error" else 0 conn.execute(""" INSERT OR IGNORE INTO vocabulary( hanzi,pinyin,meaning_fr,theme,word_type,status,mastery,successes,errors, spontaneous_uses,last_seen,created_at ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?) """, ( event.hanzi, pinyin, meaning, event.theme or "conversation", event.word_type, vocab_status(base, successes), base, successes, errors, 1 if event.result == "correct" else 0, now_iso(), now_iso() )) def update_skill_scores(grammar_delta=0, comprehension_delta=0, pronunciation_delta=0): s = get_settings() with db() as conn: conn.execute(""" UPDATE settings SET grammar_score=?,comprehension_score=?,pronunciation_score=? WHERE id=1 """, ( clamp(float(s["grammar_score"]) + grammar_delta), clamp(float(s["comprehension_score"]) + comprehension_delta), clamp(float(s["pronunciation_score"]) + pronunciation_delta) )) def apply_proficiency_evidence(level, performance_score): if level not in LEVEL_ORDER: level = "A1" performance_score = clamp(float(performance_score)) target_score = LEVEL_TARGET_SCORE[level] with db() as conn: s = dict(conn.execute("SELECT * FROM settings WHERE id=1").fetchone()) old = float(s["proficiency_score"]) # Strong evidence moves the score rapidly, weak evidence moves it more slowly. alpha = 0.32 if performance_score >= 80 else 0.18 observed = target_score * (0.65 + 0.35 * performance_score / 100) new_score = clamp(old * (1 - alpha) + observed * alpha) current_demo = s["demonstrated_level"] demo = level if LEVEL_ORDER.index(level) > LEVEL_ORDER.index(current_demo) else current_demo conn.execute( "UPDATE settings SET proficiency_score=?, demonstrated_level=? WHERE id=1", (new_score, demo) ) conn.execute( "INSERT INTO proficiency_evidence(estimated_level,performance_score,created_at) VALUES(?,?,?)", (level, performance_score, now_iso()) ) # Rapid advancement for learners who repeatedly demonstrate a higher level. current = s["current_level"] nxt = LEVELS[current]["next"] promoted = None if nxt: recent = conn.execute(""" SELECT estimated_level,performance_score FROM proficiency_evidence ORDER BY id DESC LIMIT 4 """).fetchall() if len(recent) >= 4: next_rank = LEVEL_ORDER.index(nxt) good = [ r for r in recent if r["performance_score"] >= 80 and LEVEL_ORDER.index(r["estimated_level"]) >= next_rank ] if len(good) >= 3: conn.execute("UPDATE settings SET current_level=? WHERE id=1", (nxt,)) conn.execute( "INSERT INTO learning_events(event_type,payload,created_at) VALUES(?,?,?)", ("level_up_proficiency", json.dumps({"from": current, "to": nxt}), now_iso()) ) promoted = nxt return promoted def build_chat_prompt(settings, known_words): register = ( "informal, everyday spoken Lao (friends, markets, family, travel)" if settings["mode"] == "informal" else "polite standard Lao, with appropriate particles" ) correction = "Correct mistakes clearly in French." if settings["correction_mode"] else "Avoid corrections of harmless minor mistakes." known_rule = ( "Prefer the learner's known vocabulary; introduce at most one useful new word." if settings["known_only"] else "Introduce a little new, practical vocabulary when useful." ) return f""" You are a Lao (Laotian language, ພາສາລາວ) tutor for a French-speaking learner. Use real Lao as spoken in Laos, NOT Thai, Chinese, or a blend of the two. Register: {register}. Learner may write Lao script, Latin romanization (often informal, unstandardized and without tone marks), or French. There is no single universal Lao romanization; accept reasonable phonetic variants. Prioritize practical conversation, word order, particles, verbs, daily phrases and comprehension. Whenever teaching a Lao phrase, give Lao script, a readable approximate Latin transliteration, and French meaning. Avoid overstating certainty about tonal pronunciation; explain variants where appropriate. {correction} {known_rule} Known vocabulary: {known_words} For every learner message: 1. Reply naturally at the demonstrated level, relatively briefly. 2. Give a corrected version when necessary, explaining the main issue in French. 3. Identify Lao vocabulary used or attempted. In structured vocab_events, use 'hanzi' for LAO SCRIPT and 'pinyin' for LATIN TRANSLITERATION (these are legacy field names, NEVER Chinese). Mark correct / partial / error / new truthfully, include theme and word_type. 4. Estimate a pedagogical CEFR-like level A1-C2 ONLY from the message, not a formal certification. 5. Give performance_score 0-100, based on vocabulary, grammar, naturalness and comprehension. 6. Do not fabricate Lao words or mix Thai spellings into Lao. """ def require_client(): if client is None: return jsonify({"error": "OPENAI_API_KEY n'est pas configurée sur le serveur."}), 503 return None def get_or_create_conversation(conversation_id=None): with db() as conn: if conversation_id: row = conn.execute("SELECT * FROM conversations WHERE id=?", (conversation_id,)).fetchone() if row: return dict(row) row = conn.execute("SELECT * FROM conversations ORDER BY updated_at DESC LIMIT 1").fetchone() if row: return dict(row) cur = conn.execute( "INSERT INTO conversations(title,created_at,updated_at) VALUES(?,?,?)", ("Nouvelle conversation", now_iso(), now_iso()) ) return dict(conn.execute("SELECT * FROM conversations WHERE id=?", (cur.lastrowid,)).fetchone()) @app.route("/") def dashboard(): record_progress_snapshot() with db() as conn: stats = { "total": conn.execute("SELECT COUNT(*) c FROM vocabulary").fetchone()["c"], "mastered": conn.execute("SELECT COUNT(*) c FROM vocabulary WHERE status='mastered'").fetchone()["c"], "known": conn.execute("SELECT COUNT(*) c FROM vocabulary WHERE status='known'").fetchone()["c"], "learning": conn.execute("SELECT COUNT(*) c FROM vocabulary WHERE status='learning'").fetchone()["c"], "review": conn.execute("SELECT COUNT(*) c FROM vocabulary WHERE mastery < 70").fetchone()["c"], } trouble = [dict(r) for r in conn.execute(""" SELECT * FROM vocabulary ORDER BY (errors * 8 + (100-mastery)) DESC LIMIT 6 """).fetchall()] return render_template("dashboard.html", progress=calculate_progress(), stats=stats, trouble=trouble, settings=get_settings()) @app.route("/chat") def chat_page(): requested = request.args.get("c", type=int) conv = get_or_create_conversation(requested) with db() as conn: conversations = [dict(r) for r in conn.execute( "SELECT * FROM conversations ORDER BY updated_at DESC" ).fetchall()] messages = [dict(r) for r in conn.execute( "SELECT * FROM chat_messages WHERE conversation_id=? ORDER BY id", (conv["id"],) ).fetchall()] return render_template("chat.html", messages=messages, conversations=conversations, active_conversation=conv, settings=get_settings()) @app.route("/vocabulary") def vocabulary_page(): return render_template("vocabulary.html", settings=get_settings()) @app.route("/review") def review_page(): return render_template("review.html") @app.route("/characters") def characters_page(): with db() as conn: chars = [dict(r) for r in conn.execute(""" SELECT id,hanzi,pinyin,meaning_fr,mastery FROM vocabulary WHERE hanzi<>'' ORDER BY mastery ASC,id DESC LIMIT 100 """).fetchall()] return render_template("characters.html", chars=chars) @app.route("/pronunciation") def pronunciation_page(): return render_template("pronunciation.html") @app.route("/progress") def progress_page(): record_progress_snapshot() with db() as conn: events = [dict(r) for r in conn.execute( "SELECT * FROM learning_events ORDER BY id DESC LIMIT 30" ).fetchall()] history = [dict(r) for r in conn.execute( "SELECT * FROM progress_snapshots ORDER BY snapshot_date ASC LIMIT 180" ).fetchall()] theme_stats = [dict(r) for r in conn.execute(""" SELECT theme,COUNT(*) total,ROUND(AVG(mastery),1) avg_mastery, SUM(CASE WHEN status='mastered' THEN 1 ELSE 0 END) mastered FROM vocabulary GROUP BY theme ORDER BY total DESC,theme LIMIT 20 """).fetchall()] type_stats = [dict(r) for r in conn.execute(""" SELECT word_type,COUNT(*) total,ROUND(AVG(mastery),1) avg_mastery, SUM(CASE WHEN status='mastered' THEN 1 ELSE 0 END) mastered FROM vocabulary GROUP BY word_type ORDER BY total DESC,word_type """).fetchall()] return render_template( "progress.html", progress=calculate_progress(), events=events, history=history, theme_stats=theme_stats, type_stats=type_stats, settings=get_settings() ) @app.route("/api/settings", methods=["POST"]) def api_settings(): data = request.get_json(silent=True) or {} sets, vals = [], [] if data.get("mode") in {"informal", "standard"}: sets.append("mode=?"); vals.append(data["mode"]) if isinstance(data.get("correction_mode"), bool): sets.append("correction_mode=?"); vals.append(int(data["correction_mode"])) if isinstance(data.get("known_only"), bool): sets.append("known_only=?"); vals.append(int(data["known_only"])) if sets: with db() as conn: conn.execute(f"UPDATE settings SET {', '.join(sets)} WHERE id=1", vals) return jsonify({"ok": True, "settings": get_settings()}) @app.route("/api/conversations", methods=["GET", "POST"]) def api_conversations(): if request.method == "POST": with db() as conn: cur = conn.execute( "INSERT INTO conversations(title,created_at,updated_at) VALUES(?,?,?)", ("Nouvelle conversation", now_iso(), now_iso()) ) conv_id = cur.lastrowid return jsonify({"id": conv_id, "title": "Nouvelle conversation"}) with db() as conn: rows = [dict(r) for r in conn.execute( "SELECT * FROM conversations ORDER BY updated_at DESC" ).fetchall()] return jsonify({"conversations": rows}) @app.route("/api/conversations/", methods=["PATCH", "DELETE"]) def api_conversation_item(conversation_id): if request.method == "DELETE": with db() as conn: conn.execute("DELETE FROM conversations WHERE id=?", (conversation_id,)) return jsonify({"ok": True}) data = request.get_json(silent=True) or {} title = (data.get("title") or "").strip()[:80] if title: with db() as conn: conn.execute("UPDATE conversations SET title=?,updated_at=? WHERE id=?", (title, now_iso(), conversation_id)) return jsonify({"ok": True}) @app.route("/api/chat", methods=["POST"]) def api_chat(): err = require_client() if err: return err data = request.get_json(silent=True) or {} message = (data.get("message") or "").strip() conversation_id = int(data.get("conversation_id") or 0) if not message: return jsonify({"error": "Message vide."}), 400 conv = get_or_create_conversation(conversation_id) settings = get_settings() with db() as conn: known_rows = conn.execute(""" SELECT hanzi,pinyin,meaning_fr,mastery FROM vocabulary ORDER BY mastery DESC,last_seen DESC LIMIT 180 """).fetchall() known_words = ", ".join( f"{r['pinyin']} ({r['hanzi']}) = {r['meaning_fr']}" for r in known_rows ) history = [dict(r) for r in conn.execute(""" SELECT role,content FROM chat_messages WHERE conversation_id=? ORDER BY id DESC LIMIT 16 """, (conv["id"],)).fetchall()][::-1] input_messages = [{"role": "system", "content": build_chat_prompt(settings, known_words)}] input_messages.extend(history) input_messages.append({"role": "user", "content": message}) try: response = client.responses.parse( model=OPENAI_MODEL, input=input_messages, text_format=ChatEvaluation, ) result = response.output_parsed if result is None: raise RuntimeError("Réponse structurée vide.") with db() as conn: conn.execute( "INSERT INTO chat_messages(conversation_id,role,content,created_at) VALUES(?,?,?,?)", (conv["id"], "user", message, now_iso()) ) assistant_content = result.reply if result.correction: assistant_content += f"\n\nCorrection : {result.correction}" if result.explanation_fr: assistant_content += f"\n\n{result.explanation_fr}" conn.execute( "INSERT INTO chat_messages(conversation_id,role,content,created_at) VALUES(?,?,?,?)", (conv["id"], "assistant", assistant_content, now_iso()) ) # Auto-title after first message. if conv["title"] == "Nouvelle conversation": clean = re.sub(r"\s+", " ", message).strip() title = clean[:42] + ("…" if len(clean) > 42 else "") conn.execute( "UPDATE conversations SET title=?,updated_at=? WHERE id=?", (title or "Conversation", now_iso(), conv["id"]) ) else: conn.execute("UPDATE conversations SET updated_at=? WHERE id=?", (now_iso(), conv["id"])) for event in result.vocab_events: update_vocab_from_event(event) update_skill_scores(result.grammar_delta, result.comprehension_delta) proficiency_promotion = apply_proficiency_evidence(result.estimated_cefr, result.performance_score) add_learning_event("chat", { "conversation_id": conv["id"], "message": message, "estimated_cefr": result.estimated_cefr, "performance_score": result.performance_score, "vocab_events": [e.model_dump() for e in result.vocab_events], }) regular_promotion = maybe_level_up() record_progress_snapshot() return jsonify({ "reply": result.reply, "correction": result.correction, "explanation_fr": result.explanation_fr, "progress": calculate_progress(), "level_up": proficiency_promotion or regular_promotion, "conversation_id": conv["id"] }) except Exception as exc: app.logger.exception("OpenAI chat error") return jsonify({"error": f"Erreur IA : {str(exc)}"}), 500 @app.route("/api/chat/clear", methods=["POST"]) def api_chat_clear(): data = request.get_json(silent=True) or {} conversation_id = int(data.get("conversation_id") or 0) if conversation_id: with db() as conn: conn.execute("DELETE FROM chat_messages WHERE conversation_id=?", (conversation_id,)) return jsonify({"ok": True}) @app.route("/api/vocabulary") def api_vocabulary(): q = (request.args.get("q") or "").strip() status = (request.args.get("status") or "").strip() params = [] where = ["1=1"] if q: where.append("(hanzi LIKE ? OR pinyin LIKE ? OR meaning_fr LIKE ? OR theme LIKE ? OR word_type LIKE ?)") like = f"%{q}%"; params.extend([like, like, like, like, like]) if status in {"new", "learning", "known", "mastered"}: where.append("status=?"); params.append(status) with db() as conn: rows = [dict(r) for r in conn.execute( f"SELECT * FROM vocabulary WHERE {' AND '.join(where)} ORDER BY mastery DESC,pinyin ASC", params ).fetchall()] return jsonify({"words": rows}) @app.route("/api/vocabulary/generate", methods=["POST"]) def api_vocab_generate(): err = require_client() if err: return err data = request.get_json(silent=True) or {} theme = (data.get("theme") or "conversation quotidienne").strip()[:80] count = max(3, min(int(data.get("count") or 5), 15)) with db() as conn: existing = ", ".join( r["pinyin"] for r in conn.execute("SELECT pinyin FROM vocabulary ORDER BY mastery DESC LIMIT 300").fetchall() ) prompt = f""" Generate exactly {count} useful LAO (Laotian) words for a French-speaking learner. Theme: {theme}. Avoid these already learned words: {existing}. Use authentic Lao script in the hanzi field and a clear Latin Lao romanization in the pinyin field. For example, hanzi=ສະບາຍດີ, pinyin=sabaidee. These internal field names are legacy API names, not Chinese! Avoid Thai script and Chinese. Meanings and example translations are in French. Classify each word_type as one of: nom, verbe, adjectif, adverbe, connecteur, pronom, classificateur, expression, autre. Examples must be short and natural. """ try: response = client.responses.parse( model=OPENAI_MODEL, input=[ {"role": "system", "content": "You create accurate Lao-language vocabulary packs with Lao script, readable transliteration, and French translations. Never use Thai or Chinese script."}, {"role": "user", "content": prompt}, ], text_format=VocabPack, ) pack = response.output_parsed if pack is None: raise RuntimeError("Pack vide.") inserted = [] with db() as conn: for w in pack.words[:count]: try: cur = conn.execute(""" INSERT INTO vocabulary( hanzi,pinyin,meaning_fr,theme,word_type,status,mastery,successes,errors, spontaneous_uses,last_seen,created_at ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?) """, ( w.hanzi,w.pinyin,w.meaning_fr,theme,w.word_type,"new",10,0,0,0,None,now_iso() )) inserted.append({"id": cur.lastrowid, **w.model_dump()}) except sqlite3.IntegrityError: pass add_learning_event("new_vocab", {"theme": theme, "count": len(inserted)}) record_progress_snapshot() return jsonify({"words": inserted}) except Exception as exc: app.logger.exception("Vocab generation error") return jsonify({"error": f"Erreur IA : {str(exc)}"}), 500 @app.route("/api/vocabulary/", methods=["PATCH", "DELETE"]) def api_vocab_item(word_id): if request.method == "DELETE": with db() as conn: conn.execute("DELETE FROM vocabulary WHERE id=?", (word_id,)) record_progress_snapshot() return jsonify({"ok": True}) data = request.get_json(silent=True) or {} allowed = {"hanzi", "pinyin", "meaning_fr", "theme", "word_type"} sets, vals = [], [] for key in allowed: if key in data and isinstance(data[key], str): sets.append(f"{key}=?"); vals.append(data[key].strip()[:200]) if "mastery" in data: m = clamp(float(data["mastery"])) with db() as conn: row = conn.execute("SELECT successes FROM vocabulary WHERE id=?", (word_id,)).fetchone() if row: sets.extend(["mastery=?", "status=?"]); vals.extend([m, vocab_status(m, row["successes"])]) if sets: vals.append(word_id) with db() as conn: conn.execute(f"UPDATE vocabulary SET {', '.join(sets)} WHERE id=?", vals) record_progress_snapshot() return jsonify({"ok": True}) @app.route("/api/review/next") def api_review_next(): threshold = max(0, min(request.args.get("threshold", default=100, type=int), 100)) recent_raw = (request.args.get("exclude") or "").strip() recent = [] for x in recent_raw.split(","): try: recent.append(int(x)) except Exception: pass recent = recent[-10:] with db() as conn: rows = [dict(r) for r in conn.execute( "SELECT * FROM vocabulary WHERE mastery <= ? ORDER BY id", (threshold,) ).fetchall()] if not rows: return jsonify({"word": None, "eligible": 0}) candidates = [r for r in rows if r["id"] not in recent] if not candidates: # If the filter is tiny, permit old items again but still avoid the immediate last word. last = recent[-1] if recent else None candidates = [r for r in rows if r["id"] != last] or rows # Weak words are more likely, but strong words still appear for variety. weights = [max(8, 118 - float(r["mastery"])) for r in candidates] row = random.choices(candidates, weights=weights, k=1)[0] return jsonify({"word": row, "eligible": len(rows)}) @app.route("/api/review/check", methods=["POST"]) def api_review_check(): data = request.get_json(silent=True) or {} word_id = int(data.get("word_id") or 0) answer = (data.get("answer") or "").strip() mode = data.get("mode") or "fr_to_pinyin" with db() as conn: row = conn.execute("SELECT * FROM vocabulary WHERE id=?", (word_id,)).fetchone() if not row: return jsonify({"error": "Mot introuvable."}), 404 if mode == "fr_to_pinyin": expected = row["pinyin"] correct = normalize_pinyin(answer) == normalize_pinyin(expected) elif mode in {"pinyin_to_fr", "chinese_to_fr"}: expected = row["meaning_fr"] answer_n = normalize_pinyin(answer) options = [normalize_pinyin(x) for x in re.split(r"[/,;]", expected)] correct = any(opt and (opt in answer_n or answer_n in opt) for opt in options) else: expected = row["pinyin"] correct = False mastery = clamp(float(row["mastery"]) + (7 if correct else -5)) successes = int(row["successes"]) + (1 if correct else 0) errors = int(row["errors"]) + (0 if correct else 1) status = vocab_status(mastery, successes) with db() as conn: conn.execute(""" UPDATE vocabulary SET mastery=?,successes=?,errors=?,status=?,last_seen=? WHERE id=? """, (mastery, successes, errors, status, now_iso(), word_id)) add_learning_event("review", {"word_id": word_id, "correct": correct, "mode": mode}) maybe_level_up() record_progress_snapshot() return jsonify({"correct": correct, "expected": expected, "mastery": round(mastery), "status": status}) @app.route("/api/character-attempt", methods=["POST"]) def api_character_attempt(): data = request.get_json(silent=True) or {} hanzi = (data.get("hanzi") or "").strip()[:20] self_score = max(0, min(int(data.get("self_score") or 0), 3)) if not hanzi: return jsonify({"error": "Caractère manquant."}), 400 with db() as conn: conn.execute("INSERT INTO character_attempts(hanzi,self_score,created_at) VALUES(?,?,?)", (hanzi,self_score,now_iso())) add_learning_event("character", {"hanzi": hanzi, "score": self_score}) return jsonify({"ok": True}) @app.route("/api/pronunciation/analyze", methods=["POST"]) def api_pronunciation_analyze(): err = require_client() if err: return err audio = request.files.get("audio") expected = (request.form.get("expected") or "").strip()[:300] if not audio or not expected: return jsonify({"error": "Audio ou phrase attendue manquante."}), 400 try: transcript = client.audio.transcriptions.create( model="gpt-4o-mini-transcribe", file=(audio.filename or "recording.webm", audio.stream, audio.mimetype or "audio/webm"), language="lo", ) spoken = getattr(transcript, "text", "") or "" response = client.responses.parse( model=OPENAI_MODEL, input=[ {"role": "system", "content": "You evaluate spoken Lao (Laotian), allowing romanization variants. Never claim exact acoustic tonal measurement."}, {"role": "user", "content": f"""Expected Lao: {expected} Speech transcription: {spoken} Evaluate only whether the spoken linguistic content matches. Do not claim precise acoustic tone measurement. Give concise feedback in French and an indicative score."""}, ], text_format=SpeechEvaluation, ) result = response.output_parsed if result is None: raise RuntimeError("Évaluation vide.") delta = (result.score - 50) / 50 * 2 update_skill_scores(pronunciation_delta=delta) add_learning_event("pronunciation", {"expected": expected, "transcript": spoken, "score": result.score}) maybe_level_up() record_progress_snapshot() return jsonify({ "transcript": spoken, "score": result.score, "feedback_fr": result.feedback_fr, "content_correct": result.content_correct, "progress": calculate_progress() }) except Exception as exc: app.logger.exception("Pronunciation error") return jsonify({"error": f"Erreur audio : {str(exc)}"}), 500 @app.route("/api/progress") def api_progress(): record_progress_snapshot() with db() as conn: history = [dict(r) for r in conn.execute( "SELECT * FROM progress_snapshots ORDER BY snapshot_date ASC LIMIT 180" ).fetchall()] return jsonify({"progress": calculate_progress(), "history": history}) if __name__ == "__main__": app.run(debug=True, port=5000)