shopdb-flask/plugins/knowledgebase/api/routes.py

"""Knowledge Base API endpoints."""

from flask import Blueprint, request

from shopdb.api import (
    intarg,
    db,
    Application,
    success_response,
    error_response,
    paginated_response,
    ErrorCodes,
    get_pagination_params,
    paginate_query,
)

from ..models import KnowledgeBase

from shopdb.api import require_permission, apply_import_timestamps, intarg

knowledgebase_bp = Blueprint('knowledgebase', __name__)


def _visible_articles():
    """Active articles whose topic is not a retired application.

    An article about a decommissioned application is not something anyone should
    find by browsing or searching: it describes a thing that is no longer in
    service, and presenting it alongside live documentation reads as though it
    were current.

    An article with NO topic still shows. Not every article is about an
    application, and a null topic is not a retired one.

    Expressed as a subquery rather than a join because the topic sort below joins
    Application itself, and two joins onto the same table in one query collide.
    """
    retired = db.session.query(Application.appid).filter(
        Application.isactive.is_(False))
    return KnowledgeBase.query.filter(
        KnowledgeBase.isactive.is_(True),
        db.or_(KnowledgeBase.appid.is_(None),
               KnowledgeBase.appid.notin_(retired)))


def _search_clause(search):
    """Articles containing EVERY word of the search, each word in the title, the
    keywords, or the topic's name, in any order.

    A single `ilike('%CMM Community%')` needs the whole phrase contiguous in ONE
    field. An article tagged 'community' under the CMM topic has 'CMM' only in
    its topic's name and 'Community' only in its keywords, so it matched nothing
    and the search came back empty. Splitting the query and requiring each word
    somewhere is what people mean when they type two words.

    Kept in step with `_word_match` in shopdb/core/api/search.py, which does the
    same job for global search. Two copies because the core helper is internal
    and not on the plugin contract surface (ADR-001).

    Active applications only for the topic. A retired application is not a topic
    anyone should be offered: matching its name surfaced its articles and printed
    the retired app as their subject, which reads as though it were still in
    service.
    """
    words = [w for w in search.split() if w] or ['']
    return db.and_(*[
        db.or_(
            KnowledgeBase.shortdescription.ilike(f'%{w}%'),
            KnowledgeBase.keywords.ilike(f'%{w}%'),
            KnowledgeBase.appid.in_(
                db.session.query(Application.appid).filter(
                    Application.appname.ilike(f'%{w}%'),
                    Application.isactive.is_(True))),
        )
        for w in words
    ])


@knowledgebase_bp.route('', methods=['GET'])
@require_permission('kb.view')
def list_articles():
    """List all knowledge base articles."""
    page, per_page = get_pagination_params(request)

    query = _visible_articles()

    # The topic is matched via an appid subquery instead of a join so it does not
    # collide with the sort=='topic' join below; articles with no app just miss
    # that clause and still match on title/keywords.
    if search := request.args.get('search'):
        query = query.filter(_search_clause(search))

    # Filter by topic/application
    if (appid := intarg('appid')) is not None:
        query = query.filter(KnowledgeBase.appid == appid)

    # Exact-match natural-key lookups for idempotent import. linkurl is the
    # stable natural key; shortdescription (the title) is offered as a fallback.
    if exactlinkurl := request.args.get('linkurl'):
        query = query.filter(KnowledgeBase.linkurl == exactlinkurl)
    if exacttitle := request.args.get('shortdescription'):
        query = query.filter(KnowledgeBase.shortdescription == exacttitle)

    # Sort options
    sort = request.args.get('sort', 'clicks')
    order = request.args.get('order', 'desc')

    if sort == 'clicks':
        query = query.order_by(
            KnowledgeBase.clicks.desc() if order == 'desc' else KnowledgeBase.clicks.asc(),
            KnowledgeBase.lastupdated.desc()
        )
    elif sort == 'topic':
        query = query.join(Application).order_by(
            Application.appname.desc() if order == 'desc' else Application.appname.asc()
        )
    elif sort == 'description':
        query = query.order_by(
            KnowledgeBase.shortdescription.desc() if order == 'desc' else KnowledgeBase.shortdescription.asc()
        )
    elif sort == 'lastupdated':
        query = query.order_by(
            KnowledgeBase.lastupdated.desc() if order == 'desc' else KnowledgeBase.lastupdated.asc()
        )
    else:
        query = query.order_by(KnowledgeBase.clicks.desc())

    items, total = paginate_query(query, page, per_page)
    data = []
    for article in items:
        article_dict = article.to_dict()
        if article.application:
            article_dict['application'] = {
                'appid': article.application.appid,
                'appname': article.application.appname
            }
        else:
            article_dict['application'] = None
        data.append(article_dict)

    return paginated_response(data, page, per_page, total)


