3418ed023b
Three things landed together; the message names all of them, because a commit that mentions one is a commit nobody finds the other two in. **Figures.** Thirty-four JPEG 2000 files — 21 on questions, the rest unattached in the media library — are WebP now, with `questions.image_path`, `questions.explanation_image_path` and `media_assets.path` repointed together. Serving already converted them on the way out, so nothing was broken; this removes the step and makes what is stored the same thing that is served. The originals stay: they are the only copy of what came out of the PDF, they cost a few megabytes between them, and a conversion nobody can undo is not one to run against a live bank. Paths are found by what the columns say rather than by listing a bucket, because three tables record them and updating two would be worse than none. **The openai SDK is gone.** Ten call sites — one more than the map said, the Celery article drafter — every one of them a POST with a JSON body, and not one reading usage, cost, tool calls or logprobs. Every other call to the same proxy was already plain httpx: embeddings, the ChromaDB embedding function, speech both ways, model discovery, the vision probe. So this deletes an abstraction rather than swapping one for another, and leaves one HTTP client instead of two. `chat()` and `achat()` return the message content; a `ProxyError` carries the status and the first 500 characters of the body, which is where the proxy explains itself. Behaviour is preserved deliberately, including a 600-second fallback timeout for the four call sites that were running on the SDK's ten-minute default. Lowering that is a real change and belongs in its own commit. Proved against the live proxy on both services rather than only against mocks: a completion, an async completion, a real 400 the vision probe still classifies as a refusal, 407 models read from the catalogue, and a word read off an image. **Voice.** A chosen voice is honoured whatever serves it. The prefix check only accepted a locally served one, so a site adding a hosted voice would offer it in Settings, save the learner's choice, and then quietly read every question in the default voice. The list has always come from the database — adding a voice is a row in Settings → AI models, never a code change. And the sign-in page stops offering a locked door: `signup-policy` reports whether registration is open at all, and the Sign up link goes when it is not. The switch existed and the only way to discover it was to fill the form in. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TqXevQJhxFrM7jJg82cgZN
741 lines
34 KiB
Python
741 lines
34 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)
|
||
# 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()
|
||
|
||
|
||
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 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,
|
||
)
|
||
raw = chat(
|
||
model=ai_model_id, messages=[{"role": "user", "content": prompt}],
|
||
max_tokens=4000, 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()
|
||
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
|
||
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 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()
|