<commit_before># coding=utf-8
from __future__ import absolute_import
from __future__ import unicode_literals

import logging
import sqlite3
import time as _time
import threading
import warnings

from . import sqlbase
from .exceptions import Empty

sqlite3.enable_callback_tracebacks(True)

log = logging.getLogger(__name__)

# 10 seconds internal for `wait` of event
TICK_FOR_WAIT = 10


class AckStatus(object):
    inited = '0'
    ready = '1'
    unack = '2'
    acked = '5'
    ack_failed = '9'


class SQLiteAckQueue(sqlbase.SQLiteBase):
    """SQLite3 based FIFO queue with ack support."""

    _TABLE_NAME = 'ack_queue'
    _KEY_COLUMN = '_id'  # the name of the key column, used in DB CRUD
    _MAX_ACKED_LENGTH = 1000  # depreciated
    # SQL to create a table
    _SQL_CREATE = (
        'CREATE TABLE IF NOT EXISTS {table_name} ('
        '{key_column} INTEGER PRIMARY KEY AUTOINCREMENT, '
        'data BLOB, timestamp FLOAT, status INTEGER)'
    )
    # SQL to insert a record
    _SQL_INSERT = (
        'INSERT INTO {table_name} (data, timestamp, status)'
        ' VALUES (?, ?, %s)' % AckStatus.inited
    )
    # SQL to select a record
    _SQL_SELECT_ID = (
        'SELECT {key_column}, data, timestamp, status FROM {table_name} WHERE'
        ' {key_column} = {rowid}'
    )
    _SQL_SELECT = (
        'SELECT {key_column}, data, timestamp, status FROM {table_name} '
        'WHERE {key_column} > {rowid} AND status < %s '
        'ORDER BY {key_column} ASC LIMIT 1' % AckStatus.unack
    )
    _SQL_MARK_ACK_UPDATE = (
        'UPDATE {table_name} SET status = ? WHERE {key_column} = ?'
    )
    _SQL_SELECT_WHERE = (
        'SELECT {key_column}, data, timestamp FROM {table_name}'
        ' WHERE {key_column} > {rowid} AND status < %s AND'
        ' {column} {op} ? ORDER BY {key_column} ASC'
        ' LIMIT 1 ' % AckStatus.unack
    )
    _SQL_UPDATE = 'UPDATE {table_name} SET data = ? WHERE {key_column} = ?'

    def __init__(self, path, auto_resume=True, **kwargs):
        super(SQLiteAckQueue, self).__init__(path, **kwargs)
        if not self.auto_commit:
            warnings.warn("disable auto commit is not support in ack queue")
            self.auto_commit = True
        self._unack_cache = {}
        if auto_resume:
            self.resume_unack_tasks()

    def resume_unack_tasks(self):
        unack_count = self.unack_count()
        if unack_count:
            log.warning("resume %d unack tasks", unack_count)
        sql = 'UPDATE {} set status = ?' \
              ' WHERE status = ?'.format(self._table_name)
        with self.tran_lock:
            with self._putter as tran:
                tran.execute(sql, (AckStatus.ready, AckStatus.unack,))
            self.total = self._count()

    def put(self, item):
        obj = self._serializer.dumps(item)
        _id = self._insert_into(obj, _time.time())
        self.total += 1
        self.put_event.set()
        return _id

    def _init(self):
        super(SQLiteAckQueue, self)._init()
        # Action lock to assure multiple action to be *atomic*
        self.action_lock = threading.Lock()
        self.total = self._count()

    def _count(self):
        sql = 'SELECT COUNT({}) FROM {} WHERE status < ?'.format(
            self._key_column, self._table_name
        )
        row = self._getter.execute(sql, (AckStatus.unack,)).fetchone()
        return row[0] if row else 0

    def _ack_count_via_status(self, status):
        sql = 'SELECT COUNT({}) FROM {} WHERE status = ?'.format(
            self._key_column, self._table_name
        )
        row = self._getter.execute(sql, (status,)).fetchone()
        return row[0] if row else 0

    def unack_count(self):
        return self._ack_count_via_status(AckStatus.unack)

    def acked_count(self):
        return self._ack_count_via_status(AckStatus.acked)

    def ready_count(self):
        return self._ack_count_via_status(AckStatus.ready)

    def ack_failed_count(self):
        return self._ack_count_via_status(AckStatus.ack_failed)

    @sqlbase.with_conditional_transaction
    def _mark_ack_status(self, key, status):
        return self._sql_mark_ack_status, (
            status,
            key,
        )

    @sqlbase.with_conditional_transaction
    def clear_acked_data(
        self, max_delete=1000, keep_latest=1000, clear_ack_failed=False
    ):
        acked_clear_all = ''
        acked_to_delete = ''
        acked_to_keep = ''
        if self._MAX_ACKED_LENGTH != 1000 and not max_delete:
            # Added for backward compatibility for
            # those that set the _MAX_ACKED_LENGTH
            print(
                "_MAX_ACKED_LENGTH has been depreciated.  \
                    Use clear_acked_data(keep_latest=1000, max_delete=1000)"
            )
            max_delete = self._MAX_ACKED_LENGTH
        if clear_ack_failed:
            acked_clear_all = 'OR status = %s' % AckStatus.ack_failed
        if max_delete and max_delete > 0:
            acked_to_delete = 'LIMIT %d' % max_delete
        if keep_latest and keep_latest > 0:
            acked_to_keep = 'OFFSET %d' % keep_latest
        sql = """DELETE FROM {table_name}<commit_msg>Fix misspelling of deprecated<commit_after># coding=utf-8
from __future__ import absolute_import
from __future__ import unicode_literals

import logging
import sqlite3
import time as _time
import threading
import warnings

from . import sqlbase
from .exceptions import Empty

sqlite3.enable_callback_tracebacks(True)

log = logging.getLogger(__name__)

# 10 seconds internal for `wait` of event
TICK_FOR_WAIT = 10


class AckStatus(object):
    inited = '0'
    ready = '1'
    unack = '2'
    acked = '5'
    ack_failed = '9'


class SQLiteAckQueue(sqlbase.SQLiteBase):
    """SQLite3 based FIFO queue with ack support."""

    _TABLE_NAME = 'ack_queue'
    _KEY_COLUMN = '_id'  # the name of the key column, used in DB CRUD
    _MAX_ACKED_LENGTH = 1000  # deprecated
    # SQL to create a table
    _SQL_CREATE = (
        'CREATE TABLE IF NOT EXISTS {table_name} ('
        '{key_column} INTEGER PRIMARY KEY AUTOINCREMENT, '
        'data BLOB, timestamp FLOAT, status INTEGER)'
    )
    # SQL to insert a record
    _SQL_INSERT = (
        'INSERT INTO {table_name} (data, timestamp, status)'
        ' VALUES (?, ?, %s)' % AckStatus.inited
    )
    # SQL to select a record
    _SQL_SELECT_ID = (
        'SELECT {key_column}, data, timestamp, status FROM {table_name} WHERE'
        ' {key_column} = {rowid}'
    )
    _SQL_SELECT = (
        'SELECT {key_column}, data, timestamp, status FROM {table_name} '
        'WHERE {key_column} > {rowid} AND status < %s '
        'ORDER BY {key_column} ASC LIMIT 1' % AckStatus.unack
    )
    _SQL_MARK_ACK_UPDATE = (
        'UPDATE {table_name} SET status = ? WHERE {key_column} = ?'
    )
    _SQL_SELECT_WHERE = (
        'SELECT {key_column}, data, timestamp FROM {table_name}'
        ' WHERE {key_column} > {rowid} AND status < %s AND'
        ' {column} {op} ? ORDER BY {key_column} ASC'
        ' LIMIT 1 ' % AckStatus.unack
    )
    _SQL_UPDATE = 'UPDATE {table_name} SET data = ? WHERE {key_column} = ?'

    def __init__(self, path, auto_resume=True, **kwargs):
        super(SQLiteAckQueue, self).__init__(path, **kwargs)
        if not self.auto_commit:
            warnings.warn("disable auto commit is not support in ack queue")
            self.auto_commit = True
        self._unack_cache = {}
        if auto_resume:
            self.resume_unack_tasks()

    def resume_unack_tasks(self):
        unack_count = self.unack_count()
        if unack_count:
            log.warning("resume %d unack tasks", unack_count)
        sql = 'UPDATE {} set status = ?' \
              ' WHERE status = ?'.format(self._table_name)
        with self.tran_lock:
            with self._putter as tran:
                tran.execute(sql, (AckStatus.ready, AckStatus.unack,))
            self.total = self._count()

    def put(self, item):
        obj = self._serializer.dumps(item)
        _id = self._insert_into(obj, _time.time())
        self.total += 1
        self.put_event.set()
        return _id

    def _init(self):
        super(SQLiteAckQueue, self)._init()
        # Action lock to assure multiple action to be *atomic*
        self.action_lock = threading.Lock()
        self.total = self._count()

    def _count(self):
        sql = 'SELECT COUNT({}) FROM {} WHERE status < ?'.format(
            self._key_column, self._table_name
        )
        row = self._getter.execute(sql, (AckStatus.unack,)).fetchone()
        return row[0] if row else 0

    def _ack_count_via_status(self, status):
        sql = 'SELECT COUNT({}) FROM {} WHERE status = ?'.format(
            self._key_column, self._table_name
        )
        row = self._getter.execute(sql, (status,)).fetchone()
        return row[0] if row else 0

    def unack_count(self):
        return self._ack_count_via_status(AckStatus.unack)

    def acked_count(self):
        return self._ack_count_via_status(AckStatus.acked)

    def ready_count(self):
        return self._ack_count_via_status(AckStatus.ready)

    def ack_failed_count(self):
        return self._ack_count_via_status(AckStatus.ack_failed)

    @sqlbase.with_conditional_transaction
    def _mark_ack_status(self, key, status):
        return self._sql_mark_ack_status, (
            status,
            key,
        )

    @sqlbase.with_conditional_transaction
    def clear_acked_data(
        self, max_delete=1000, keep_latest=1000, clear_ack_failed=False
    ):
        acked_clear_all = ''
        acked_to_delete = ''
        acked_to_keep = ''
        if self._MAX_ACKED_LENGTH != 1000 and not max_delete:
            # Added for backward compatibility for
            # those that set the _MAX_ACKED_LENGTH
            print(
                "_MAX_ACKED_LENGTH has been deprecated.  \
                    Use clear_acked_data(keep_latest=1000, max_delete=1000)"
            )
            max_delete = self._MAX_ACKED_LENGTH
        if clear_ack_failed:
            acked_clear_all = 'OR status = %s' % AckStatus.ack_failed
        if max_delete and max_delete > 0:
            acked_to_delete = 'LIMIT %d' % max_delete
        if keep_latest and keep_latest > 0:
            acked_to_keep = 'OFFSET %d' % keep_latest
        sql = """DELETE FROM {table_name}
{'fraction_non_alphanumeric': 0.09928391716663441, 'fraction_numerical': 0.008902651441842461, 'mean_word_length': 2.4404127829560585, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 11, 'https://': 0, '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': '5366733', 'n_tokens_mistral': 3411, 'n_tokens_neox': 3250, 'n_words': 816}