# # This file is part of SickGear. # # SickGear 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. # # SickGear 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 SickGear. If not, see <http://www.gnu.org/licenses/>. import copy import datetime import threading from . import db, logger from .scheduler import Job from exceptions_helper import ex from six import integer_types # noinspection PyUnreachableCode if False: from typing import AnyStr, Callable, Dict, List, Optional, Tuple, Union from .search_queue import BaseSearchQueueItem from .show_queue import ShowQueueItem from .people_queue import CastQueueItem class QueuePriorities(object): LOW = 10 NORMAL = 20 HIGH = 30 VERYHIGH = 40 class GenericQueue(Job): def __init__(self, cache_db_tables=None, main_db_tables=None): # type: (List[AnyStr], List[AnyStr]) -> None super(GenericQueue, self).__init__(self.job_run, silent=True, kwargs={}, reentrant_lock=True) self.currentItem = None # type: QueueItem or None self.queue = [] # type: List[Union[QueueItem, BaseSearchQueueItem, ShowQueueItem]] self.queue_name = 'QUEUE' # type: AnyStr self.min_priority = 0 # type: int self.events = {} # type: Dict[int, List[Callable]] self.cache_db_tables = cache_db_tables or [] # type: List[AnyStr] self.main_db_tables = main_db_tables or [] # type: List[AnyStr] self._id_counter = self._load_init_id() # type: integer_types def job_run(self): # only start a new task if one isn't already going with self.lock: if None is self.currentItem or not self.currentItem.is_alive(): # if the thread is dead then the current item should be finished if self.currentItem: self.currentItem.finish() try: self.delete_item(self.currentItem, finished_run=True) except (BaseException, Exception): pass self.currentItem = None # if there's something in the queue then run it in a thread and take it out of the queue if 0 < len(self.queue): self.queue.sort(key=lambda y: (-y.priority, y.added)) if self.queue[0].priority < self.min_priority: return # launch the queue item in a thread self.currentItem = self.queue.pop(0) if 'SEARCHQUEUE' != self.queue_name: self.currentItem.name = self.queue_name + '-' + self.currentItem.name self.currentItem.start() self.check_events() def _load_init_id(self): # type: (...) -> integer_types """ fetch highest uid for queue type to initialize the class """ my_db = db.DBConnection('cache.db') cr = my_db.mass_action([['SELECT max(uid) as max_id FROM %s' % t] for t in self.cache_db_tables]) my_db = db.DBConnection() mr = my_db.mass_action([['SELECT max(uid) as max_id FROM %s' % t] for t in self.main_db_tables]) return max([c[0]['max_id'] or 0 for c in cr] + [s[0]['max_id'] or 0 for s in mr] + [0]) def _get_new_id(self): # type: (...) -> integer_types self._id_counter += 1 return self._id_counter def load_queue(self): pass def save_queue(self): cl = self._clear_sql() try: with self.lock: for item in ((self.currentItem and [self.currentItem]) or []) + self.queue: cl.extend(self._get_item_sql(item)) if cl: my_db = db.DBConnection('cache.db') my_db.mass_action(cl) except (BaseException, Exception) as e: logger.error('Exception saving queue %s to db: %s' % (self.__class__.__name__, ex(e))) def _clear_sql(self): # type: (...) -> List[List] return [] def save_item(self, item): try: if item: item_sql = self._get_item_sql(item) if item_sql: my_db = db.DBConnection('cache.db') my_db.mass_action(item_sql) except (BaseException, Exception) as e: logger.error('Exception saving item %s to db: %s' % (item, ex(e))) def delete_item(self, item, finished_run=False): # type: (Union[QueueItem, CastQueueItem], bool) -> None """ :param item: :param finished_run: set to True when queue item has run """ if item: try: item_sql = self._delete_item_from_db_sql(item) if item_sql: my_db = db.DBConnection('cache.db') my_db.mass_action(item_sql) except (BaseException, Exception) as e: logger.error('Exception deleting item %s from db: %s' % (item, ex(e))) def _get_item_sql(self, item): # type: (Union[QueueItem, CastQueueItem]) -> List[List] return [] def _delete_item_from_db_sql(self, item): # type: (Union[QueueItem, CastQueueItem]) -> List[List] pass def remove_from_queue(self, to_remove=None, force=False): # type: (List[AnyStr], bool) -> None """ remove given uid items from queue :param to_remove: list of uids to remove from queue :param force: force removal from db """ self._remove_from_queue(to_remove=to_remove, excluded_types=[], force=force) def _remove_from_queue(self, to_remove=None, excluded_types=None, force=False): # type: (List[AnyStr], List, bool) -> None """ remove given uid items from queue :param to_remove: list of uids to remove from queue :param force: force removal from db """ if to_remove: excluded_types = excluded_types or [] with self.lock: if not force: to_remove = [r for r in to_remove for q in self.queue if r == q.uid and (q.action_id not in excluded_types)] del_sql = [ ['DELETE FROM %s WHERE uid IN (%s)' % (t, ','.join(['?'] * len(to_remove))), to_remove] for t in self.cache_db_tables ] del_main_sql = [ ['DELETE FROM %s WHERE uid IN (%s)' % (t, ','.join(['?'] * len(to_remove))), to_remove] for t in self.main_db_tables ] self.queue = [q for q in self.queue if q.uid not in to_remove] if del_sql: my_db = db.DBConnection('cache.db') my_db.mass_action(del_sql) if del_main_sql: my_db = db.DBConnection() my_db.mass_action(del_main_sql) def clear_queue(self, action_types=None): # type: (integer_types) -> None """ clear queue excluding internal defined types :param action_types: only clear supplied action types """ if not isinstance(action_types, list): action_types = [action_types] return self._clear_queue(action_types=action_types) def _clear_queue(self, action_types=None, excluded_types=None): # type: (List[integer_types], List) -> None excluded_types = excluded_types or [] with self.lock: if action_types: self.queue = [q for q in self.queue if q.action_id in excluded_types or q.action_id not in action_types] del_sql = [ ['DELETE FROM %s WHERE action_id IN (%s)' % (t, ','.join(['?'] * len(action_types))), action_types] for t in self.cache_db_tables ] del_main_sql = [ ['DELETE FROM %s WHERE action_id IN (%s)' % (t, ','.join(['?'] * len(action_types))), action_types] for t in self.main_db_tables ] else: self.queue = [q for q in self.queue if q.action_id in excluded_types] del_sql = [ ['DELETE FROM %s' % t] for t in self.cache_db_tables ] del_main_sql = [ ['DELETE FROM %s' % t] for t in self.main_db_tables ] if del_sql: my_db = db.DBConnection('cache.db') my_db.mass_action(del_sql) if del_main_sql: my_db = db.DBConnection() my_db.mass_action(del_main_sql) def pause(self): logger.log('Pausing queue') if self.lock: self.min_priority = 999999999999 def unpause(self): logger.log('Un-pausing queue') with self.lock: self.min_priority = 0 def add_item(self, item, add_to_db=True): """ :param item: Queue Item :type item: QueueItem :param add_to_db: add to db :return: Queue Item :rtype: QueueItem """ with self.lock: item.added = datetime.datetime.now() item.uid = item.uid or self._get_new_id() self.queue.append(item) if add_to_db: self.save_item(item) return item def check_events(self): pass def add_event(self, event_type, method): # type: (int, Callable) -> None if isinstance(event_type, integer_types) and callable(method): if event_type not in self.events: self.events[event_type] = [] if method not in self.events[event_type]: self.events[event_type].append(method) def remove_event(self, event_type, method): # type: (int, Callable) -> None if isinstance(event_type, integer_types) and callable(method): if event_type in self.events and method in self.events[event_type]: try: self.events[event_type].remove(method) if 0 == len(self.events[event_type]): del self.events[event_type] except (BaseException, Exception) as e: logger.error('Error removing event method from queue: %s' % ex(e)) def execute_events(self, event_type, *args, **kwargs): # type: (int, Tuple, Dict) -> None if event_type in self.events: for event in self.events.get(event_type): try: event(*args, **kwargs) except (BaseException, Exception) as e: logger.error('Error executing Event: %s' % ex(e)) class QueueItem(threading.Thread): def __init__(self, name, action_id=0, uid=None): # type: (AnyStr, int, integer_types) -> None """ :param name: name :param action_id: :param uid: """ super(QueueItem, self).__init__() self.name = name.replace(' ', '-').upper() # type: AnyStr self.inProgress = False # type: bool self.priority = QueuePriorities.NORMAL # type: int self.action_id = action_id # type: int self.stop = threading.Event() self.added = None # type: Optional[datetime.datetime] self.uid = uid # type: integer_types def copy(self, deepcopy_obj=None): """ :param deepcopy_obj: List of properties to be deep copied :type deepcopy_obj: List :return: a shallow copy of QueueItem with optional deepcopy of in deepcopy_obj listed objects :rtype: QueueItem """ cls = self.__class__ result = cls.__new__(cls) result.__dict__.update(self.__dict__) if deepcopy_obj: for o in deepcopy_obj: if self.__dict__.get(o): new_seg = copy.deepcopy(self.__dict__.get(o)) result.__dict__[o] = new_seg return result def run(self): """Implementing classes should call this""" self.inProgress = True def finish(self): """Implementing Classes should call this""" self.inProgress = False threading.current_thread().name = self.name