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

"""An article's standing claim over a category, and the links it produces.

The question this answers, which came up the moment topic-linking existed:
*if the Cardiology article covers the Cardiology questions, does it cover the
one filed there tomorrow?*

It does. "This article covers this topic" is a standing statement about the
material, not a snapshot of who happened to be filed where on the afternoon
somebody pressed the button — and a copy stops being true silently, with
nothing on any screen to say so.

The shape that gets both halves right:

* `question_article_links` stays the **only** table anything reads. Every
  count, the QBank button, the mirror panel on each question, the AI Mode
  curated boost — none of them learn a second question to ask.
* `article_topic_claims` records **why** some of those rows exist, and is the
  one place that writes them: when the claim is staked, when a question is
  filed into the category, and on a sweep that catches whatever the other two
  missed.

A link made this way is an ordinary row and can be deleted one at a time. A
sweep will put it back, which is the honest consequence of a standing claim —
dropping the claim is how you stop it, and that is a sentence the interface can
say out loud.
"""
import logging

from sqlalchemy import or_, select
from sqlalchemy.orm import Session

from app.models.article import Article, ArticleTopicClaim, QuestionArticleLink
from app.models.question import Question
from app.models.question_category import QuestionCategory, QuestionCategoryLink
from app.services.quiz_builder import category_descendants, general_question_predicate

logger = logging.getLogger(__name__)

#: How many links one claim may create in a single pass. A top-level discipline
#: holds hundreds; a runaway claim over the root of the tree should stop and be
#: reported rather than attach the whole bank to one article.
MAX_LINKS_PER_CLAIM = 300


def categories_under(db: Session, category_id: int, include_subtopics: bool) -> set[int]:
    if not include_subtopics:
        return {category_id}
    return category_descendants(db.query(QuestionCategory).all(), {category_id})


def question_ids_in(db: Session, category_ids: set[int]) -> set[int]:
    """Every question filed under these categories, primary or additional.

    Both, because a question's primary category is one fact about it and its
    extra ones are others: a question filed under Cardiology and also tagged
    Emergency Medicine is in both, and an educator claiming either expects it.
    """
    rows = db.query(Question.id).filter(
        general_question_predicate(),
        or_(Question.question_category_id.in_(category_ids),
            Question.id.in_(select(QuestionCategoryLink.question_id).where(
                QuestionCategoryLink.category_id.in_(category_ids)))),
    ).all()
    return {row[0] for row in rows}


def apply(db: Session, claim: ArticleTopicClaim, *, limit: int = MAX_LINKS_PER_CLAIM) -> int:
    """Make the links this claim implies. Returns how many were new."""
    wanted = question_ids_in(db, categories_under(db, claim.category_id, claim.include_subtopics))
    if not wanted:
        return 0
    existing = {row[0] for row in db.query(QuestionArticleLink.question_id).filter(
        QuestionArticleLink.article_id == claim.article_id,
        QuestionArticleLink.section_id == claim.section_id,
        QuestionArticleLink.question_id.in_(wanted)).all()}
    pending = sorted(wanted - existing)[:limit]
    if not pending:
        return 0
    db.bulk_save_objects([
        QuestionArticleLink(question_id=question_id, article_id=claim.article_id,
                            section_id=claim.section_id, user_id=claim.user_id)
        for question_id in pending
    ])
    db.commit()
    return len(pending)


def preview(db: Session, claim_like) -> dict:
    """What staking this claim would do, before it is staked."""
    wanted = question_ids_in(
        db, categories_under(db, claim_like.category_id, claim_like.include_subtopics))
    existing = {row[0] for row in db.query(QuestionArticleLink.question_id).filter(
        QuestionArticleLink.article_id == claim_like.article_id,
        QuestionArticleLink.section_id == claim_like.section_id).all()} if wanted else set()
    return {"total": len(wanted), "already_linked": len(wanted & existing),
            "would_link": len(wanted - existing), "limit": MAX_LINKS_PER_CLAIM}


def claims_for_category(db: Session, category_ids) -> list[ArticleTopicClaim]:
    """Claims that reach any of these categories, ancestors included.

    A claim on Cardiology with subtopics covers a question filed under
    Cardiology › Arrhythmias, so the lookup has to walk *up* from the question's
    category, not down from the claim.
    """
    if not category_ids:
        return []
    parents = {row.id: row.parent_id for row in db.query(QuestionCategory).all()}
    reach = set()
    for category_id in category_ids:
        node = category_id
        seen = set()
        while node is not None and node not in seen:
            seen.add(node)
            reach.add(node)
            node = parents.get(node)
    direct = set(category_ids)
    rows = db.query(ArticleTopicClaim).filter(ArticleTopicClaim.category_id.in_(reach)).all()
    # A claim without subtopics only covers its own category.
    return [claim for claim in rows
            if claim.include_subtopics or claim.category_id in direct]


def link_new_question(db: Session, question: Question) -> int:
    """Link a question to every article whose claim now covers it.

    Called wherever a question's filing changes. Best-effort by design: a
    failure here must not stop somebody saving a question, and the sweep will
    catch whatever this missed.
    """
    try:
        categories = {question.question_category_id} if question.question_category_id else set()
        categories |= {row[0] for row in db.query(QuestionCategoryLink.category_id).filter(
            QuestionCategoryLink.question_id == question.id).all()}
        claims = claims_for_category(db, categories)
        made = 0
        for claim in claims:
            exists = db.query(QuestionArticleLink.id).filter_by(
                question_id=question.id, article_id=claim.article_id,
                section_id=claim.section_id).first()
            if exists:
                continue
            db.add(QuestionArticleLink(question_id=question.id, article_id=claim.article_id,
                                       section_id=claim.section_id, user_id=claim.user_id))
            made += 1
        if made:
            db.commit()
        return made
    except Exception:
        db.rollback()
        logger.warning("Could not apply topic claims to question %s", question.id, exc_info=True)
        return 0


def sweep(db: Session) -> dict:
    """Re-apply every claim. The safety net, not the mechanism.

    Runs periodically because the two live paths — staking a claim, filing a
    question — can both be bypassed: a bulk category change written in SQL, an
    import, a restored article. A claim that is true should stay true without
    anybody remembering to press anything.
    """
    made = 0
    claims = db.query(ArticleTopicClaim).all()
    alive = {row[0] for row in db.query(Article.id).filter(Article.deleted_at.is_(None)).all()}
    for claim in claims:
        if claim.article_id not in alive:
            continue
        made += apply(db, claim)
    return {"claims": len(claims), "links_added": made}