<commit_before>from __future__ import unicode_literals, division, absolute_import
from builtins import *  # noqa pylint: disable=unused-import, redefined-builtin

import logging
import pickle
from datetime import datetime

from sqlalchemy import Column, Integer, String, Unicode, DateTime, Index, select

from flexget import db_schema, plugin
from flexget.event import event
from flexget.utils import json
from flexget.manager import Session
from flexget.utils.database import entry_synonym, with_session
from flexget.utils.tools import parse_timedelta
from flexget.utils.sqlalchemy_utils import table_schema, table_add_column

log = logging.getLogger('backlog')
Base = db_schema.versioned_base('backlog', 2)


@db_schema.upgrade('backlog')
def upgrade(ver, session):
    if ver is None:
        # Make sure there is no data we can't load in the backlog table
        backlog_table = table_schema('backlog', session)
        try:
            for item in session.query('entry').select_from(backlog_table).all():
                pickle.loads(item.entry)
        except (ImportError, TypeError):
            # If there were problems, we can drop the data.
            log.info('Backlog table contains unloadable data, clearing old data.')
            session.execute(backlog_table.delete())
        ver = 0
    if ver == 0:
        backlog_table = table_schema('backlog', session)
        log.info('Creating index on backlog table.')
        Index('ix_backlog_feed_expire', backlog_table.c.feed, backlog_table.c.expire).create(bind=session.bind)
        ver = 1
    if ver == 1:
        table = table_schema('backlog', session)
        table_add_column(table, 'json', Unicode, session)
        # Make sure we get the new schema with the added column
        table = table_schema('backlog', session)
        for row in session.execute(select([table.c.id, table.c.entry])):
            try:
                p = pickle.loads(row['entry'])
                session.execute(table.update().where(table.c.id == row['id']).values(
                    json=json.dumps(p, encode_datetime=True)))
            except KeyError as e:
                log.error('Unable error upgrading backlog pickle object due to %s' % str(e))

        ver = 2
    return ver


class BacklogEntry(Base):
    __tablename__ = 'backlog'

    id = Column(Integer, primary_key=True)
    task = Column('feed', String)
    title = Column(String)
    expire = Column(DateTime)
    _json = Column('json', Unicode)
    entry = entry_synonym('_json')

    def __repr__(self):
        return '<BacklogEntry(title=%s)>' % (self.title)


Index('ix_backlog_feed_expire', BacklogEntry.task, BacklogEntry.expire)


@with_session
def get_entries(task=None, session=None):
    query = session.query(BacklogEntry)
    if task:
        query = query.filter(BacklogEntry.task == task)
    return query.all()


@with_session
def clear_entries(task=None, all=False, session=None):
    """
    Clear entries from backlog.

    :param task: If given, only entries from specified task will be cleared.
    :param all: If True, all entries will be cleared, otherwise only expired entries will be cleared.
    :returns: Number of entries cleared from backlog.
    """
    query = session.query(BacklogEntry)
    if task:
        query = query.filter(BacklogEntry.task == task)
    if not all:
        query = query.filter(BacklogEntry.expire < datetime.now())
    return query.delete()


