259 lines
8.7 KiB
Python
259 lines
8.7 KiB
Python
"""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')
|