openmedialibrary/oml/tasks.py

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

from websocket import trigger_event
import state
import settings

import logging
logger = logging.getLogger(__name__)

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
        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
                        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 == 'scan':
                            item.scan.run_scan()
                        elif action == 'scanimport':
                            item.scan.import_folder()
                        elif action == 'syncmetadata':
                            sync_metadata(data)
                        else:
                            trigger_event('error', {'error': 'unknown action'})
                        logger.debug('%s done', action)
                    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:
                        self.q.put(task)
                    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') as f:
                json.dump(self._tasks, f)

    def join(self):
        self.q.put((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'):
                priority -= 1
            if action in ('peering', 'changelibrarypath'):
                priority -= 2
        if not state.shutdown:
            #logger.debug('queue: %s (%s)', action, data)
            self.q.put((priority, (action, data)))