feat(bot): the Telegram bot follows the current taxonomy
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
This commit is contained in:
Daniel
parent
2c5d3c67b4
commit
0d4042169a
1 changed files with 288 additions and 176 deletions
|
|
@ -28,6 +28,10 @@ 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*$")
|
||||
|
|
@ -44,6 +48,10 @@ class QuizState:
|
|||
|
||||
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:
|
||||
|
|
@ -84,14 +92,48 @@ def answer_index(question: dict[str, Any], options: list[str]) -> int | None:
|
|||
return None
|
||||
|
||||
|
||||
def question_base_sql(where: str = "") -> str:
|
||||
#: 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 q.question_type = 'mcq'
|
||||
and q.options is not null
|
||||
and q.is_shared = 1
|
||||
where {SERVEABLE}
|
||||
{where}
|
||||
order by random()
|
||||
limit %s
|
||||
|
|
@ -113,80 +155,104 @@ def valid_questions(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|||
return clean
|
||||
|
||||
|
||||
def random_questions(limit: int) -> list[dict[str, Any]]:
|
||||
rows = db_query(question_base_sql(), (limit * 2,))
|
||||
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) -> list[dict[str, Any]]:
|
||||
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("""
|
||||
question_base_sql(f"""
|
||||
{clause}
|
||||
and (
|
||||
q.question_text ilike %s
|
||||
or q.explanation ilike %s
|
||||
or qc.name ilike %s
|
||||
or exists (
|
||||
select 1
|
||||
from question_tag_links qtl
|
||||
join question_tags qt on qt.id = qtl.tag_id
|
||||
where qtl.question_id = q.id and qt.name ilike %s
|
||||
select 1 from question_categories c
|
||||
where c.id = q.question_category_id and c.name ilike %s
|
||||
)
|
||||
)
|
||||
"""),
|
||||
(pattern, pattern, pattern, pattern, limit * 2),
|
||||
(*params, pattern, pattern, pattern, limit * 2),
|
||||
)
|
||||
return valid_questions(rows)[:limit]
|
||||
|
||||
|
||||
def category_questions(category_id: int, limit: int) -> list[dict[str, Any]]:
|
||||
rows = db_query(question_base_sql("and q.question_category_id = %s"), (category_id, limit * 2))
|
||||
return valid_questions(rows)[:limit]
|
||||
|
||||
|
||||
def tag_questions(tag_id: int, limit: int) -> list[dict[str, Any]]:
|
||||
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("and exists (select 1 from question_tag_links qtl where qtl.question_id = q.id and qtl.tag_id = %s)"),
|
||||
(tag_id, limit * 2),
|
||||
question_base_sql(f"and {IN_SUBTREE} {clause}", prefix=SUBTREE),
|
||||
(category_id, *params, limit * 2),
|
||||
)
|
||||
return valid_questions(rows)[:limit]
|
||||
|
||||
|
||||
def list_categories() -> list[dict[str, Any]]:
|
||||
return db_query(
|
||||
"""
|
||||
select qc.id, qc.name, count(q.id)::int as count
|
||||
from question_categories qc
|
||||
join questions q on q.question_category_id = qc.id and q.question_type = 'mcq' and q.is_shared = 1
|
||||
group by qc.id, qc.name
|
||||
having count(q.id) > 0
|
||||
order by qc.name
|
||||
"""
|
||||
)
|
||||
|
||||
|
||||
def search_tags(term: str | None = None, limit: int = 20) -> list[dict[str, Any]]:
|
||||
where = ""
|
||||
params: list[Any] = []
|
||||
if term:
|
||||
where = "where t.name ilike %s"
|
||||
params.append(f"%{term}%")
|
||||
params.append(limit)
|
||||
def list_exams() -> list[dict[str, Any]]:
|
||||
"""The objectives, with how many serveable questions each covers."""
|
||||
return db_query(
|
||||
f"""
|
||||
select t.id, t.name, t.type, count(qtl.question_id)::int as count
|
||||
from question_tags t
|
||||
join question_tag_links qtl on qtl.tag_id = t.id
|
||||
join questions q on q.id = qtl.question_id and q.question_type = 'mcq' and q.is_shared = 1
|
||||
{where}
|
||||
group by t.id, t.name, t.type
|
||||
order by count(qtl.question_id) desc, t.name
|
||||
limit %s
|
||||
""",
|
||||
tuple(params),
|
||||
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
|
||||
|
|
@ -224,47 +290,29 @@ async def respond(update: Update, context: ContextTypes.DEFAULT_TYPE | None, tex
|
|||
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() -> InlineKeyboardMarkup:
|
||||
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("Categories", callback_data="list:categories:0")],
|
||||
[InlineKeyboardButton("Top keywords", callback_data="list:tags:0")],
|
||||
[InlineKeyboardButton("Browse topics", callback_data="browse:0")],
|
||||
[InlineKeyboardButton(objective, callback_data="exams")],
|
||||
])
|
||||
|
||||
|
||||
def count_menu(kind: str, item_id: int) -> InlineKeyboardMarkup:
|
||||
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 format_question_review(state: QuizState, question: dict[str, Any], chosen_idx: int, correct_idx: int, ok: bool) -> str:
|
||||
lines = [
|
||||
f"<b>Question {state.index + 1}/{len(state.questions)}</b>",
|
||||
html.escape(question["question_text"]),
|
||||
"",
|
||||
]
|
||||
for idx, option in enumerate(question["options"][:8]):
|
||||
marker = ""
|
||||
if idx == correct_idx:
|
||||
marker = " correct"
|
||||
elif idx == chosen_idx:
|
||||
marker = " your answer"
|
||||
lines.append(f"{LETTERS[idx]}. {html.escape(option)}{marker}")
|
||||
lines.extend([
|
||||
"",
|
||||
"Correct." if ok else f"Incorrect. Correct answer: {LETTERS[correct_idx]}",
|
||||
"",
|
||||
f"<b>Explanation:</b> {html.escape(question.get('explanation') or 'No explanation available.')}",
|
||||
])
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def truncate_text(value: str, limit: int) -> str:
|
||||
value = (value or "").strip()
|
||||
if len(value) <= limit:
|
||||
|
|
@ -272,6 +320,32 @@ def truncate_text(value: str, limit: int) -> str:
|
|||
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],
|
||||
|
|
@ -344,20 +418,14 @@ async def send_answer_feedback(
|
|||
ok: bool,
|
||||
) -> None:
|
||||
header = answer_feedback_header(chosen_idx, correct_idx, ok)
|
||||
explanation = (question.get("explanation") or "No explanation available.").strip()
|
||||
first_prefix = f"{header}\n\n<b>Explanation:</b> "
|
||||
next_prefix = "<b>Explanation continued:</b> "
|
||||
first_limit = TELEGRAM_MESSAGE_LIMIT - len(first_prefix) - 200
|
||||
next_limit = TELEGRAM_MESSAGE_LIMIT - len(next_prefix) - 200
|
||||
chunks = split_plain_text(explanation, max(1000, first_limit))
|
||||
for idx, chunk in enumerate(chunks):
|
||||
prefix = first_prefix if idx == 0 else next_prefix
|
||||
if idx > 0 and len(chunk) > next_limit:
|
||||
# First chunk has a smaller budget because it includes answer feedback.
|
||||
for subchunk in split_plain_text(chunk, next_limit):
|
||||
await context.bot.send_message(chat_id, f"{next_prefix}{html.escape(subchunk)}", parse_mode=ParseMode.HTML)
|
||||
continue
|
||||
await context.bot.send_message(chat_id, f"{prefix}{html.escape(chunk)}", parse_mode=ParseMode.HTML)
|
||||
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:
|
||||
|
|
@ -365,27 +433,40 @@ def answered_question_text(state: QuizState, question: dict[str, Any], chosen_id
|
|||
|
||||
|
||||
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>PedQuiz bot</b>",
|
||||
"<b>PedsHub bot</b>",
|
||||
"Send <code>20</code> for a random 20-question study quiz.",
|
||||
"Use /random 20 exam for exam mode.",
|
||||
"Use /categories to choose from PREP/category buckets.",
|
||||
"Use /keywords fever to search generated keywords/subjects.",
|
||||
"Use /search sepsis 20 to quiz by text search.",
|
||||
"Use /stop to end the current 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.",
|
||||
"",
|
||||
"Study mode shows each answer immediately. Exam mode shows answers at the end.",
|
||||
]), main_menu())
|
||||
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:
|
||||
await start_quiz(update, random_questions(parse_count(context.args)), parse_mode(context.args))
|
||||
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()
|
||||
|
|
@ -394,79 +475,93 @@ async def search_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None
|
|||
return
|
||||
if update.message:
|
||||
await update.message.chat.send_action(ChatAction.TYPING)
|
||||
await start_quiz(update, search_questions(term, count), mode, label=f"Search: {term}")
|
||||
await start_quiz(update, search_questions(term, count, chat_exam.get(chat_id)), mode, label=f"Search: {term}")
|
||||
|
||||
|
||||
async def categories_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||||
await show_categories(update, page=0)
|
||||
async def topics_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||||
await show_browse(update, 0)
|
||||
|
||||
|
||||
async def keywords_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
|
||||
term = " ".join(context.args).strip() if context.args else None
|
||||
rows = search_tags(term, 30)
|
||||
if not rows:
|
||||
await send_text(update, "No keywords found.")
|
||||
return
|
||||
buttons = [
|
||||
[InlineKeyboardButton(f"{row['name']} ({row['type']}, {row['count']})", callback_data=f"pick:tag:{row['id']}")]
|
||||
for row in rows[:20]
|
||||
]
|
||||
await send_text(update, "Choose a keyword/subject, then choose quiz size:", InlineKeyboardMarkup(buttons))
|
||||
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.", main_menu())
|
||||
await send_text(update, "Quiz stopped. Nothing was recorded.",
|
||||
main_menu(update.effective_chat.id if update.effective_chat else 0))
|
||||
|
||||
|
||||
async def show_categories(update: Update, page: int) -> None:
|
||||
rows = list_categories()
|
||||
per_page = 10
|
||||
start = page * per_page
|
||||
chunk = rows[start:start + per_page]
|
||||
if not chunk:
|
||||
await send_text(update, "No categories found.")
|
||||
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
|
||||
buttons = [
|
||||
[InlineKeyboardButton(f"{row['name']} ({row['count']})", callback_data=f"pick:category:{row['id']}")]
|
||||
for row in chunk
|
||||
]
|
||||
|
||||
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"list:categories:{page - 1}"))
|
||||
if start + per_page < len(rows):
|
||||
nav.append(InlineKeyboardButton("Next", callback_data=f"list:categories:{page + 1}"))
|
||||
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)
|
||||
text = "Choose a category, then choose quiz size and mode."
|
||||
if update.callback_query:
|
||||
await update.callback_query.edit_message_text(text, reply_markup=InlineKeyboardMarkup(buttons))
|
||||
|
||||
if node is None:
|
||||
title = "Choose a topic."
|
||||
else:
|
||||
await send_text(update, text, InlineKeyboardMarkup(buttons))
|
||||
# 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_tags(update: Update, page: int) -> None:
|
||||
rows = search_tags(limit=80)
|
||||
per_page = 10
|
||||
start = page * per_page
|
||||
chunk = rows[start:start + per_page]
|
||||
buttons = [
|
||||
[InlineKeyboardButton(f"{row['name']} ({row['type']}, {row['count']})", callback_data=f"pick:tag:{row['id']}")]
|
||||
for row in chunk
|
||||
]
|
||||
nav = []
|
||||
if page > 0:
|
||||
nav.append(InlineKeyboardButton("Prev", callback_data=f"list:tags:{page - 1}"))
|
||||
if start + per_page < len(rows):
|
||||
nav.append(InlineKeyboardButton("Next", callback_data=f"list:tags:{page + 1}"))
|
||||
if nav:
|
||||
buttons.append(nav)
|
||||
text = "Choose a keyword/subject, then choose quiz size and mode."
|
||||
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=InlineKeyboardMarkup(buttons))
|
||||
await update.callback_query.edit_message_text(text, reply_markup=markup)
|
||||
else:
|
||||
await send_text(update, text, InlineKeyboardMarkup(buttons))
|
||||
await send_text(update, text, markup)
|
||||
|
||||
|
||||
async def start_quiz(
|
||||
|
|
@ -479,7 +574,10 @@ async def start_quiz(
|
|||
if not update.effective_chat:
|
||||
return
|
||||
if not questions:
|
||||
await respond(update, context, "No usable MCQ questions found for that selection. Try /keywords, /categories, or a broader /search term.", main_menu())
|
||||
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)
|
||||
|
|
@ -524,7 +622,7 @@ async def finish_quiz(chat_id: int, context: ContextTypes.DEFAULT_TYPE) -> None:
|
|||
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(), disable_web_page_preview=True)
|
||||
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:
|
||||
|
|
@ -535,36 +633,49 @@ async def callback_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -
|
|||
data = query.data or ""
|
||||
chat_id = query.message.chat_id if query.message else update.effective_chat.id
|
||||
|
||||
if data.startswith("list:categories:"):
|
||||
await show_categories(update, int(data.rsplit(":", 1)[1]))
|
||||
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.startswith("list:tags:"):
|
||||
await show_tags(update, int(data.rsplit(":", 1)[1]))
|
||||
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))), mode, context=context)
|
||||
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("How many questions?", reply_markup=count_menu("category", category_id))
|
||||
return
|
||||
if data.startswith("pick:tag:"):
|
||||
tag_id = int(data.rsplit(":", 1)[1])
|
||||
await query.edit_message_text("How many questions?", reply_markup=count_menu("tag", tag_id))
|
||||
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))), mode, "Category", context)
|
||||
return
|
||||
if data.startswith("start:tag:"):
|
||||
_, _, tag_id, count, mode = data.split(":", 4)
|
||||
await start_quiz(update, tag_questions(int(tag_id), clamp_count(int(count))), mode, "Keyword", context)
|
||||
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 /categories.")
|
||||
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]
|
||||
|
|
@ -596,15 +707,16 @@ async def text_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> No
|
|||
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 /categories, /keywords, /search, /random.", main_menu())
|
||||
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("categories", "Browse categories"),
|
||||
BotCommand("keywords", "Browse/search keywords: /keywords fever"),
|
||||
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"),
|
||||
])
|
||||
|
|
@ -615,8 +727,8 @@ def main() -> None:
|
|||
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("categories", categories_cmd))
|
||||
app.add_handler(CommandHandler(["keywords", "tags"], keywords_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))
|
||||
|
|
|
|||
Loading…
Reference in a new issue