2bc1b9fbd8
Four things a deck got wrong. The category. Every deck written from an article is filed where the article is filed — and the list drew all of them "Uncategorized", because the router defines its own FlashcardDeckResponse that shadows the one in schemas/, and that one has no category_id. So the field was set on the row, returned by nothing, and an educator refiled by hand what the system had already filed correctly. The shared schema was imported by no module at all, so it is gone rather than left as a second definition to read past next time. The size. Fifteen cards is the per-chunk default, and an article is one chunk however long it is — a ten-section piece and a two-paragraph stub both asked for fifteen. Now roughly a card per 150 words, floored at 12 so a short article still makes a deck and capped at 30 so one call stays inside the model's output. The card. Set at list-item size inside a frame that fills the window, so a two-line question sat in the middle of an acre of white. The face scales with the window and stops at a comfortable measure; the back is set smaller than the front, as prose rather than a headline. And the contract snapshot, which still owed the jobs endpoint from the last commit. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TqXevQJhxFrM7jJg82cgZN
1259 lines
60 KiB
Python
1259 lines
60 KiB
Python
"""Async quiz extraction task with step-by-step progress reporting via Redis."""
|
||
import json
|
||
import logging
|
||
import re
|
||
import time
|
||
import os
|
||
|
||
from sqlalchemy import or_, text as sa_text
|
||
|
||
from app.tasks import celery_app
|
||
from app.database import SessionLocal
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
EXPIRE_SECONDS = 3600
|
||
CHUNK_PAGES = 50
|
||
|
||
|
||
def _redis():
|
||
import redis
|
||
from app.config import settings
|
||
return redis.from_url(settings.REDIS_URL, decode_responses=True)
|
||
|
||
|
||
def _push_step(r, job_id: str, step: str, message: str):
|
||
key = f"extraction:steps:{job_id}"
|
||
entry = json.dumps({"step": step, "message": message, "ts": time.time()})
|
||
r.rpush(key, entry)
|
||
r.expire(key, EXPIRE_SECONDS)
|
||
|
||
|
||
def _normalize_ocr(text: str) -> str:
|
||
"""Fix common OCR artifacts in the source PDFs."""
|
||
return (text
|
||
.replace("Pref erred", "Preferred")
|
||
.replace("Pre ferred", "Preferred")
|
||
.replace("Prefer red", "Preferred")
|
||
.replace("ltem", "Item")
|
||
.replace("ltcm", "Item"))
|
||
|
||
|
||
@celery_app.task(name="extract_quiz", bind=True)
|
||
def extract_quiz(
|
||
self,
|
||
job_id: str,
|
||
user_id: int,
|
||
section_id: int,
|
||
title: str,
|
||
mode: str,
|
||
time_limit_minutes: int | None,
|
||
model_id: str | None,
|
||
question_category_id: int | None,
|
||
extraction_mode: str = "standard",
|
||
):
|
||
r = _redis()
|
||
r.set(f"extraction:status:{job_id}", "running", ex=EXPIRE_SECONDS)
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
from app.models.section import Section
|
||
from app.models.pdf_document import PDFDocument
|
||
from app.services import ai_service, vector_service, pdf_service, embedding_service
|
||
from app.models.quiz import Quiz
|
||
from app.models.question import Question
|
||
from app.models.quiz_question_link import QuizQuestionLink
|
||
from app.config import settings
|
||
|
||
_push_step(r, job_id, "start", "Starting extraction…")
|
||
|
||
section = db.query(Section).filter(Section.id == section_id).first()
|
||
if not section:
|
||
raise ValueError("Section not found")
|
||
document = db.query(PDFDocument).filter(PDFDocument.id == section.document_id).first()
|
||
if not document:
|
||
raise ValueError("Document not found")
|
||
|
||
total_pages = section.end_page - section.start_page + 1
|
||
_push_step(r, job_id, "text", f"Loading text from pages {section.start_page}–{section.end_page} ({total_pages} pages)…")
|
||
|
||
# Determine model
|
||
if model_id:
|
||
from app.models.ai_model_config import AIModelConfig
|
||
config = db.query(AIModelConfig).filter(AIModelConfig.model_id == model_id).first()
|
||
api_key = config.api_key if config and config.api_key else None
|
||
model_name = config.name if config else model_id
|
||
else:
|
||
model_id, api_key = ai_service.get_model_for_task(db, "extraction")
|
||
model_name = model_id
|
||
|
||
# Split into 50-page chunks
|
||
if total_pages <= CHUNK_PAGES:
|
||
chunks = [(section.start_page, section.end_page)]
|
||
else:
|
||
chunks = []
|
||
p = section.start_page
|
||
while p <= section.end_page:
|
||
end = min(p + CHUNK_PAGES - 1, section.end_page)
|
||
chunks.append((p, end))
|
||
p = end + 1
|
||
|
||
n_chunks = len(chunks)
|
||
if n_chunks > 1:
|
||
_push_step(r, job_id, "text", f"Large section: splitting into {n_chunks} chunks of up to {CHUNK_PAGES} pages each.")
|
||
|
||
all_valid_questions = []
|
||
# A page whose text is missing is a different failure from a page the
|
||
# model found nothing in, and the two were reported as one.
|
||
pages_without_text = 0
|
||
all_skipped = []
|
||
|
||
# ── Non-standard extraction modes ─────────────────────────────────────
|
||
if extraction_mode != "standard":
|
||
from app.services.extraction_modes import (
|
||
extract_questions_only, extract_two_step,
|
||
extract_with_regex, ai_decide_strategy, generate_from_text,
|
||
ai_answer_questions,
|
||
)
|
||
|
||
resolved_mode = extraction_mode
|
||
|
||
if extraction_mode == "ai_decide":
|
||
_push_step(r, job_id, "ai", "AI is analysing the document to choose the best strategy…")
|
||
resolved_mode, reasoning = ai_decide_strategy(
|
||
section.document_id, section.start_page, section.end_page,
|
||
model_id, api_key,
|
||
)
|
||
_push_step(r, job_id, "ai", f"AI chose: {resolved_mode} — {reasoning}")
|
||
|
||
if resolved_mode == "questions_only":
|
||
_push_step(r, job_id, "ai", "Mode: Questions Only — extracting questions without answers.")
|
||
for chunk_idx, (start_p, end_p) in enumerate(chunks, 1):
|
||
if r.get(f"extraction:status:{job_id}") == "cancelled":
|
||
_push_step(r, job_id, "cancelled", "Job cancelled.")
|
||
return
|
||
_push_step(r, job_id, "ai", f"Chunk {chunk_idx}/{n_chunks}: pages {start_p}–{end_p}…")
|
||
chunk_content = vector_service.get_pages_text(
|
||
document_id=section.document_id, start_page=start_p, end_page=end_p)
|
||
if not chunk_content:
|
||
continue
|
||
try:
|
||
qs = extract_questions_only(_normalize_ocr(chunk_content),
|
||
f"{start_p}-{end_p}", start_p, model_id, api_key)
|
||
all_valid_questions.extend(qs)
|
||
_push_step(r, job_id, "ai", f" Pages {start_p}–{end_p}: {len(qs)} questions. Total: {len(all_valid_questions)}.")
|
||
except Exception as e:
|
||
_push_step(r, job_id, "ai", f" Pages {start_p}–{end_p} failed: {e}")
|
||
|
||
elif resolved_mode == "ai_answer":
|
||
_push_step(r, job_id, "ai", "Mode: AI Answer — extracting questions and using AI to determine correct answers.")
|
||
for chunk_idx, (start_p, end_p) in enumerate(chunks, 1):
|
||
if r.get(f"extraction:status:{job_id}") == "cancelled":
|
||
_push_step(r, job_id, "cancelled", "Job cancelled.")
|
||
return
|
||
_push_step(r, job_id, "ai", f"Chunk {chunk_idx}/{n_chunks}: pages {start_p}–{end_p} — extracting questions…")
|
||
chunk_content = vector_service.get_pages_text(
|
||
document_id=section.document_id, start_page=start_p, end_page=end_p)
|
||
if not chunk_content:
|
||
continue
|
||
try:
|
||
normalized = _normalize_ocr(chunk_content)
|
||
qs = extract_questions_only(normalized, f"{start_p}-{end_p}", start_p, model_id, api_key)
|
||
if qs:
|
||
_push_step(r, job_id, "ai", f" Pages {start_p}–{end_p}: {len(qs)} questions found, AI determining answers…")
|
||
qs = ai_answer_questions(qs, normalized, f"{start_p}-{end_p}", model_id, api_key)
|
||
answered = sum(1 for q in qs if q.get("correct_answer") and q["correct_answer"] != "PENDING")
|
||
all_valid_questions.extend(qs)
|
||
_push_step(r, job_id, "ai", f" Pages {start_p}–{end_p}: {answered}/{len(qs)} answered. Total: {len(all_valid_questions)}.")
|
||
except Exception as e:
|
||
_push_step(r, job_id, "ai", f" Pages {start_p}–{end_p} failed: {e}")
|
||
|
||
elif resolved_mode == "two_step":
|
||
_push_step(r, job_id, "ai", "Mode: Two-Step (separate answer key section).")
|
||
all_valid_questions, all_skipped = extract_two_step(
|
||
section.document_id, section.start_page, section.end_page,
|
||
model_id, api_key,
|
||
push_step=lambda step, msg: _push_step(r, job_id, step, msg),
|
||
chunk_pages=CHUNK_PAGES,
|
||
)
|
||
|
||
elif resolved_mode == "regex":
|
||
_push_step(r, job_id, "ai", "Mode: AI+Regex — analysing format then applying regex.")
|
||
all_valid_questions, all_skipped = extract_with_regex(
|
||
section.document_id, section.start_page, section.end_page,
|
||
model_id, api_key,
|
||
push_step=lambda step, msg: _push_step(r, job_id, step, msg),
|
||
chunk_pages=CHUNK_PAGES,
|
||
)
|
||
|
||
elif resolved_mode == "generate":
|
||
_push_step(r, job_id, "ai", "Mode: Generate — AI creates questions from the text.")
|
||
for chunk_idx, (start_p, end_p) in enumerate(chunks, 1):
|
||
if r.get(f"extraction:status:{job_id}") == "cancelled":
|
||
_push_step(r, job_id, "cancelled", "Job cancelled.")
|
||
return
|
||
_push_step(r, job_id, "ai", f"Chunk {chunk_idx}/{n_chunks}: pages {start_p}–{end_p}…")
|
||
chunk_content = vector_service.get_pages_text(
|
||
document_id=section.document_id, start_page=start_p, end_page=end_p)
|
||
if not chunk_content:
|
||
continue
|
||
try:
|
||
qs = generate_from_text(_normalize_ocr(chunk_content),
|
||
f"{start_p}-{end_p}", start_p, model_id, api_key)
|
||
all_valid_questions.extend(qs)
|
||
_push_step(r, job_id, "ai", f" Pages {start_p}–{end_p}: {len(qs)} questions generated. Total: {len(all_valid_questions)}.")
|
||
except Exception as e:
|
||
_push_step(r, job_id, "ai", f" Pages {start_p}–{end_p} failed: {e}")
|
||
|
||
else:
|
||
# ai_decide resolved to standard — fall through to standard loop below
|
||
extraction_mode = "standard"
|
||
|
||
if extraction_mode == "standard":
|
||
for chunk_idx, (start_p, end_p) in enumerate(chunks, 1):
|
||
if r.get(f"extraction:status:{job_id}") == "cancelled":
|
||
_push_step(r, job_id, "cancelled", "Job cancelled.")
|
||
return
|
||
if n_chunks > 1:
|
||
_push_step(r, job_id, "ai", f"Chunk {chunk_idx}/{n_chunks}: pages {start_p}–{end_p} → {model_name}…")
|
||
else:
|
||
_push_step(r, job_id, "ai", f"Sending pages {start_p}–{end_p} to {model_name}…")
|
||
chunk_content = vector_service.get_pages_text(
|
||
document_id=section.document_id, start_page=start_p, end_page=end_p,
|
||
)
|
||
if not chunk_content:
|
||
pages_without_text += 1
|
||
_push_step(r, job_id, "ai",
|
||
f" No stored text for pages {start_p}–{end_p}. The document's text is"
|
||
f" read from the search index, not the file, so this usually means it"
|
||
f" was never processed or its index was lost.")
|
||
continue
|
||
try:
|
||
chunk_data = ai_service.extract_questions(
|
||
_normalize_ocr(chunk_content),
|
||
page_info=f"{start_p}-{end_p}",
|
||
page_ref=start_p,
|
||
model_id=model_id,
|
||
api_key=api_key,
|
||
)
|
||
chunk_skipped = chunk_data[0].pop("skipped", []) if chunk_data else []
|
||
chunk_valid = [q for q in chunk_data if q.get("correct_answer")]
|
||
all_valid_questions.extend(chunk_valid)
|
||
all_skipped.extend(chunk_skipped)
|
||
_push_step(r, job_id, "ai",
|
||
f" Pages {start_p}–{end_p}: {len(chunk_valid)} questions"
|
||
f"{f', {len(chunk_skipped)} skipped' if chunk_skipped else ''}. "
|
||
f"Total: {len(all_valid_questions)}.")
|
||
except Exception as e:
|
||
_push_step(r, job_id, "ai", f" Pages {start_p}–{end_p} failed: {e}. Continuing…")
|
||
|
||
valid_questions = all_valid_questions
|
||
skipped = all_skipped
|
||
|
||
_push_step(r, job_id, "ai", f"Extraction complete: {len(valid_questions)} valid questions{f', {len(skipped)} skipped' if skipped else ''}.")
|
||
|
||
if not valid_questions:
|
||
# Blaming the model for a document that was never indexed sent
|
||
# people to change the model, the prompt and the page range, none
|
||
# of which was the problem.
|
||
if pages_without_text:
|
||
raise ValueError(
|
||
"This document has no stored text to read. Its pages are indexed when it is"
|
||
" uploaded, and that index is what extraction reads — not the file. Re-process"
|
||
" the document and try again.")
|
||
raise ValueError(
|
||
"The model found no questions with a marked correct answer in these pages.")
|
||
|
||
# Refresh DB connection — it may have gone stale during long LLM extraction
|
||
from sqlalchemy import text as _text
|
||
try:
|
||
db.execute(_text("SELECT 1"))
|
||
except Exception:
|
||
db.rollback()
|
||
db.close()
|
||
db = SessionLocal()
|
||
# Re-fetch objects that were bound to the old session
|
||
section = db.query(Section).filter(Section.id == section_id).first()
|
||
document = db.query(PDFDocument).filter(PDFDocument.id == section.document_id).first()
|
||
|
||
# Extract images
|
||
_push_step(r, job_id, "images", "Extracting question images…")
|
||
file_path = os.path.join(settings.UPLOAD_DIR, document.filename)
|
||
page_images = {}
|
||
if os.path.exists(file_path):
|
||
try:
|
||
page_images = pdf_service.extract_all_images(
|
||
file_path, document.id, section.start_page, section.end_page
|
||
)
|
||
except Exception as e:
|
||
_push_step(r, job_id, "images", f"Image extraction skipped: {e}")
|
||
|
||
# A batch, not a quiz. What a model pulled out of a PDF is a proposal:
|
||
# it is read, corrected and decided before it is anything, and a
|
||
# question id — which comes from a sequence and is never reissued — is
|
||
# taken at acceptance rather than at extraction.
|
||
from app.models.draft_question import DraftBatch, DraftQuestion
|
||
|
||
batch = DraftBatch(
|
||
title=title,
|
||
document_id=document.id,
|
||
section_id=section_id,
|
||
job_id=job_id,
|
||
model_id=model_id,
|
||
extraction_mode=extraction_mode,
|
||
category_id=question_category_id,
|
||
created_by=user_id,
|
||
status="open",
|
||
)
|
||
db.add(batch)
|
||
db.flush()
|
||
|
||
_push_step(r, job_id, "save", f"Saving {len(valid_questions)} drafts for review…")
|
||
|
||
for pos, q in enumerate(valid_questions):
|
||
page_ref = q.get("page_reference")
|
||
image_path = None
|
||
# Only link an image if the AI flagged the question as having a figure
|
||
if q.get("has_figure") and page_ref and page_ref in page_images and page_images[page_ref]:
|
||
image_path = page_images[page_ref].pop(0)
|
||
if not page_images[page_ref]:
|
||
del page_images[page_ref]
|
||
|
||
db.add(DraftQuestion(
|
||
batch_id=batch.id,
|
||
position=pos,
|
||
question_text=q["question_text"],
|
||
question_type=q["question_type"],
|
||
options=q.get("options"),
|
||
correct_answer=q["correct_answer"],
|
||
explanation=q.get("explanation", ""),
|
||
page_reference=page_ref,
|
||
image_path=image_path,
|
||
))
|
||
|
||
# No embedding here. A vector is for finding a question in the bank,
|
||
# and a draft is not in the bank; it is generated when one is accepted.
|
||
db.commit()
|
||
db.refresh(batch)
|
||
|
||
_push_step(r, job_id, "done",
|
||
f"{len(valid_questions)} drafts ready to review. Nothing is in the bank yet.")
|
||
r.set(f"extraction:status:{job_id}", "completed", ex=EXPIRE_SECONDS)
|
||
r.set(f"extraction:batch_id:{job_id}", str(batch.id), ex=EXPIRE_SECONDS)
|
||
return batch.id
|
||
|
||
except Exception as e:
|
||
logger.exception(f"Quiz extraction failed for job {job_id}")
|
||
_push_step(r, job_id, "error", f"Extraction failed: {e}")
|
||
r.set(f"extraction:status:{job_id}", "failed", ex=EXPIRE_SECONDS)
|
||
r.set(f"extraction:error:{job_id}", str(e)[:500], ex=EXPIRE_SECONDS)
|
||
raise
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
#: What each label means, in the prompt and in the docs, so the three words
|
||
#: mean the same thing to the model, to the learner reading a filter, and to
|
||
#: whoever disagrees with a label later.
|
||
DIFFICULTY_RUBRIC = """easy — one step. The stem names a classic presentation and the answer is
|
||
recall: a diagnosis with a pathognomonic finding, a first-line drug, a
|
||
standard schedule. A prepared candidate answers without working anything out.
|
||
medium — two steps, or one step under noise. The finding has to be interpreted
|
||
before it can be used, a distractor is genuinely plausible, or the question
|
||
asks for the *next* action rather than the diagnosis.
|
||
hard — several steps, or a judgement. An atypical presentation, a decision
|
||
where two answers are defensible and one is better, an exception to the rule
|
||
the candidate learned, or arithmetic on top of interpretation."""
|
||
|
||
DIFFICULTY_PROMPT = """You are labelling pediatric board-exam questions by how hard they are to
|
||
answer correctly, for a candidate who has studied the material.
|
||
|
||
Judge the *question*, not the topic. A rare disease with a give-away finding is
|
||
easy; a common one where two managements are defensible is not.
|
||
|
||
""" + DIFFICULTY_RUBRIC + """
|
||
|
||
Return ONLY a JSON array, one object per question, no prose:
|
||
[{"id": 12, "difficulty": "easy"}, ...]
|
||
Every id you were given must appear exactly once.
|
||
|
||
QUESTIONS
|
||
"""
|
||
|
||
#: Small enough that one bad reply costs little and the JSON stays inside the
|
||
#: reply limit; large enough that 3,000 questions is a hundred calls and not
|
||
#: three thousand.
|
||
DIFFICULTY_BATCH = 25
|
||
|
||
|
||
#: Judging a figure against its question needs a model that can see. The job
|
||
#: model is passed in rather than read from the admin's `tool` setting, because
|
||
#: this is a one-off audit and not a standing capability — it should not
|
||
#: quietly become the thing that decides what "the tool model" means.
|
||
FIGURE_AUDIT_MODEL = "openrouter-gemini-2.5-flash"
|
||
|
||
#: Two tries per figure, a second apart. A 502 from the proxy is usually a
|
||
#: moment rather than a state.
|
||
RETRIES = 3
|
||
#: And if it *is* a state, stop. A run that reports every figure unreadable has
|
||
#: told you nothing and cost an hour.
|
||
GIVE_UP_AFTER = 12
|
||
|
||
FIGURE_AUDIT_PROMPT = """This figure is attached to the exam question below. Decide whether it belongs
|
||
to it.
|
||
|
||
It belongs if a candidate would need it, or could reasonably use it, to answer
|
||
*this* question. It does not belong if it is about some other topic entirely,
|
||
which happens when a figure is lifted from the wrong page of a source document.
|
||
|
||
**The rule that matters, learned by getting it wrong.** Two kinds of figure sit
|
||
on these questions and they are not judged the same way.
|
||
|
||
A *clinical* figure — a photograph, radiograph, ultrasound, ECG, fundoscopy,
|
||
otoscopy, a microscopy slide, a specimen — is almost always the right one, and
|
||
the connection is often indirect: a tick on a leaf against a July fever, fungal
|
||
hyphae against a scaly rash, a recessed chin in the notes of a two-week-old.
|
||
Answer "no" for one of these ONLY when it is anatomically or clinically
|
||
impossible — a shoulder radiograph on a knee injury. Otherwise "yes", or
|
||
"unsure" if you truly cannot tell. Detaching a clinical figure a question needs
|
||
is the worst outcome here.
|
||
|
||
A *non-clinical* figure — a table, a citation, a reference list, a bar chart, a
|
||
nomogram, a page of text — belongs only when it is about the same subject as the
|
||
question. These are where mis-extraction shows up, and where "no" is usually
|
||
right.
|
||
|
||
Answer ONLY with JSON:
|
||
{"belongs": "yes" | "no" | "unsure",
|
||
"kind": "clinical" | "table" | "other",
|
||
"shows": "one sentence describing the figure"}
|
||
|
||
QUESTION
|
||
"""
|
||
|
||
|
||
@celery_app.task(name="audit_question_figures", bind=True)
|
||
def audit_question_figures(self, job_id: str = "", limit: int | None = None,
|
||
detach: bool = True, model_id: str | None = None) -> dict:
|
||
"""Look at every figure attached to a question and say whether it belongs.
|
||
|
||
Extraction took figures off the page a question was printed on, which is
|
||
usually right and sometimes catastrophically wrong: an HPV vaccination
|
||
question was carrying a table of craniofacial reconstruction by age. Nobody
|
||
can find those by reading stems, because the mistake is only visible in the
|
||
picture.
|
||
|
||
Every figure is also *described* while it is being looked at, and the
|
||
description is kept — most of them said "Figure from question #1206", which
|
||
is a filename with extra steps and made the image bank unsearchable.
|
||
|
||
Detaching is reversible in the sense that matters: the asset stays in the
|
||
bank with a note saying what happened, and only the link goes.
|
||
"""
|
||
import json as _json
|
||
|
||
from app.models.media import MediaAsset
|
||
from app.models.question import Question
|
||
from app.models.question_media import QuestionMedia
|
||
from app.services import storage_service, thumbnails, vision_service
|
||
from app.services.ai_service import chat
|
||
|
||
r = _redis()
|
||
if job_id:
|
||
r.set(f"extraction:status:{job_id}", "running", ex=EXPIRE_SECONDS)
|
||
db = SessionLocal()
|
||
model = model_id or FIGURE_AUDIT_MODEL
|
||
seen = kept = detached = unsure = unreadable = described = 0
|
||
consecutive_failures = 0
|
||
last = None
|
||
mismatches: list[dict] = []
|
||
try:
|
||
rows = db.query(Question).filter(
|
||
Question.image_path.isnot(None), Question.image_path != "").order_by(Question.id)
|
||
questions = rows.limit(limit).all() if limit else rows.all()
|
||
total = len(questions)
|
||
if job_id:
|
||
_push_step(r, job_id, "start", f"{total} figures to look at")
|
||
|
||
for position, question in enumerate(questions, start=1):
|
||
seen += 1
|
||
data = None
|
||
try:
|
||
data = storage_service.load(question.image_path)
|
||
except Exception:
|
||
data = None
|
||
if not data:
|
||
unreadable += 1
|
||
continue
|
||
# 640px is what the reader is shown and plenty to judge a figure by;
|
||
# sending the original would be megabytes a question for no gain.
|
||
small = thumbnails.render(data, 640) or data
|
||
image = vision_service.prepare(small, "image/webp", caption="a figure attached to a question")
|
||
if image is None:
|
||
unreadable += 1
|
||
continue
|
||
|
||
stem = (question.question_text or "")[:900]
|
||
verdict = None
|
||
for attempt in range(RETRIES):
|
||
try:
|
||
raw = (chat(model=model, max_tokens=300, temperature=0, messages=[{
|
||
"role": "user", "content": [
|
||
vision_service.image_part(image),
|
||
{"type": "text", "text": FIGURE_AUDIT_PROMPT + stem},
|
||
]}]) or "").strip()
|
||
if raw.startswith("```"):
|
||
raw = raw.split("\n", 1)[1] if "\n" in raw else raw[3:]
|
||
raw = raw[:-3] if raw.endswith("```") else raw
|
||
verdict = _json.loads(raw.strip())
|
||
break
|
||
except Exception as exc:
|
||
last = exc
|
||
# A 502 from the proxy is usually a moment, not a state.
|
||
if attempt + 1 < RETRIES:
|
||
time.sleep(2 ** attempt)
|
||
if verdict is None:
|
||
unreadable += 1
|
||
consecutive_failures += 1
|
||
logger.warning("Figure audit failed for question %s: %s", question.id, last)
|
||
# A whole corpus of "unreadable" is not a result, it is an
|
||
# outage — and the first run of this marched through 327
|
||
# questions reporting nothing while the proxy was down. Stop and
|
||
# say so, so the run can be repeated when it is back.
|
||
if consecutive_failures >= GIVE_UP_AFTER:
|
||
raise RuntimeError(
|
||
f"{consecutive_failures} figures in a row could not be read — "
|
||
f"the model or the proxy is unavailable. Last error: {last}")
|
||
continue
|
||
consecutive_failures = 0
|
||
|
||
belongs = str(verdict.get("belongs", "unsure")).strip().lower()
|
||
kind = str(verdict.get("kind", "")).strip().lower()
|
||
shows = str(verdict.get("shows", "")).strip()[:600]
|
||
|
||
# The prompt says it and this enforces it: a clinical figure is
|
||
# never detached on the model's say-so alone. Sixteen good ones went
|
||
# in the first run — a tick on a leaf, fungal hyphae, an ECG on a
|
||
# tachypnoeic neonate — because a judgement about relevance was
|
||
# allowed to act on a photograph. It becomes a flag for a person.
|
||
if belongs == "no" and kind == "clinical":
|
||
belongs = "unsure"
|
||
|
||
# The description is worth keeping whatever the verdict: "Figure
|
||
# from question #1206" is a filename with extra steps.
|
||
asset = db.query(MediaAsset).filter(MediaAsset.path == question.image_path).first()
|
||
if asset and shows:
|
||
placeholder = not asset.alt_text or asset.alt_text.lower().startswith("figure from question")
|
||
if placeholder:
|
||
asset.alt_text = shows
|
||
described += 1
|
||
if not asset.caption:
|
||
asset.caption = shows
|
||
|
||
if belongs == "no":
|
||
mismatches.append({"question_id": question.id, "path": question.image_path,
|
||
"shows": shows, "stem": stem[:160], "kind": kind,
|
||
"verdict": "no"})
|
||
if detach:
|
||
question.image_path = None
|
||
db.query(QuestionMedia).filter(
|
||
QuestionMedia.question_id == question.id).delete(synchronize_session=False)
|
||
if asset:
|
||
note = f"Detached from question #{question.id}: the figure shows {shows}"
|
||
asset.caption = (asset.caption or shows)
|
||
asset.alt_text = asset.alt_text or note
|
||
detached += 1
|
||
else:
|
||
kept += 1
|
||
elif belongs == "unsure":
|
||
unsure += 1
|
||
# Named, not just counted: "2 unsure" is a number nobody can
|
||
# act on. These stay attached and go on the list for a person.
|
||
mismatches.append({"question_id": question.id, "path": question.image_path,
|
||
"shows": shows, "stem": stem[:160], "kind": kind,
|
||
"verdict": "unsure"})
|
||
else:
|
||
kept += 1
|
||
|
||
db.commit()
|
||
if job_id and position % 10 == 0:
|
||
_push_step(r, job_id, "batch",
|
||
f"{position} of {total} · {detached} detached · {unsure} unsure")
|
||
|
||
result = {"looked_at": seen, "kept": kept, "detached": detached, "unsure": unsure,
|
||
"unreadable": unreadable, "described": described,
|
||
"mismatches": mismatches[:60]}
|
||
if job_id:
|
||
r.set(f"extraction:status:{job_id}", "completed", ex=EXPIRE_SECONDS)
|
||
_push_step(r, job_id, "done",
|
||
f"{detached} figures detached, {unsure} to check by hand")
|
||
return result
|
||
except Exception as exc:
|
||
logger.warning("Figure audit failed: %s", exc)
|
||
if job_id:
|
||
r.set(f"extraction:status:{job_id}", "failed", ex=EXPIRE_SECONDS)
|
||
raise
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
@celery_app.task(name="classify_question_difficulty", bind=True)
|
||
def classify_question_difficulty(self, job_id: str = "", limit: int | None = None,
|
||
relabel: bool = False) -> dict:
|
||
"""Give every question a difficulty, in batches, resumably.
|
||
|
||
The column existed and was NULL on all 2,924 rows, which made the filter in
|
||
the test builder a control that could only ever empty the bank, and made
|
||
"move the session along the difficulty range" impossible to build. Nothing
|
||
was ever going to write it by hand.
|
||
|
||
Resumable by construction: only rows with no difficulty are asked about
|
||
unless `relabel` is set, so a run that dies halfway is continued by running
|
||
it again. Each batch is committed on its own — a failure late in a long run
|
||
keeps everything the earlier batches decided.
|
||
"""
|
||
from app.models.question import Question
|
||
from app.services.ai_service import get_model_for_task, chat
|
||
|
||
r = _redis()
|
||
if job_id:
|
||
r.set(f"extraction:status:{job_id}", "running", ex=EXPIRE_SECONDS)
|
||
db = SessionLocal()
|
||
labelled, failed_batches = 0, 0
|
||
try:
|
||
model_id, api_key = get_model_for_task(db, "keyword")
|
||
query = db.query(Question).order_by(Question.id)
|
||
if not relabel:
|
||
query = query.filter(or_(Question.difficulty.is_(None), Question.difficulty == ""))
|
||
rows = query.limit(limit).all() if limit else query.all()
|
||
total = len(rows)
|
||
if job_id:
|
||
_push_step(r, job_id, "start", f"{total} questions to label")
|
||
|
||
for start in range(0, total, DIFFICULTY_BATCH):
|
||
batch = rows[start:start + DIFFICULTY_BATCH]
|
||
payload = []
|
||
for question in batch:
|
||
options = question.options if isinstance(question.options, list) else []
|
||
payload.append({
|
||
"id": question.id,
|
||
# Trimmed: the shape of the reasoning is in the first
|
||
# paragraph and the options, and a full vignette times
|
||
# twenty-five is a reply nobody needs to pay for.
|
||
"stem": (question.question_text or "")[:700],
|
||
"options": [str(option)[:120] for option in options[:6]],
|
||
"answer": (question.correct_answer or "")[:120],
|
||
})
|
||
prompt = DIFFICULTY_PROMPT + json.dumps(payload, ensure_ascii=False)
|
||
try:
|
||
raw = (chat(model=model_id, messages=[{"role": "user", "content": prompt}],
|
||
max_tokens=1500, temperature=0, api_key=api_key) or "").strip()
|
||
if raw.startswith("```"):
|
||
raw = raw.split("\n", 1)[1] if "\n" in raw else raw[3:]
|
||
raw = raw[:-3] if raw.endswith("```") else raw
|
||
verdicts = json.loads(raw.strip())
|
||
except Exception as exc:
|
||
failed_batches += 1
|
||
logger.warning("Difficulty batch at %s failed: %s", start, exc)
|
||
if job_id:
|
||
_push_step(r, job_id, "warn", f"A batch of {len(batch)} could not be read; skipped")
|
||
continue
|
||
|
||
by_id = {question.id: question for question in batch}
|
||
for verdict in verdicts if isinstance(verdicts, list) else []:
|
||
question = by_id.get(verdict.get("id"))
|
||
level = str(verdict.get("difficulty", "")).strip().lower()
|
||
# Anything that is not one of the three words is not a label.
|
||
# A model that answers "moderate" has not answered.
|
||
if question is None or level not in ("easy", "medium", "hard"):
|
||
continue
|
||
question.difficulty = level
|
||
labelled += 1
|
||
db.commit()
|
||
if job_id:
|
||
_push_step(r, job_id, "batch",
|
||
f"{min(start + DIFFICULTY_BATCH, total)} of {total} · {labelled} labelled")
|
||
|
||
if job_id:
|
||
r.set(f"extraction:status:{job_id}", "completed", ex=EXPIRE_SECONDS)
|
||
_push_step(r, job_id, "done", f"{labelled} questions labelled")
|
||
return {"labelled": labelled, "considered": total, "failed_batches": failed_batches}
|
||
except Exception as exc:
|
||
logger.warning("Difficulty run failed: %s", exc)
|
||
if job_id:
|
||
r.set(f"extraction:status:{job_id}", "failed", ex=EXPIRE_SECONDS)
|
||
r.set(f"extraction:error:{job_id}", str(exc)[:300], ex=EXPIRE_SECONDS)
|
||
raise
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
@celery_app.task(name="apply_topic_claims")
|
||
def apply_topic_claims() -> dict:
|
||
"""Re-apply every article's topic claim. The safety net, not the mechanism.
|
||
|
||
A claim is applied when it is staked and again whenever a question is filed
|
||
into the category, which covers the two ways it normally changes. Both can
|
||
be bypassed — a bulk update written in SQL, an import, an article restored
|
||
from the trash — and a claim that is true should stay true without anybody
|
||
remembering to press anything. Normally finds nothing.
|
||
"""
|
||
db = SessionLocal()
|
||
try:
|
||
from app.services import topic_claims
|
||
|
||
result = topic_claims.sweep(db)
|
||
if result["links_added"]:
|
||
logger.info("Topic claims: %s links added across %s claims",
|
||
result["links_added"], result["claims"])
|
||
return result
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
@celery_app.task(name="retry_missing_embeddings")
|
||
def retry_missing_embeddings(batch: int = 200) -> dict:
|
||
"""Backfill questions that have no usable vector.
|
||
|
||
Embedding at creation time is best effort: if the encoder is briefly
|
||
unavailable the question is still saved, and without this it would stay
|
||
invisible to semantic search forever. Runs on a schedule and normally finds
|
||
nothing. Also catches rows left by an embedding-model change.
|
||
"""
|
||
db = SessionLocal()
|
||
try:
|
||
from app.services import embedding_service
|
||
|
||
active = embedding_service._get_embedding_model()
|
||
pending_total, embedded_total = 0, 0
|
||
# Every embeddable corpus, so an article or card is not left behind.
|
||
for kind, model in embedding_service.embeddable_models().items():
|
||
pending = (
|
||
db.query(model)
|
||
.filter(
|
||
(model.embedding.is_(None))
|
||
| (model.embedding_model.is_(None))
|
||
| (model.embedding_model != active)
|
||
)
|
||
.limit(batch)
|
||
.all()
|
||
)
|
||
pending_total += len(pending)
|
||
for row in pending:
|
||
try:
|
||
if embedding_service.embed_record(row, kind):
|
||
embedded_total += 1
|
||
except Exception:
|
||
logger.warning("Retry embedding failed for %s %s", kind, row.id, exc_info=True)
|
||
# And the other direction. A section row survives its article being
|
||
# deleted or unpublished only if something removes it, and nothing did:
|
||
# the index is rebuilt when an article is *saved*, so an article that is
|
||
# never saved again keeps its rows forever. Half-written prose then
|
||
# stays in search results and in the shortlist the assistant answers
|
||
# from, long after a reader could open the page it came from.
|
||
dropped = _drop_unreachable_sections(db)
|
||
if embedded_total or dropped:
|
||
db.commit()
|
||
logger.info("Backfilled %s embeddings (%s pending), dropped %s stale section rows",
|
||
embedded_total, pending_total, dropped)
|
||
return {"pending": pending_total, "embedded": embedded_total,
|
||
"dropped": dropped, "model": active}
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
def _drop_unreachable_sections(db) -> int:
|
||
"""Remove index rows whose article is gone or no longer published."""
|
||
from sqlalchemy import select
|
||
|
||
from app.models.article import Article, ArticleSectionIndex
|
||
|
||
reachable = select(Article.id).where(Article.status == "published")
|
||
return (db.query(ArticleSectionIndex)
|
||
.filter(~ArticleSectionIndex.article_id.in_(reachable))
|
||
.delete(synchronize_session=False))
|
||
|
||
|
||
@celery_app.task(name="regenerate_embeddings", bind=True)
|
||
def regenerate_embeddings(self, job_id: str, user_id: int, stale_only: bool = True):
|
||
"""Re-embed questions with the current model.
|
||
|
||
`stale_only` (the default) covers exactly what breaks semantic search: rows
|
||
with no vector, and rows whose vector came from a different model and so sits
|
||
in an incomparable space. Pass False to rebuild the whole bank.
|
||
"""
|
||
r = _redis()
|
||
r.set(f"extraction:status:{job_id}", "running", ex=EXPIRE_SECONDS)
|
||
r.set(f"extraction:job_title:{job_id}", "Regenerate Embeddings", ex=EXPIRE_SECONDS)
|
||
r.lpush(f"extraction:user_jobs:{user_id}", job_id)
|
||
r.expire(f"extraction:user_jobs:{user_id}", 86400)
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
from app.models.question import Question
|
||
from app.services import embedding_service
|
||
|
||
active = embedding_service._get_embedding_model()
|
||
pending = []
|
||
for kind, model in embedding_service.embeddable_models().items():
|
||
query = db.query(model)
|
||
if stale_only:
|
||
query = query.filter(
|
||
(model.embedding.is_(None))
|
||
| (model.embedding_model.is_(None))
|
||
| (model.embedding_model != active)
|
||
)
|
||
pending.extend((kind, row) for row in query.all())
|
||
total = len(pending)
|
||
scope = "missing or stale" if stale_only else "all"
|
||
_push_step(r, job_id, "start", f"Regenerating embeddings for {total} {scope} records…")
|
||
|
||
ok = 0
|
||
for i, (kind, row) in enumerate(pending):
|
||
try:
|
||
if embedding_service.embed_record(row, kind):
|
||
ok += 1
|
||
if (i + 1) % 50 == 0:
|
||
db.commit()
|
||
_push_step(r, job_id, "progress", f"{i + 1}/{total} processed ({ok} embedded)")
|
||
except Exception as e:
|
||
logger.warning(f"Embedding failed for {kind} {row.id}: {e}")
|
||
|
||
db.commit()
|
||
_push_step(r, job_id, "done", f"Done — {ok}/{total} questions re-embedded.")
|
||
r.set(f"extraction:status:{job_id}", "completed", ex=EXPIRE_SECONDS)
|
||
except Exception as e:
|
||
logger.exception(f"Embedding regeneration failed for job {job_id}")
|
||
_push_step(r, job_id, "error", f"Failed: {e}")
|
||
r.set(f"extraction:status:{job_id}", "failed", ex=EXPIRE_SECONDS)
|
||
raise
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
@celery_app.task(name="generate_flashcard_deck", bind=True)
|
||
def generate_flashcard_deck(self, job_id: str, section_id: int, user_id: int,
|
||
title: str, model_id: str | None = None):
|
||
"""Generate flashcards from a document section using AI."""
|
||
r = _redis()
|
||
r.set(f"extraction:status:{job_id}", "running", ex=EXPIRE_SECONDS)
|
||
db = SessionLocal()
|
||
try:
|
||
from app.models.section import Section
|
||
from app.models.pdf_document import PDFDocument
|
||
from app.services import vector_service
|
||
from app.services import extraction_modes
|
||
|
||
section = db.query(Section).filter(Section.id == section_id).first()
|
||
if not section:
|
||
r.set(f"extraction:status:{job_id}", "failed", ex=EXPIRE_SECONDS)
|
||
_push_step(r, job_id, "error", "Section not found")
|
||
return
|
||
document = db.query(PDFDocument).filter(PDFDocument.id == section.document_id).first()
|
||
|
||
from app.services.ai_service import get_model_for_task
|
||
ai_model_id, ai_api_key = get_model_for_task(db, "flashcard")
|
||
if model_id:
|
||
ai_model_id = model_id
|
||
|
||
total_pages = section.end_page - section.start_page + 1
|
||
_push_step(r, job_id, "start", f"Generating flashcards from {total_pages} pages…")
|
||
|
||
all_cards = []
|
||
|
||
if total_pages <= CHUNK_PAGES:
|
||
content = vector_service.get_pages_text(section.document_id, section.start_page, section.end_page)
|
||
if content:
|
||
_push_step(r, job_id, "ai", f"Generating flashcards from pages {section.start_page}–{section.end_page}…")
|
||
cards = extraction_modes.generate_flashcards(
|
||
content, f"{section.start_page}–{section.end_page}",
|
||
section.start_page, ai_model_id, ai_api_key,
|
||
)
|
||
all_cards.extend(cards)
|
||
_push_step(r, job_id, "ai", f"Generated {len(cards)} cards")
|
||
else:
|
||
n_chunks = (total_pages + CHUNK_PAGES - 1) // CHUNK_PAGES
|
||
_push_step(r, job_id, "ai", f"Large section: splitting into {n_chunks} chunks")
|
||
for chunk_idx in range(1, n_chunks + 1):
|
||
start_p = section.start_page + (chunk_idx - 1) * CHUNK_PAGES
|
||
end_p = min(start_p + CHUNK_PAGES - 1, section.end_page)
|
||
content = vector_service.get_pages_text(section.document_id, start_p, end_p)
|
||
if not content or len(content.strip()) < 100:
|
||
_push_step(r, job_id, "ai", f"Chunk {chunk_idx}/{n_chunks}: no text, skipping")
|
||
continue
|
||
_push_step(r, job_id, "ai", f"Chunk {chunk_idx}/{n_chunks}: pages {start_p}–{end_p}…")
|
||
cards = extraction_modes.generate_flashcards(
|
||
content, f"{start_p}–{end_p}", start_p, ai_model_id, ai_api_key,
|
||
)
|
||
all_cards.extend(cards)
|
||
_push_step(r, job_id, "ai", f"Chunk {chunk_idx}/{n_chunks}: {len(cards)} cards")
|
||
|
||
if not all_cards:
|
||
r.set(f"extraction:status:{job_id}", "failed", ex=EXPIRE_SECONDS)
|
||
_push_step(r, job_id, "error", "No flashcards could be generated")
|
||
return
|
||
|
||
# Refresh DB connection for save phase
|
||
from sqlalchemy import text as _text
|
||
try:
|
||
db.execute(_text("SELECT 1"))
|
||
except Exception:
|
||
db.rollback()
|
||
db.close()
|
||
db = SessionLocal()
|
||
|
||
_push_step(r, job_id, "save", f"Saving {len(all_cards)} flashcards…")
|
||
|
||
from app.models.flashcard import FlashcardDeck, Flashcard
|
||
deck = FlashcardDeck(
|
||
title=title,
|
||
section_id=section_id,
|
||
user_id=user_id,
|
||
card_count=len(all_cards),
|
||
)
|
||
db.add(deck)
|
||
db.flush()
|
||
|
||
for c in all_cards:
|
||
card = Flashcard(
|
||
deck_id=deck.id,
|
||
front=c["front"],
|
||
back=c["back"],
|
||
page_reference=c.get("page_reference"),
|
||
)
|
||
db.add(card)
|
||
|
||
db.commit()
|
||
r.set(f"extraction:status:{job_id}", "completed", ex=EXPIRE_SECONDS)
|
||
r.set(f"extraction:deck_id:{job_id}", str(deck.id), ex=EXPIRE_SECONDS)
|
||
_push_step(r, job_id, "done", f"Created deck '{title}' with {len(all_cards)} cards")
|
||
|
||
except Exception as e:
|
||
logger.exception(f"Flashcard generation failed: {e}")
|
||
r.set(f"extraction:status:{job_id}", "failed", ex=EXPIRE_SECONDS)
|
||
r.set(f"extraction:error:{job_id}", str(e)[:500], ex=EXPIRE_SECONDS)
|
||
_push_step(r, job_id, "error", f"Failed: {str(e)[:200]}")
|
||
try:
|
||
db.rollback()
|
||
except Exception:
|
||
pass
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
#: The three readings of a topic the reader offers, and what each one is for.
|
||
#: Written into the prompt because a model that is not told about them writes
|
||
#: one article and the other two tabs stay empty — which is what happened to
|
||
#: every generated article until now.
|
||
#: Room for the whole thing. An article with a long view, a high-yield view and
|
||
#: a clinical one runs well past four thousand tokens, and a reply that stops
|
||
#: mid-string is not a JSON document.
|
||
ARTICLE_MAX_TOKENS = 16000
|
||
|
||
|
||
class ArticleDraftError(RuntimeError):
|
||
"""A refusal with a sentence the educator can act on."""
|
||
|
||
|
||
def _why(exc: Exception) -> str:
|
||
"""One sentence a person can act on, out of whatever the provider said.
|
||
|
||
A proxy answers a missing model with three hundred characters of JSON. The
|
||
part that matters is which model, and that it does not exist — everything
|
||
else belongs in the log.
|
||
"""
|
||
text = str(exc)
|
||
found = re.search(r"Invalid model name passed in model=([\w.:-]+)", text)
|
||
if found:
|
||
return (f"The model configured for this job — {found.group(1)} — is not on "
|
||
"the AI proxy any more. Pick another in Admin → Models.")
|
||
if "429" in text or "rate limit" in text.lower():
|
||
return "The AI provider is rate-limiting us. Try again in a minute."
|
||
if "timed out" in text.lower() or "timeout" in text.lower():
|
||
return "The model took too long to answer."
|
||
return text[:160]
|
||
|
||
|
||
def _draft_grounding(r, job_id: str, topic: str, use_library: bool, use_pubmed: bool):
|
||
"""Source material for a draft, and the references that come with it.
|
||
|
||
Both lookups are optional, both are best-effort, and neither can fail the
|
||
draft: an educator who ticks a box and gets nothing is told so on the
|
||
progress line and still gets their article.
|
||
|
||
Returned as a block to append to the prompt, so the prompt itself — which
|
||
is working — does not change shape when a source is added or removed.
|
||
"""
|
||
from app.services import clinical_corpus, pubmed
|
||
|
||
blocks: list[str] = []
|
||
references: list[str] = []
|
||
|
||
if use_library:
|
||
_push_step(r, job_id, "ai", "Reading the clinical library…")
|
||
found = clinical_corpus.search(topic)
|
||
if found["results"]:
|
||
blocks.append(
|
||
"Excerpts from this institution's clinical library, for you to write "
|
||
"from. Use them for the facts, the structure and the emphasis, and "
|
||
"write the article in your own words — do not copy sentences out of "
|
||
"them. They are reference material, not a draft.\n\n"
|
||
+ clinical_corpus.for_prompt(found["results"]))
|
||
_push_step(r, job_id, "ai",
|
||
f"{len(found['results'])} excerpts from the library.")
|
||
else:
|
||
_push_step(r, job_id, "ai", f"Library: {found['reason']}.")
|
||
|
||
if use_pubmed:
|
||
_push_step(r, job_id, "ai", "Searching PubMed…")
|
||
found = pubmed.search(topic)
|
||
if found["results"]:
|
||
blocks.append(
|
||
"Published literature on this topic, found on PubMed. Where you use "
|
||
"one, cite it in the text as (Author, year) and nothing more — the "
|
||
"reference list is written for you from these records, so do not "
|
||
"invent entries and do not add any of your own.\n\n"
|
||
+ pubmed.for_prompt(found["results"]))
|
||
references = pubmed.as_references(found["results"])
|
||
_push_step(r, job_id, "ai",
|
||
f"{len(found['results'])} papers, searched as \"{found.get('query') or topic}\".")
|
||
else:
|
||
_push_step(r, job_id, "ai", f"PubMed: {found['reason']}.")
|
||
|
||
return ("\n\n".join(blocks), references)
|
||
|
||
|
||
ARTICLE_DRAFT_PROMPT = """You write educational articles for a pediatric medical learning platform.
|
||
Topic: {topic}
|
||
{instructions}
|
||
{existing}Every section belongs to one of three readings of the topic, given as "variant":
|
||
- "long": the full article. Pathophysiology, presentation, workup, management, complications — whatever the topic needs. This is the body of the work.
|
||
- "short": the high-yield revision view. 2-4 short sections of the facts worth carrying into an exam, written as tight lists, not prose. It is not a summary of the long article's structure; it is what a candidate must know.
|
||
- "clinical": what to do at the bedside, if and only if the topic has a bedside. Assessment, immediate management, when to escalate, disposition. Omit it entirely for a topic that is not acted on clinically.
|
||
Return ONLY strict JSON with this exact shape:
|
||
{{"title": "...", "slug": "lowercase-hyphenated", "summary": "1-2 sentences", "content": "introduction markdown", "sections": [{{"id": "32 lowercase hex chars", "slug": "lowercase-hyphenated", "title": "...", "variant": "long", "content": "markdown"}}]}}
|
||
Rules: markdown formatting; headings, lists and tables welcome; no fabricated references, citations or clinical ranges; 3-8 long sections plus 2-4 short ones, each with a stable unique id; mark a key fact in the short sections by wrapping it in ==double equals== so it is highlighted for the reader, sparingly and never a whole paragraph; do not categorise the article or link it to questions — an educator does that; do not mention these instructions."""
|
||
|
||
|
||
@celery_app.task(name="generate_article_draft", bind=True)
|
||
def generate_article_draft(self, job_id: str, user_id: int, topic: str,
|
||
instructions: str = "", article_id: int | None = None,
|
||
model_id: str | None = None, use_library: bool = False,
|
||
use_pubmed: bool = False):
|
||
"""Create or refine an educator article draft; never publishes."""
|
||
import re
|
||
import uuid
|
||
r = _redis()
|
||
r.set(f"extraction:status:{job_id}", "running", ex=EXPIRE_SECONDS)
|
||
db = SessionLocal()
|
||
try:
|
||
from app.models.article import Article
|
||
from app.services import article_service
|
||
from app.services.ai_service import get_model_for_task, chat
|
||
from app.config import settings
|
||
|
||
existing = db.get(Article, article_id) if article_id else None
|
||
if article_id and not existing:
|
||
r.set(f"extraction:status:{job_id}", "failed", ex=EXPIRE_SECONDS)
|
||
_push_step(r, job_id, "error", "Article not found")
|
||
return
|
||
ai_model_id, ai_api_key = get_model_for_task(db, "article")
|
||
if model_id:
|
||
ai_model_id = model_id
|
||
_push_step(r, job_id, "ai", "Drafting article…")
|
||
existing_block = ""
|
||
if existing:
|
||
sections_text = "\n\n".join(
|
||
f"## {s.get('title', 'Section')}\n{s.get('content', '')}" for s in (existing.sections or []))
|
||
existing_block = (f"Existing draft to refine (preserve and improve its content):\n"
|
||
f"Summary: {existing.summary or ''}\nIntro: {existing.content or ''}\n"
|
||
f"{sections_text}\n\n")
|
||
prompt = ARTICLE_DRAFT_PROMPT.format(
|
||
topic=topic,
|
||
instructions=f"Refine this existing draft: {existing.title}\n{instructions}" if existing else instructions or "",
|
||
existing=existing_block,
|
||
)
|
||
# Anything the educator asked us to draw on, added after the prompt
|
||
# rather than woven into it: a draft with nothing to draw on is exactly
|
||
# the draft that has been working, and this is the only way to keep
|
||
# that true as sources are added.
|
||
grounding, references = _draft_grounding(r, job_id, topic, use_library, use_pubmed)
|
||
if grounding:
|
||
prompt = f"{prompt}\n\n{grounding}"
|
||
# 4,000 was the cap, and it is what broke this: the prompt asks for a
|
||
# full article *plus* a high-yield view plus a clinical one, the reply
|
||
# ran past the ceiling, and `json.loads` failed on a string the model
|
||
# never got to close — reported to the educator as "the model may need
|
||
# an 'article' configuration", which was nothing to do with it.
|
||
raw = chat(
|
||
model=ai_model_id, messages=[{"role": "user", "content": prompt}],
|
||
max_tokens=ARTICLE_MAX_TOKENS, temperature=0.4, api_key=ai_api_key).strip()
|
||
if raw.startswith("```"):
|
||
raw = raw.split("\n", 1)[1] if "\n" in raw else raw[3:]
|
||
if raw.endswith("```"):
|
||
raw = raw[:-3]
|
||
raw = raw.strip()
|
||
try:
|
||
data = json.loads(raw)
|
||
except json.JSONDecodeError as broken:
|
||
# A truncated reply is a different problem from a badly configured
|
||
# one, and telling an educator to check their model settings when
|
||
# the model worked fine is how an afternoon is wasted.
|
||
truncated = not raw.rstrip().endswith("}")
|
||
raise ArticleDraftError(
|
||
"The model's reply was cut off before it finished the article. "
|
||
"Try a narrower topic, or ask for fewer sections in the instructions."
|
||
if truncated else
|
||
f"The model did not return usable JSON ({broken})."
|
||
) from broken
|
||
title = str(data.get("title", topic)).strip()[:300]
|
||
slug = re.sub(r"[^a-z0-9]+", "-", str(data.get("slug", topic)).strip().lower()).strip("-")[:120] or "topic"
|
||
sections = []
|
||
for section in data.get("sections", []):
|
||
section_id = str(section.get("id") or "").strip().lower()
|
||
if not re.fullmatch(r"[0-9a-f]{32}", section_id):
|
||
section_id = uuid.uuid4().hex
|
||
# An unknown or missing variant is the full article, which is what
|
||
# every section written before the model was told about the three
|
||
# readings already is.
|
||
variant = str(section.get("variant") or "long").strip().lower()
|
||
if variant not in ("short", "long", "clinical"):
|
||
variant = "long"
|
||
sections.append({
|
||
"id": section_id,
|
||
"slug": re.sub(r"[^a-z0-9]+", "-", str(section.get("slug", "section")).strip().lower()).strip("-")[:120] or "section",
|
||
"title": str(section.get("title", "Section")).strip()[:300] or "Section",
|
||
"variant": variant,
|
||
"content": str(section.get("content", "")),
|
||
})
|
||
if existing:
|
||
existing.title, existing.slug, existing.summary = title, slug, str(data.get("summary", "") or "")[:2000]
|
||
existing.content, existing.sections = str(data.get("content", "") or ""), sections
|
||
article = existing
|
||
else:
|
||
base_slug = slug
|
||
n = 2
|
||
while db.query(Article.id).filter(Article.slug == slug).first():
|
||
slug = f"{base_slug}-{n}"
|
||
n += 1
|
||
# Stamped with the model that wrote it. The column existed and
|
||
# nothing set it, so an AI draft was indistinguishable from one a
|
||
# person typed — and the editorial queue that lists generated
|
||
# drafts never showed a single one of them.
|
||
article = Article(slug=slug, title=title, summary=str(data.get("summary", "") or "")[:2000],
|
||
content=str(data.get("content", "") or ""), sections=sections,
|
||
user_id=user_id, status="draft",
|
||
generated_by=str(ai_model_id or "ai")[:80])
|
||
db.add(article)
|
||
# The reference list is written from the records rather than by the
|
||
# model, which is the whole reason for searching PubMed rather than
|
||
# asking it what the literature says: every line here is a paper that
|
||
# exists, with a PMID somebody can look up.
|
||
if references:
|
||
kept = list(article.references_json or [])
|
||
for line in references:
|
||
if line not in kept:
|
||
kept.append(line)
|
||
article.references_json = kept
|
||
db.commit()
|
||
# A draft that is not indexed is a draft nobody can find. Every writer of
|
||
# `Article.sections` has to do this; the ones that did not left 323
|
||
# articles with no section rows and a vector built from the title alone.
|
||
article_service.reindex(db, article)
|
||
# The id, so whoever asked for the draft can be taken straight to it.
|
||
r.set(f"extraction:article:{job_id}", str(article.id), ex=EXPIRE_SECONDS)
|
||
r.set(f"extraction:status:{job_id}", "completed", ex=EXPIRE_SECONDS)
|
||
_push_step(r, job_id, "done", f"Draft saved: {title}")
|
||
except ArticleDraftError as refusal:
|
||
logger.warning("Article draft job %s: %s", job_id, refusal)
|
||
r.set(f"extraction:status:{job_id}", "failed", ex=EXPIRE_SECONDS)
|
||
r.set(f"extraction:error:{job_id}", str(refusal)[:300], ex=EXPIRE_SECONDS)
|
||
_push_step(r, job_id, "error", str(refusal)[:300])
|
||
except Exception as exc:
|
||
logger.warning("Article draft job %s failed: %s", job_id, exc)
|
||
r.set(f"extraction:status:{job_id}", "failed", ex=EXPIRE_SECONDS)
|
||
r.set(f"extraction:error:{job_id}", str(exc)[:300], ex=EXPIRE_SECONDS)
|
||
_push_step(r, job_id, "error", f"Drafting failed: {str(exc)[:200]}")
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
@celery_app.task(name="generate_article_cards", bind=True)
|
||
def generate_article_cards(self, job_id: str, user_id: int, article_id: int,
|
||
model_id: str | None = None):
|
||
"""Generate cards from an article into an unshared educator deck; links stay private until shared."""
|
||
r = _redis()
|
||
r.set(f"extraction:status:{job_id}", "running", ex=EXPIRE_SECONDS)
|
||
db = SessionLocal()
|
||
try:
|
||
from app.models.article import Article
|
||
from app.models.flashcard import Flashcard, FlashcardDeck, FlashcardArticleLink
|
||
from app.services import extraction_modes
|
||
from app.services.ai_service import get_model_for_task
|
||
|
||
article = db.get(Article, article_id)
|
||
if not article:
|
||
r.set(f"extraction:status:{job_id}", "failed", ex=EXPIRE_SECONDS)
|
||
_push_step(r, job_id, "error", "Article not found")
|
||
return
|
||
ai_model_id, ai_api_key = get_model_for_task(db, "flashcard")
|
||
if model_id:
|
||
ai_model_id = model_id
|
||
_push_step(r, job_id, "ai", f"Generating cards from {article.title}…")
|
||
content = "\n\n".join(filter(None, [
|
||
article.title, article.summary, article.content,
|
||
*[f"## {s['title']}\n{s['content']}" for s in (article.sections or [])],
|
||
]))
|
||
# As many cards as the article has substance for. Fifteen is the
|
||
# per-chunk default, and an article is one chunk however long it is —
|
||
# so a ten-section piece and a two-paragraph stub asked for the same
|
||
# fifteen. Roughly a card per 150 words, floored so a short article
|
||
# still makes a usable deck and capped so one call stays inside the
|
||
# model's output.
|
||
words = len(content.split())
|
||
wanted = max(12, min(30, round(words / 150)))
|
||
cards = extraction_modes.generate_flashcards(
|
||
content, "article", None, ai_model_id, ai_api_key, n=wanted)
|
||
if not cards:
|
||
r.set(f"extraction:status:{job_id}", "failed", ex=EXPIRE_SECONDS)
|
||
_push_step(r, job_id, "error", "No cards could be generated")
|
||
return
|
||
deck = db.query(FlashcardDeck).filter(
|
||
FlashcardDeck.title == f"Cards: {article.title}",
|
||
FlashcardDeck.user_id == user_id,
|
||
FlashcardDeck.deleted_at.is_(None),
|
||
).first()
|
||
if not deck:
|
||
# Filed where the article is filed. A deck written from an
|
||
# article belongs to the same topic, and asking somebody to
|
||
# choose the category again is asking them to repeat a fact
|
||
# the system already knows.
|
||
deck = FlashcardDeck(title=f"Cards: {article.title}", user_id=user_id,
|
||
category_id=article.category_id, card_count=0, is_shared=0)
|
||
db.add(deck)
|
||
db.flush()
|
||
new_cards = []
|
||
for card in cards:
|
||
item = Flashcard(deck_id=deck.id, front=card["front"], back=card["back"],
|
||
page_reference=card.get("page_reference"))
|
||
db.add(item)
|
||
new_cards.append(item)
|
||
db.flush()
|
||
for item in new_cards:
|
||
db.add(FlashcardArticleLink(flashcard_id=item.id, article_id=article.id))
|
||
deck.card_count = db.query(Flashcard).filter(Flashcard.deck_id == deck.id).count()
|
||
db.commit()
|
||
r.set(f"extraction:status:{job_id}", "completed", ex=EXPIRE_SECONDS)
|
||
_push_step(r, job_id, "done", f"{len(cards)} cards saved to private deck {deck.title}")
|
||
except Exception as exc:
|
||
logger.warning("Article cards job %s failed: %s", job_id, exc)
|
||
db.rollback()
|
||
r.set(f"extraction:status:{job_id}", "failed", ex=EXPIRE_SECONDS)
|
||
r.set(f"extraction:error:{job_id}", str(exc)[:300], ex=EXPIRE_SECONDS)
|
||
# The reason, on the line somebody actually reads. "Card generation
|
||
# failed." was all the jobs panel showed for a fortnight while the
|
||
# answer — a model the proxy had dropped — sat in a field nothing
|
||
# displayed.
|
||
_push_step(r, job_id, "error", f"Card generation failed. {_why(exc)}")
|
||
finally:
|
||
db.close()
|