"""
Library for sending batch notifications from the Luigi scheduler. This module
is internal to Luigi and not designed for use in other contexts.
"""

import collections
import time
from datetime import datetime

import luigi
import luigi.parameter
import luigi.task
from luigi.notifications import email, send_email


class batch_email(luigi.task.Config):
    email_interval = luigi.parameter.IntParameter(
        default=60,
        config_path=dict(section="batch-notifier", name="email-interval-minutes"),
        description="Number of minutes between e-mail sends (default: 60)",
    )
    batch_mode = luigi.parameter.ChoiceParameter(
        default="unbatched_params",
        choices=("family", "all", "unbatched_params"),
        description='Method used for batching failures in e-mail. If "family" all failures for '
        'tasks with the same family will be batched. If "unbatched_params", all '
        "failures for tasks with the same family and non-batched parameters will be "
        'batched. If "all", tasks will only be batched if they have identical names.',
    )
    error_lines = luigi.parameter.IntParameter(default=20, description="Number of lines to show from each error message. 0 means show all")
    error_messages = luigi.parameter.IntParameter(default=1, description="Number of error messages to show for each group")
    group_by_error_messages = luigi.parameter.BoolParameter(default=True, description="Group items with the same error messages together")


class ExplQueue(collections.OrderedDict):
    def __init__(self, num_items):
        self.num_items = num_items
        super(ExplQueue, self).__init__()

    def enqueue(self, item):
        self.pop(item, None)
        self[item] = datetime.now()
        if len(self) > self.num_items:
            self.popitem(last=False)  # pop first item if past length


def _fail_queue(num_messages):
    return lambda: collections.defaultdict(lambda: ExplQueue(num_messages))


def _plural_format(template, number, plural="s"):
    if number == 0:
        return ""
    return template.format(number, "" if number == 1 else plural)


