pdf-quiz-generator/backend/app/services/article_writer.py

"""Write a topic article from what the clinical library returns.

The division of labour is deliberate and matches AI Mode: **retrieval supplies
the facts and the provenance; the model supplies the prose.** References are
built from the metadata of the passages that were actually retrieved, never from
the model, so a reference cannot be invented — the same property that makes an
AI Mode citation trustworthy.

What is generated is an original piece of writing grounded in the library, not
an extract from it. A textbook's sentences belong to its publisher; its facts do
not, and an article that reproduced the sentences would be redistribution
whatever produced the copy. The prompt says this plainly, and the length limits
make a long verbatim passage impossible to hide.

Three views come out of one call rather than three, because they are three
readings of one topic and generating them separately lets them contradict each
other on the numbers.
"""
import json
import logging
import re
import uuid

from sqlalchemy.orm import Session

from app.models.article import Article
from app.services import article_service, clinical_library
from app.services.ai_service import _call_model, get_model_for_task

logger = logging.getLogger(__name__)

PASSAGES = 14
# Enough retrieved prose to write from. Below this the model is being asked to
# write a medical article out of fragments, and it will either refuse or invent.
MIN_SOURCE_CHARS = 3000
# Generous for a long article, short enough that a stalled call is noticed in
# minutes rather than discovered hours later with nothing written since.
WRITE_TIMEOUT = 150
# Three views of one topic, the long one about 550 words. Four thousand tokens is
# comfortable for that and keeps each request small enough to be affordable.
WRITE_MAX_TOKENS = 4000
# Long enough to be worth reading, short enough that nobody skims past the point.
LONG_WORDS = 550
SHELF = "Pediatrics"


def slugify(name: str) -> str:
    slug = re.sub(r"[^a-z0-9]+", "-", (name or "").lower()).strip("-")
    return slug[:120] or "topic"


def _prompt(topic: str, passages: list[dict]) -> str:
    sources = "\n\n".join(
        f"[{i + 1}] {p['source'].get('title', 'source')}"
        f"{', p. ' + str(p['source']['page']) if p['source'].get('page') else ''}\n{p['text']}"
        for i, p in enumerate(passages)
    )
    return f"""You are writing a study article about **{topic}** for a paediatric exam-revision platform.

Write it from the reference passages below. They are extracts from standard
textbooks and guidelines. Use them for the facts, the numbers and the structure —
then write the article in your own words. Do not reproduce sentences from the
passages: this is an article about the topic, not an extract from a book.

If the passages do not cover something, leave it out rather than filling the gap
from memory. A gap a reader can see is safer than a confident sentence nobody
can check.

Produce three views of the same topic:

1. "long" — the full article, about {LONG_WORDS} words, in 4 to 7 sections.
   Typical headings: Definition, Epidemiology, Etiology, Pathophysiology,
   Clinical features, Diagnostics, Differential diagnosis, Treatment,
   Complications, Prevention. Use only the ones the passages support.
2. "short" — one section titled "In short", 6 to 10 tight bullets a learner
   could revise from the night before an exam. Every bullet must carry a fact:
   no bullet that only says a topic is important.
3. "clinical" — one or two sections covering management at the bedside:
   what to do, in what order, with drug doses and routes where the passages give
   them. Include units and per-kilogram dosing exactly as stated. If a dose is
   not in the passages, do not state one.

Return ONLY valid JSON, no markdown fence:

{{
  "summary": "one or two sentences, no heading",
  "long": [{{"title": "Definition", "content": "markdown"}}],
  "short": [{{"title": "In short", "content": "- bullet\\n- bullet"}}],
  "clinical": [{{"title": "Management", "content": "markdown"}}]
}}

Markdown may use lists, bold and tables. Do not write citation markers, footnote
numbers or URLs anywhere: the reference list is attached separately.

REFERENCE PASSAGES

{sources}"""


