# Copyright: Ankitects Pty Ltd and contributors
# License: GNU AGPL, version 3 or later; http://www.gnu.org/licenses/agpl.html
from __future__ import annotations
import json
import random
import re
from dataclasses import dataclass
from enum import Enum, auto
from typing import Any, Callable, Literal, Match, Sequence, cast
import aqt
import aqt.browser
import aqt.operations
from anki import hooks
from anki.cards import Card, CardId
from anki.collection import Config, OpChanges, OpChangesWithCount
from anki.scheduler.base import ScheduleCardsAsNew
from anki.scheduler.v3 import CardAnswer, QueuedCards
from anki.scheduler.v3 import Scheduler as V3Scheduler
from anki.scheduler.v3 import SchedulingStates
from anki.tags import MARKED_TAG
from anki.types import assert_exhaustive
from aqt import AnkiQt, gui_hooks
from aqt.browser.card_info import PreviousReviewerCardInfo, ReviewerCardInfo
from aqt.deckoptions import confirm_deck_then_display_options
from aqt.operations.card import set_card_flag
from aqt.operations.note import remove_notes
from aqt.operations.scheduling import (
answer_card,
bury_cards,
bury_notes,
forget_cards,
set_due_date_dialog,
suspend_cards,
suspend_note,
)
from aqt.operations.tag import add_tags_to_notes, remove_tags_from_notes
from aqt.profiles import VideoDriver
from aqt.qt import *
from aqt.sound import av_player, play_clicked_audio, record_audio
from aqt.theme import theme_manager
from aqt.toolbar import BottomBar
from aqt.utils import (
askUserDialog,
downArrow,
qtMenuShortcutWorkaround,
show_warning,
tooltip,
tr,
)
class RefreshNeeded(Enum):
NOTE_TEXT = auto()
QUEUES = auto()
FLAG = auto()
class ReviewerBottomBar:
def __init__(self, reviewer: Reviewer) -> None:
self.reviewer = reviewer
def replay_audio(card: Card, question_side: bool) -> None:
if question_side:
av_player.play_tags(card.question_av_tags())
else:
tags = card.answer_av_tags()
if card.replay_question_audio_on_answer_side():
tags = card.question_av_tags() + tags
av_player.play_tags(tags)
@dataclass
class V3CardInfo:
"""Stores the top of the card queue for the v3 scheduler.
This includes current and potential next states of the displayed card,
which may be mutated by a user's custom scheduling.
"""
queued_cards: QueuedCards
states: SchedulingStates
@staticmethod
def from_queue(queued_cards: QueuedCards) -> V3CardInfo:
top_card = queued_cards.cards[0]
states = top_card.states
states.current.custom_data = top_card.card.custom_data
return V3CardInfo(
queued_cards=queued_cards,
states=states,
)
def top_card(self) -> QueuedCards.QueuedCard:
return self.queued_cards.cards[0]
def counts(self) -> tuple[int, list[int]]:
"Returns (idx, counts)."
counts = [
self.queued_cards.new_count,
self.queued_cards.learning_count,
self.queued_cards.review_count,
]
card = self.top_card()
if card.queue == QueuedCards.NEW:
idx = 0
elif card.queue == QueuedCards.LEARNING:
idx = 1
else:
idx = 2
return idx, counts
@staticmethod
def rating_from_ease(ease: int) -> CardAnswer.Rating.V:
if ease == 1:
return CardAnswer.AGAIN
elif ease == 2:
return CardAnswer.HARD
elif ease == 3:
return CardAnswer.GOOD
else:
return CardAnswer.EASY
class Reviewer:
def __init__(self, mw: AnkiQt) -> None:
self.mw = mw
self.web = mw.web
self.card: Card | None = None
self.cardQueue: list[Card] = []
self.previous_card: Card | None = None
self.hadCardQueue = False
self._answeredIds: list[CardId] = []
self._recordedAudio: str | None = None
self.typeCorrect: str = None # web init happens before this is set
self.state: Literal["question", "answer", "transition", None] = None
self._refresh_needed: RefreshNeeded | None = None
self._v3: V3CardInfo | None = None
self._state_mutation_key = str(random.randint(0, 2**64 - 1))
self.bottom = BottomBar(mw, mw.bottomWeb)
self._card_info = ReviewerCardInfo(self.mw)
self._previous_card_info = PreviousReviewerCardInfo(self.mw)
hooks.card_did_leech.append(self.onLeech)
def show(self) -> None:
if self.mw.col.sched_ver() == 1:
self.mw.moveToState("deckBrowser")
show_warning(tr.scheduling_update_required())
return
self.mw.setStateShortcuts(self._shortcutKeys()) # type: ignore
self.web.set_bridge_command(self._linkHandler, self)
self.bottom.web.set_bridge_command(self._linkHandler, ReviewerBottomBar(self))
self._state_mutation_js = self.mw.col.get_config("cardStateCustomizer")
self._reps: int = None
self._refresh_needed = RefreshNeeded.QUEUES
self.refresh_if_needed()
# this is only used by add-ons
def lastCard(self) -> Card | None:
if self._answeredIds:
if not self.card or self._answeredIds[-1] != self.card.id:
try:
return self.mw.col.get_card(self._answeredIds[-1])
except TypeError:
# id was deleted
return None
return None
def cleanup(self) -> None:
gui_hooks.reviewer_will_end()
self.card = None
def refresh_if_needed(self) -> None:
if self._refresh_needed is RefreshNeeded.QUEUES:
self.mw.col.reset()
self.nextCard()
self.mw.fade_in_webview()
self._refresh_needed = None
elif self._refresh_needed is RefreshNeeded.NOTE_TEXT:
self._redraw_current_card()
self.mw.fade_in_webview()
self._refresh_needed = None
elif self._refresh_needed is RefreshNeeded.FLAG:
self.card.load()
self._update_flag_icon()
# for when modified in browser
self.mw.fade_in_webview()
self._refresh_needed = None
elif self._refresh_needed:
assert_exhaustive(self._refresh_needed)
def op_executed(
self, changes: OpChanges, handler: object | None, focused: bool
) -> bool:
if handler is not self:
if changes.study_queues:
self._refresh_needed = RefreshNeeded.QUEUES
elif changes.note_text:
self._refresh_needed = RefreshNeeded.NOTE_TEXT
elif changes.card:
self._refresh_needed = RefreshNeeded.FLAG
if focused and self._refresh_needed:
self.refresh_if_needed()
return bool(self._refresh_needed)
def _redraw_current_card(self) -> None:
self.card.load()
if self.state == "answer":
self._showAnswer()
else:
self._showQuestion()
# Fetching a card
##########################################################################
def nextCard(self) -> None:
self.previous_card = self.card
self.card = None
self._v3 = None
if self.mw.col.sched.version < 3:
self._get_next_v1_v2_card()
else:
self._get_next_v3_card()
self._previous_card_info.set_card(self.previous_card)
self._card_info.set_card(self.card)
if not self.card:
self.mw.moveToState("overview")
return
if self._reps is None:
self._initWeb()
self._showQuestion()
def _get_next_v1_v2_card(self) -> None:
if self.cardQueue:
# undone/edited cards to show
card = self.cardQueue.pop()
card.start_timer()
self.hadCardQueue = True
else:
if self.hadCardQueue:
# the undone/edited cards may be sitting in the regular queue;
# need to reset
self.mw.col.reset()
self.hadCardQueue = False
card = self.mw.col.sched.getCard()
self.card = card
def _get_next_v3_card(self) -> None:
assert isinstance(self.mw.col.sched, V3Scheduler)
output = self.mw.col.sched.get_queued_cards()
if not output.cards:
return
self._v3 = V3CardInfo.from_queue(output)
self.card = Card(self.mw.col, backend_card=self._v3.top_card().card)
self.card.start_timer()
def get_scheduling_states(self) -> SchedulingStates | None:
if v3 := self._v3:
return v3.states
else:
return None
def set_scheduling_states(self, key: str, states: SchedulingStates) -> None:
if key != self._state_mutation_key:
return
if v3 := self._v3:
v3.states = states
def _run_state_mutation_hook(self) -> None:
if self._v3 and (js := self._state_mutation_js):
self.web.eval(
f"anki.mutateNextCardStates('{self._state_mutation_key}', (states, customData) => {{ {js} }})"
)
# Audio
##########################################################################
def replayAudio(self) -> None:
if self.state == "question":
replay_audio(self.card, True)
elif self.state == "answer":
replay_audio(self.card, False)
# Initializing the webview
##########################################################################
def revHtml(self) -> str:
extra = self.mw.col.conf.get("reviewExtra", "")
fade = ""
if self.mw.pm.video_driver() == VideoDriver.Software:
fade = ""
return f"""
★
⚑
{fade}
{extra}
"""
def _initWeb(self) -> None:
self._reps = 0
# main window
self.web.stdHtml(
self.revHtml(),
css=["css/reviewer.css"],
js=[
"js/mathjax.js",
"js/vendor/mathjax/tex-chtml.js",
"js/reviewer.js",
],
context=self,
)
# block default drag & drop behavior while allowing drop events to be received by JS handlers
self.web.allow_drops = True
self.web.eval("_blockDefaultDragDropBehavior();")
# show answer / ease buttons
self.bottom.web.show()
self.bottom.web.stdHtml(
self._bottomHTML(),
css=["css/toolbar-bottom.css", "css/reviewer-bottom.css"],
js=["js/vendor/jquery.min.js", "js/reviewer-bottom.js"],
context=ReviewerBottomBar(self),
)
# Showing the question
##########################################################################
def _mungeQA(self, buf: str) -> str:
return self.typeAnsFilter(self.mw.prepare_card_text_for_display(buf))
def _showQuestion(self) -> None:
self._reps += 1
self.state = "question"
self.typedAnswer: str = None
c = self.card
# grab the question and play audio
q = c.question()
# play audio?
if c.autoplay():
self.web.setPlaybackRequiresGesture(False)
sounds = c.question_av_tags()
gui_hooks.reviewer_will_play_question_sounds(c, sounds)
else:
self.web.setPlaybackRequiresGesture(True)
sounds = []
gui_hooks.reviewer_will_play_question_sounds(c, sounds)
gui_hooks.av_player_will_play_tags(sounds, self.state, self)
av_player.play_tags(sounds)
# render & update bottom
q = self._mungeQA(q)
q = gui_hooks.card_will_show(q, c, "reviewQuestion")
self._run_state_mutation_hook()
bodyclass = theme_manager.body_classes_for_card_ord(c.ord)
a = self.mw.col.media.escape_media_filenames(c.answer())
self.web.eval(
f"_showQuestion({json.dumps(q)}, {json.dumps(a)}, '{bodyclass}');"
)
self._update_flag_icon()
self._update_mark_icon()
self._showAnswerButton()
self.mw.web.setFocus()
# user hook
gui_hooks.reviewer_did_show_question(c)
def autoplay(self, card: Card) -> bool:
print("use card.autoplay() instead of reviewer.autoplay(card)")
return card.autoplay()
def _update_flag_icon(self) -> None:
self.web.eval(f"_drawFlag({self.card.user_flag()});")
def _update_mark_icon(self) -> None:
self.web.eval(f"_drawMark({json.dumps(self.card.note().has_tag(MARKED_TAG))});")
_drawMark = _update_mark_icon
_drawFlag = _update_flag_icon
# Showing the answer
##########################################################################
def _showAnswer(self) -> None:
if self.mw.state != "review":
# showing resetRequired screen; ignore space
return
self.state = "answer"
c = self.card
a = c.answer()
# play audio?
if c.autoplay():
sounds = c.answer_av_tags()
gui_hooks.reviewer_will_play_answer_sounds(c, sounds)
else:
sounds = []
gui_hooks.reviewer_will_play_answer_sounds(c, sounds)
gui_hooks.av_player_will_play_tags(sounds, self.state, self)
av_player.play_tags(sounds)
a = self._mungeQA(a)
a = gui_hooks.card_will_show(a, c, "reviewAnswer")
# render and update bottom
self.web.eval(f"_showAnswer({json.dumps(a)});")
self._showEaseButtons()
self.mw.web.setFocus()
# user hook
gui_hooks.reviewer_did_show_answer(c)
# Answering a card
############################################################
def _answerCard(self, ease: Literal[1, 2, 3, 4]) -> None:
"Reschedule card and show next."
if self.mw.state != "review":
# showing resetRequired screen; ignore key
return
if self.state != "answer":
return
if self.mw.col.sched.answerButtons(self.card) < ease:
return
proceed, ease = gui_hooks.reviewer_will_answer_card(
(True, ease), self, self.card
)
if not proceed:
return
if (v3 := self._v3) and (sched := cast(V3Scheduler, self.mw.col.sched)):
answer = sched.build_answer(
card=self.card,
states=v3.states,
rating=v3.rating_from_ease(ease),
)
def after_answer(changes: OpChanges) -> None:
if gui_hooks.reviewer_did_answer_card.count() > 0:
self.card.load()
self._after_answering(ease)
if sched.state_is_leech(answer.new_state):
self.onLeech()
self.state = "transition"
answer_card(parent=self.mw, answer=answer).success(
after_answer
).run_in_background(initiator=self)
else:
self.mw.col.sched.answerCard(self.card, ease)
self._after_answering(ease)
def _after_answering(self, ease: Literal[1, 2, 3, 4]) -> None:
gui_hooks.reviewer_did_answer_card(self, self.card, ease)
self._answeredIds.append(self.card.id)
self.mw.autosave()
if not self.check_timebox():
self.nextCard()
# Handlers
############################################################
def _shortcutKeys(
self,
) -> Sequence[Union[tuple[str, Callable], tuple[Qt.Key, Callable]]]:
return [
("e", self.mw.onEditCurrent),
(" ", self.onEnterKey),
(Qt.Key.Key_Return, self.onEnterKey),
(Qt.Key.Key_Enter, self.onEnterKey),
("m", self.showContextMenu),
("r", self.replayAudio),
(Qt.Key.Key_F5, self.replayAudio),
*(
(f"Ctrl+{flag.index}", self.set_flag_func(flag.index))
for flag in self.mw.flags.all()
),
("*", self.toggle_mark_on_current_note),
("=", self.bury_current_note),
("-", self.bury_current_card),
("!", self.suspend_current_note),
("@", self.suspend_current_card),
("Ctrl+Alt+N", self.forget_current_card),
("Ctrl+Alt+E", self.on_create_copy),
("Ctrl+Delete", self.delete_current_note),
("Ctrl+Shift+D", self.on_set_due),
("v", self.onReplayRecorded),
("Shift+v", self.onRecordVoice),
("o", self.onOptions),
("i", self.on_card_info),
("Ctrl+Alt+i", self.on_previous_card_info),
("1", lambda: self._answerCard(1)),
("2", lambda: self._answerCard(2)),
("3", lambda: self._answerCard(3)),
("4", lambda: self._answerCard(4)),
("5", self.on_pause_audio),
("6", self.on_seek_backward),
("7", self.on_seek_forward),
]
def on_pause_audio(self) -> None:
av_player.toggle_pause()
seek_secs = 5
def on_seek_backward(self) -> None:
av_player.seek_relative(-self.seek_secs)
def on_seek_forward(self) -> None:
av_player.seek_relative(self.seek_secs)
def onEnterKey(self) -> None:
if self.state == "question":
self._getTypedAnswer()
elif self.state == "answer":
self.bottom.web.evalWithCallback(
"selectedAnswerButton()", self._onAnswerButton
)
def _onAnswerButton(self, val: str) -> None:
# button selected?
if val and val in "1234":
val2: Literal[1, 2, 3, 4] = int(val) # type: ignore
self._answerCard(val2)
else:
self._answerCard(self._defaultEase())
def _linkHandler(self, url: str) -> None:
if url == "ans":
self._getTypedAnswer()
elif url.startswith("ease"):
val: Literal[1, 2, 3, 4] = int(url[4:]) # type: ignore
self._answerCard(val)
elif url == "edit":
self.mw.onEditCurrent()
elif url == "more":
self.showContextMenu()
elif url.startswith("play:"):
play_clicked_audio(url, self.card)
else:
print("unrecognized anki link:", url)
# Type in the answer
##########################################################################
typeAnsPat = r"\[\[type:(.+?)\]\]"
def typeAnsFilter(self, buf: str) -> str:
if self.state == "question":
return self.typeAnsQuestionFilter(buf)
else:
return self.typeAnsAnswerFilter(buf)
def typeAnsQuestionFilter(self, buf: str) -> str:
self.typeCorrect = None
clozeIdx = None
m = re.search(self.typeAnsPat, buf)
if not m:
return buf
fld = m.group(1)
# if it's a cloze, extract data
if fld.startswith("cloze:"):
# get field and cloze position
clozeIdx = self.card.ord + 1
fld = fld.split(":")[1]
# loop through fields for a match
for f in self.card.note_type()["flds"]:
if f["name"] == fld:
self.typeCorrect = self.card.note()[f["name"]]
if clozeIdx:
# narrow to cloze
self.typeCorrect = self._contentForCloze(self.typeCorrect, clozeIdx)
self.typeFont = f["font"]
self.typeSize = f["size"]
break
if not self.typeCorrect:
if self.typeCorrect is None:
if clozeIdx:
warn = tr.studying_please_run_toolsempty_cards()
else:
warn = tr.studying_type_answer_unknown_field(val=fld)
return re.sub(self.typeAnsPat, warn, buf)
else:
# empty field, remove type answer pattern
return re.sub(self.typeAnsPat, "", buf)
return re.sub(
self.typeAnsPat,
f"""
""",
buf,
)
def typeAnsAnswerFilter(self, buf: str) -> str:
if not self.typeCorrect:
return re.sub(self.typeAnsPat, "", buf)
origSize = len(buf)
buf = buf.replace("", "")
hadHR = len(buf) != origSize
# munge correct value
expected = self.typeCorrect
provided = self.typedAnswer
# compare with typed answer
output = self.mw.col.compare_answer(expected, provided)
# and update the type answer area
def repl(match: Match) -> str:
# can't pass a string in directly, and can't use re.escape as it
# escapes too much
s = """
{}
""".format(
self.typeFont,
self.typeSize,
output,
)
if hadHR:
# a hack to ensure the q/a separator falls before the answer
# comparison when user is using {{FrontSide}}
s = f"{s}"
return s
return re.sub(self.typeAnsPat, repl, buf)
def _contentForCloze(self, txt: str, idx: int) -> str:
matches = re.findall(r"\{\{c%s::(.+?)\}\}" % idx, txt, re.DOTALL)
if not matches:
return None
def noHint(txt: str) -> str:
if "::" in txt:
return txt.split("::")[0]
return txt
matches = [noHint(txt) for txt in matches]
uniqMatches = set(matches)
if len(uniqMatches) == 1:
txt = matches[0]
else:
txt = ", ".join(matches)
return txt
def _getTypedAnswer(self) -> None:
self.web.evalWithCallback("getTypedAnswer();", self._onTypedAnswer)
def _onTypedAnswer(self, val: None) -> None:
self.typedAnswer = val or ""
self._showAnswer()
# Bottom bar
##########################################################################
def _bottomHTML(self) -> str:
return """
""" % dict(
rem=self._remaining(),
edit=tr.studying_edit(),
editkey=tr.actions_shortcut_key(val="E"),
more=tr.studying_more(),
downArrow=downArrow(),
time=self.card.time_taken() // 1000,
)
def _showAnswerButton(self) -> None:
middle = """
""".format(
tr.actions_shortcut_key(val=tr.studying_space()),
tr.studying_show_answer(),
self._remaining(),
)
# wrap it in a table so it has the same top margin as the ease buttons
middle = (
"