125 lines
4.3 KiB
Python
125 lines
4.3 KiB
Python
# -*- 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
|
|
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 == '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'})
|
|
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') 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)))
|