def _sections(blocks: list[dict], variant: str) -> list[dict]:
    """Turn the model's blocks into stable sections of one view."""
    out = []
    for block in blocks or []:
        # The model occasionally returns a bare string where a block was asked
        # for. Skipping it costs one section; letting it through ends the run.
        if not isinstance(block, dict):
            continue
        title = str(block.get("title") or "").strip()
        content = str(block.get("content") or "").strip()
        if not title or not content:
            continue
        out.append({
            # A fresh id per section, because links are made against these and a
            # regenerated article must not silently inherit another's links.
            "id": uuid.uuid4().hex,
            "slug": slugify(title)[:60] or f"section-{len(out) + 1}",
            "title": title[:300],
            "content": content,
            "parent_id": None,
            "variant": variant,
        })
    return out


def _unique_slugs(sections: list[dict]) -> list[dict]:
    """Slugs are unique within an article, and three views repeat titles."""
    seen: set[str] = set()
    for section in sections:
        base = section["slug"]
        candidate, n = base, 2
        while candidate in seen:
            candidate = f"{base}-{n}"
            n += 1
        section["slug"] = candidate
        seen.add(candidate)
    return sections


def write_article(db: Session, topic: str, category_id: int | None = None,
                  user_id: int | None = None) -> dict:
    """Retrieve, write, and store one article as a draft.

    Returns a small report rather than the article, because the caller is a
    background run over hundreds of topics and wants to know what happened.
    """
    passages = clinical_library.search(
        f"{topic} in children: definition, causes, clinical features, diagnosis and management",
        limit=PASSAGES, folder_contains=SHELF)
    source_chars = sum(len(p["text"]) for p in passages)
    if len(passages) < 3 or source_chars < MIN_SOURCE_CHARS:
        # Too little to ground an article. Writing one anyway would produce
        # exactly the confident unverifiable prose this is designed to avoid,
        # and a topic named after a shelf rather than a condition lands here.
        return {"topic": topic, "status": "skipped",
                "reason": f"only {source_chars} chars from {len(passages)} passages",
                "passages": len(passages)}

    model_id, api_key = get_model_for_task(db, "extraction")
    # One topic must not be able to stall a run of five hundred.
    raw = _call_model(_prompt(topic, passages), model_id, api_key,
                      timeout=WRITE_TIMEOUT, max_tokens=WRITE_MAX_TOKENS)
    text = raw.strip()
    if text.startswith("```"):
        text = re.sub(r"^```[a-z]*\n?|```$", "", text).strip()
    try:
        data = json.loads(text)
    except ValueError:
        logger.warning("Article writer returned unparseable JSON for %s", topic)
        return {"topic": topic, "status": "failed", "reason": "model did not return JSON"}
    if not isinstance(data, dict):
        return {"topic": topic, "status": "failed", "reason": "model returned the wrong shape"}

    sections = _unique_slugs([
        *_sections(data.get("long"), "long"),
        *_sections(data.get("short"), "short"),
        *_sections(data.get("clinical"), "clinical"),
    ])
    if not sections:
        return {"topic": topic, "status": "failed", "reason": "no usable sections"}

    slug = slugify(topic)
    existing = db.query(Article).filter(Article.slug == slug).first()
    if existing:
        return {"topic": topic, "status": "exists", "article_id": existing.id}

    article = Article(
        slug=slug, title=topic[:300],
        summary=(data.get("summary") or "").strip()[:2000] or None,
        content=None, sections=sections, category_id=category_id, user_id=user_id,
        # Never published by generation. A person decides that.
        status="draft",
        # References come from the retrieved metadata, not from the model, so a
        # source cannot be invented.
        references_json=clinical_library.references_from(passages),
        generated_by=f"clinical-library:{model_id}",
    )
    from datetime import datetime

    article.generated_at = datetime.utcnow()
    db.add(article)
    db.flush()
    article_service.record_slug(db, article)
    db.commit()
    db.refresh(article)
    return {"topic": topic, "status": "written", "article_id": article.id,
            "sections": len(sections), "references": len(article.references_json or []),
            "passages": len(passages)}