calibre-web/cps/tasks_status.py
2023-01-21 15:23:18 +01:00

107 lines
3.9 KiB
Python

# This file is part of the Calibre-Web (https://github.com/janeczku/calibre-web)
# Copyright (C) 2022 OzzieIsaacs
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from markupsafe import escape
from flask import Blueprint, jsonify
from flask_login import login_required, current_user
from flask_babel import gettext as _
from flask_babel import format_datetime
from babel.units import format_unit
from . import logger
from .render_template import render_title_template
from .services.worker import WorkerThread, STAT_WAITING, STAT_FAIL, STAT_STARTED, STAT_FINISH_SUCCESS, STAT_ENDED, \
STAT_CANCELLED
tasks = Blueprint('tasks', __name__)
log = logger.create()
@tasks.route("/ajax/emailstat")
@login_required
def get_email_status_json():
tasks = WorkerThread.get_instance().tasks
return jsonify(render_task_status(tasks))
@tasks.route("/tasks")
@login_required
def get_tasks_status():
# if current user admin, show all email, otherwise only own emails
tasks = WorkerThread.get_instance().tasks
answer = render_task_status(tasks)
return render_title_template('tasks.html', entries=answer, title=_("Tasks"), page="tasks")
# helper function to apply localize status information in tasklist entries
def render_task_status(tasklist):
rendered_tasklist = list()
for __, user, __, task, __ in tasklist:
if user == current_user.name or current_user.role_admin():
ret = {}
if task.start_time:
ret['starttime'] = format_datetime(task.start_time, format='short')
ret['runtime'] = format_runtime(task.runtime)
# localize the task status
if isinstance(task.stat, int):
if task.stat == STAT_WAITING:
ret['status'] = _('Waiting')
elif task.stat == STAT_FAIL:
ret['status'] = _('Failed')
elif task.stat == STAT_STARTED:
ret['status'] = _('Started')
elif task.stat == STAT_FINISH_SUCCESS:
ret['status'] = _('Finished')
elif task.stat == STAT_ENDED:
ret['status'] = _('Ended')
elif task.stat == STAT_CANCELLED:
ret['status'] = _('Cancelled')
else:
ret['status'] = _('Unknown Status')
ret['taskMessage'] = "{}: {}".format(task.name, task.message) if task.message else task.name
ret['progress'] = "{} %".format(int(task.progress * 100))
ret['user'] = escape(user) # prevent xss
# Hidden fields
ret['task_id'] = task.id
ret['stat'] = task.stat
ret['is_cancellable'] = task.is_cancellable
rendered_tasklist.append(ret)
return rendered_tasklist
# helper function for displaying the runtime of tasks
def format_runtime(runtime):
ret_val = ""
if runtime.days:
ret_val = format_unit(runtime.days, 'duration-day', length="long") + ', '
minutes, seconds = divmod(runtime.seconds, 60)
hours, minutes = divmod(minutes, 60)
# ToDo: locale.number_symbols._data['timeSeparator'] -> localize time separator ?
if hours:
ret_val += '{:d}:{:02d}:{:02d}s'.format(hours, minutes, seconds)
elif minutes:
ret_val += '{:2d}:{:02d}s'.format(minutes, seconds)
else:
ret_val += '{:2d}s'.format(seconds)
return ret_val