class BatchNotifier:
    def __init__(self, **kwargs):
        self._config = batch_email(**kwargs)
        self._fail_counts = collections.defaultdict(collections.Counter)
        self._disabled_counts = collections.defaultdict(collections.Counter)
        self._scheduling_fail_counts = collections.defaultdict(collections.Counter)
        self._fail_expls = collections.defaultdict(_fail_queue(self._config.error_messages))
        self._update_next_send()

        self._email_format = email().format
        if email().receiver:
            self._default_owner = set(filter(None, email().receiver.split(",")))
        else:
            self._default_owner = set()

    def _update_next_send(self):
        self._next_send = time.time() + 60 * self._config.email_interval

    def _key(self, task_name, family, unbatched_args):
        if self._config.batch_mode == "all":
            return task_name
        elif self._config.batch_mode == "family":
            return family
        elif self._config.batch_mode == "unbatched_params":
            param_str = ", ".join("{}={}".format(k, v) for k, v in unbatched_args.items())
            return "{}({})".format(family, param_str)
        else:
            raise ValueError("Unknown batch mode for batch notifier: {}".format(self._config.batch_mode))

    def _format_expl(self, expl):
        lines = expl.rstrip().split("\n")[-self._config.error_lines :]
        if self._email_format == "html":
            return "<pre>{}</pre>".format("\n".join(lines))
        else:
            return "\n{}".format("\n".join(map("      {}".format, lines)))

    def _expl_body(self, expls):
        lines = [self._format_expl(expl) for expl in expls]
        if lines and self._email_format != "html":
            lines.append("")
        return "\n".join(lines)

    def _format_task(self, task_tuple):
        task, failure_count, disable_count, scheduling_count = task_tuple
        counts = [
            _plural_format("{} failure{}", failure_count),
            _plural_format("{} disable{}", disable_count),
            _plural_format("{} scheduling failure{}", scheduling_count),
        ]
        count_str = ", ".join(filter(None, counts))
        return "{} ({})".format(task, count_str)

    def _format_tasks(self, tasks):
        lines = map(self._format_task, sorted(tasks, key=self._expl_key))
        if self._email_format == "html":
            return "<li>{}".format("\n<br>".join(lines))
        else:
            return "- {}".format("\n  ".join(lines))

    def _owners(self, owners):
        return self._default_owner | set(owners)

    def add_failure(self, task_name, family, unbatched_args, expl, owners):
        key = self._key(task_name, family, unbatched_args)
        for owner in self._owners(owners):
            self._fail_counts[owner][key] += 1
            self._fail_expls[owner][key].enqueue(expl)

    def add_disable(self, task_name, family, unbatched_args, owners):
        key = self._key(task_name, family, unbatched_args)
        for owner in self._owners(owners):
            self._disabled_counts[owner][key] += 1
            self._fail_counts[owner].setdefault(key, 0)

    def add_scheduling_fail(self, task_name, family, unbatched_args, expl, owners):
        key = self._key(task_name, family, unbatched_args)
        for owner in self._owners(owners):
            self._scheduling_fail_counts[owner][key] += 1
            self._fail_expls[owner][key].enqueue(expl)
            self._fail_counts[owner].setdefault(key, 0)

    def _task_expl_groups(self, expls):
        if not self._config.group_by_error_messages:
            return [((task,), msg) for task, msg in expls.items()]

        groups = collections.defaultdict(list)
        for task, msg in expls.items():
            groups[msg].append(task)
        return [(tasks, msg) for msg, tasks in groups.items()]

    def _expls_key(self, expls_tuple):
        expls = expls_tuple[0]
        num_failures = sum(failures + scheduling_fails for (_1, failures, _2, scheduling_fails) in expls)
        num_disables = sum(disables for (_1, _2, disables, _3) in expls)
        min_name = min(expls)[0]
        return -num_failures, -num_disables, min_name

    def _expl_key(self, expl):
        return self._expls_key(((expl,), None))

    def _email_body(self, fail_counts, disable_counts, scheduling_counts, fail_expls):
        expls = {
            (name, fail_count, disable_counts[name], scheduling_counts[name]): self._expl_body(fail_expls[name]) for name, fail_count in fail_counts.items()
        }
        expl_groups = sorted(self._task_expl_groups(expls), key=self._expls_key)
        body_lines = []
        for tasks, msg in expl_groups:
            body_lines.append(self._format_tasks(tasks))
            body_lines.append(msg)
        body = "\n".join(filter(None, body_lines)).rstrip()
        if self._email_format == "html":
            return "<ul>\n{}\n</ul>".format(body)
        else:
            return body

    def _send_email(self, fail_counts, disable_counts, scheduling_counts, fail_expls, owner):
        num_failures = sum(fail_counts.values())
        num_disables = sum(disable_counts.values())
        num_scheduling_failures = sum(scheduling_counts.values())
        subject_parts = [
            _plural_format("{} failure{}", num_failures),
            _plural_format("{} disable{}", num_disables),
            _plural_format("{} scheduling failure{}", num_scheduling_failures),
        ]
        subject_base = ", ".join(filter(None, subject_parts))
        if subject_base:
            prefix = "" if owner in self._default_owner else "Your tasks have "
            subject = "Luigi: {}{} in the last {} minutes".format(prefix, subject_base, self._config.email_interval)
            email_body = self._email_body(fail_counts, disable_counts, scheduling_counts, fail_expls)
            send_email(subject, email_body, email().sender, (owner,))

    def send_email(self):
        try:
            for owner, failures in self._fail_counts.items():
                self._send_email(
                    fail_counts=failures,
                    disable_counts=self._disabled_counts[owner],
                    scheduling_counts=self._scheduling_fail_counts[owner],
                    fail_expls=self._fail_expls[owner],
                    owner=owner,
                )
        finally:
            self._update_next_send()
            self._fail_counts.clear()
            self._disabled_counts.clear()
            self._scheduling_fail_counts.clear()
            self._fail_expls.clear()

    def update(self):
        if time.time() >= self._next_send:
            self.send_email()
