openmedialibrary/oml/tasks.py

# -*- coding: utf-8 -*-
# vi:si:et:sw=4:sts=4:ts=4
from queue import PriorityQueue
from threading import Thread
import json
import os
import time

from websocket import trigger_event
import state
import settings

import logging
logger = logging.getLogger(__name__)

DEUBG_TASKS = False

class QueueElement(tuple):
    def __lt__(self, rhs):
        return self[0] < rhs[0]
    def __gt__(self, rhs):
        return self[0] > rhs[0]
    def __le__(self, rhs):
        return self[0] <= rhs[0]
    def __ge__(self, rhs):
        return self[0] >= rhs[0]

class Tasks(Thread):
    _tasks = []

    def __init__(self):
        self._taskspath = os.path.join(settings.data_path, 'tasks.json')
        self.q = PriorityQueue()
        Thread.__init__(self)
        self.daemon = True
        self.start()

    def run(self):
        self.load_tasks()
        self.queue('scan')
        import item.scan
        from item.models import sync_metadata, get_preview, get_cover
        from user.models import (
                export_list, update_user_peering,
                add_local_info, remove_local_info,
                upload
        )
        shutdown = False
        while not shutdown:
            p, m = self.q.get()
            if m:
                if state.shutdown:
                    self._tasks.append((p, m))
                else:
                    try:
                        action, data = m
                        if DEUBG_TASKS:
                            t0 = time.time()
                            logger.debug('%s start', action)

                        if action == 'changelibrarypath':
                            item.scan.change_path(data[0], data[1])
                        elif action == 'export':
                            export_list(data)
                        elif action == 'getcover':
                            get_cover(data)
                        elif action == 'getpreview':
                            get_preview(data)
                        elif action == 'import':
                            item.scan.run_import(data)
                        elif action == 'peering':
                            update_user_peering(*data)
                        elif action == 'ping':
                            trigger_event('pong', data)
                        elif action == 'addlocalinfo':
                            add_local_info(data)
                        elif action == 'removelocalinfo':
                            remove_local_info(data)
                        elif action == 'scan':
                            item.scan.run_scan()
                        elif action == 'scanimport':
                            item.scan.import_folder()
                        elif action == 'syncmetadata':
                            sync_metadata(data)
                        elif action == 'upload':
                            upload(data)
                        else:
                            trigger_event('error', {'error': 'unknown action'})
                        if DEUBG_TASKS:
                            logger.debug('%s done (%s)', action, time.time()-t0)
                    except:
                        logger.debug('task failed', exc_info=True)
            else:
                shutdown = True
            self.q.task_done()

    def load_tasks(self):
        if os.path.exists(self._taskspath):
            try:
                with open(self._taskspath) as f:
                    tasks = json.load(f)
                    for task in tasks:
                        if len(task) == 2:
                            self.q.put(QueueElement((task[0], tuple(task[1]))))
                    logger.debug('loaded %s tasks', len(tasks))
            except:
                logger.debug('failed to load saved tasks', exc_info=True)
            os.unlink(self._taskspath)

    def save_tasks(self):
        if self._tasks:
            logger.debug('saving %s tasks for later', len(self._tasks))
            with open(self._taskspath, 'w', encoding='utf-8') as f:
                json.dump(self._tasks, f)

    def join(self):
        self.q.put(QueueElement((1000, None)))
        r = Thread.join(self)
        self.save_tasks()
        return r

    def queue(self, action, data=None, priority=None):
        if priority is None:
            priority = 100
            if action in ('getcover', 'getpreview'):
                priority += 1
            if action in ('import', 'export', 'scanimport'):
                priority -= 1
            if action in ('peering', 'changelibrarypath'):
                priority -= 2
        if not state.shutdown:
            #logger.debug('queue: %s (%s)', action, data)
            self.q.put(QueueElement((priority, (action, data))))
        else:
            self._tasks.append((priority, (action, data)))