john
05e173b293
Replace the WebView shell with SwiftUI screens, add account-scoped Wiki and TTS APIs with adaptive review and photo scan support, and keep web/iOS pages usable while data loads asynchronously. Co-authored-by: Cursor <cursoragent@cursor.com>
402 lines
14 KiB
Python
402 lines
14 KiB
Python
import re
|
|
from datetime import datetime, timezone
|
|
from typing import Optional
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from models import QuizRecord, User, Word
|
|
from schemas import CoachSessionResponse, CoachTurnResponse, CoachWordBrief, MemoryToken
|
|
from services.adaptive_review_service import adaptive_review_service, retrievability
|
|
from services.memory_visual_service import (
|
|
memory_visual_service,
|
|
parse_iso,
|
|
retention_percent,
|
|
stability_hours,
|
|
word_en,
|
|
word_zh,
|
|
)
|
|
from services.quiz_service import quiz_service
|
|
from services.text_utils import en_answers_match
|
|
from services.word_service import word_service
|
|
|
|
|
|
def _mask_zh(zh: str) -> str:
|
|
zh = zh.strip()
|
|
if len(zh) <= 1:
|
|
return zh
|
|
return zh[0] + "※" * (len(zh) - 1)
|
|
|
|
|
|
def _blank_example(example: str, en: str) -> str:
|
|
if not example:
|
|
return ""
|
|
pattern = re.compile(re.escape(en), re.IGNORECASE)
|
|
return pattern.sub("______", example, count=1)
|
|
|
|
|
|
def _retention_now(word: Word) -> float:
|
|
now = datetime.now(timezone.utc)
|
|
last_at = parse_iso(word.last_reviewed_at or word.created_at)
|
|
hours = (now - last_at).total_seconds() / 3600
|
|
return round(retention_percent(hours, stability_hours(word)), 1)
|
|
|
|
|
|
def _last_quiz_hint(db: Session, word_id: int) -> Optional[str]:
|
|
r = (
|
|
db.query(QuizRecord)
|
|
.filter(QuizRecord.word_id == word_id)
|
|
.order_by(QuizRecord.created_at.desc())
|
|
.first()
|
|
)
|
|
if not r:
|
|
return None
|
|
return "上次答对" if r.is_correct else "上次答错"
|
|
|
|
|
|
class MemoryCoachService:
|
|
def build_tokens(self, db: Session, word: Word, reveal_extra: int = 0) -> list[MemoryToken]:
|
|
en = word_en(word).strip()
|
|
zh = word_zh(word).strip()
|
|
retention = _retention_now(word)
|
|
hint = _last_quiz_hint(db, word.id)
|
|
|
|
tokens: list[MemoryToken] = [
|
|
MemoryToken(
|
|
role="K",
|
|
key="retention",
|
|
label="记忆保持",
|
|
value=f"{retention}%",
|
|
revealed=True,
|
|
),
|
|
MemoryToken(
|
|
role="K",
|
|
key="mastery",
|
|
label="掌握率",
|
|
value=f"{word.mastery_score}%",
|
|
revealed=True,
|
|
),
|
|
MemoryToken(
|
|
role="K",
|
|
key="status",
|
|
label="词库状态",
|
|
value=word.status,
|
|
revealed=True,
|
|
),
|
|
]
|
|
if hint:
|
|
tokens.append(
|
|
MemoryToken(
|
|
role="K",
|
|
key="last_quiz",
|
|
label="练习记录",
|
|
value=hint,
|
|
revealed=True,
|
|
)
|
|
)
|
|
|
|
q_tokens: list[MemoryToken] = [
|
|
MemoryToken(
|
|
role="Q",
|
|
key="zh_full",
|
|
label="中文释义",
|
|
value=zh,
|
|
revealed=True,
|
|
),
|
|
MemoryToken(
|
|
role="Q",
|
|
key="length",
|
|
label="字母数",
|
|
value=str(len(en)),
|
|
revealed=True,
|
|
),
|
|
]
|
|
if zh:
|
|
q_tokens.append(
|
|
MemoryToken(
|
|
role="Q",
|
|
key="zh_hint",
|
|
label="释义提示",
|
|
value=_mask_zh(zh),
|
|
revealed=reveal_extra > 0,
|
|
)
|
|
)
|
|
if word.phonetic:
|
|
q_tokens.append(
|
|
MemoryToken(
|
|
role="Q",
|
|
key="phonetic",
|
|
label="音标",
|
|
value=word.phonetic,
|
|
revealed=reveal_extra > 0,
|
|
)
|
|
)
|
|
if len(en) >= 2:
|
|
q_tokens.append(
|
|
MemoryToken(
|
|
role="Q",
|
|
key="prefix",
|
|
label="英文前缀",
|
|
value=en[:2] + "…",
|
|
revealed=reveal_extra > 1,
|
|
)
|
|
)
|
|
if len(en) >= 4:
|
|
q_tokens.append(
|
|
MemoryToken(
|
|
role="Q",
|
|
key="suffix",
|
|
label="英文尾缀",
|
|
value="…" + en[-2:],
|
|
revealed=reveal_extra > 2,
|
|
)
|
|
)
|
|
if word.example_en:
|
|
blanked = _blank_example(word.example_en, en)
|
|
if blanked and blanked != word.example_en:
|
|
q_tokens.append(
|
|
MemoryToken(
|
|
role="Q",
|
|
key="example",
|
|
label="例句挖空",
|
|
value=blanked,
|
|
revealed=reveal_extra > 3,
|
|
)
|
|
)
|
|
|
|
tokens.extend(q_tokens)
|
|
tokens.extend(
|
|
[
|
|
MemoryToken(
|
|
role="V",
|
|
key="en",
|
|
label="英文",
|
|
value=en,
|
|
revealed=False,
|
|
),
|
|
MemoryToken(
|
|
role="V",
|
|
key="zh",
|
|
label="中文",
|
|
value=zh,
|
|
revealed=False,
|
|
),
|
|
]
|
|
)
|
|
if word.example_en:
|
|
tokens.append(
|
|
MemoryToken(
|
|
role="V",
|
|
key="example_en",
|
|
label="例句",
|
|
value=word.example_en,
|
|
revealed=False,
|
|
)
|
|
)
|
|
if word.example_cn:
|
|
tokens.append(
|
|
MemoryToken(
|
|
role="V",
|
|
key="example_cn",
|
|
label="例句译文",
|
|
value=word.example_cn,
|
|
revealed=False,
|
|
)
|
|
)
|
|
return tokens
|
|
|
|
def _reveal_q_tokens(self, tokens: list[MemoryToken], count: int) -> list[MemoryToken]:
|
|
hidden_q = [t for t in tokens if t.role == "Q" and not t.revealed]
|
|
for t in hidden_q[:count]:
|
|
t.revealed = True
|
|
return tokens
|
|
|
|
def _reveal_all_v(self, tokens: list[MemoryToken]) -> list[MemoryToken]:
|
|
for t in tokens:
|
|
if t.role == "V":
|
|
t.revealed = True
|
|
if t.role == "Q":
|
|
t.revealed = True
|
|
return tokens
|
|
|
|
def start_session(self, db: Session, user: User, track: str = "accumulation") -> CoachSessionResponse:
|
|
from services.book_service import book_service
|
|
|
|
if track == "book":
|
|
book = book_service.get_active_book(db, user)
|
|
if not book:
|
|
return CoachSessionResponse(words=[], total=0)
|
|
book_service.ensure_user_book_words(db, user, book)
|
|
practice = book_service.practice_settings_dict(db, user, book)
|
|
book_id = book.id
|
|
words = quiz_service.select_coach_words(
|
|
db, user, book_id, practice["daily_target"]
|
|
)
|
|
else:
|
|
settings = quiz_service.get_settings(db, user)
|
|
book_id = 0
|
|
words = quiz_service.select_coach_words(
|
|
db, user, book_id, settings.daily_target
|
|
)
|
|
if not words:
|
|
return CoachSessionResponse(words=[], total=0)
|
|
|
|
summaries = memory_visual_service.get_visualization(db, user, book_id=book_id)["words"]
|
|
risk_map = {w["id"]: w.get("retention_now", 0) for w in summaries}
|
|
|
|
items = [
|
|
CoachWordBrief(
|
|
word_id=w.id,
|
|
zh=word_zh(w),
|
|
phonetic=w.phonetic,
|
|
retention_now=risk_map.get(w.id, _retention_now(w)),
|
|
mastery_score=w.mastery_score,
|
|
status=w.status,
|
|
retrievability=round(retrievability(w) * 100.0, 1),
|
|
next_review_at=w.next_review_at,
|
|
)
|
|
for w in words
|
|
]
|
|
return CoachSessionResponse(words=items, total=len(items))
|
|
|
|
def handle_turn(
|
|
self,
|
|
db: Session,
|
|
user: User,
|
|
word_id: int,
|
|
stage: str,
|
|
user_message: str = "",
|
|
hints_used: int = 0,
|
|
duration_seconds: int = 0,
|
|
) -> CoachTurnResponse:
|
|
word = word_service.get_word(db, user, word_id)
|
|
en = word_en(word).strip()
|
|
zh = word_zh(word).strip()
|
|
phonetic = word.phonetic
|
|
tokens = self.build_tokens(db, word, reveal_extra=hints_used)
|
|
messages: list[str] = []
|
|
expect_input = False
|
|
is_correct: Optional[bool] = None
|
|
quiz_recorded = False
|
|
word_complete = False
|
|
next_stage = stage
|
|
input_hint = f"请输入「{zh}」的英文"
|
|
review_message: Optional[str] = None
|
|
review_retrievability: Optional[float] = None
|
|
next_review_at: Optional[str] = None
|
|
|
|
if stage == "intro":
|
|
next_stage = "derive"
|
|
expect_input = True
|
|
|
|
elif stage == "derive":
|
|
answer = user_message.strip()
|
|
if not answer:
|
|
expect_input = True
|
|
next_stage = "derive"
|
|
else:
|
|
is_correct = en_answers_match(answer, en)
|
|
if is_correct:
|
|
attempts = hints_used + 1
|
|
before_r = retrievability(word)
|
|
tokens = self._reveal_all_v(tokens)
|
|
messages.append(f"正确：{en}")
|
|
quiz_service.submit_answer(
|
|
db,
|
|
user,
|
|
word.id,
|
|
"memory_coach",
|
|
answer,
|
|
en,
|
|
duration_seconds,
|
|
)
|
|
update = adaptive_review_service.update_word(
|
|
word,
|
|
is_correct=True,
|
|
attempts=attempts,
|
|
hints_used=hints_used,
|
|
duration_seconds=duration_seconds,
|
|
observed_retrievability=before_r,
|
|
)
|
|
db.commit()
|
|
if adaptive_review_service.is_plan_completed(word):
|
|
review_message = "本词已练会，已移出当前练习计划；可在学习设置中重置后继续复习"
|
|
else:
|
|
review_message = update.message
|
|
review_retrievability = update.retrievability
|
|
next_review_at = update.next_review_at
|
|
quiz_recorded = True
|
|
word_complete = True
|
|
next_stage = "done"
|
|
else:
|
|
before_r = retrievability(word)
|
|
quiz_service.record_attempt(
|
|
db,
|
|
user,
|
|
word.id,
|
|
"memory_coach",
|
|
answer,
|
|
en,
|
|
duration_seconds,
|
|
)
|
|
hints_used += 1
|
|
tokens = self.build_tokens(db, word, reveal_extra=hints_used)
|
|
tokens = self._reveal_q_tokens(tokens, 1)
|
|
hidden_left = sum(1 for t in tokens if t.role == "Q" and not t.revealed)
|
|
messages.append("不对，再试一次。")
|
|
if hidden_left == 0:
|
|
tokens = self._reveal_all_v(tokens)
|
|
messages.append(f"答案：{en}")
|
|
update = adaptive_review_service.update_word(
|
|
word,
|
|
is_correct=False,
|
|
attempts=hints_used,
|
|
hints_used=hints_used,
|
|
duration_seconds=duration_seconds,
|
|
observed_retrievability=before_r,
|
|
)
|
|
word.consecutive_correct_count = 0
|
|
word.status = "weak"
|
|
word.last_reviewed_at = datetime.now(timezone.utc).strftime(
|
|
"%Y-%m-%dT%H:%M:%SZ"
|
|
)
|
|
db.commit()
|
|
review_message = update.message
|
|
review_retrievability = update.retrievability
|
|
next_review_at = update.next_review_at
|
|
quiz_recorded = True
|
|
word_complete = True
|
|
next_stage = "done"
|
|
else:
|
|
expect_input = True
|
|
next_stage = "derive"
|
|
|
|
elif stage == "done":
|
|
word_complete = True
|
|
next_stage = "done"
|
|
|
|
else:
|
|
next_stage = "intro"
|
|
expect_input = True
|
|
|
|
return CoachTurnResponse(
|
|
assistant_messages=messages,
|
|
tokens=tokens,
|
|
stage=next_stage,
|
|
expect_input=expect_input,
|
|
prompt_zh=zh,
|
|
prompt_phonetic=phonetic,
|
|
input_hint=input_hint,
|
|
is_correct=is_correct,
|
|
quiz_recorded=quiz_recorded,
|
|
word_complete=word_complete,
|
|
hints_used=hints_used,
|
|
target_en=en if word_complete else None,
|
|
target_zh=zh if word_complete else None,
|
|
review_message=review_message,
|
|
retrievability=review_retrievability,
|
|
next_review_at=next_review_at,
|
|
)
|
|
|
|
|
|
memory_coach_service = MemoryCoachService()
|