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, require_role

knowledgebase_bp = Blueprint('knowledgebase', __name__)


@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 = KnowledgeBase.query.filter_by(isactive=True)

    # Search
    if search := request.args.get('search'):
        query = query.filter(
            db.or_(
                KnowledgeBase.shortdescription.ilike(f'%{search}%'),
                KnowledgeBase.keywords.ilike(f'%{search}%')
            )
        )

    # Filter by topic/application
    if appid := request.args.get('appid'):
        query = query.filter(KnowledgeBase.appid == int(appid))

    # 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."""
    total_clicks = db.session.query(
        db.func.coalesce(db.func.sum(KnowledgeBase.clicks), 0)
    ).filter(KnowledgeBase.isactive == True).scalar()

    total_articles = KnowledgeBase.query.filter_by(isactive=True).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)
    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])

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