500e8430a5
The rule was in my head and in a commit message; it needed to be in the task. The first run detached sixteen good figures — a tick on a leaf against a July fever, fungal hyphae against a scaly rash, a recessed chin in a two-week-old's notes, an ECG on a tachypnoeic neonate — because a judgement about relevance was allowed to act on a photograph, and the connection between a clinical figure and its question is often indirect. Two changes. The prompt now separates the two kinds of figure and says how each is judged: a clinical one — photograph, radiograph, ultrasound, ECG, fundoscopy, otoscopy, microscopy — is refused only when it is anatomically impossible, and a table, citation, chart or nomogram is refused whenever it is about another subject. And the model must now say which kind it is looking at, so the code can enforce it: "no" on anything clinical becomes "unsure", which flags it for a person instead of detaching it. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TqXevQJhxFrM7jJg82cgZN
1120 lines
53 KiB
Python
1120 lines
53 KiB
Python
"""Async quiz extraction task with step-by-step progress reporting via Redis."""
|
||
import json
|
||
import logging
|
||
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"
|
||
|
||
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
|
||
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]
|
||
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())
|
||
except Exception as exc:
|
||
unreadable += 1
|
||
logger.warning("Figure audit failed for question %s: %s", question.id, exc)
|
||
continue
|
||
|
||
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."""
|
||
|
||
|
||
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):
|
||
"""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,
|
||
)
|
||
# 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
|
||
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")
|
||
db.add(article)
|
||
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)
|
||
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 [])],
|
||
]))
|
||
cards = extraction_modes.generate_flashcards(content, "article", None, ai_model_id, ai_api_key)
|
||
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:
|
||
deck = FlashcardDeck(title=f"Cards: {article.title}", user_id=user_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)
|
||
_push_step(r, job_id, "error", "Card generation failed.")
|
||
finally:
|
||
db.close()
|