class InputBacklog(object):
    """
    Keeps task history for given amount of time.

    Example::

      backlog: 4 days

    Rarely useful for end users, mainly used by other plugins.
    """

    schema = {'type': 'string', 'format': 'interval'}

    @plugin.priority(-255)
    def on_task_input(self, task, config):
        # Get a list of entries to inject
        injections = self.get_injections(task)
        # Take a snapshot of the entries' states after the input event in case we have to store them to backlog
        for entry in task.entries + injections:
            entry.take_snapshot('after_input')
        if config:
            # If backlog is manually enabled for this task, learn the entries.
            self.learn_backlog(task, config)
        # Return the entries from backlog that are not already in the task
        return injections

    def on_task_abort(self, task, config):
        """Remember all entries until next execution when task gets aborted."""
        if task.entries:
            log.debug('Remembering all entries to backlog because of task abort.')
            self.learn_backlog(task)

    @with_session
    def add_backlog(self, task, entry, amount='', session=None):
        """Add single entry to task backlog

        If :amount: is not specified, entry will only be injected on next execution."""
        snapshot = entry.snapshots.get('after_input')
        if not snapshot:
            if task.current_phase != 'input':
                # Not having a snapshot is normal during input phase, don't display a warning
                log.warning('No input snapshot available for `%s`, using current state' % entry['title'])
            snapshot = entry
        expire_time = datetime.now() + parse_timedelta(amount)
        backlog_entry = session.query(BacklogEntry).filter(BacklogEntry.title == entry['title']). \
            filter(BacklogEntry.task == task.name).first()<commit_msg>series: save entry state before metainfo phase to avoid lazy lookups during backlog (timeframe) (#2215)<commit_after>from __future__ import unicode_literals, division, absolute_import
from builtins import *  # noqa pylint: disable=unused-import, redefined-builtin

import logging
import pickle
from datetime import datetime

from sqlalchemy import Column, Integer, String, Unicode, DateTime, Index, select

from flexget import db_schema, plugin
from flexget.event import event
from flexget.utils import json
from flexget.manager import Session
from flexget.utils.database import entry_synonym, with_session
from flexget.utils.tools import parse_timedelta
from flexget.utils.sqlalchemy_utils import table_schema, table_add_column

log = logging.getLogger('backlog')
Base = db_schema.versioned_base('backlog', 2)


@db_schema.upgrade('backlog')
def upgrade(ver, session):
    if ver is None:
        # Make sure there is no data we can't load in the backlog table
        backlog_table = table_schema('backlog', session)
        try:
            for item in session.query('entry').select_from(backlog_table).all():
                pickle.loads(item.entry)
        except (ImportError, TypeError):
            # If there were problems, we can drop the data.
            log.info('Backlog table contains unloadable data, clearing old data.')
            session.execute(backlog_table.delete())
        ver = 0
    if ver == 0:
        backlog_table = table_schema('backlog', session)
        log.info('Creating index on backlog table.')
        Index('ix_backlog_feed_expire', backlog_table.c.feed, backlog_table.c.expire).create(bind=session.bind)
        ver = 1
    if ver == 1:
        table = table_schema('backlog', session)
        table_add_column(table, 'json', Unicode, session)
        # Make sure we get the new schema with the added column
        table = table_schema('backlog', session)
        for row in session.execute(select([table.c.id, table.c.entry])):
            try:
                p = pickle.loads(row['entry'])
                session.execute(table.update().where(table.c.id == row['id']).values(
                    json=json.dumps(p, encode_datetime=True)))
            except KeyError as e:
                log.error('Unable error upgrading backlog pickle object due to %s' % str(e))

        ver = 2
    return ver


class BacklogEntry(Base):
    __tablename__ = 'backlog'

    id = Column(Integer, primary_key=True)
    task = Column('feed', String)
    title = Column(String)
    expire = Column(DateTime)
    _json = Column('json', Unicode)
    entry = entry_synonym('_json')

    def __repr__(self):
        return '<BacklogEntry(title=%s)>' % (self.title)


Index('ix_backlog_feed_expire', BacklogEntry.task, BacklogEntry.expire)


@with_session
def get_entries(task=None, session=None):
    query = session.query(BacklogEntry)
    if task:
        query = query.filter(BacklogEntry.task == task)
    return query.all()


@with_session
def clear_entries(task=None, all=False, session=None):
    """
    Clear entries from backlog.

    :param task: If given, only entries from specified task will be cleared.
    :param all: If True, all entries will be cleared, otherwise only expired entries will be cleared.
    :returns: Number of entries cleared from backlog.
    """
    query = session.query(BacklogEntry)
    if task:
        query = query.filter(BacklogEntry.task == task)
    if not all:
        query = query.filter(BacklogEntry.expire < datetime.now())
    return query.delete()


class InputBacklog(object):
    """
    Keeps task history for given amount of time.

    Example::

      backlog: 4 days

    Rarely useful for end users, mainly used by other plugins.
    """

    schema = {'type': 'string', 'format': 'interval'}

    @plugin.priority(-255)
    def on_task_input(self, task, config):
        # Get a list of entries to inject
        injections = self.get_injections(task)
        # Take a snapshot of the entries' states after the input event in case we have to store them to backlog
        for entry in task.entries + injections:
            entry.take_snapshot('after_input')
        if config:
            # If backlog is manually enabled for this task, learn the entries.
            self.learn_backlog(task, config)
        # Return the entries from backlog that are not already in the task
        return injections

    @plugin.priority(255)
    def on_task_metainfo(self, task, config):
        # Take a snapshot of any new entries' states before metainfo event in case we have to store them to backlog
        # This is really a hack to avoid unnecessary lazy lookups causing db locks. Ideally, saving a snapshot
        # should not cause lazy lookups, but we currently have no other way of saving a lazy field than performing its
        # action.
        # https://github.com/Flexget/Flexget/issues/1000
        for entry in task.entries:
            snapshot = entry.snapshots.get('after_input')
            if snapshot:
                continue
            entry.take_snapshot('after_input')

    def on_task_abort(self, task, config):
        """Remember all entries until next execution when task gets aborted."""
        if task.entries:
            log.debug('Remembering all entries to backlog because of task abort.')
            self.learn_backlog(task)

    @with_session
    def add_backlog(self, task, entry, amount='', session=None):
        """Add single entry to task backlog

        If :amount: is not specified, entry will only be injected on next execution."""
        snapshot = entry.snapshots.get('after_input')
        if not snapshot:
            if task.current_phase != 'input':
                # Not having a snapshot is normal during input phase, don't display a warning
                log.warning('No input snapshot available for `%s`, using current state' % entry['title'])
            snapshot = entry
        expire_time = datetime.now() + parse_timedelta(amount)
        backlog_entry = session.query(BacklogEntry).filter(BacklogEntry.title == entry['title']). \
            filter(BacklogEntry.task == task.name).first()
{'fraction_non_alphanumeric': 0.09416884247171453, 'fraction_numerical': 0.0026979982593559617, 'mean_word_length': 2.903192934782609, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 5, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4717028', 'n_tokens_mistral': 3257, 'n_tokens_neox': 3156, 'n_words': 1047}