<commit_before>import os
import random
import re
import thread
import threading
import traceback
import time
from gettext import gettext as _

import mailpile.util
from mailpile.util import *
from mailpile.eventlog import Event
from mailpile.mailboxes import *
from mailpile.mailutils import FormatMbxId


__all__ = ['mbox', 'maildir', 'imap']


class BaseMailSource(threading.Thread):
    """
    MailSources take care of managing a group of mailboxes, synchronizing
    the source with Mailpile's local metadata and/or caches.
    """
    DEFAULT_JITTER = 15         # Fudge factor to tame thundering herds
    SAVE_STATE_INTERVAL = 3600  # How frequently we pickle our state
    INTERNAL_ERROR_SLEEP = 900  # Pause time on error, in seconds
    RESCAN_BATCH_SIZE = 2500    # Index at most this many new e-mails at once
    MAX_PATHS = 50000           # Abort if asked to scan too many directories

    # This is a helper for the events.
    __classname__ = 'mailpile.mail_source.BaseMailSource'

    def __init__(self, session, my_config):
        threading.Thread.__init__(self)
        self._lock = threading.Lock()
        self.my_config = my_config
        self.session = session
        self.alive = None
        self.event = None
        self.jitter = self.DEFAULT_JITTER
        self._state = 'Idle'
        self._sleeping = None
        self._interrupt = None
        self._rescanning = False
        self._rescan_waiters = []
        self._last_rescan_count = 0
        self._last_saved = time.time()  # Saving right away would be silly

        # Make locked versions of a few functions
        self._load_state = self._locked(self._unlocked_load_state)
        self.open = self._locked(self._unlocked_open)
        self.sync_mail = self._locked(self._unlocked_sync_mail)
        self.rescan_mailbox = self._locked(self._unlocked_rescan_mailbox)
        self.take_over_mailbox = self._locked(self._unlocked_take_over_mailbox)
        self.discover_mailboxes = self._locked(
            self._unlocked_discover_mailboxes)

    def __str__(self):
        rv = ': '.join([threading.Thread.__str__(self), self._state])
        if self._sleeping > 0:
            rv += '(%s)' % self._sleeping
        return rv

    def _locked(self, func):
        def locked_func(*args, **kwargs):
            try:
                self._lock.acquire()
                ostate, self._state = self._state, func.__name__
                return func(*args, **kwargs)
            except:
                traceback.print_exc()
                raise
            finally:
                self._state = ostate
                try:
                    self._lock.release()
                except thread.error:
                    pass
        return locked_func

    def _pfn(self):
        return 'mail-source.%s' % self.my_config._key

    def _unlocked_load_state(self):
        config, my_config = self.session.config, self.my_config
        events = list(config.event_log.incomplete(source=self,
                                                  data_id=my_config._key))
        if events:
            self.event = events[0]
        else:
            self.event = config.event_log.log(source=self,
                                              flags=Event.RUNNING,
                                              message=_('Starting up'),
                                              data={'id': my_config._key})

    def _save_state(self):
        self.session.config.event_log.log_event(self.event)

    def _log_status(self, message):
        self.event.message = message
        self.session.config.event_log.log_event(self.event)
        if 'sources' in self.session.config.sys.debug:
            self.session.ui.debug('%s: %s' % (self, message))

    def _unlocked_open(self):
        """Open mailboxes or connect to the remote mail source."""
        raise NotImplemented('Please override _open in %s' % self)

    def _has_mailbox_changed(self, mbx, state):
        """For the default sync_mail routine, report if mailbox changed."""
        raise NotImplemented('Please override _open in %s' % self)

    def _mark_mailbox_rescanned(self, mbx, state):
        """For the default sync_mail routine, note mailbox was rescanned."""
        raise NotImplemented('Please override _open in %s' % self)

    def _path(self, mbx):
        if mbx.path.startswith('@'):
            return self.session.config.sys.mailbox[mbx.path[1:]]
        else:
            return mbx.path

    def _unlocked_sync_mail(self):
        """Iterates through all the mailboxes and scans if necessary."""
        def unsorted(l):
            l.sort(key=lambda k: random.randint(0, 500))
            return l
        config = self.session.config
        self._last_rescan_count = rescanned = errors = 0
        self._interrupt = None
        batch = self.RESCAN_BATCH_SIZE
        errors = rescanned = 0
        if self.session.config.sys.debug:
            batch //= 10
        for mbx in unsorted(self.my_config.mailbox.values()):
            if mailpile.util.QUITTING or self._interrupt:
                self._log_status(_('Interrupted: %s'
                                   ) % (self._interrupt or _('Quitting')))
                self._interrupt = None
                break
            try:
                state = {}
                # Generally speaking, we only rescan if a mailbox looks like
                # it has changed. However, 1/20th of the time we take a look
                # anyway just in case looks are deceiving.
                if batch > 0 and (self._has_mailbox_changed(mbx, state) or
                                  random.randint(0, 20) == 10):

                    count = self._unlocked_rescan_mailbox(mbx._key,
                                                          stop_after=batch)
                    if count >= 0:
                        batch -= count
                        if (count and batch > 0 and
                                not self._interrupt and
                                not mailpile.util.QUITTING):
                            self._mark_mailbox_rescanned(mbx, state)
                            rescanned += 1
                    else:
                        errors += 1
            except (NoSuchMailboxError, IOError, OSError):
                errors += 1
            except:
                self._log_status(_('Internal error'))
                raise

        if batch > 0 and not errors:
            self._log_status(_('Checking for new mailboxes'))
            self._unlocked_discover_mailboxes(self.my_config.discovery.paths)

        if errors:
            self._log_status(_('Rescanned %d mailboxes, failed to rescan %d'
                               ) % (rescanned, errors))
        else:
            self._log_status(_('Rescanned %d mailboxes') % rescanned)
        self._last_rescan_count = rescanned
        return rescanned

    def _jitter(self, seconds):
        return seconds + random.randint(0, self.jitter)

    def _sleep(self, seconds):
        if self.session.config.sys.debug:
            seconds //= 10
        if self._sleeping != 0:
            self._sleeping = seconds<commit_msg>Discover new mailboxes even if we have errors on old ones<commit_after>import os
