399 lines
14 KiB
Python
399 lines
14 KiB
Python
"""
|
|
Murphy Grammar Engine & Handwritten Journaling (Task 16).
|
|
Structured micro-lessons based on Raymond Murphy's Essential Grammar in Use (A1-A2)
|
|
and English Grammar in Use (B1-B2).
|
|
100% offline, zero-dependency, pure Python stdlib + SQLite.
|
|
Features:
|
|
- Evening Deep Work Mode (5-7 min bite-sized lessons).
|
|
- Contrast-based pedagogical explanations (I do vs I am doing).
|
|
- Golden formulas for physical notebook & handwritten personal journaling prompts.
|
|
- 5 Cloze Tap micro-exercises per unit.
|
|
- Automated SRS Ingestion for grammar mistakes.
|
|
"""
|
|
|
|
import json
|
|
import sqlite3
|
|
import random
|
|
from typing import Dict, Any, List, Optional
|
|
from datetime import datetime
|
|
|
|
|
|
from pathlib import Path
|
|
from app import config
|
|
|
|
_CURATED_MURPHY_UNITS_CACHE = None
|
|
|
|
def get_curated_murphy_units():
|
|
global _CURATED_MURPHY_UNITS_CACHE
|
|
if _CURATED_MURPHY_UNITS_CACHE is None:
|
|
seeds_file = config.BASE_DIR / 'data' / 'seeds' / 'grammar_units.json'
|
|
if seeds_file.is_file():
|
|
try:
|
|
with open(seeds_file, 'r', encoding='utf-8') as f:
|
|
_CURATED_MURPHY_UNITS_CACHE = json.load(f)
|
|
except Exception:
|
|
_CURATED_MURPHY_UNITS_CACHE = []
|
|
else:
|
|
_CURATED_MURPHY_UNITS_CACHE = []
|
|
return _CURATED_MURPHY_UNITS_CACHE
|
|
|
|
class _LazyUnitsList(list):
|
|
def __iter__(self):
|
|
return iter(get_curated_murphy_units())
|
|
def __len__(self):
|
|
return len(get_curated_murphy_units())
|
|
def __getitem__(self, idx):
|
|
return get_curated_murphy_units()[idx]
|
|
|
|
CURATED_MURPHY_UNITS = _LazyUnitsList()
|
|
|
|
|
|
def seed_grammar_units(conn: sqlite3.Connection) -> int:
|
|
"""
|
|
Populates grammar_units table with curated Murphy units.
|
|
Safe, idempotent upsert (ON CONFLICT DO UPDATE).
|
|
"""
|
|
inserted = 0
|
|
for u in CURATED_MURPHY_UNITS:
|
|
contrast_json = json.dumps(u["contrast_explanation"], ensure_ascii=False)
|
|
exercises_json = json.dumps(u["exercises"], ensure_ascii=False)
|
|
cursor = conn.execute(
|
|
"""
|
|
INSERT INTO grammar_units (
|
|
unit_num, level, category, title, subtitle, golden_formula,
|
|
cheatsheet_table, detailed_notes,
|
|
contrast_explanation_json, journaling_prompt, exercises_json
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(unit_num) DO UPDATE SET
|
|
level = excluded.level,
|
|
category = excluded.category,
|
|
title = excluded.title,
|
|
subtitle = excluded.subtitle,
|
|
golden_formula = excluded.golden_formula,
|
|
cheatsheet_table = excluded.cheatsheet_table,
|
|
detailed_notes = excluded.detailed_notes,
|
|
contrast_explanation_json = excluded.contrast_explanation_json,
|
|
journaling_prompt = excluded.journaling_prompt,
|
|
exercises_json = excluded.exercises_json
|
|
""",
|
|
(
|
|
u["unit_num"],
|
|
u["level"],
|
|
u["category"],
|
|
u["title"],
|
|
u.get("subtitle", ""),
|
|
u["golden_formula"],
|
|
u.get("cheatsheet_table", ""),
|
|
u.get("detailed_notes", ""),
|
|
contrast_json,
|
|
u["journaling_prompt"],
|
|
exercises_json,
|
|
),
|
|
)
|
|
if cursor.rowcount > 0:
|
|
inserted += 1
|
|
return inserted
|
|
|
|
|
|
def get_grammar_units_list(conn: sqlite3.Connection, user_id: int) -> List[Dict[str, Any]]:
|
|
"""
|
|
Returns list of all grammar units with user progress and handwritten status.
|
|
"""
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT
|
|
u.id, u.unit_num, u.level, u.category, u.title, u.subtitle,
|
|
p.is_completed, p.score, p.handwritten_logged, p.completed_at
|
|
FROM grammar_units u
|
|
LEFT JOIN grammar_progress p ON u.id = p.unit_id AND p.user_id = ?
|
|
ORDER BY u.unit_num ASC
|
|
""",
|
|
(user_id,),
|
|
).fetchall()
|
|
|
|
units = []
|
|
for r in rows:
|
|
units.append({
|
|
"id": r["id"],
|
|
"unit_num": r["unit_num"],
|
|
"level": r["level"],
|
|
"category": r["category"],
|
|
"title": r["title"],
|
|
"subtitle": r["subtitle"],
|
|
"is_completed": bool(r["is_completed"]),
|
|
"score": r["score"] or 0,
|
|
"handwritten_logged": bool(r["handwritten_logged"]),
|
|
"completed_at": r["completed_at"],
|
|
})
|
|
return units
|
|
|
|
|
|
def get_grammar_unit_detail(conn: sqlite3.Connection, user_id: int, unit_id: int) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Returns full content of a unit for lesson mode.
|
|
Supports looking up by database PK or unit_num.
|
|
"""
|
|
row = conn.execute(
|
|
"""
|
|
SELECT
|
|
u.id, u.unit_num, u.level, u.category, u.title, u.subtitle,
|
|
u.golden_formula, u.cheatsheet_table, u.detailed_notes,
|
|
u.contrast_explanation_json, u.journaling_prompt, u.exercises_json,
|
|
p.is_completed, p.score, p.handwritten_logged, p.completed_at
|
|
FROM grammar_units u
|
|
LEFT JOIN grammar_progress p ON u.id = p.unit_id AND p.user_id = ?
|
|
WHERE u.id = ? OR u.unit_num = ?
|
|
ORDER BY (u.id = ?) DESC
|
|
LIMIT 1
|
|
""",
|
|
(user_id, unit_id, unit_id, unit_id),
|
|
).fetchone()
|
|
|
|
if not row:
|
|
return None
|
|
|
|
try:
|
|
contrasts = json.loads(row["contrast_explanation_json"])
|
|
except Exception:
|
|
contrasts = []
|
|
|
|
try:
|
|
raw_exercises = json.loads(row["exercises_json"])
|
|
except Exception:
|
|
raw_exercises = []
|
|
|
|
# Clean exercises for client (shuffle options, don't expose target directly in payload)
|
|
exercises = []
|
|
for idx, ex in enumerate(raw_exercises):
|
|
opts = list(ex.get("options", []))
|
|
random.shuffle(opts)
|
|
exercises.append({
|
|
"idx": idx,
|
|
"sentence": ex["sentence"],
|
|
"clue": ex.get("clue", ""),
|
|
"options": opts,
|
|
})
|
|
|
|
return {
|
|
"id": row["id"],
|
|
"unit_num": row["unit_num"],
|
|
"level": row["level"],
|
|
"category": row["category"],
|
|
"title": row["title"],
|
|
"subtitle": row["subtitle"] or "",
|
|
"golden_formula": row["golden_formula"],
|
|
"cheatsheet_table": row["cheatsheet_table"] or "",
|
|
"detailed_notes": row["detailed_notes"] or "",
|
|
"contrasts": contrasts,
|
|
"contrast_explanation": contrasts,
|
|
"journaling_prompt": row["journaling_prompt"],
|
|
"exercises": exercises,
|
|
"is_completed": bool(row["is_completed"]),
|
|
"score": row["score"] or 0,
|
|
"handwritten_logged": bool(row["handwritten_logged"]),
|
|
"completed_at": row["completed_at"],
|
|
}
|
|
|
|
|
|
def process_grammar_answer(
|
|
conn: sqlite3.Connection,
|
|
user_id: int,
|
|
unit_id: int,
|
|
exercise_idx: int,
|
|
selected_option: str,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Validates answer to a micro-exercise.
|
|
If wrong, performs Automated Gap Ingestion into user's SRS active vocabulary
|
|
with tags 'grammar, murphy-unit-{unit_num}' and srs_state='learning'.
|
|
"""
|
|
unit_row = conn.execute(
|
|
"SELECT * FROM grammar_units WHERE id = ? OR unit_num = ? ORDER BY (id = ?) DESC LIMIT 1",
|
|
(unit_id, unit_id, unit_id),
|
|
).fetchone()
|
|
|
|
if not unit_row:
|
|
return {"error": "Unit not found"}
|
|
|
|
try:
|
|
exercises = json.loads(unit_row["exercises_json"])
|
|
ex = exercises[exercise_idx]
|
|
except Exception:
|
|
return {"error": "Exercise not found"}
|
|
|
|
target = ex["target"].strip()
|
|
selected_clean = (selected_option or "").strip()
|
|
is_correct = selected_clean.lower() == target.lower()
|
|
explanation = ex.get("explanation", "")
|
|
full_sentence = ex["sentence"].replace("___", target)
|
|
unit_num = unit_row["unit_num"]
|
|
level = unit_row["level"]
|
|
|
|
card_created = False
|
|
card_updated = False
|
|
target_card_id = None
|
|
|
|
if not is_correct:
|
|
# =====================================================================
|
|
# AUTOMATED GAP INGESTION FOR GRAMMAR MISTAKE
|
|
# =====================================================================
|
|
# Term represents the grammatical pattern or key phrase
|
|
term_label = f"{target} ({unit_row['title'].split('(')[0].strip()})"
|
|
collection_name = "Грамматика Мерфи"
|
|
tags_str = f"grammar, murphy-unit-{unit_num}, {level.lower().replace('-', '_')}"
|
|
|
|
existing_card = conn.execute(
|
|
"SELECT * FROM vocabulary_cards WHERE user_id = ? AND lower(trim(term)) = ?",
|
|
(user_id, term_label.lower()),
|
|
).fetchone()
|
|
|
|
if not existing_card:
|
|
cur = conn.execute(
|
|
"""
|
|
INSERT INTO vocabulary_cards (
|
|
user_id, term, translation, context_sentence, phonetic, audio_url,
|
|
part_of_speech, definition_en, tags, notes, collection,
|
|
srs_reps, srs_interval, srs_ease, srs_state, error_count,
|
|
srs_next_review
|
|
) VALUES (?, ?, ?, ?, '', '', 'grammar', ?, ?, ?, ?, 0, 0, 2.5, 'learning', 1, CURRENT_TIMESTAMP)
|
|
""",
|
|
(
|
|
user_id,
|
|
term_label,
|
|
ex.get("clue", target),
|
|
full_sentence,
|
|
explanation,
|
|
tags_str,
|
|
f"Ошибка из упражнения Юнита {unit_num} (Мерфи)",
|
|
collection_name,
|
|
),
|
|
)
|
|
target_card_id = cur.lastrowid
|
|
card_created = True
|
|
else:
|
|
target_card_id = existing_card["id"]
|
|
conn.execute(
|
|
"""
|
|
UPDATE vocabulary_cards
|
|
SET srs_state = 'learning',
|
|
srs_next_review = CURRENT_TIMESTAMP,
|
|
error_count = error_count + 1
|
|
WHERE id = ?
|
|
""",
|
|
(target_card_id,),
|
|
)
|
|
card_updated = True
|
|
|
|
# Log into srs_logs
|
|
if target_card_id:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO srs_logs (user_id, card_id, rating, interval, ease, reviewed_at)
|
|
VALUES (?, ?, 1, 0, 2.5, CURRENT_TIMESTAMP)
|
|
""",
|
|
(user_id, target_card_id),
|
|
)
|
|
|
|
return {
|
|
"correct": is_correct,
|
|
"correct_option": target,
|
|
"explanation": explanation,
|
|
"full_sentence": full_sentence,
|
|
"card_created": card_created,
|
|
"card_updated": card_updated,
|
|
"card_id": target_card_id,
|
|
}
|
|
|
|
|
|
def record_grammar_completion(
|
|
conn: sqlite3.Connection,
|
|
user_id: int,
|
|
unit_id: int,
|
|
score: int,
|
|
handwritten_logged: bool,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Saves unit completion score and handwritten journaling status.
|
|
"""
|
|
u_row = conn.execute(
|
|
"SELECT id FROM grammar_units WHERE id = ? OR unit_num = ? ORDER BY (id = ?) DESC LIMIT 1",
|
|
(unit_id, unit_id, unit_id),
|
|
).fetchone()
|
|
real_unit_id = u_row["id"] if u_row else unit_id
|
|
|
|
prog_row = conn.execute(
|
|
"SELECT id FROM grammar_progress WHERE user_id = ? AND unit_id = ?",
|
|
(user_id, real_unit_id),
|
|
).fetchone()
|
|
|
|
hw_val = 1 if handwritten_logged else 0
|
|
|
|
if prog_row:
|
|
conn.execute(
|
|
"""
|
|
UPDATE grammar_progress
|
|
SET is_completed = 1,
|
|
score = max(score, ?),
|
|
handwritten_logged = max(handwritten_logged, ?),
|
|
handwritten_at = CASE WHEN ? = 1 THEN CURRENT_TIMESTAMP ELSE handwritten_at END,
|
|
completed_at = CURRENT_TIMESTAMP
|
|
WHERE id = ?
|
|
""",
|
|
(score, hw_val, hw_val, prog_row["id"]),
|
|
)
|
|
else:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO grammar_progress (
|
|
user_id, unit_id, is_completed, score, handwritten_logged, handwritten_at, completed_at
|
|
) VALUES (?, ?, 1, ?, ?, CASE WHEN ? = 1 THEN CURRENT_TIMESTAMP ELSE NULL END, CURRENT_TIMESTAMP)
|
|
""",
|
|
(user_id, real_unit_id, score, hw_val, hw_val),
|
|
)
|
|
|
|
return {"status": "ok", "unit_id": real_unit_id, "score": score, "handwritten_logged": handwritten_logged}
|
|
|
|
|
|
def update_handwritten_status(
|
|
conn: sqlite3.Connection,
|
|
user_id: int,
|
|
unit_id: int,
|
|
handwritten: bool,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Toggles handwritten paper notebook checkbox for a unit.
|
|
"""
|
|
u_row = conn.execute(
|
|
"SELECT id FROM grammar_units WHERE id = ? OR unit_num = ? ORDER BY (id = ?) DESC LIMIT 1",
|
|
(unit_id, unit_id, unit_id),
|
|
).fetchone()
|
|
real_unit_id = u_row["id"] if u_row else unit_id
|
|
|
|
hw_val = 1 if handwritten else 0
|
|
prog_row = conn.execute(
|
|
"SELECT id FROM grammar_progress WHERE user_id = ? AND unit_id = ?",
|
|
(user_id, real_unit_id),
|
|
).fetchone()
|
|
|
|
if prog_row:
|
|
conn.execute(
|
|
"""
|
|
UPDATE grammar_progress
|
|
SET handwritten_logged = ?,
|
|
handwritten_at = CASE WHEN ? = 1 THEN CURRENT_TIMESTAMP ELSE NULL END
|
|
WHERE id = ?
|
|
""",
|
|
(hw_val, hw_val, prog_row["id"]),
|
|
)
|
|
else:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO grammar_progress (
|
|
user_id, unit_id, is_completed, score, handwritten_logged, handwritten_at
|
|
) VALUES (?, ?, 0, 0, ?, CASE WHEN ? = 1 THEN CURRENT_TIMESTAMP ELSE NULL END)
|
|
""",
|
|
(user_id, real_unit_id, hw_val, hw_val),
|
|
)
|
|
|
|
return {"status": "ok", "unit_id": real_unit_id, "handwritten_logged": handwritten}
|