"""
Exercise generators for text, dialogue, visual, and pronunciation features.
All feedback uses encouraging language — see prompts.py for tone constraints.
"""
import json
import base64
import io
import logging
import os
from db import get_cursor
import gamify
import llm
import prompts
logger = logging.getLogger(__name__)
# ── Text exercise (Day 6) — kept for app.py's themed-Blocks fallback ──────────
def generate_text_exercise(lesson_text: str, user_id: str) -> dict:
result = llm.chat_json(
prompts.TEXT_EXERCISE_SYSTEM,
f"Lesson text:\n{lesson_text[:800]}",
fallback=_FALLBACK_EXERCISES["fill_blank"],
)
_save_exercise(user_id, None, "text", result.get("sentence_with_blank"), result.get("answer"), result)
return result
def render_text_exercise(ex: dict) -> str:
if not ex:
return ""
return (
f'
'
f'
{ex.get("instruction","")}
'
f'
'
f'{ex.get("sentence_with_blank","")}
'
f'
Hint: {ex.get("hint","")}
'
f'
'
)
def render_exercise_feedback(correct: bool, answer: str, explanation: str) -> str:
color = "#2e7d32" if correct else "#1565c0"
icon = "✅" if correct else "💡"
label = "Exactly right!" if correct else f"The answer is: {answer}"
return (
f''
f'
{icon} {label}
'
f'
{explanation}
'
f'
'
)
# ── Coach Agent: plan → generate → critique → revise (Day 3) ──────────────────
_SYLLABUS_PATH = os.path.join(os.path.dirname(__file__), "syllabus_full_a1_c2.json")
_syllabus_concepts: list[dict] | None = None
EXERCISE_TYPES = ["fill_blank", "multiple_choice", "error_detection", "reorder", "translation"]
# Used if the LLM is unreachable — still gives a real, varied 5-item set
# (degrade gracefully, per CLAUDE.md §1).
_FALLBACK_EXERCISES = {
"fill_blank": {
"type": "fill_blank", "instruction": "Fill in the blank:",
"sentence_with_blank": "Le ___ dort sur la table.",
"answer": "chat", "hint": "a small household animal (masculine noun)",
"explanation": "Chat (masc.) = cat. Articles: le chat, un chat.",
},
"multiple_choice": {
"type": "multiple_choice", "instruction": "Choose the correct answer:",
"question": "Comment dit-on « I am » en français ?",
"options": ["Je suis", "Tu es", "Il est", "Nous sommes"],
"answer": "Je suis",
"explanation": "« Je suis » = « I am » — first-person singular of être.",
},
"error_detection": {
"type": "error_detection", "instruction": "Find and fix the mistake:",
"sentence": "Elle est un étudiante.",
"answer": "Elle est une étudiante.",
"explanation": "« Étudiante » is feminine, so it takes « une », not « un ».",
},
"reorder": {
"type": "reorder", "instruction": "Put the words in the correct order:",
"words": ["je", "le", "matin", "café", "bois", "un"],
"answer": "Je bois un café le matin.",
"explanation": "Subject + verb + object + time expression is the typical French word order.",
},
"translation": {
"type": "translation", "instruction": "Translate to French:",
"prompt": "I would like a coffee, please.",
"answer": "Je voudrais un café, s'il vous plaît.",
"explanation": "« Je voudrais » (I would like) is a polite, common way to make a request.",
},
}
def _load_a1_a2_concepts() -> list[dict]:
"""A1/A2 concepts from the CEFR syllabus — the Coach Agent's grounding menu."""
global _syllabus_concepts
if _syllabus_concepts is None:
try:
with open(_SYLLABUS_PATH, encoding="utf-8") as f:
concepts = json.load(f)["concepts"]
_syllabus_concepts = [c for c in concepts if c.get("cefr_level") in ("A1", "A2")]
except Exception as e:
logger.warning("_load_a1_a2_concepts failed: %s", e)
_syllabus_concepts = []
return _syllabus_concepts
def generate_exercise_set(lesson_text: str, user_id: str, page_id: str | None = None, topic: str = "") -> dict:
"""Coach Agent: PLAN -> GENERATE -> CRITIQUE -> REVISE -> RETURN.
Returns {"concepts": [...], "exercises": [...]} — 5-7 mixed, self-checked
exercises grounded in the lesson and the A1/A2 syllabus. `topic` is an
optional learner-chosen focus; if blank, the agent picks from the lesson.
"""
concepts_menu = _load_a1_a2_concepts()
menu_ids = {c["id"] for c in concepts_menu}
plan = llm.chat_json(
prompts.COACH_PLAN_SYSTEM,
prompts.coach_plan_user(lesson_text, concepts_menu, topic),
fallback={"concepts": [], "plan": [{"type": t, "focus": "general practice from this lesson"} for t in EXERCISE_TYPES]},
)
chosen_concepts = [c for c in concepts_menu if c["id"] in (plan.get("concepts") or []) and c["id"] in menu_ids]
items_plan = [
spec for spec in (plan.get("plan") or [])
if isinstance(spec, dict) and spec.get("type") in EXERCISE_TYPES
][:7]
if not items_plan:
items_plan = [{"type": t, "focus": "general practice from this lesson"} for t in EXERCISE_TYPES]
exercises = [
_generate_and_critique(lesson_text, spec["type"], spec.get("focus", "general practice from this lesson"), topic=topic)
for spec in items_plan
]
if chosen_concepts:
_mark_concepts_covered(chosen_concepts)
result = {"concepts": chosen_concepts, "exercises": exercises}
_save_exercise(user_id, page_id, "coach_set", None, None, result)
return result
def _generate_and_critique(lesson_text: str, ex_type: str, focus: str, max_attempts: int = 2, topic: str = "") -> dict:
"""GENERATE -> CRITIQUE -> REVISE, bounded to max_attempts generations."""
revise_note = ""
exercise = _FALLBACK_EXERCISES[ex_type]
for _ in range(max_attempts):
exercise = llm.chat_json(
prompts.COACH_EXERCISE_SYSTEM,
prompts.coach_exercise_user(lesson_text, ex_type, focus, revise_note, topic),
fallback=_FALLBACK_EXERCISES[ex_type],
)
exercise.setdefault("type", ex_type)
critique = llm.chat_json(
prompts.COACH_CRITIQUE_SYSTEM,
prompts.coach_critique_user(exercise),
fallback={"valid": True, "issue": ""},
)
if critique.get("valid", True):
break
revise_note = critique.get("issue", "")
return exercise
def _mark_concepts_covered(concepts: list[dict]) -> None:
"""UPSERT concepts as covered today, so the Summary tab can draw on them."""
try:
with get_cursor() as cur:
for c in concepts:
cur.execute(
"""INSERT INTO concepts (id, name, cefr_level, family, covered_on)
VALUES (%s, %s, %s, %s, CURRENT_DATE)
ON CONFLICT (id) DO UPDATE SET covered_on = CURRENT_DATE""",
(c["id"], c["name"], c["cefr_level"], c["family"]),
)
except Exception as e:
logger.warning("_mark_concepts_covered failed: %s", e)
def check_coach_exercise(exercise: dict, user_answer: str, user_id: str) -> dict:
"""Check one answer. Always awards participation points — never deducts."""
ex_type = exercise.get("type", "fill_blank")
correct_answer = (exercise.get("answer") or "").strip()
user_answer = (user_answer or "").strip()
if ex_type in ("fill_blank", "multiple_choice"):
correct = user_answer.lower() == correct_answer.lower()
feedback = "Exactly right!" if correct else exercise.get("explanation", "")
else:
graded = llm.chat_json(
prompts.COACH_CHECK_SYSTEM,
prompts.coach_check_user(exercise, user_answer),
fallback={
"correct": user_answer.lower() == correct_answer.lower(),
"feedback": exercise.get("explanation", ""),
},
)
correct = bool(graded.get("correct"))
feedback = graded.get("feedback") or exercise.get("explanation", "")
gamify.add_points(user_id, "exercise_done")
return {"correct": correct, "feedback": feedback, "answer": correct_answer}
# ── Dialogue exercise (Day 7) ─────────────────────────────────────────────────
def generate_dialogue(lesson_text: str, user_id: str, topic: str = "") -> dict:
result = llm.chat_json(
prompts.DIALOGUE_SYSTEM,
prompts.dialogue_user(lesson_text, topic),
fallback={
"scene": "At a café in Montréal",
"agent_role": "Serveur",
"user_role": "Client",
"turns": [
{"speaker": "agent", "text": "Bonjour! Qu'est-ce que vous désirez?",
"translation": "Hello! What would you like?"},
{"speaker": "user", "hint": "Order a coffee"},
{"speaker": "agent", "text": "Très bien! Un café pour vous.",
"translation": "Very well! A coffee for you."},
{"speaker": "user", "hint": "Say thank you"},
],
},
)
_save_exercise(user_id, None, "dialogue", None, None, result)
return result
def _visible_turns(turns: list[dict], replies_count: int) -> list[dict]:
"""Turns revealed so far: stop at the first user turn the learner hasn't
replied to yet, so future agent lines aren't shown before their time —
keeps the dialogue to one exchange at a time."""
reply_idx = 0
for i, turn in enumerate(turns):
if turn["speaker"] == "user":
if reply_idx < replies_count:
reply_idx += 1
else:
return turns[:i]
return turns
def get_new_agent_lines(dialogue: dict, replies_count: int) -> list[str]:
"""Agent lines newly revealed by reaching `replies_count` completed
replies (for replies_count == 0, this is the opening agent line(s))."""
turns = dialogue.get("turns", [])
visible = _visible_turns(turns, replies_count)
prev_visible = _visible_turns(turns, replies_count - 1) if replies_count else []
return [t["text"] for t in visible[len(prev_visible):] if t["speaker"] == "agent"]
def render_dialogue(dialogue: dict, completed_replies: list[str]) -> str:
"""Render the dialogue up to the current exchange as a chat-style HTML
transcript — one turn at a time (see _visible_turns)."""
if not dialogue:
return ""
scene = dialogue.get("scene", "")
agent_role = dialogue.get("agent_role", "Agent")
user_role = dialogue.get("user_role", "You")
turns = _visible_turns(dialogue.get("turns", []), len(completed_replies))
reply_idx = 0
parts = [
f'📍 {scene}
'
]
for turn in turns:
if turn["speaker"] == "agent":
parts.append(
f''
f'{agent_role}
'
f'{turn["text"]}'
f''
f'({turn.get("translation","")})'
f'
'
)
else: # user turn — _visible_turns guarantees a reply exists
reply = completed_replies[reply_idx]
parts.append(
f''
f'{user_role}
'
f'{reply}'
f'
'
)
reply_idx += 1
return f'{"".join(parts)}
'
def get_next_user_hint(dialogue: dict, replies_count: int) -> str:
"""Return the hint for the next user turn."""
user_turns = [t for t in dialogue.get("turns", []) if t["speaker"] == "user"]
if replies_count < len(user_turns):
return user_turns[replies_count].get("hint", "Respond naturally")
return ""
def dialogue_feedback(user_reply: str, hint: str, scene: str, user_id: str) -> dict:
result = llm.chat_json(
prompts.DIALOGUE_FEEDBACK_SYSTEM,
prompts.dialogue_feedback_user(user_reply, hint, scene),
fallback={
"feedback": "Bien joué ! (Well done!) Replying in French takes courage — keep it up, you're making great progress.",
"natural_version": "",
},
)
_save_points_for_dialogue(user_id)
return result
def _save_points_for_dialogue(user_id: str):
try:
from gamify import add_points
add_points(user_id, "dialogue_turn")
except Exception:
pass
# ── Visual exercise (Day 8) ───────────────────────────────────────────────────
def generate_visual_exercise(pil_image, user_id: str) -> dict:
"""Upload image → vision LLM describes → text LLM builds exercises."""
import io, base64
buf = io.BytesIO()
pil_image.save(buf, format="JPEG", quality=85)
image_b64 = base64.b64encode(buf.getvalue()).decode()
description = llm.vision_chat(image_b64, prompts.VISUAL_DESCRIBE_PROMPT)
if description.startswith("⚠"):
return {"error": description, "exercises": []}
result = llm.chat_json(
prompts.VISUAL_EXERCISE_SYSTEM,
f"Image content:\n{description}",
fallback={"image_summary": description, "exercises": []},
)
_save_exercise(user_id, None, "visual", None, None, result)
return result
def render_visual_exercises(result: dict) -> str:
if "error" in result:
return f'{result["error"]}
'
exercises = result.get("exercises", [])
summary = result.get("image_summary", "")
if not exercises:
return 'No exercises generated.
'
parts = [f'📷 {summary}
']
for i, ex in enumerate(exercises, 1):
hint = ex.get("hint", "")
hint_html = f'Hint: {hint}
' if hint else ""
parts.append(
f''
f'
Exercise {i} — {ex.get("type","").title()}
'
f'
{ex.get("instruction","")}
'
f'
{ex.get("content","")}
'
f'{hint_html}'
f'
'
f'Show answer
'
f'{ex.get("answer","")} — {ex.get("explanation","")}
'
f' '
)
return "".join(parts)
# ── Visual exercise from a matched sample image (Day 4) ───────────────────────
_SAMPLE_IMAGES_DIR = os.path.join(os.path.dirname(__file__), "frontend", "public", "sample_images")
_sample_images: list[dict] | None = None
def _load_sample_images() -> list[dict]:
"""Pre-generated images + hand-written descriptions (see generate_sample_images.py)."""
global _sample_images
if _sample_images is None:
try:
with open(os.path.join(_SAMPLE_IMAGES_DIR, "manifest.json"), encoding="utf-8") as f:
_sample_images = json.load(f)["images"]
except Exception as e:
logger.warning("_load_sample_images failed: %s", e)
_sample_images = []
return _sample_images
def pick_sample_image(topic: str, user_id: str) -> dict | None:
"""Pick a sample image matching the lesson's topic that this user hasn't
seen yet. If every image for that topic has already been seen, cycle back
to the least-recently-used one *for that topic* — matching what the
learner asked for matters more than image variety. Only falls back to an
unrelated topic if no image exists for the requested one at all."""
images = _load_sample_images()
if not images:
return None
try:
with get_cursor() as cur:
cur.execute(
"""SELECT image_id, MAX(used_at) AS last_used FROM user_image_usage
WHERE user_id = %s GROUP BY image_id""",
(user_id,),
)
last_used = {r["image_id"]: r["last_used"] for r in cur.fetchall()}
except Exception:
last_used = {}
seen = set(last_used)
topic_matches = [img for img in images if img["topic"] == topic]
unseen_topic_matches = [img for img in topic_matches if img["id"] not in seen]
if unseen_topic_matches:
return unseen_topic_matches[0]
if topic_matches:
return min(topic_matches, key=lambda img: last_used[img["id"]])
unseen = [img for img in images if img["id"] not in seen]
if unseen:
return unseen[0]
if last_used:
lru_id = min(last_used, key=last_used.get)
by_id = {img["id"]: img for img in images}
if lru_id in by_id:
return by_id[lru_id]
return images[0]
def _mark_image_used(user_id: str, image_id: str) -> None:
try:
with get_cursor() as cur:
cur.execute(
"INSERT INTO user_image_usage (user_id, image_id) VALUES (%s, %s)",
(user_id, image_id),
)
except Exception as e:
logger.warning("_mark_image_used failed: %s", e)
def generate_visual_topic_exercise(image: dict, lesson_text: str, user_id: str, topic: str = "") -> dict:
"""Build 5-6 exercises grounded in a pre-generated sample image's
description (no vision call at request time)."""
result = llm.chat_json(
prompts.VISUAL_TOPIC_EXERCISE_SYSTEM,
prompts.visual_topic_exercise_user(image["description"], lesson_text, topic),
fallback={"image_summary": image["description"], "exercises": []},
max_tokens=1536,
)
_mark_image_used(user_id, image["id"])
_save_exercise(user_id, None, "visual", None, None, result)
return result
# ── Pronunciation (Day 9) ─────────────────────────────────────────────────────
# Several distinct fallback phrases (used if the LLM is unreachable) so that
# repeated "New phrase" clicks don't show the exact same sentence — cycles
# through this pool based on how many phrases have already been seen.
_PRONUNCIATION_FALLBACKS = [
{"phrase": "Bonjour, je m'appelle Marie.", "translation": "Hello, my name is Marie.",
"tip": "The French 'r' is pronounced at the back of the throat."},
{"phrase": "Je voudrais un café, s'il vous plaît.", "translation": "I would like a coffee, please.",
"tip": "Round your lips for the 'ou' sound in 'voudrais' and 'vous'."},
{"phrase": "J'habite à Montréal.", "translation": "I live in Montreal.",
"tip": "The 'h' in 'habite' is silent — start straight on the vowel sound."},
{"phrase": "Quel temps fait-il aujourd'hui ?", "translation": "What's the weather like today?",
"tip": "Nasalize the 'en' in 'temps' — let the sound come through your nose."},
{"phrase": "À quelle heure commence le cours ?", "translation": "What time does the class start?",
"tip": "Link 'quelle' and 'heure' smoothly — French often blends words together."},
]
def generate_pronunciation_target(lesson_text: str, topic: str = "", avoid: list[str] | None = None) -> dict:
avoid = [p for p in (avoid or []) if p]
parts = []
if topic.strip():
parts.append(f"Focus topic requested by the learner: {topic.strip()}")
if lesson_text.strip():
parts.append(f"Lesson: {lesson_text[:300]}")
if avoid:
recent = ", ".join(f'"{p}"' for p in avoid[-5:])
parts.append(f"The learner already practiced these phrases — pick a different one: {recent}")
context = "\n".join(parts) if parts else "A common A1 phrase"
fallback_pool = [f for f in _PRONUNCIATION_FALLBACKS if f["phrase"] not in avoid] or _PRONUNCIATION_FALLBACKS
fallback = fallback_pool[len(avoid) % len(fallback_pool)]
return llm.chat_json(
prompts.PRONUNCIATION_TARGET_SYSTEM,
context,
fallback=fallback,
)
def get_pronunciation_feedback(target: str, transcription: str) -> dict:
return llm.chat_json(
prompts.PRONUNCIATION_FEEDBACK_SYSTEM,
prompts.pronunciation_feedback_user(target, transcription),
fallback={
"feedback": "Excellent effort — every attempt builds your pronunciation muscle memory!",
"focus": "Try to match the rhythm of the phrase.",
"tip": "Read slowly first, then speed up gradually.",
},
)
# ── Shared DB helper ──────────────────────────────────────────────────────────
def _save_exercise(user_id, page_id, kind, prompt_text, answer, content):
try:
with get_cursor() as cur:
cur.execute(
"""INSERT INTO exercises (user_id, page_id, kind, prompt, model_answer, content)
VALUES (%s, %s, %s, %s, %s, %s)""",
(user_id, page_id, kind, prompt_text, answer, json.dumps(content)),
)
except Exception as e:
logger.warning("_save_exercise failed: %s", e)