ebcc11ae12
claims to hold them The scaffolding is down. 203 subject, 2,275 disease and 4,281 keyword tags, and 25,356 links, deleted — backed up first to a 1.9MB JSON of replayable rows, because "we can always put it back" should be true rather than said. The 16 system rows stay: categories point at them. With them go the things that only existed to feed them — the classify_questions task, its snapshot helpers, POST /tags/classify and its status poll — and the three Taxonomy tabs that would now always read zero. A tab showing 0 forever teaches people the page is broken. The organ-system filter in the session builder moved onto categories with the rest, including everything beneath a matched topic, so it groups the way the analysis does. Registration: `settings:registration_enabled` was set to false, and there was no switch anywhere on the site to set it back. The API had always accepted it; the Site policy page had never shown it. So the site could be closed to new members with the admin looking at three switches, all correct, and no way to see the one that was actually refusing them. It is now the first switch on that page, and says plainly that the ones below it have nothing to act on while it is off. The SSO-only flag was hidden the same way and is shown when SSO is configured. Deleting a topic no longer silently unfiles its questions. It asks where they go, and says how many are waiting, unless the topic is empty — the same rule promotion now follows. Its extra category links move too, minus any that would duplicate a pair the destination already has. Back links: Trash, Extraction jobs, Taxonomy and the Handbook had none at all, and Access pointed at the wrong section. They are one component now, each returning one step to the section it was opened from. Editorial has its own entry in the section bar, so its Tools card is gone rather than being a second door to the same room. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TqXevQJhxFrM7jJg82cgZN
721 lines
33 KiB
Python
721 lines
33 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 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()
|
||
|
||
|
||
@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)
|
||
if embedded_total:
|
||
db.commit()
|
||
logger.info("Backfilled %s embeddings (%s pending)", embedded_total, pending_total)
|
||
return {"pending": pending_total, "embedded": embedded_total, "model": active}
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
@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()
|
||
|
||
|
||
ARTICLE_DRAFT_PROMPT = """You write educational articles for a pediatric medical learning platform.
|
||
Topic: {topic}
|
||
{instructions}
|
||
{existing}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": "...", "content": "markdown"}}]}}
|
||
Rules: markdown formatting; headings, lists and tables welcome; no fabricated references, citations or clinical ranges; keep 2-6 sections with stable unique ids; 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.ai_service import get_model_for_task, _proxy_model
|
||
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,
|
||
)
|
||
import litellm
|
||
kwargs = {"model": _proxy_model(ai_model_id), "messages": [{"role": "user", "content": prompt}],
|
||
"max_tokens": 4000, "temperature": 0.4}
|
||
if ai_api_key:
|
||
kwargs["api_key"] = ai_api_key
|
||
elif settings.LITELLM_API_KEY:
|
||
kwargs["api_key"] = settings.LITELLM_API_KEY
|
||
if settings.LITELLM_API_BASE:
|
||
kwargs["api_base"] = settings.LITELLM_API_BASE
|
||
response = litellm.completion(**kwargs)
|
||
raw = response.choices[0].message.content.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()
|
||
data = json.loads(raw)
|
||
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
|
||
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",
|
||
"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
|
||
else:
|
||
base_slug = slug
|
||
n = 2
|
||
while db.query(Article.id).filter(Article.slug == slug).first():
|
||
slug = f"{base_slug}-{n}"
|
||
n += 1
|
||
db.add(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.commit()
|
||
r.set(f"extraction:status:{job_id}", "completed", ex=EXPIRE_SECONDS)
|
||
_push_step(r, job_id, "done", f"Draft saved: {title}")
|
||
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", "Drafting failed; the model may need an 'article' configuration.")
|
||
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()
|