645 lines
23 KiB
Python
645 lines
23 KiB
Python
"""
|
|
CEFR Level Vocabulary Tests & Assessment Engine (A1, A2, B1, B2).
|
|
Oxford 3000 / 5000 based curated benchmark Cloze Tap questions.
|
|
100% offline, zero-dependency, pure Python stdlib + SQLite.
|
|
Features:
|
|
- Fast 1-tap Cloze Tap quiz format.
|
|
- Automated Gap Ingestion: mistakes automatically create/update vocabulary cards
|
|
with status 'learning' in the user's SRS active deck with level tags.
|
|
- Per-level progress tracking (mastered, learning, percentage).
|
|
"""
|
|
|
|
import re
|
|
import json
|
|
import sqlite3
|
|
import random
|
|
from typing import Dict, Any, List, Optional
|
|
from datetime import datetime
|
|
from app.dictionary import clean_russian_translation
|
|
|
|
CEFR_LEVELS = ["A1", "A2", "B1", "B2"]
|
|
|
|
|
|
from pathlib import Path
|
|
from app import config
|
|
|
|
_CURATED_CEFR_QUESTIONS_CACHE = None
|
|
|
|
def get_curated_cefr_questions():
|
|
global _CURATED_CEFR_QUESTIONS_CACHE
|
|
if _CURATED_CEFR_QUESTIONS_CACHE is None:
|
|
seeds_file = config.BASE_DIR / 'data' / 'seeds' / 'cefr_questions.json'
|
|
if seeds_file.is_file():
|
|
try:
|
|
with open(seeds_file, 'r', encoding='utf-8') as f:
|
|
_CURATED_CEFR_QUESTIONS_CACHE = json.load(f)
|
|
except Exception:
|
|
_CURATED_CEFR_QUESTIONS_CACHE = []
|
|
else:
|
|
_CURATED_CEFR_QUESTIONS_CACHE = []
|
|
return _CURATED_CEFR_QUESTIONS_CACHE
|
|
|
|
class _LazyQuestionsList(list):
|
|
def __iter__(self):
|
|
return iter(get_curated_cefr_questions())
|
|
def __len__(self):
|
|
return len(get_curated_cefr_questions())
|
|
def __getitem__(self, idx):
|
|
return get_curated_cefr_questions()[idx]
|
|
|
|
CURATED_CEFR_QUESTIONS = _LazyQuestionsList()
|
|
|
|
|
|
def seed_cefr_questions(conn: sqlite3.Connection) -> int:
|
|
"""
|
|
Populates cefr_questions table with curated Oxford 3000/5000 benchmark questions.
|
|
Safe and idempotent (using INSERT OR IGNORE).
|
|
"""
|
|
inserted = 0
|
|
for q in CURATED_CEFR_QUESTIONS:
|
|
options_str = json.dumps(q["options"], ensure_ascii=False)
|
|
cursor = conn.execute(
|
|
"""
|
|
INSERT OR IGNORE INTO cefr_questions (
|
|
level, term, translation, pos, phonetic, sentence, options_json, explanation
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
q["level"],
|
|
q["term"],
|
|
q["translation"],
|
|
q.get("pos", ""),
|
|
q.get("phonetic", ""),
|
|
q["sentence"],
|
|
options_str,
|
|
q.get("explanation", ""),
|
|
),
|
|
)
|
|
if cursor.rowcount > 0:
|
|
inserted += 1
|
|
return inserted
|
|
|
|
|
|
def get_cefr_level_summaries(conn: sqlite3.Connection, user_id: int) -> List[Dict[str, Any]]:
|
|
"""
|
|
Computes real-time progress metrics for each CEFR level (A1, A2, B1, B2) for user.
|
|
"""
|
|
summaries = []
|
|
level_titles = {
|
|
"A1": "Beginner (Начальный)",
|
|
"A2": "Elementary (Базовый)",
|
|
"B1": "Intermediate (Средний)",
|
|
"B2": "Upper-Intermediate (Продвинутый)",
|
|
}
|
|
|
|
for lvl in CEFR_LEVELS:
|
|
total_q = conn.execute(
|
|
"SELECT COUNT(*) FROM cefr_questions WHERE level = ?", (lvl,)
|
|
).fetchone()[0]
|
|
|
|
row = conn.execute(
|
|
"""
|
|
SELECT
|
|
SUM(CASE WHEN is_mastered = 1 THEN 1 ELSE 0 END) as mastered_cnt,
|
|
SUM(CASE WHEN mistakes > 0 AND is_mastered = 0 THEN 1 ELSE 0 END) as in_learning_cnt,
|
|
COUNT(*) as attempted_cnt
|
|
FROM cefr_progress
|
|
WHERE user_id = ? AND level = ?
|
|
""",
|
|
(user_id, lvl),
|
|
).fetchone()
|
|
|
|
mastered = row["mastered_cnt"] or 0
|
|
in_learning = row["in_learning_cnt"] or 0
|
|
attempted = row["attempted_cnt"] or 0
|
|
|
|
# Also count active SRS cards with this level tag
|
|
srs_tag = f"level-{lvl.lower()}"
|
|
active_vocab_cards = conn.execute(
|
|
"""
|
|
SELECT COUNT(*) FROM vocabulary_cards
|
|
WHERE user_id = ? AND tags LIKE ?
|
|
""",
|
|
(user_id, f"%{srs_tag}%"),
|
|
).fetchone()[0]
|
|
|
|
pct = int(round((mastered / total_q) * 100)) if total_q > 0 else 0
|
|
|
|
summaries.append({
|
|
"level": lvl,
|
|
"title": level_titles.get(lvl, lvl),
|
|
"total_questions": total_q,
|
|
"mastered_count": mastered,
|
|
"in_learning_count": in_learning,
|
|
"attempted_count": attempted,
|
|
"srs_cards_count": active_vocab_cards,
|
|
"progress_percent": pct,
|
|
})
|
|
|
|
return summaries
|
|
|
|
|
|
def get_cefr_quiz_batch(
|
|
conn: sqlite3.Connection,
|
|
user_id: int,
|
|
level: str,
|
|
count: int = 10,
|
|
mode: str = "all",
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
Fetches next batch of Cloze Tap questions for the selected level.
|
|
If mode == 'mistakes', returns questions where user made mistakes and hasn't mastered yet.
|
|
Otherwise prioritizes unattempted questions, then review of mistakes.
|
|
Options are randomized per question. Correct answer is kept hidden from client.
|
|
"""
|
|
clean_level = level.upper().strip()
|
|
if clean_level not in CEFR_LEVELS:
|
|
clean_level = "A1"
|
|
count = max(1, min(count, 30))
|
|
|
|
if mode == "mistakes":
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT q.id, q.level, q.sentence, q.translation, q.pos, q.options_json,
|
|
q.source_book_id, q.source_book_title, q.source_chapter_title
|
|
FROM cefr_questions q
|
|
JOIN cefr_progress p ON q.id = p.question_id
|
|
WHERE p.user_id = ? AND q.level = ? AND p.is_mastered = 0 AND p.mistakes > 0
|
|
ORDER BY p.last_tested_at ASC
|
|
LIMIT ?
|
|
""",
|
|
(user_id, clean_level, count),
|
|
).fetchall()
|
|
else:
|
|
# 1. Unattempted questions
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT q.id, q.level, q.sentence, q.translation, q.pos, q.options_json,
|
|
q.source_book_id, q.source_book_title, q.source_chapter_title
|
|
FROM cefr_questions q
|
|
LEFT JOIN cefr_progress p ON q.id = p.question_id AND p.user_id = ?
|
|
WHERE q.level = ? AND (p.question_id IS NULL OR p.is_mastered = 0)
|
|
ORDER BY CASE WHEN p.question_id IS NULL THEN 0 ELSE 1 END,
|
|
p.last_tested_at ASC,
|
|
q.id DESC
|
|
LIMIT ?
|
|
""",
|
|
(user_id, clean_level, count),
|
|
).fetchall()
|
|
|
|
# If not enough, fill with any from this level
|
|
if len(rows) < count:
|
|
needed = count - len(rows)
|
|
existing_ids = [r["id"] for r in rows]
|
|
placeholders = ",".join(["?"] * len(existing_ids)) if existing_ids else "-1"
|
|
extra = conn.execute(
|
|
f"""
|
|
SELECT q.id, q.level, q.sentence, q.translation, q.pos, q.options_json,
|
|
q.source_book_id, q.source_book_title, q.source_chapter_title
|
|
FROM cefr_questions q
|
|
WHERE q.level = ? AND q.id NOT IN ({placeholders})
|
|
ORDER BY q.id DESC
|
|
LIMIT ?
|
|
""",
|
|
[clean_level] + existing_ids + [needed],
|
|
).fetchall()
|
|
rows = list(rows) + list(extra)
|
|
|
|
quiz_items = []
|
|
for r in rows:
|
|
try:
|
|
options = json.loads(r["options_json"])
|
|
except Exception:
|
|
options = []
|
|
# Randomize options order
|
|
shuffled_options = list(options)
|
|
random.shuffle(shuffled_options)
|
|
|
|
item = {
|
|
"id": r["id"],
|
|
"level": r["level"],
|
|
"sentence": r["sentence"],
|
|
"clue": r["translation"],
|
|
"pos": r["pos"] or "",
|
|
"options": shuffled_options,
|
|
}
|
|
if "source_book_title" in r.keys() and r["source_book_title"]:
|
|
item["source_book_title"] = r["source_book_title"]
|
|
item["source_chapter_title"] = r["source_chapter_title"] or ""
|
|
quiz_items.append(item)
|
|
|
|
return quiz_items
|
|
|
|
|
|
def resolve_context_sentence_for_term(
|
|
conn: sqlite3.Connection,
|
|
term: str,
|
|
candidate_sentence: str = "",
|
|
) -> str:
|
|
"""
|
|
Resolves the best authentic context sentence for a given term:
|
|
1. If candidate_sentence contains the term, use it (with ___ replaced).
|
|
2. Look up in cefr_questions specifically for this term.
|
|
3. Look up in books table for an authentic sentence containing this term.
|
|
4. Fallback to clean natural sentence.
|
|
"""
|
|
clean_term = term.strip()
|
|
if not clean_term:
|
|
return ""
|
|
|
|
# 1. If candidate_sentence already contains this term (and is a complete sentence)
|
|
if candidate_sentence:
|
|
cand_clean = candidate_sentence.strip()
|
|
is_complete = bool(re.search(r"[.!?]['\"”’»)]*$", cand_clean)) and not bool(
|
|
re.search(r"\b(?:Mr|Mrs|Ms|Miss|Dr|Prof|St|Jr|Sr|[A-Za-z])\.['\"”’»)]*$", cand_clean, re.IGNORECASE)
|
|
)
|
|
if is_complete and re.search(rf"\b{re.escape(clean_term)}\b", cand_clean, re.IGNORECASE):
|
|
return re.sub(r"\s+", " ", cand_clean.replace("___", clean_term)).strip()
|
|
|
|
# 2. Look up in cefr_questions specifically for this term
|
|
try:
|
|
q_row = conn.execute(
|
|
"SELECT sentence FROM cefr_questions WHERE lower(trim(term)) = lower(trim(?)) ORDER BY id ASC LIMIT 1",
|
|
(clean_term,),
|
|
).fetchone()
|
|
if q_row and q_row["sentence"]:
|
|
return re.sub(r"\s+", " ", q_row["sentence"].replace("___", clean_term)).strip()
|
|
except Exception:
|
|
pass
|
|
|
|
# 3. Look up in books table
|
|
try:
|
|
books = conn.execute("SELECT chapters_json FROM books ORDER BY id DESC LIMIT 50").fetchall()
|
|
for b in books:
|
|
try:
|
|
chapters = json.loads(b["chapters_json"])
|
|
except Exception:
|
|
continue
|
|
for ch in chapters:
|
|
content = ch.get("content", "")
|
|
if clean_term.lower() in content.lower():
|
|
from app.miner import robust_sentence_split
|
|
for s_clean in robust_sentence_split(content):
|
|
if 20 <= len(s_clean) <= 240 and re.search(rf"\b{re.escape(clean_term)}\b", s_clean, re.IGNORECASE):
|
|
return s_clean.strip()
|
|
except Exception:
|
|
pass
|
|
|
|
# 4. If candidate_sentence had '___' and came from this term
|
|
if candidate_sentence and "___" in candidate_sentence:
|
|
return re.sub(r"\s+", " ", candidate_sentence.replace("___", clean_term)).strip()
|
|
|
|
return f"Using '{clean_term}' in meaningful sentences reinforces active recall and long-term fluency."
|
|
|
|
|
|
def process_cefr_answer(
|
|
conn: sqlite3.Connection,
|
|
user_id: int,
|
|
question_id: int,
|
|
selected_option: str,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Validates user's Cloze Tap choice.
|
|
- If correct: marks question mastered.
|
|
- If mistake:
|
|
- marks mistake in cefr_progress
|
|
- AUTOMATIC GAP INGESTION: creates or updates vocabulary_card in user's dictionary
|
|
with srs_state='learning', interval=0, next_review=now, level tags so it appears
|
|
in daily morning SRS training!
|
|
"""
|
|
q_row = conn.execute(
|
|
"SELECT * FROM cefr_questions WHERE id = ?", (question_id,)
|
|
).fetchone()
|
|
|
|
if not q_row:
|
|
return {"error": "Question not found"}
|
|
|
|
target_term = q_row["term"].strip()
|
|
selected_clean = (selected_option or "").strip()
|
|
is_correct = selected_clean.lower() == target_term.lower()
|
|
explanation = q_row["explanation"] or ""
|
|
level = q_row["level"]
|
|
phonetic = q_row["phonetic"] or ""
|
|
pos = q_row["pos"] or ""
|
|
translation = q_row["translation"] or ""
|
|
sentence_with_term = q_row["sentence"].replace("___", target_term)
|
|
|
|
# 1. Update or Insert into cefr_progress
|
|
prog_row = conn.execute(
|
|
"SELECT * FROM cefr_progress WHERE user_id = ? AND question_id = ?",
|
|
(user_id, question_id),
|
|
).fetchone()
|
|
|
|
if prog_row:
|
|
new_attempts = prog_row["attempts"] + 1
|
|
new_mistakes = prog_row["mistakes"] + (0 if is_correct else 1)
|
|
new_mastered = 1 if is_correct else 0
|
|
conn.execute(
|
|
"""
|
|
UPDATE cefr_progress
|
|
SET is_mastered = ?, attempts = ?, mistakes = ?, last_tested_at = CURRENT_TIMESTAMP
|
|
WHERE id = ?
|
|
""",
|
|
(new_mastered, new_attempts, new_mistakes, prog_row["id"]),
|
|
)
|
|
else:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO cefr_progress (user_id, question_id, level, is_mastered, attempts, mistakes, last_tested_at)
|
|
VALUES (?, ?, ?, ?, 1, ?, CURRENT_TIMESTAMP)
|
|
""",
|
|
(user_id, question_id, level, 1 if is_correct else 0, 0 if is_correct else 1),
|
|
)
|
|
|
|
# 2. Activity log entry for heatmap streak calculation
|
|
target_card_info = None
|
|
selected_card_info = None
|
|
|
|
if not is_correct:
|
|
# =====================================================================
|
|
# DUAL AUTOMATED GAP INGESTION (Crucial Pedagogical Requirement)
|
|
# 1. Correct target word -> goes 100% to vocabulary with its sentence
|
|
# 2. Chosen wrong word -> also goes to vocabulary with its OWN context!
|
|
# =====================================================================
|
|
target_ctx = resolve_context_sentence_for_term(conn, target_term, sentence_with_term)
|
|
target_card_info = _ingest_card_for_user(
|
|
conn,
|
|
user_id=user_id,
|
|
term=target_term,
|
|
translation=translation,
|
|
phonetic=phonetic,
|
|
pos=pos,
|
|
context_sentence=target_ctx,
|
|
explanation=explanation,
|
|
level=level,
|
|
tag_suffix="test-mistake",
|
|
custom_note=f"Верный ответ из теста уровня {level}",
|
|
)
|
|
|
|
# Ingest the erroneously selected option (the user's mistaken choice)
|
|
sel_row = conn.execute(
|
|
"SELECT term, translation, phonetic, pos FROM offline_dictionary WHERE term = ? COLLATE NOCASE",
|
|
(selected_clean,),
|
|
).fetchone()
|
|
sel_trans = clean_russian_translation(sel_row["translation"]) if sel_row else selected_clean
|
|
sel_phon = sel_row["phonetic"] if sel_row else ""
|
|
sel_pos = sel_row["pos"] if sel_row else ""
|
|
sel_ctx = resolve_context_sentence_for_term(conn, selected_clean)
|
|
|
|
selected_card_info = _ingest_card_for_user(
|
|
conn,
|
|
user_id=user_id,
|
|
term=selected_clean,
|
|
translation=sel_trans,
|
|
phonetic=sel_phon,
|
|
pos=sel_pos,
|
|
context_sentence=sel_ctx,
|
|
explanation=f"Выбрано по ошибке в вопросе про '{target_term}'",
|
|
level=level,
|
|
tag_suffix="test-mistake-chosen",
|
|
custom_note=f"Ошибочно выбрано вместо '{target_term}' (тест {level})",
|
|
)
|
|
else:
|
|
# On correct answer, if user has this card in library, record a positive review log
|
|
existing_card = conn.execute(
|
|
"SELECT id FROM vocabulary_cards WHERE user_id = ? AND lower(trim(term)) = ?",
|
|
(user_id, target_term.lower()),
|
|
).fetchone()
|
|
if existing_card:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO srs_logs (user_id, card_id, rating, interval, ease, reviewed_at)
|
|
VALUES (?, ?, 4, 1, 2.5, CURRENT_TIMESTAMP)
|
|
""",
|
|
(user_id, existing_card["id"]),
|
|
)
|
|
|
|
# 3. Build rich options breakdown for ALL 4 options
|
|
try:
|
|
raw_options = json.loads(q_row["options_json"])
|
|
except Exception:
|
|
raw_options = [target_term]
|
|
|
|
options_breakdown = []
|
|
for opt in raw_options:
|
|
opt_clean = opt.strip()
|
|
is_target = (opt_clean.lower() == target_term.lower())
|
|
was_selected = (opt_clean.lower() == selected_clean.lower())
|
|
|
|
if is_target:
|
|
opt_trans = translation
|
|
opt_phon = phonetic
|
|
opt_pos = pos
|
|
else:
|
|
opt_row = conn.execute(
|
|
"SELECT term, translation, phonetic, pos FROM offline_dictionary WHERE term = ? COLLATE NOCASE",
|
|
(opt_clean,),
|
|
).fetchone()
|
|
opt_trans = clean_russian_translation(opt_row["translation"]) if opt_row else opt_clean
|
|
opt_phon = opt_row["phonetic"] if opt_row else ""
|
|
opt_pos = opt_row["pos"] if opt_row else ""
|
|
|
|
user_card = conn.execute(
|
|
"SELECT id, srs_state FROM vocabulary_cards WHERE user_id = ? AND lower(trim(term)) = ?",
|
|
(user_id, opt_clean.lower()),
|
|
).fetchone()
|
|
|
|
is_auto_added = False
|
|
if not is_correct and (is_target or was_selected):
|
|
is_auto_added = True
|
|
|
|
opt_ctx = resolve_context_sentence_for_term(conn, opt_clean, sentence_with_term if is_target else "")
|
|
|
|
options_breakdown.append({
|
|
"term": opt_clean,
|
|
"translation": opt_trans,
|
|
"phonetic": opt_phon,
|
|
"pos": opt_pos,
|
|
"context_sentence": opt_ctx,
|
|
"is_correct": is_target,
|
|
"was_selected": was_selected,
|
|
"in_library": bool(user_card) or is_auto_added,
|
|
"card_id": user_card["id"] if user_card else None,
|
|
"auto_added": is_auto_added,
|
|
})
|
|
|
|
return {
|
|
"correct": is_correct,
|
|
"correct_option": target_term,
|
|
"selected_option": selected_clean,
|
|
"phonetic": phonetic,
|
|
"translation": translation,
|
|
"explanation": explanation,
|
|
"card_created": target_card_info["card_created"] if target_card_info else False,
|
|
"card_updated": target_card_info["card_updated"] if target_card_info else False,
|
|
"card_id": target_card_info["card_id"] if target_card_info else None,
|
|
"card_term": target_term,
|
|
"selected_card_id": selected_card_info["card_id"] if selected_card_info else None,
|
|
"selected_card_term": selected_clean if not is_correct else None,
|
|
"selected_translation": selected_card_info["translation"] if selected_card_info else None,
|
|
"source_book_title": q_row["source_book_title"] if "source_book_title" in q_row.keys() else None,
|
|
"source_chapter_title": q_row["source_chapter_title"] if "source_chapter_title" in q_row.keys() else None,
|
|
"options_breakdown": options_breakdown,
|
|
}
|
|
|
|
|
|
def _ingest_card_for_user(
|
|
conn: sqlite3.Connection,
|
|
user_id: int,
|
|
term: str,
|
|
translation: str,
|
|
phonetic: str,
|
|
pos: str,
|
|
context_sentence: str,
|
|
explanation: str,
|
|
level: str,
|
|
tag_suffix: str = "test-mistake",
|
|
custom_note: str = "",
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Internal helper to insert or update a vocabulary card in 'learning' SRS state.
|
|
"""
|
|
level_tag = f"level-{level.lower()}"
|
|
col_name = f"Уровни CEFR ({level.upper()})"
|
|
term_clean = term.strip()
|
|
|
|
existing_card = conn.execute(
|
|
"SELECT * FROM vocabulary_cards WHERE user_id = ? AND lower(trim(term)) = ?",
|
|
(user_id, term_clean.lower()),
|
|
).fetchone()
|
|
|
|
card_created = False
|
|
card_updated = False
|
|
card_id = None
|
|
|
|
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 (?, ?, ?, ?, ?, '', ?, ?, ?, ?, ?, 0, 0, 2.5, 'learning', 1, CURRENT_TIMESTAMP)
|
|
""",
|
|
(
|
|
user_id,
|
|
term_clean,
|
|
translation or term_clean,
|
|
context_sentence or "",
|
|
phonetic or "",
|
|
pos or "",
|
|
explanation or "",
|
|
f"{level_tag}, cefr-test, {tag_suffix}",
|
|
custom_note or f"Добавлено из теста уровня {level}",
|
|
col_name,
|
|
),
|
|
)
|
|
card_id = cur.lastrowid
|
|
card_created = True
|
|
else:
|
|
card_id = existing_card["id"]
|
|
existing_tags = set([t.strip() for t in (existing_card["tags"] or "").split(",") if t.strip()])
|
|
existing_tags.add(level_tag)
|
|
existing_tags.add("cefr-test")
|
|
existing_tags.add(tag_suffix)
|
|
updated_tags = ", ".join(sorted(existing_tags))
|
|
|
|
updated_context = (existing_card["context_sentence"] or "").strip()
|
|
# If existing context has "___" or lacks the term, overwrite with new authentic context!
|
|
if ("___" in updated_context or not re.search(rf"\b{re.escape(term_clean)}\b", updated_context, re.IGNORECASE)) and context_sentence:
|
|
updated_context = context_sentence
|
|
elif context_sentence and context_sentence not in updated_context:
|
|
updated_context = f"{updated_context}\n\n{context_sentence}".strip()
|
|
|
|
conn.execute(
|
|
"""
|
|
UPDATE vocabulary_cards
|
|
SET srs_state = 'learning',
|
|
srs_next_review = CURRENT_TIMESTAMP,
|
|
error_count = error_count + 1,
|
|
tags = ?,
|
|
context_sentence = ?
|
|
WHERE id = ?
|
|
""",
|
|
(updated_tags, updated_context, card_id),
|
|
)
|
|
card_updated = True
|
|
|
|
# Log in srs_logs
|
|
if 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, card_id),
|
|
)
|
|
|
|
return {
|
|
"card_id": card_id,
|
|
"term": term_clean,
|
|
"translation": translation,
|
|
"card_created": card_created,
|
|
"card_updated": card_updated,
|
|
}
|
|
|
|
|
|
def ingest_cefr_word(
|
|
conn: sqlite3.Connection,
|
|
user_id: int,
|
|
level: str,
|
|
term: str,
|
|
context_sentence: str = "",
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Manually ingests an unselected option word from a CEFR test into user's SRS deck.
|
|
"""
|
|
clean_term = term.strip()
|
|
clean_level = level.upper().strip()
|
|
|
|
resolved_ctx = resolve_context_sentence_for_term(conn, clean_term, context_sentence)
|
|
|
|
row = conn.execute(
|
|
"SELECT term, translation, phonetic, pos FROM offline_dictionary WHERE term = ? COLLATE NOCASE",
|
|
(clean_term,),
|
|
).fetchone()
|
|
trans = clean_russian_translation(row["translation"]) if row else clean_term
|
|
phon = row["phonetic"] if row else ""
|
|
pos = row["pos"] if row else ""
|
|
|
|
info = _ingest_card_for_user(
|
|
conn,
|
|
user_id=user_id,
|
|
term=clean_term,
|
|
translation=trans,
|
|
phonetic=phon,
|
|
pos=pos,
|
|
context_sentence=resolved_ctx,
|
|
explanation=f"Добавлено вручную из вариантов вопроса теста CEFR ({clean_level})",
|
|
level=clean_level,
|
|
tag_suffix="manual-test-add",
|
|
custom_note=f"Добавлено из вариантов вопроса теста {clean_level}",
|
|
)
|
|
return {
|
|
"status": "ok",
|
|
"card_id": info["card_id"],
|
|
"term": clean_term,
|
|
"translation": trans,
|
|
"phonetic": phon,
|
|
"context_sentence": resolved_ctx,
|
|
}
|
|
|
|
|
|
|
|
def reset_user_cefr_level(conn: sqlite3.Connection, user_id: int, level: str) -> int:
|
|
"""
|
|
Resets progress for a specific CEFR level for user.
|
|
"""
|
|
clean_level = level.upper().strip()
|
|
cursor = conn.execute(
|
|
"DELETE FROM cefr_progress WHERE user_id = ? AND level = ?",
|
|
(user_id, clean_level),
|
|
)
|
|
return cursor.rowcount
|