# -*- coding: utf-8 -*- 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() if time.mktime(time.gmtime()) - settings.server.get('last_scan', 0) > 24*60*60: settings.server['last_scan'] = time.mktime(time.gmtime()) 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): logger.debug('loading tasks') 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)))