<commit_before>"""
Auto-download of ancillary files.

It allows Operations to specify a serious of source locations (http/ftp/rss URLs)
and destination locations to download to.

This is intended to replace Operations maintenance of many diverse and
complicated scripts with a single, central configuration file.
"""
from __future__ import print_function
import fcntl
import logging
import os
import stat
import sys
import heapq
import time
import multiprocessing
import signal
from setproctitle import setproctitle

import arrow
from croniter import croniter

from . import FetchReporter, TaskFailureEmailer, RemoteFetchException
from .load import load_yaml


_log = logging.getLogger(__name__)


def _attempt_lock(lock_file):
    """
    Use the given file as a lock.

    Return true if successful.

    :type lock_file: str
    :rtype: bool
    """
    umask_original = os.umask(0)
    try:
        fp = os.open(lock_file, os.O_WRONLY | os.O_CREAT, stat.S_IWUSR | stat.S_IWGRP | stat.S_IWOTH)
    finally:
        os.umask(umask_original)

    try:
        fcntl.lockf(fp, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except IOError:
        return False

    return True


def _redirect_output(log_file):
    """
    Redirect all output to the given file.

    :type log_file: str
    """
    output = open(log_file, 'w')
    sys.stdout = output
    sys.stderr = output
    logging.getLogger().removeHandler(_LOG_HANDLER)
    handler = logging.StreamHandler(stream=output)
    handler.setFormatter(_LOG_FORMATTER)
    logging.getLogger().addHandler(handler)


def _run_module(reporter, name, module, scheduled_time, log_directory, lock_directory):
    """
    Run the given module in a subprocess
    :type reporter: FetchReporter
    :type name: str
    :type module: DataSource
    :rtype: ScheduledProcess
    """
    p = ScheduledProcess(
        reporter, name, module, scheduled_time, log_directory, lock_directory
    )

    _log.debug('Module info %r', module)
    _log.info('Starting %r. Log %r, Lock %r', p.name, p.log_file, p.lock_file)
    p.start()
    return p


class ScheduledProcess(multiprocessing.Process):
    """
    A subprocess to run a module.
    """

    def __init__(self, reporter, name, module, scheduled_time, log_directory, lock_directory):
        """
        :type reporter: onreceipt.fetch.FetchReporter
        :type name: str
        :type module: onreceipt.fetch.DataSource
        :type scheduled_time: float
        :type log_directory: str
        :type lock_directory: str
        """
        super(ScheduledProcess, self).__init__()
        _id = _sanitize_for_filename(name)
        lock_file = os.path.join(
            lock_directory,
            '{id}.lck'.format(id=_id)
        )
        scheduled_time_st = time.strftime('%H%M', time.localtime(scheduled_time))
        log_file = os.path.join(
            log_directory,
            '{time}-{id}.log'.format(
                id=_id,
                time=scheduled_time_st
            )
        )

        self.log_file = log_file
        self.lock_file = lock_file
        self.id = _id
        self.name = 'fetch {} {}'.format(scheduled_time_st, name)
        self.module = module
        self.reporter = reporter

    def run(self):
        """
        Configure the environment and run our module.
        """
        _init_signals()
        _redirect_output(self.log_file)

        if not _attempt_lock(self.lock_file):
            _log.debug('Lock is activated. Skipping run. %r', self.name)
            sys.exit(0)

        setproctitle(self.id)
        _log.debug('Triggering %s: %r', self.name, self.module)
        try:
            self.module.trigger(self.reporter)
        except RemoteFetchException as e:
            print('-' * 10)
            print(e.message)
            print('-' * 10)
            print(e.detailed)
            sys.exit(1)


def _init_signals(trigger_exit=None, trigger_reload=None):
    """
    Set signal handlers.

    :param trigger_reload: Handler for reload
    :type trigger_exit: Handler for exit
    """
    # For a SIGINT signal (Ctrl-C) or SIGTERM signal (`kill <pid>` command), we start a graceful shutdown.
    signal.signal(signal.SIGINT, trigger_exit if trigger_exit else signal.SIG_DFL)
    signal.signal(signal.SIGTERM, trigger_exit if trigger_exit else signal.SIG_DFL)

    # SIGHUP triggers a reload of config (following conventions of many daemons).
    signal.signal(signal.SIGHUP, trigger_reload if trigger_reload else signal.SIG_DFL)

<commit_msg>Fix pylint warning: reserved word as attribute name.<commit_after>"""
Auto-download of ancillary files.

It allows Operations to specify a serious of source locations (http/ftp/rss URLs)
and destination locations to download to.

This is intended to replace Operations maintenance of many diverse and
complicated scripts with a single, central configuration file.
"""
from __future__ import print_function
import fcntl
import logging
import os
import stat
import sys
import heapq
import time
import multiprocessing
import signal
from setproctitle import setproctitle

import arrow
from croniter import croniter

from . import FetchReporter, TaskFailureEmailer, RemoteFetchException
from .load import load_yaml


_log = logging.getLogger(__name__)


def _attempt_lock(lock_file):
    """
    Use the given file as a lock.

    Return true if successful.

    :type lock_file: str
    :rtype: bool
    """
    umask_original = os.umask(0)
    try:
        fp = os.open(lock_file, os.O_WRONLY | os.O_CREAT, stat.S_IWUSR | stat.S_IWGRP | stat.S_IWOTH)
    finally:
        os.umask(umask_original)

    try:
        fcntl.lockf(fp, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except IOError:
        return False

    return True


def _redirect_output(log_file):
    """
    Redirect all output to the given file.

    :type log_file: str
    """
    output = open(log_file, 'w')
    sys.stdout = output
    sys.stderr = output
    logging.getLogger().removeHandler(_LOG_HANDLER)
    handler = logging.StreamHandler(stream=output)
    handler.setFormatter(_LOG_FORMATTER)
    logging.getLogger().addHandler(handler)


def _run_module(reporter, name, module, scheduled_time, log_directory, lock_directory):
    """
    Run the given module in a subprocess
    :type reporter: FetchReporter
    :type name: str
    :type module: DataSource
    :rtype: ScheduledProcess
    """
    p = ScheduledProcess(
        reporter, name, module, scheduled_time, log_directory, lock_directory
    )

    _log.debug('Module info %r', module)
    _log.info('Starting %r. Log %r, Lock %r', p.name, p.log_file, p.lock_file)
    p.start()
    return p


class ScheduledProcess(multiprocessing.Process):
    """
    A subprocess to run a module.
    """

    def __init__(self, reporter, name, module, scheduled_time, log_directory, lock_directory):
        """
        :type reporter: onreceipt.fetch.FetchReporter
        :type name: str
        :type module: onreceipt.fetch.DataSource
        :type scheduled_time: float
        :type log_directory: str
        :type lock_directory: str
        """
        super(ScheduledProcess, self).__init__()
        id_ = _sanitize_for_filename(name)
        lock_file = os.path.join(
            lock_directory,
            '{id}.lck'.format(id=id_)
        )
        scheduled_time_st = time.strftime('%H%M', time.localtime(scheduled_time))
        log_file = os.path.join(
            log_directory,
            '{time}-{id}.log'.format(
                id=id_,
                time=scheduled_time_st
            )
        )

        self.log_file = log_file
        self.lock_file = lock_file
        self.id_ = id_
        self.name = 'fetch {} {}'.format(scheduled_time_st, name)
        self.module = module
        self.reporter = reporter

    def run(self):
        """
        Configure the environment and run our module.
        """
        _init_signals()
        _redirect_output(self.log_file)

        if not _attempt_lock(self.lock_file):
            _log.debug('Lock is activated. Skipping run. %r', self.name)
            sys.exit(0)

        setproctitle(self.id_)
        _log.debug('Triggering %s: %r', self.name, self.module)
        try:
            self.module.trigger(self.reporter)
        except RemoteFetchException as e:
            print('-' * 10)
            print(e.message)
            print('-' * 10)
            print(e.detailed)
            sys.exit(1)


def _init_signals(trigger_exit=None, trigger_reload=None):
    """
    Set signal handlers.

    :param trigger_reload: Handler for reload
    :type trigger_exit: Handler for exit
    """
    # For a SIGINT signal (Ctrl-C) or SIGTERM signal (`kill <pid>` command), we start a graceful shutdown.
    signal.signal(signal.SIGINT, trigger_exit if trigger_exit else signal.SIG_DFL)
    signal.signal(signal.SIGTERM, trigger_exit if trigger_exit else signal.SIG_DFL)

    # SIGHUP triggers a reload of config (following conventions of many daemons).
    signal.signal(signal.SIGHUP, trigger_reload if trigger_reload else signal.SIG_DFL)


{'fraction_non_alphanumeric': 0.09250857585481907, 'fraction_numerical': 0.001549186676994578, 'mean_word_length': 2.696523517382413, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3251145', 'n_tokens_mistral': 2826, 'n_tokens_neox': 2729, 'n_words': 737}