490 lines
16 KiB
Python
490 lines
16 KiB
Python
"""
|
|
Comprehensive Lesson Builder & Runner Module (Beta).
|
|
Enables customizable, multi-step learning workflows (Words -> Grammar -> Listening -> Reading -> AI Debrief).
|
|
100% offline, zero-dependency, pure Python stdlib + SQLite.
|
|
"""
|
|
|
|
import json
|
|
import sqlite3
|
|
import random
|
|
from typing import Dict, Any, List, Optional
|
|
from datetime import datetime
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Curated High-Yield Bite-Sized Reading Articles & Facts
|
|
# -----------------------------------------------------------------------------
|
|
|
|
from pathlib import Path
|
|
from app import config
|
|
|
|
_CURATED_ARTICLES_CACHE = None
|
|
|
|
def get_curated_articles_list():
|
|
global _CURATED_ARTICLES_CACHE
|
|
if _CURATED_ARTICLES_CACHE is None:
|
|
seeds_file = config.BASE_DIR / 'data' / 'seeds' / 'reading_articles.json'
|
|
if seeds_file.is_file():
|
|
try:
|
|
with open(seeds_file, 'r', encoding='utf-8') as f:
|
|
_CURATED_ARTICLES_CACHE = json.load(f)
|
|
except Exception:
|
|
_CURATED_ARTICLES_CACHE = []
|
|
else:
|
|
_CURATED_ARTICLES_CACHE = []
|
|
return _CURATED_ARTICLES_CACHE
|
|
|
|
class _LazyArticlesList(list):
|
|
def __iter__(self):
|
|
return iter(get_curated_articles_list())
|
|
def __len__(self):
|
|
return len(get_curated_articles_list())
|
|
def __getitem__(self, idx):
|
|
return get_curated_articles_list()[idx]
|
|
|
|
CURATED_ARTICLES = _LazyArticlesList()
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Default Lesson Presets (Ready-to-use Configurations)
|
|
# -----------------------------------------------------------------------------
|
|
DEFAULT_PRESETS: List[Dict[str, Any]] = [
|
|
{
|
|
"id": "express",
|
|
"name": "⚡ Утренний экспресс (10 мин)",
|
|
"icon": "⚡",
|
|
"description": "Быстрый разгон словарного запаса, ретро-тест грамматики и 1 аудио-ситуация.",
|
|
"estimated_min": 10,
|
|
"is_default": 1,
|
|
"steps": [
|
|
{
|
|
"id": "step_vocab",
|
|
"type": "vocab",
|
|
"title": "Вспоминаем слова (SRS)",
|
|
"enabled": True,
|
|
"count": 10,
|
|
"mode": "flashcards",
|
|
"filter": "due"
|
|
},
|
|
{
|
|
"id": "step_grammar_review",
|
|
"type": "grammar_review",
|
|
"title": "Закрепление грамматики",
|
|
"enabled": True,
|
|
"count": 5,
|
|
"topic_mode": "weakest"
|
|
},
|
|
{
|
|
"id": "step_listening",
|
|
"type": "listening",
|
|
"title": "Аудирование на слух",
|
|
"enabled": True,
|
|
"count": 1,
|
|
"topic": "all"
|
|
},
|
|
{
|
|
"id": "step_ai_debrief",
|
|
"type": "ai_debrief",
|
|
"title": "ИИ-Итог и рекомендации",
|
|
"enabled": True
|
|
}
|
|
]
|
|
},
|
|
{
|
|
"id": "balanced",
|
|
"name": "🎯 Сбалансированный урок (20 мин)",
|
|
"icon": "🎯",
|
|
"description": "Слова + закрепление прошлой темы + новая теория Мерфи + 2 ситуации на слух.",
|
|
"estimated_min": 20,
|
|
"is_default": 1,
|
|
"steps": [
|
|
{
|
|
"id": "step_vocab",
|
|
"type": "vocab",
|
|
"title": "Вспоминаем слова (SRS)",
|
|
"enabled": True,
|
|
"count": 10,
|
|
"mode": "quiz",
|
|
"filter": "due"
|
|
},
|
|
{
|
|
"id": "step_grammar_review",
|
|
"type": "grammar_review",
|
|
"title": "Закрепление грамматики",
|
|
"enabled": True,
|
|
"count": 5,
|
|
"topic_mode": "weakest"
|
|
},
|
|
{
|
|
"id": "step_grammar_new",
|
|
"type": "grammar_new",
|
|
"title": "Новая тема Мерфи",
|
|
"enabled": True,
|
|
"unit_mode": "auto_next",
|
|
"exercises_count": 5
|
|
},
|
|
{
|
|
"id": "step_listening",
|
|
"type": "listening",
|
|
"title": "Аудирование на слух",
|
|
"enabled": True,
|
|
"count": 2,
|
|
"topic": "all"
|
|
},
|
|
{
|
|
"id": "step_ai_debrief",
|
|
"type": "ai_debrief",
|
|
"title": "ИИ-Итог и рекомендации",
|
|
"enabled": True
|
|
}
|
|
]
|
|
},
|
|
{
|
|
"id": "deep",
|
|
"name": "📖 Глубокое погружение (30 мин)",
|
|
"icon": "📖",
|
|
"description": "Полный комплексный цикл: слова, закрепление, новая грамматика, аудирование и чтение.",
|
|
"estimated_min": 30,
|
|
"is_default": 1,
|
|
"steps": [
|
|
{
|
|
"id": "step_vocab",
|
|
"type": "vocab",
|
|
"title": "Вспоминаем слова (SRS)",
|
|
"enabled": True,
|
|
"count": 15,
|
|
"mode": "typing",
|
|
"filter": "due"
|
|
},
|
|
{
|
|
"id": "step_grammar_review",
|
|
"type": "grammar_review",
|
|
"title": "Закрепление грамматики",
|
|
"enabled": True,
|
|
"count": 5,
|
|
"topic_mode": "weakest"
|
|
},
|
|
{
|
|
"id": "step_grammar_new",
|
|
"type": "grammar_new",
|
|
"title": "Новая тема Мерфи",
|
|
"enabled": True,
|
|
"unit_mode": "auto_next",
|
|
"exercises_count": 5
|
|
},
|
|
{
|
|
"id": "step_listening",
|
|
"type": "listening",
|
|
"title": "Аудирование на слух",
|
|
"enabled": True,
|
|
"count": 2,
|
|
"topic": "all"
|
|
},
|
|
{
|
|
"id": "step_reading",
|
|
"type": "reading",
|
|
"title": "Чтение и словарная работа",
|
|
"enabled": True,
|
|
"source": "curated_article",
|
|
"article_id": "random"
|
|
},
|
|
{
|
|
"id": "step_ai_debrief",
|
|
"type": "ai_debrief",
|
|
"title": "ИИ-Итог и рекомендации",
|
|
"enabled": True
|
|
}
|
|
]
|
|
},
|
|
{
|
|
"id": "audio_read",
|
|
"name": "🎧 Аудио + Чтение (15 мин)",
|
|
"icon": "🎧",
|
|
"description": "Фокус на понимании на слух и чтении статьи с пополнением словаря.",
|
|
"estimated_min": 15,
|
|
"is_default": 1,
|
|
"steps": [
|
|
{
|
|
"id": "step_vocab",
|
|
"type": "vocab",
|
|
"title": "Вспоминаем слова (SRS)",
|
|
"enabled": True,
|
|
"count": 10,
|
|
"mode": "flashcards",
|
|
"filter": "due"
|
|
},
|
|
{
|
|
"id": "step_listening",
|
|
"type": "listening",
|
|
"title": "Аудирование на слух",
|
|
"enabled": True,
|
|
"count": 3,
|
|
"topic": "all"
|
|
},
|
|
{
|
|
"id": "step_reading",
|
|
"type": "reading",
|
|
"title": "Чтение и словарная работа",
|
|
"enabled": True,
|
|
"source": "curated_article",
|
|
"article_id": "random"
|
|
},
|
|
{
|
|
"id": "step_ai_debrief",
|
|
"type": "ai_debrief",
|
|
"title": "ИИ-Итог и рекомендации",
|
|
"enabled": True
|
|
}
|
|
]
|
|
}
|
|
]
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Database & Helper Functions
|
|
# -----------------------------------------------------------------------------
|
|
|
|
def get_user_presets(conn: sqlite3.Connection, user_id: int) -> List[Dict[str, Any]]:
|
|
"""Returns built-in system presets combined with user custom saved presets."""
|
|
result = []
|
|
# 1. System defaults
|
|
for p in DEFAULT_PRESETS:
|
|
item = dict(p)
|
|
item["is_custom"] = False
|
|
result.append(item)
|
|
|
|
# 2. User custom presets from SQLite
|
|
try:
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT id, name, icon, description, estimated_min, config_json, created_at
|
|
FROM lesson_presets
|
|
WHERE user_id = ?
|
|
ORDER BY id ASC
|
|
""",
|
|
(user_id,),
|
|
).fetchall()
|
|
for r in rows:
|
|
cfg = json.loads(r["config_json"]) if r["config_json"] else []
|
|
result.append({
|
|
"id": f"custom_{r['id']}",
|
|
"custom_id": r["id"],
|
|
"name": r["name"],
|
|
"icon": r["icon"] or "🚀",
|
|
"description": r["description"] or "",
|
|
"estimated_min": r["estimated_min"] or 15,
|
|
"is_default": 0,
|
|
"is_custom": True,
|
|
"steps": cfg,
|
|
"created_at": r["created_at"],
|
|
})
|
|
except Exception:
|
|
pass
|
|
|
|
return result
|
|
|
|
|
|
def save_user_preset(
|
|
conn: sqlite3.Connection,
|
|
user_id: int,
|
|
name: str,
|
|
steps: List[Dict[str, Any]],
|
|
icon: str = "🚀",
|
|
description: str = "",
|
|
estimated_min: int = 15,
|
|
) -> Dict[str, Any]:
|
|
"""Saves a custom user lesson configuration to SQLite."""
|
|
config_json = json.dumps(steps, ensure_ascii=False)
|
|
cur = conn.execute(
|
|
"""
|
|
INSERT INTO lesson_presets (user_id, name, icon, description, estimated_min, config_json)
|
|
VALUES (?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(user_id, name.strip(), icon, description.strip(), estimated_min, config_json),
|
|
)
|
|
new_id = cur.lastrowid
|
|
return {
|
|
"id": f"custom_{new_id}",
|
|
"custom_id": new_id,
|
|
"name": name.strip(),
|
|
"icon": icon,
|
|
"description": description.strip(),
|
|
"estimated_min": estimated_min,
|
|
"steps": steps,
|
|
"is_custom": True,
|
|
}
|
|
|
|
|
|
def delete_user_preset(conn: sqlite3.Connection, user_id: int, custom_id: int) -> bool:
|
|
"""Deletes a custom user lesson preset."""
|
|
cur = conn.execute(
|
|
"DELETE FROM lesson_presets WHERE id = ? AND user_id = ?",
|
|
(custom_id, user_id),
|
|
)
|
|
return cur.rowcount > 0
|
|
|
|
|
|
def get_grammar_recommendations(conn: sqlite3.Connection, user_id: int) -> Dict[str, Any]:
|
|
"""
|
|
Calculates intelligent grammar suggestions for the lesson:
|
|
- weakest_unit: unit with lowest score or most errors recorded in srs_logs
|
|
- previous_unit: the last unit completed by user
|
|
- next_unit: the first uncompleted unit in pedagogical sequence
|
|
- all_units: complete list of units for manual selection
|
|
"""
|
|
# 1. Fetch all grammar units
|
|
all_units_rows = conn.execute(
|
|
"""
|
|
SELECT u.id, u.unit_num, u.level, u.category, u.title, u.subtitle, u.golden_formula,
|
|
COALESCE(p.score, 0) as user_score,
|
|
COALESCE(p.is_completed, 0) as is_completed,
|
|
p.completed_at
|
|
FROM grammar_units u
|
|
LEFT JOIN grammar_progress p ON p.unit_id = u.id AND p.user_id = ?
|
|
ORDER BY u.unit_num ASC
|
|
""",
|
|
(user_id,),
|
|
).fetchall()
|
|
|
|
all_units = []
|
|
completed_units = []
|
|
uncompleted_units = []
|
|
|
|
for r in all_units_rows:
|
|
unit_info = {
|
|
"id": r["id"],
|
|
"unit_num": r["unit_num"],
|
|
"level": r["level"],
|
|
"category": r["category"],
|
|
"title": r["title"],
|
|
"subtitle": r["subtitle"] or "",
|
|
"golden_formula": r["golden_formula"] or "",
|
|
"score": r["user_score"],
|
|
"is_completed": bool(r["is_completed"]),
|
|
"completed_at": r["completed_at"],
|
|
}
|
|
all_units.append(unit_info)
|
|
if unit_info["is_completed"]:
|
|
completed_units.append(unit_info)
|
|
else:
|
|
uncompleted_units.append(unit_info)
|
|
|
|
# 2. Determine Weakest Unit (lowest score among completed units, or unit with lowest score)
|
|
weakest_unit = None
|
|
if completed_units:
|
|
weakest_unit = min(completed_units, key=lambda u: u["score"])
|
|
elif all_units:
|
|
weakest_unit = all_units[0]
|
|
|
|
# 3. Determine Previous Unit (most recently completed unit)
|
|
previous_unit = None
|
|
if completed_units:
|
|
# Sort by completed_at desc
|
|
sorted_completed = sorted(
|
|
completed_units,
|
|
key=lambda u: u["completed_at"] or "",
|
|
reverse=True,
|
|
)
|
|
previous_unit = sorted_completed[0]
|
|
elif all_units:
|
|
previous_unit = all_units[0]
|
|
|
|
# 4. Determine Next Unit (first uncompleted unit)
|
|
next_unit = uncompleted_units[0] if uncompleted_units else (all_units[0] if all_units else None)
|
|
|
|
return {
|
|
"weakest_unit": weakest_unit,
|
|
"previous_unit": previous_unit,
|
|
"next_unit": next_unit,
|
|
"all_units": all_units,
|
|
}
|
|
|
|
|
|
def get_curated_articles(topic: Optional[str] = None) -> List[Dict[str, Any]]:
|
|
"""Returns curated reading articles, optionally filtered by category."""
|
|
if not topic or topic.lower() == "all":
|
|
return CURATED_ARTICLES
|
|
return [a for a in CURATED_ARTICLES if a.get("topic", "").lower() == topic.lower()]
|
|
|
|
|
|
def record_lesson_session(
|
|
conn: sqlite3.Connection,
|
|
user_id: int,
|
|
preset_name: str,
|
|
duration_seconds: int,
|
|
total_score: int,
|
|
grade: str,
|
|
steps_summary: List[Dict[str, Any]],
|
|
errors_list: List[Dict[str, Any]],
|
|
ai_report: Dict[str, Any],
|
|
) -> int:
|
|
"""Stores the finished composite lesson session in SQLite."""
|
|
cur = conn.execute(
|
|
"""
|
|
INSERT INTO lesson_sessions (
|
|
user_id, preset_name, duration_seconds, total_score, grade,
|
|
steps_summary_json, errors_json, ai_report_json
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
user_id,
|
|
preset_name,
|
|
duration_seconds,
|
|
total_score,
|
|
grade,
|
|
json.dumps(steps_summary, ensure_ascii=False),
|
|
json.dumps(errors_list, ensure_ascii=False),
|
|
json.dumps(ai_report, ensure_ascii=False),
|
|
),
|
|
)
|
|
return cur.lastrowid
|
|
|
|
|
|
def get_lesson_history(conn: sqlite3.Connection, user_id: int, limit: int = 15) -> Dict[str, Any]:
|
|
"""Retrieves previous comprehensive lesson sessions for user analytics."""
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT id, preset_name, duration_seconds, total_score, grade,
|
|
steps_summary_json, errors_json, ai_report_json, completed_at
|
|
FROM lesson_sessions
|
|
WHERE user_id = ?
|
|
ORDER BY completed_at DESC
|
|
LIMIT ?
|
|
""",
|
|
(user_id, limit),
|
|
).fetchall()
|
|
|
|
history = []
|
|
total_lessons = 0
|
|
total_time_sec = 0
|
|
avg_score = 0
|
|
score_sum = 0
|
|
|
|
for r in rows:
|
|
steps = json.loads(r["steps_summary_json"]) if r["steps_summary_json"] else []
|
|
errors = json.loads(r["errors_json"]) if r["errors_json"] else []
|
|
ai_rep = json.loads(r["ai_report_json"]) if r["ai_report_json"] else {}
|
|
history.append({
|
|
"id": r["id"],
|
|
"preset_name": r["preset_name"],
|
|
"duration_seconds": r["duration_seconds"],
|
|
"total_score": r["total_score"],
|
|
"grade": r["grade"],
|
|
"steps": steps,
|
|
"errors_count": len(errors),
|
|
"headline": ai_rep.get("headline", ""),
|
|
"completed_at": r["completed_at"],
|
|
})
|
|
total_lessons += 1
|
|
total_time_sec += r["duration_seconds"] or 0
|
|
score_sum += r["total_score"] or 0
|
|
|
|
if total_lessons > 0:
|
|
avg_score = round(score_sum / total_lessons)
|
|
|
|
return {
|
|
"total_lessons": total_lessons,
|
|
"total_time_minutes": round(total_time_sec / 60),
|
|
"avg_score": avg_score,
|
|
"history": history,
|
|
}
|