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

"""Knowledge Base API endpoints."""

from flask import Blueprint, request
from flask_jwt_extended import jwt_required

from shopdb.api import (
    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

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)))


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

    query = _visible_articles()

    # Search: title, keywords, and the topic (its Application's name). 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'):
        like = f'%{search}%'
        # Active applications only. 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.
        topic_appids = db.session.query(Application.appid).filter(
            Application.appname.ilike(like),
            Application.isactive.is_(True))
        query = query.filter(
            db.or_(
                KnowledgeBase.shortdescription.ilike(like),
                KnowledgeBase.keywords.ilike(like),
                KnowledgeBase.appid.in_(topic_appids)
            )
        )

    # Filter by topic/application
    if appid := request.args.get('appid'):
        query = query.filter(KnowledgeBase.appid == int(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'])
@jwt_required(optional=True)
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'])
@jwt_required(optional=True)
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'])
@jwt_required(optional=True)
def track_click(link_id: int):
    """Increment click counter and return the URL to redirect to."""
    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)

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

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


@knowledgebase_bp.route('', methods=['POST'])
@jwt_required()
@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'])
@jwt_required()
@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'])
@jwt_required()
@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')