import random
import re
import thread
import threading
import traceback
import time
from gettext import gettext as _

import mailpile.util
from mailpile.util import *
from mailpile.eventlog import Event
from mailpile.mailboxes import *
from mailpile.mailutils import FormatMbxId


__all__ = ['mbox', 'maildir', 'imap']


class BaseMailSource(threading.Thread):
    """
    MailSources take care of managing a group of mailboxes, synchronizing
    the source with Mailpile's local metadata and/or caches.
    """
    DEFAULT_JITTER = 15         # Fudge factor to tame thundering herds
    SAVE_STATE_INTERVAL = 3600  # How frequently we pickle our state
    INTERNAL_ERROR_SLEEP = 900  # Pause time on error, in seconds
    RESCAN_BATCH_SIZE = 2500    # Index at most this many new e-mails at once
    MAX_PATHS = 50000           # Abort if asked to scan too many directories

    # This is a helper for the events.
    __classname__ = 'mailpile.mail_source.BaseMailSource'

    def __init__(self, session, my_config):
        threading.Thread.__init__(self)
        self._lock = threading.Lock()
        self.my_config = my_config
        self.session = session
        self.alive = None
        self.event = None
        self.jitter = self.DEFAULT_JITTER
        self._state = 'Idle'
        self._sleeping = None
        self._interrupt = None
        self._rescanning = False
        self._rescan_waiters = []
        self._last_rescan_count = 0
        self._last_saved = time.time()  # Saving right away would be silly

        # Make locked versions of a few functions
        self._load_state = self._locked(self._unlocked_load_state)
        self.open = self._locked(self._unlocked_open)
        self.sync_mail = self._locked(self._unlocked_sync_mail)
        self.rescan_mailbox = self._locked(self._unlocked_rescan_mailbox)
        self.take_over_mailbox = self._locked(self._unlocked_take_over_mailbox)
        self.discover_mailboxes = self._locked(
            self._unlocked_discover_mailboxes)

    def __str__(self):
        rv = ': '.join([threading.Thread.__str__(self), self._state])
        if self._sleeping > 0:
            rv += '(%s)' % self._sleeping
        return rv

    def _locked(self, func):
        def locked_func(*args, **kwargs):
            try:
                self._lock.acquire()
                ostate, self._state = self._state, func.__name__
                return func(*args, **kwargs)
            except:
                traceback.print_exc()
                raise
            finally:
                self._state = ostate
                try:
                    self._lock.release()
                except thread.error:
                    pass
        return locked_func

    def _pfn(self):
        return 'mail-source.%s' % self.my_config._key

    def _unlocked_load_state(self):
        config, my_config = self.session.config, self.my_config
        events = list(config.event_log.incomplete(source=self,
                                                  data_id=my_config._key))
        if events:
            self.event = events[0]
        else:
            self.event = config.event_log.log(source=self,
                                              flags=Event.RUNNING,
                                              message=_('Starting up'),
                                              data={'id': my_config._key})

    def _save_state(self):
        self.session.config.event_log.log_event(self.event)

    def _log_status(self, message):
        self.event.message = message
        self.session.config.event_log.log_event(self.event)
        if 'sources' in self.session.config.sys.debug:
            self.session.ui.debug('%s: %s' % (self, message))

    def _unlocked_open(self):
        """Open mailboxes or connect to the remote mail source."""
        raise NotImplemented('Please override _open in %s' % self)

    def _has_mailbox_changed(self, mbx, state):
        """For the default sync_mail routine, report if mailbox changed."""
        raise NotImplemented('Please override _open in %s' % self)

    def _mark_mailbox_rescanned(self, mbx, state):
        """For the default sync_mail routine, note mailbox was rescanned."""
        raise NotImplemented('Please override _open in %s' % self)

    def _path(self, mbx):
        if mbx.path.startswith('@'):
            return self.session.config.sys.mailbox[mbx.path[1:]]
        else:
            return mbx.path

    def _unlocked_sync_mail(self):
        """Iterates through all the mailboxes and scans if necessary."""
        def unsorted(l):
            l.sort(key=lambda k: random.randint(0, 500))
            return l
        config = self.session.config
        self._last_rescan_count = rescanned = errors = 0
        self._interrupt = None
        batch = self.RESCAN_BATCH_SIZE
        errors = rescanned = 0
        if self.session.config.sys.debug:
            batch //= 10
        for mbx in unsorted(self.my_config.mailbox.values()):
            if mailpile.util.QUITTING or self._interrupt:
                self._log_status(_('Interrupted: %s'
                                   ) % (self._interrupt or _('Quitting')))
                self._interrupt = None
                break
            try:
                state = {}
                # Generally speaking, we only rescan if a mailbox looks like
                # it has changed. However, 1/20th of the time we take a look
                # anyway just in case looks are deceiving.
                if batch > 0 and (self._has_mailbox_changed(mbx, state) or
                                  random.randint(0, 20) == 10):

                    count = self._unlocked_rescan_mailbox(mbx._key,
                                                          stop_after=batch)
                    if count >= 0:
                        batch -= count
                        if (count and batch > 0 and
                                not self._interrupt and
                                not mailpile.util.QUITTING):
                            self._mark_mailbox_rescanned(mbx, state)
                            rescanned += 1
                    else:
                        errors += 1
            except (NoSuchMailboxError, IOError, OSError):
                errors += 1
            except:
                self._log_status(_('Internal error'))
                raise

        if batch > 0:
            self._log_status(_('Checking for new mailboxes'))
            self._unlocked_discover_mailboxes(self.my_config.discovery.paths)

        if errors:
            self._log_status(_('Rescanned %d mailboxes, failed to rescan %d'
                               ) % (rescanned, errors))
        else:
            self._log_status(_('Rescanned %d mailboxes') % rescanned)
        self._last_rescan_count = rescanned
        return rescanned

    def _jitter(self, seconds):
        return seconds + random.randint(0, self.jitter)

    def _sleep(self, seconds):
        if self.session.config.sys.debug:
            seconds //= 10
        if self._sleeping != 0:
            self._sleeping = seconds
{'fraction_non_alphanumeric': 0.08484170801593403, 'fraction_numerical': 0.006848836396673422, 'mean_word_length': 1.7466410748560461, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 30, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '342018', 'n_tokens_mistral': 4042, 'n_tokens_neox': 3825, 'n_words': 1006}