@knowledgebase_bp.route('/stats', methods=['GET'])
@require_permission('kb.view')
def get_stats():
    """Get knowledge base statistics."""
    # Counted over the same set the list shows. A total that includes articles
    # nobody can see is a total nobody can reconcile.
    visible = _visible_articles()
    total_clicks = sum(article.clicks or 0 for article in visible)
    total_articles = visible.count()

    return success_response({
        'totalclicks': int(total_clicks),
        'totalarticles': total_articles
    })


@knowledgebase_bp.route('/<int:link_id>', methods=['GET'])
@require_permission('kb.view')
def get_article(link_id: int):
    """Get a single knowledge base article."""
    article = db.session.get(KnowledgeBase, link_id)

    if not article or not article.isactive:
        return error_response(ErrorCodes.NOT_FOUND, 'Article not found', http_code=404)

    data = article.to_dict()
    if article.application:
        data['application'] = {
            'appid': article.application.appid,
            'appname': article.application.appname
        }
    else:
        data['application'] = None

    return success_response(data)


@knowledgebase_bp.route('/<int:link_id>/click', methods=['POST'])
@require_permission('kb.view')
def track_click(link_id: int):
    """Increment the click counter for an article the caller is allowed to read.

    Gated on kb.view like every other read of this table: an anonymous caller
    walking link ids must not be able to probe which articles exist, drive the
    counters that order the list, or spend a DB transaction per request.

    Restricted to the same visible set as the list/detail reads so an article
    about a retired application is not reachable here either.

    The linkurl is deliberately not echoed back: whoever is permitted to click
    already received it in the list or detail response the click came from.
    """
    article = _visible_articles().filter(KnowledgeBase.linkid == link_id).first()

    if not article:
        return error_response(ErrorCodes.NOT_FOUND, 'Article not found', http_code=404)

    article.increment_clicks()
    db.session.commit()

    return success_response({'clicks': article.clicks})


@knowledgebase_bp.route('', methods=['POST'])
@require_permission('kb.create')
def create_article():
    """Create a new knowledge base article."""
    data = request.get_json()

    if not data or not data.get('shortdescription'):
        return error_response(ErrorCodes.VALIDATION_ERROR, 'shortdescription is required')

    if not data.get('linkurl'):
        return error_response(ErrorCodes.VALIDATION_ERROR, 'linkurl is required')

    # Validate application if provided
    if data.get('appid'):
        app = db.session.get(Application, data['appid'])
        if not app:
            return error_response(ErrorCodes.NOT_FOUND, 'Application not found', http_code=404)

    article = KnowledgeBase(
        shortdescription=data['shortdescription'],
        linkurl=data['linkurl'],
        appid=data.get('appid'),
        keywords=data.get('keywords'),
        clicks=0
    )

    db.session.add(article)
    apply_import_timestamps(article, data)
    db.session.commit()

    return success_response(article.to_dict(), message='Article created', http_code=201)


@knowledgebase_bp.route('/<int:link_id>', methods=['PUT'])
@require_permission('kb.edit')
def update_article(link_id: int):
    """Update a knowledge base article."""
    article = db.session.get(KnowledgeBase, link_id)

    if not article:
        return error_response(ErrorCodes.NOT_FOUND, 'Article not found', http_code=404)

    data = request.get_json()
    if not data:
        return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')

    # Validate application if being changed
    if 'appid' in data and data['appid']:
        app = db.session.get(Application, data['appid'])
        if not app:
            return error_response(ErrorCodes.NOT_FOUND, 'Application not found', http_code=404)

    fields = ['shortdescription', 'linkurl', 'appid', 'keywords', 'isactive']
    for key in fields:
        if key in data:
            setattr(article, key, data[key])

    apply_import_timestamps(article, data)
    db.session.commit()
    return success_response(article.to_dict(), message='Article updated')


@knowledgebase_bp.route('/<int:link_id>', methods=['DELETE'])
@require_permission('kb.delete')
def delete_article(link_id: int):
    """Delete (deactivate) a knowledge base article."""
    article = db.session.get(KnowledgeBase, link_id)

    if not article:
        return error_response(ErrorCodes.NOT_FOUND, 'Article not found', http_code=404)

    article.isactive = False
    db.session.commit()

    return success_response(message='Article deleted')