0d4042169a
It was still browsing the flat category list and the free-text keyword tags, neither of which is how questions are arranged any more. A discipline holding six hundred questions three levels down showed as "(0)" because only what was filed directly on the node was counted. /topics now walks the tree — discipline, sub-topic, condition — with each node counted over its whole subtree, and any node can be quizzed whole. /objective scopes every quiz to one exam, listing only objectives that actually have questions. The keyword browse is gone with the tags. Chat is where you answer, not where you read the teaching: explanations are abridged to about a screenful, cut at a sentence, with markdown and cross-reference markers stripped and a link to the site when there is more. Per-option reasoning and figures are not sent — the bot never did, and now says where they are. Still read-only, and now says so: no account, no history, and a quiz that lives only until it ends or /stop. `is_shared` remains the only gate between an anonymous stranger and the bank, applied in one place. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TqXevQJhxFrM7jJg82cgZN
739 lines
29 KiB
Python
739 lines
29 KiB
Python
import html
|
||
import json
|
||
import logging
|
||
import os
|
||
import random
|
||
import re
|
||
from dataclasses import dataclass, field
|
||
from typing import Any
|
||
|
||
import psycopg
|
||
from psycopg.rows import dict_row
|
||
from telegram import BotCommand, InlineKeyboardButton, InlineKeyboardMarkup, Update
|
||
from telegram.constants import ChatAction, ParseMode
|
||
from telegram.ext import Application, CallbackQueryHandler, CommandHandler, ContextTypes, MessageHandler, filters
|
||
|
||
|
||
logging.basicConfig(level=os.getenv("LOG_LEVEL", "INFO"))
|
||
logger = logging.getLogger("quiz-telegram-bot")
|
||
logging.getLogger("httpx").setLevel(logging.WARNING)
|
||
logging.getLogger("httpcore").setLevel(logging.WARNING)
|
||
|
||
TELEGRAM_BOT_TOKEN = os.environ["TELEGRAM_BOT_TOKEN"]
|
||
DATABASE_URL = os.environ["DATABASE_URL"]
|
||
DEFAULT_QUIZ_SIZE = int(os.getenv("DEFAULT_QUIZ_SIZE", "20"))
|
||
MAX_QUIZ_SIZE = int(os.getenv("MAX_QUIZ_SIZE", "50"))
|
||
PUBLIC_APP_URL = os.getenv("PUBLIC_APP_URL", "https://pedshub.com").rstrip("/")
|
||
TELEGRAM_MESSAGE_LIMIT = 4096
|
||
QUESTION_TEXT_LIMIT = 3200
|
||
OPTION_TEXT_LIMIT = 3200
|
||
TELEGRAM_MESSAGE_MARGIN = 200
|
||
# Telegram is where you answer a question on the bus, not where you read the
|
||
# teaching. The explanation is abridged to roughly a screenful and the site
|
||
# link carries the rest; per-option explanations are not sent at all.
|
||
EXPLANATION_LIMIT = int(os.getenv("EXPLANATION_LIMIT", "700"))
|
||
|
||
LETTERS = "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
|
||
NUMBER_RE = re.compile(r"^\s*(\d{1,3})\s*$")
|
||
|
||
|
||
@dataclass
|
||
class QuizState:
|
||
questions: list[dict[str, Any]]
|
||
mode: str = "study"
|
||
index: int = 0
|
||
score: int = 0
|
||
answers: list[tuple[int, str, str, bool]] = field(default_factory=list)
|
||
|
||
|
||
active_quizzes: dict[int, QuizState] = {}
|
||
|
||
# What each chat is revising for, by exam id. In memory and per chat, like the
|
||
# quiz itself: the bot has no accounts and writes nothing down.
|
||
chat_exam: dict[int, int] = {}
|
||
|
||
|
||
def db_query(sql: str, params: tuple[Any, ...] = ()) -> list[dict[str, Any]]:
|
||
with psycopg.connect(DATABASE_URL, row_factory=dict_row) as conn:
|
||
with conn.cursor() as cur:
|
||
cur.execute(sql, params)
|
||
return list(cur.fetchall())
|
||
|
||
|
||
def normalize_options(value: Any) -> list[str]:
|
||
if value is None:
|
||
return []
|
||
if isinstance(value, list):
|
||
return [str(item) for item in value]
|
||
if isinstance(value, str):
|
||
try:
|
||
parsed = json.loads(value)
|
||
except json.JSONDecodeError:
|
||
return []
|
||
if isinstance(parsed, list):
|
||
return [str(item) for item in parsed]
|
||
return []
|
||
|
||
|
||
def answer_index(question: dict[str, Any], options: list[str]) -> int | None:
|
||
raw = str(question.get("correct_answer") or "").strip()
|
||
if not raw:
|
||
return None
|
||
upper = raw.upper()
|
||
if len(upper) == 1 and upper in LETTERS:
|
||
idx = LETTERS.index(upper)
|
||
return idx if idx < len(options) else None
|
||
if raw.isdigit():
|
||
idx = int(raw) - 1
|
||
return idx if 0 <= idx < len(options) else None
|
||
for idx, option in enumerate(options):
|
||
if option.strip().lower() == raw.lower():
|
||
return idx
|
||
return None
|
||
|
||
|
||
#: Questions the bot may serve at all. `is_shared` is the only gate between an
|
||
#: anonymous stranger on Telegram and the bank, so it is applied in one place.
|
||
SERVEABLE = """
|
||
q.question_type = 'mcq'
|
||
and q.options is not null
|
||
and q.is_shared = 1
|
||
"""
|
||
|
||
#: Every category at or below one node. The taxonomy is a tree now — discipline
|
||
#: above subdiscipline above condition — so "Cardiology" has to mean everything
|
||
#: under Cardiology, not only what was filed directly on it.
|
||
SUBTREE = """
|
||
with recursive subtree(id) as (
|
||
select id from question_categories where id = %s
|
||
union all
|
||
select c.id from question_categories c join subtree s on c.parent_id = s.id
|
||
)
|
||
"""
|
||
|
||
#: A question belongs to a category by its primary column or by a link row.
|
||
IN_SUBTREE = """
|
||
(q.question_category_id in (select id from subtree)
|
||
or exists (select 1 from question_category_links l
|
||
where l.question_id = q.id and l.category_id in (select id from subtree)))
|
||
"""
|
||
|
||
|
||
def exam_clause(exam_id: int | None) -> tuple[str, tuple[Any, ...]]:
|
||
"""Narrow to one study objective, or to nothing if none is chosen."""
|
||
if not exam_id:
|
||
return "", ()
|
||
return ("and exists (select 1 from question_exam_links qel "
|
||
"where qel.question_id = q.id and qel.exam_id = %s)", (exam_id,))
|
||
|
||
|
||
def question_base_sql(where: str = "", prefix: str = "") -> str:
|
||
return f"""
|
||
{prefix}
|
||
select q.id, q.question_text, q.options, q.correct_answer, q.explanation, qc.name as category
|
||
from questions q
|
||
left join question_categories qc on qc.id = q.question_category_id
|
||
where {SERVEABLE}
|
||
{where}
|
||
order by random()
|
||
limit %s
|
||
"""
|
||
|
||
|
||
def valid_questions(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
clean = []
|
||
for row in rows:
|
||
options = normalize_options(row.get("options"))
|
||
if len(options) < 2:
|
||
continue
|
||
idx = answer_index(row, options)
|
||
if idx is None:
|
||
continue
|
||
row["options"] = options
|
||
row["correct_index"] = idx
|
||
clean.append(row)
|
||
return clean
|
||
|
||
|
||
def random_questions(limit: int, exam_id: int | None = None) -> list[dict[str, Any]]:
|
||
clause, params = exam_clause(exam_id)
|
||
rows = db_query(question_base_sql(clause), (*params, limit * 2))
|
||
return valid_questions(rows)[:limit]
|
||
|
||
|
||
def search_questions(term: str, limit: int, exam_id: int | None = None) -> list[dict[str, Any]]:
|
||
pattern = f"%{term}%"
|
||
clause, params = exam_clause(exam_id)
|
||
rows = db_query(
|
||
question_base_sql(f"""
|
||
{clause}
|
||
and (
|
||
q.question_text ilike %s
|
||
or q.explanation ilike %s
|
||
or exists (
|
||
select 1 from question_categories c
|
||
where c.id = q.question_category_id and c.name ilike %s
|
||
)
|
||
)
|
||
"""),
|
||
(*params, pattern, pattern, pattern, limit * 2),
|
||
)
|
||
return valid_questions(rows)[:limit]
|
||
|
||
|
||
def category_questions(category_id: int, limit: int, exam_id: int | None = None) -> list[dict[str, Any]]:
|
||
clause, params = exam_clause(exam_id)
|
||
rows = db_query(
|
||
question_base_sql(f"and {IN_SUBTREE} {clause}", prefix=SUBTREE),
|
||
(category_id, *params, limit * 2),
|
||
)
|
||
return valid_questions(rows)[:limit]
|
||
|
||
|
||
def list_exams() -> list[dict[str, Any]]:
|
||
"""The objectives, with how many serveable questions each covers."""
|
||
return db_query(
|
||
f"""
|
||
select e.id, e.name, e.family, count(distinct q.id)::int as count
|
||
from exams e
|
||
join question_exam_links qel on qel.exam_id = e.id
|
||
join questions q on q.id = qel.question_id and {SERVEABLE}
|
||
where e.is_active = 1
|
||
group by e.id, e.name, e.family, e.sort_order
|
||
having count(distinct q.id) > 0
|
||
order by e.sort_order, e.name
|
||
"""
|
||
)
|
||
|
||
|
||
def category_children(parent_id: int | None, exam_id: int | None = None) -> list[dict[str, Any]]:
|
||
"""The children of one node, each counted over its own whole subtree.
|
||
|
||
Counting only what is filed directly on a node would show "Cardiology (0)"
|
||
for a discipline holding six hundred questions three levels down.
|
||
"""
|
||
clause, params = exam_clause(exam_id)
|
||
parent_test = "c.parent_id is null" if parent_id is None else "c.parent_id = %s"
|
||
parent_params: tuple[Any, ...] = () if parent_id is None else (parent_id,)
|
||
return db_query(
|
||
f"""
|
||
with recursive tree(root, id) as (
|
||
select c.id, c.id from question_categories c where {parent_test}
|
||
union all
|
||
select t.root, c.id
|
||
from question_categories c join tree t on c.parent_id = t.id
|
||
),
|
||
hits(root, question_id) as (
|
||
select t.root, q.id
|
||
from tree t join questions q on q.question_category_id = t.id
|
||
where {SERVEABLE} {clause}
|
||
union
|
||
select t.root, q.id
|
||
from tree t
|
||
join question_category_links l on l.category_id = t.id
|
||
join questions q on q.id = l.question_id
|
||
where {SERVEABLE} {clause}
|
||
)
|
||
select r.id, r.name,
|
||
count(distinct h.question_id)::int as count,
|
||
exists (select 1 from question_categories k where k.parent_id = r.id) as has_children
|
||
from question_categories r
|
||
left join hits h on h.root = r.id
|
||
where r.id in (select distinct root from tree)
|
||
group by r.id, r.name
|
||
having count(distinct h.question_id) > 0
|
||
order by r.name
|
||
""",
|
||
(*parent_params, *params, *params),
|
||
)
|
||
|
||
|
||
def category_name(category_id: int) -> str:
|
||
rows = db_query("select name from question_categories where id = %s", (category_id,))
|
||
return rows[0]["name"] if rows else "Category"
|
||
|
||
|
||
def clamp_count(value: int | None) -> int:
|
||
if value is None:
|
||
return DEFAULT_QUIZ_SIZE
|
||
return max(1, min(MAX_QUIZ_SIZE, value))
|
||
|
||
|
||
def parse_count(args: list[str]) -> int:
|
||
for arg in args:
|
||
if arg.isdigit():
|
||
return clamp_count(int(arg))
|
||
return DEFAULT_QUIZ_SIZE
|
||
|
||
|
||
def parse_mode(args: list[str]) -> str:
|
||
lowered = {arg.lower() for arg in args}
|
||
return "exam" if "exam" in lowered else "study"
|
||
|
||
|
||
async def send_text(update: Update, text: str, reply_markup: InlineKeyboardMarkup | None = None) -> None:
|
||
if update.message:
|
||
await update.message.reply_text(
|
||
text,
|
||
parse_mode=ParseMode.HTML,
|
||
disable_web_page_preview=True,
|
||
reply_markup=reply_markup,
|
||
)
|
||
|
||
|
||
async def respond(update: Update, context: ContextTypes.DEFAULT_TYPE | None, text: str, reply_markup: InlineKeyboardMarkup | None = None) -> None:
|
||
if update.message:
|
||
await send_text(update, text, reply_markup)
|
||
elif update.callback_query and update.callback_query.message:
|
||
await update.callback_query.edit_message_text(text, parse_mode=ParseMode.HTML, reply_markup=reply_markup, disable_web_page_preview=True)
|
||
elif context and update.effective_chat:
|
||
await context.bot.send_message(update.effective_chat.id, text, parse_mode=ParseMode.HTML, reply_markup=reply_markup, disable_web_page_preview=True)
|
||
|
||
|
||
def main_menu(chat_id: int | None = None) -> InlineKeyboardMarkup:
|
||
exam_id = chat_exam.get(chat_id) if chat_id else None
|
||
objective = "Change objective" if exam_id else "Choose an objective"
|
||
return InlineKeyboardMarkup([
|
||
[InlineKeyboardButton("Random 20 study", callback_data="quiz:random:20:study")],
|
||
[InlineKeyboardButton("Random 20 exam", callback_data="quiz:random:20:exam")],
|
||
[InlineKeyboardButton("Browse topics", callback_data="browse:0")],
|
||
[InlineKeyboardButton(objective, callback_data="exams")],
|
||
])
|
||
|
||
|
||
def count_menu(kind: str, item_id: int, back_to: int | None = None) -> InlineKeyboardMarkup:
|
||
buttons = []
|
||
for count in (5, 10, 20, 30, 50):
|
||
buttons.append([
|
||
InlineKeyboardButton(f"{count} study", callback_data=f"start:{kind}:{item_id}:{count}:study"),
|
||
InlineKeyboardButton(f"{count} exam", callback_data=f"start:{kind}:{item_id}:{count}:exam"),
|
||
])
|
||
if back_to is not None:
|
||
buttons.append([InlineKeyboardButton("⬅ Back", callback_data=f"browse:{back_to}")])
|
||
return InlineKeyboardMarkup(buttons)
|
||
|
||
|
||
def truncate_text(value: str, limit: int) -> str:
|
||
value = (value or "").strip()
|
||
if len(value) <= limit:
|
||
return value
|
||
return value[:limit].rsplit(" ", 1)[0].rstrip() + "..."
|
||
|
||
|
||
def strip_markup(value: str) -> str:
|
||
"""Explanations are markdown on the site; Telegram gets the words."""
|
||
value = re.sub(r"!?\[\[?\d*\|?([^\]]*)\]\]?(\([^)]*\))?", r"\1", value) # links, cross-refs
|
||
value = re.sub(r"[*_`>#]+", "", value)
|
||
value = re.sub(r"\n{3,}", "\n\n", value)
|
||
return value.strip()
|
||
|
||
|
||
def abridge(explanation: str | None) -> str:
|
||
"""About a screenful, cut at a sentence rather than mid-word."""
|
||
text = strip_markup(explanation or "")
|
||
if not text:
|
||
return "No explanation available."
|
||
if len(text) <= EXPLANATION_LIMIT:
|
||
return text
|
||
window = text[:EXPLANATION_LIMIT]
|
||
stop = max(window.rfind(". "), window.rfind("! "), window.rfind("? "))
|
||
if stop > EXPLANATION_LIMIT // 2:
|
||
return window[:stop + 1]
|
||
return window.rsplit(" ", 1)[0].rstrip() + "…"
|
||
|
||
|
||
def was_abridged(explanation: str | None) -> bool:
|
||
return len(strip_markup(explanation or "")) > EXPLANATION_LIMIT
|
||
|
||
|
||
def question_message_text(
|
||
state: QuizState,
|
||
question: dict[str, Any],
|
||
chosen_idx: int | None = None,
|
||
correct_idx: int | None = None,
|
||
) -> str:
|
||
def build(question_limit: int, option_limit: int) -> str:
|
||
lines = [
|
||
f"<b>Question {state.index + 1}/{len(state.questions)}</b>",
|
||
html.escape(truncate_text(question["question_text"], question_limit)),
|
||
"",
|
||
]
|
||
for idx, option in enumerate(question["options"][:8]):
|
||
marker = ""
|
||
if correct_idx is not None and idx == correct_idx:
|
||
marker = " ✓ correct"
|
||
elif chosen_idx is not None and idx == chosen_idx:
|
||
marker = " ✗ your answer"
|
||
lines.append(f"{LETTERS[idx]}. {html.escape(truncate_text(option, option_limit))}{marker}")
|
||
lines.extend([
|
||
"",
|
||
f"Category: {html.escape(str(question.get('category') or 'Uncategorized'))}",
|
||
])
|
||
return "\n".join(lines)
|
||
|
||
for question_limit, option_limit in (
|
||
(QUESTION_TEXT_LIMIT, OPTION_TEXT_LIMIT),
|
||
(QUESTION_TEXT_LIMIT, 700),
|
||
(1600, 420),
|
||
):
|
||
text = build(question_limit, option_limit)
|
||
if len(text) <= TELEGRAM_MESSAGE_LIMIT - TELEGRAM_MESSAGE_MARGIN:
|
||
return text
|
||
return build(1200, 280)
|
||
|
||
|
||
def answer_feedback_header(chosen_idx: int, correct_idx: int, ok: bool) -> str:
|
||
chosen = LETTERS[chosen_idx] if chosen_idx < len(LETTERS) else "?"
|
||
correct = LETTERS[correct_idx] if correct_idx < len(LETTERS) else "?"
|
||
return "Correct." if ok else f"Incorrect. You chose {chosen}; correct answer: {correct}."
|
||
|
||
|
||
def split_plain_text(text: str, limit: int) -> list[str]:
|
||
if len(text) <= limit:
|
||
return [text]
|
||
chunks = []
|
||
remaining = text
|
||
while remaining:
|
||
if len(remaining) <= limit:
|
||
chunks.append(remaining)
|
||
break
|
||
split_at = remaining.rfind("\n\n", 0, limit)
|
||
if split_at < limit // 2:
|
||
split_at = remaining.rfind("\n", 0, limit)
|
||
if split_at < limit // 2:
|
||
split_at = remaining.rfind(" ", 0, limit)
|
||
if split_at < limit // 2:
|
||
split_at = limit
|
||
chunks.append(remaining[:split_at].rstrip())
|
||
remaining = remaining[split_at:].lstrip()
|
||
return chunks
|
||
|
||
|
||
async def send_answer_feedback(
|
||
context: ContextTypes.DEFAULT_TYPE,
|
||
chat_id: int,
|
||
question: dict[str, Any],
|
||
chosen_idx: int,
|
||
correct_idx: int,
|
||
ok: bool,
|
||
) -> None:
|
||
header = answer_feedback_header(chosen_idx, correct_idx, ok)
|
||
body = abridge(question.get("explanation"))
|
||
lines = [header, "", f"<b>Why:</b> {html.escape(body)}"]
|
||
if was_abridged(question.get("explanation")):
|
||
# Per-option reasoning and the figures stay on the site. Chat is where
|
||
# you answer; it is not where you read the teaching.
|
||
lines.append(f'\n<a href="{PUBLIC_APP_URL}">Full explanation on PedsHub</a>')
|
||
await context.bot.send_message(chat_id, "\n".join(lines),
|
||
parse_mode=ParseMode.HTML, disable_web_page_preview=True)
|
||
|
||
|
||
def answered_question_text(state: QuizState, question: dict[str, Any], chosen_idx: int, correct_idx: int) -> str:
|
||
return question_message_text(state, question, chosen_idx, correct_idx)
|
||
|
||
|
||
async def help_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||
chat_id = update.effective_chat.id if update.effective_chat else 0
|
||
exam_id = chat_exam.get(chat_id)
|
||
current = next((e["name"] for e in list_exams() if e["id"] == exam_id), None) if exam_id else None
|
||
await send_text(update, "\n".join([
|
||
"<b>PedsHub bot</b>",
|
||
"Send <code>20</code> for a random 20-question study quiz.",
|
||
"/random 20 exam — random questions in exam mode.",
|
||
"/topics — browse the subject tree and quiz any part of it.",
|
||
"/objective — pick the exam you are revising for.",
|
||
"/search sepsis 20 — quiz by text search.",
|
||
"/stop — end the current quiz.",
|
||
"",
|
||
f"Objective: <b>{html.escape(current)}</b>" if current else "Objective: everything.",
|
||
"",
|
||
"Study mode shows each answer immediately; exam mode shows them at the end.",
|
||
"Explanations here are abridged — the full teaching, the figures and the",
|
||
f"per-option reasoning are on the site: {PUBLIC_APP_URL}",
|
||
"",
|
||
"<i>Nothing you do here is recorded. The bot keeps no account and writes",
|
||
"nothing to your history — a quiz lives only until you finish or /stop.</i>",
|
||
]), main_menu(chat_id))
|
||
|
||
|
||
async def random_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||
chat_id = update.effective_chat.id if update.effective_chat else 0
|
||
await start_quiz(update, random_questions(parse_count(context.args), chat_exam.get(chat_id)),
|
||
parse_mode(context.args))
|
||
|
||
|
||
async def search_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||
if not context.args:
|
||
await send_text(update, "Usage: /search <topic> [number] [study|exam]")
|
||
return
|
||
chat_id = update.effective_chat.id if update.effective_chat else 0
|
||
count = parse_count(context.args)
|
||
mode = parse_mode(context.args)
|
||
term = " ".join(arg for arg in context.args if not arg.isdigit() and arg.lower() not in {"study", "exam"}).strip()
|
||
if not term:
|
||
await send_text(update, "Usage: /search <topic> [number] [study|exam]")
|
||
return
|
||
if update.message:
|
||
await update.message.chat.send_action(ChatAction.TYPING)
|
||
await start_quiz(update, search_questions(term, count, chat_exam.get(chat_id)), mode, label=f"Search: {term}")
|
||
|
||
|
||
async def topics_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||
await show_browse(update, 0)
|
||
|
||
|
||
async def objective_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||
await show_exams(update)
|
||
|
||
|
||
async def stop_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||
if update.effective_chat:
|
||
active_quizzes.pop(update.effective_chat.id, None)
|
||
await send_text(update, "Quiz stopped. Nothing was recorded.",
|
||
main_menu(update.effective_chat.id if update.effective_chat else 0))
|
||
|
||
|
||
PER_PAGE = 10
|
||
|
||
|
||
async def show_browse(update: Update, parent_id: int, page: int = 0) -> None:
|
||
"""One level of the taxonomy. 0 means the top."""
|
||
chat_id = update.effective_chat.id if update.effective_chat else 0
|
||
exam_id = chat_exam.get(chat_id)
|
||
node = None if parent_id == 0 else parent_id
|
||
rows = category_children(node, exam_id)
|
||
|
||
if not rows:
|
||
await respond(update, None, "Nothing here for the objective you have chosen.", main_menu(chat_id))
|
||
return
|
||
|
||
chunk = rows[page * PER_PAGE:(page + 1) * PER_PAGE]
|
||
buttons = []
|
||
# A node that goes deeper opens; a leaf goes straight to the size menu.
|
||
for row in chunk:
|
||
action = "browse" if row["has_children"] else "pick:category"
|
||
target = f"{action}:{row['id']}" if row["has_children"] else f"pick:category:{row['id']}"
|
||
arrow = " ›" if row["has_children"] else ""
|
||
buttons.append([InlineKeyboardButton(f"{row['name']} ({row['count']}){arrow}",
|
||
callback_data=target)])
|
||
|
||
nav = []
|
||
if page > 0:
|
||
nav.append(InlineKeyboardButton("Prev", callback_data=f"browse:{parent_id}:{page - 1}"))
|
||
if (page + 1) * PER_PAGE < len(rows):
|
||
nav.append(InlineKeyboardButton("Next", callback_data=f"browse:{parent_id}:{page + 1}"))
|
||
if nav:
|
||
buttons.append(nav)
|
||
|
||
if node is None:
|
||
title = "Choose a topic."
|
||
else:
|
||
# Everything under this node, not only what is filed directly on it.
|
||
title = f"<b>{html.escape(category_name(node))}</b> — choose a sub-topic, or quiz the whole of it."
|
||
buttons.append([InlineKeyboardButton("▶ All of this topic", callback_data=f"pick:category:{node}")])
|
||
buttons.append([InlineKeyboardButton("⬅ Top", callback_data="browse:0")])
|
||
|
||
markup = InlineKeyboardMarkup(buttons)
|
||
if update.callback_query:
|
||
await update.callback_query.edit_message_text(title, parse_mode=ParseMode.HTML, reply_markup=markup)
|
||
else:
|
||
await send_text(update, title, markup)
|
||
|
||
|
||
async def show_exams(update: Update) -> None:
|
||
"""Which exam the questions should come from. Optional — none means all."""
|
||
chat_id = update.effective_chat.id if update.effective_chat else 0
|
||
rows = list_exams()
|
||
if not rows:
|
||
await respond(update, None, "No study objectives are set up yet.", main_menu(chat_id))
|
||
return
|
||
current = chat_exam.get(chat_id)
|
||
buttons = []
|
||
for row in rows:
|
||
mark = "✓ " if row["id"] == current else ""
|
||
family = f"{row['family']} — " if row.get("family") else ""
|
||
buttons.append([InlineKeyboardButton(f"{mark}{family}{row['name']} ({row['count']})",
|
||
callback_data=f"exam:{row['id']}")])
|
||
buttons.append([InlineKeyboardButton(("✓ " if not current else "") + "Everything",
|
||
callback_data="exam:0")])
|
||
text = "Which exam are you revising for? It scopes every quiz the bot gives you."
|
||
markup = InlineKeyboardMarkup(buttons)
|
||
if update.callback_query:
|
||
await update.callback_query.edit_message_text(text, reply_markup=markup)
|
||
else:
|
||
await send_text(update, text, markup)
|
||
|
||
|
||
async def start_quiz(
|
||
update: Update,
|
||
questions: list[dict[str, Any]],
|
||
mode: str,
|
||
label: str = "Random",
|
||
context: ContextTypes.DEFAULT_TYPE | None = None,
|
||
) -> None:
|
||
if not update.effective_chat:
|
||
return
|
||
if not questions:
|
||
await respond(update, context,
|
||
"No questions there for the objective you have chosen. Try /topics, "
|
||
"a broader /search, or /objective to widen it.",
|
||
main_menu(update.effective_chat.id if update.effective_chat else 0))
|
||
return
|
||
random.shuffle(questions)
|
||
active_quizzes[update.effective_chat.id] = QuizState(questions=questions, mode=mode)
|
||
start_text = f"{html.escape(label)} quiz started: {len(questions)} questions, {mode} mode."
|
||
if update.message:
|
||
await send_text(update, start_text)
|
||
await send_current_question(update.effective_chat.id, update, None)
|
||
elif context:
|
||
await context.bot.send_message(update.effective_chat.id, start_text, parse_mode=ParseMode.HTML)
|
||
await send_current_question(update.effective_chat.id, None, context)
|
||
|
||
|
||
async def send_current_question(chat_id: int, update: Update | None, context: ContextTypes.DEFAULT_TYPE | None) -> None:
|
||
state = active_quizzes.get(chat_id)
|
||
if not state:
|
||
return
|
||
question = state.questions[state.index]
|
||
options = question["options"]
|
||
buttons = [
|
||
[InlineKeyboardButton(LETTERS[idx], callback_data=f"answer:{idx}")]
|
||
for idx, option in enumerate(options[:8])
|
||
]
|
||
text = question_message_text(state, question)
|
||
markup = InlineKeyboardMarkup(buttons)
|
||
if update and update.message:
|
||
await update.message.reply_text(text, parse_mode=ParseMode.HTML, reply_markup=markup)
|
||
elif context:
|
||
await context.bot.send_message(chat_id, text, parse_mode=ParseMode.HTML, reply_markup=markup)
|
||
|
||
|
||
async def finish_quiz(chat_id: int, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||
state = active_quizzes.pop(chat_id, None)
|
||
if not state:
|
||
return
|
||
total = len(state.questions)
|
||
lines = [f"Quiz complete. Score: {state.score}/{total}"]
|
||
if state.mode == "exam":
|
||
lines.append("")
|
||
lines.append("Answers:")
|
||
for question_id, chosen, correct, ok in state.answers:
|
||
marker = "OK" if ok else "MISS"
|
||
lines.append(f"Q{question_id}: {marker}. You: {chosen}. Correct: {correct}")
|
||
lines.append("")
|
||
lines.append(f"Full question bank: {PUBLIC_APP_URL}")
|
||
await context.bot.send_message(chat_id, "\n".join(lines), reply_markup=main_menu(chat_id), disable_web_page_preview=True)
|
||
|
||
|
||
async def callback_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||
query = update.callback_query
|
||
if not query:
|
||
return
|
||
await query.answer()
|
||
data = query.data or ""
|
||
chat_id = query.message.chat_id if query.message else update.effective_chat.id
|
||
|
||
if data.startswith("browse:"):
|
||
parts = data.split(":")
|
||
parent_id = int(parts[1])
|
||
page = int(parts[2]) if len(parts) > 2 else 0
|
||
await show_browse(update, parent_id, page)
|
||
return
|
||
if data == "exams":
|
||
await show_exams(update)
|
||
return
|
||
if data.startswith("exam:"):
|
||
exam_id = int(data.rsplit(":", 1)[1])
|
||
if exam_id:
|
||
chat_exam[chat_id] = exam_id
|
||
name = next((e["name"] for e in list_exams() if e["id"] == exam_id), "that objective")
|
||
await query.edit_message_text(f"Revising for {html.escape(name)}.",
|
||
parse_mode=ParseMode.HTML, reply_markup=main_menu(chat_id))
|
||
else:
|
||
chat_exam.pop(chat_id, None)
|
||
await query.edit_message_text("Questions will come from everything.",
|
||
reply_markup=main_menu(chat_id))
|
||
return
|
||
if data.startswith("quiz:random:"):
|
||
_, _, count, mode = data.split(":", 3)
|
||
await start_quiz(update, random_questions(clamp_count(int(count)), chat_exam.get(chat_id)),
|
||
mode, context=context)
|
||
return
|
||
if data.startswith("pick:category:"):
|
||
category_id = int(data.rsplit(":", 1)[1])
|
||
await query.edit_message_text(
|
||
f"<b>{html.escape(category_name(category_id))}</b> — how many questions?",
|
||
parse_mode=ParseMode.HTML,
|
||
reply_markup=count_menu("category", category_id, back_to=category_id))
|
||
return
|
||
if data.startswith("start:category:"):
|
||
_, _, category_id, count, mode = data.split(":", 4)
|
||
await start_quiz(update,
|
||
category_questions(int(category_id), clamp_count(int(count)), chat_exam.get(chat_id)),
|
||
mode, category_name(int(category_id)), context)
|
||
return
|
||
if data.startswith("answer:"):
|
||
state = active_quizzes.get(chat_id)
|
||
if not state:
|
||
await query.edit_message_text("This quiz expired. Start a new one with /random or /topics.")
|
||
return
|
||
chosen_idx = int(data.split(":", 1)[1])
|
||
question = state.questions[state.index]
|
||
correct_idx = question["correct_index"]
|
||
ok = chosen_idx == correct_idx
|
||
if ok:
|
||
state.score += 1
|
||
chosen = LETTERS[chosen_idx] if chosen_idx < len(LETTERS) else "?"
|
||
correct = LETTERS[correct_idx] if correct_idx < len(LETTERS) else "?"
|
||
state.answers.append((state.index + 1, chosen, correct, ok))
|
||
if state.mode == "study":
|
||
await query.edit_message_text(
|
||
answered_question_text(state, question, chosen_idx, correct_idx),
|
||
parse_mode=ParseMode.HTML,
|
||
)
|
||
await send_answer_feedback(context, chat_id, question, chosen_idx, correct_idx, ok)
|
||
else:
|
||
await query.edit_message_reply_markup(reply_markup=None)
|
||
state.index += 1
|
||
if state.index >= len(state.questions):
|
||
await finish_quiz(chat_id, context)
|
||
else:
|
||
await send_current_question(chat_id, None, context)
|
||
|
||
|
||
async def text_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||
text = update.message.text if update.message else ""
|
||
match = NUMBER_RE.match(text or "")
|
||
if match:
|
||
await start_quiz(update, random_questions(clamp_count(int(match.group(1)))), "study")
|
||
return
|
||
await send_text(update, "Send a number like 20, or use /topics, /objective, /search, /random.",
|
||
main_menu(update.effective_chat.id if update.effective_chat else 0))
|
||
|
||
|
||
async def post_init(app: Application) -> None:
|
||
await app.bot.set_my_commands([
|
||
BotCommand("start", "Show help and quick quiz buttons"),
|
||
BotCommand("random", "Start a random quiz: /random 20 exam"),
|
||
BotCommand("topics", "Browse the subject tree"),
|
||
BotCommand("objective", "Pick the exam you are revising for"),
|
||
BotCommand("search", "Search text/topic: /search sepsis 20"),
|
||
BotCommand("stop", "Stop the current quiz"),
|
||
])
|
||
|
||
|
||
def main() -> None:
|
||
app = Application.builder().token(TELEGRAM_BOT_TOKEN).post_init(post_init).build()
|
||
app.add_handler(CommandHandler(["start", "help"], help_cmd))
|
||
app.add_handler(CommandHandler("random", random_cmd))
|
||
app.add_handler(CommandHandler("search", search_cmd))
|
||
app.add_handler(CommandHandler(["topics", "categories"], topics_cmd))
|
||
app.add_handler(CommandHandler(["objective", "exam"], objective_cmd))
|
||
app.add_handler(CommandHandler("stop", stop_cmd))
|
||
app.add_handler(CallbackQueryHandler(callback_handler))
|
||
app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, text_handler))
|
||
app.run_polling(allowed_updates=Update.ALL_TYPES)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|