<commit_before>"""This is Gratipay's payday algorithm.

Exchanges (moving money between Gratipay and the outside world) and transfers
(moving money amongst Gratipay users) happen within an isolated event called
payday. This event has duration (it's not punctiliar).

Payday is designed to be crash-resistant. Everything that can be rolled back
happens inside a single DB transaction. Exchanges cannot be rolled back, so they
immediately affect the participant's balance.

"""
from __future__ import unicode_literals

import itertools
from multiprocessing.dummy import Pool as ThreadPool

from balanced import CardHold

import aspen.utils
from aspen import log
from gratipay.billing.exchanges import (
    ach_credit, cancel_card_hold, capture_card_hold, create_card_hold, upcharge
)
from gratipay.exceptions import NegativeBalance
from gratipay.models import check_db
from psycopg2 import IntegrityError


with open('sql/payday.sql') as f:
    PAYDAY = f.read()

with open('sql/fake_payday.sql') as f:
    FAKE_PAYDAY = f.read()


class ExceptionWrapped(Exception): pass


def threaded_map(func, iterable, threads=5):
    pool = ThreadPool(threads)
    def g(*a, **kw):
        # Without this wrapper we get a traceback from inside multiprocessing.
        try:
            return func(*a, **kw)
        except Exception as e:
            import traceback
            raise ExceptionWrapped(e, traceback.format_exc())
    try:
        r = pool.map(g, iterable)
    except ExceptionWrapped as e:
        print(e.args[1])
        raise e.args[0]
    pool.close()
    pool.join()
    return r


class NoPayday(Exception):
    __str__ = lambda self: "No payday found where one was expected."


class Payday(object):
    """Represent an abstract event during which money is moved.

    On Payday, we want to use a participant's Gratipay balance to settle their
    tips due (pulling in more money via credit card as needed), but we only
    want to use their balance at the start of Payday. Balance changes should be
    atomic globally per-Payday.

    Here's the call structure of the Payday.run method:

        run
            payin
                prepare
                create_card_holds
                process_subscriptions
                transfer_takes
                process_draws
                settle_card_holds
                update_balances
                take_over_balances
            payout
            update_stats
            update_cached_amounts
            end

    """


    @classmethod
    def start(cls):
        """Try to start a new Payday.

        If there is a Payday that hasn't finished yet, then the UNIQUE
        constraint on ts_end will kick in and notify us of that. In that case
        we load the existing Payday and work on it some more. We use the start
        time of the current Payday to synchronize our work.

        """
        try:
            d = cls.db.one("""
                INSERT INTO paydays DEFAULT VALUES
                RETURNING id, (ts_start AT TIME ZONE 'UTC') AS ts_start, stage
            """, back_as=dict)
            log("Starting a new payday.")
        except IntegrityError:  # Collision, we have a Payday already.
            d = cls.db.one("""
                SELECT id, (ts_start AT TIME ZONE 'UTC') AS ts_start, stage
                  FROM paydays
                 WHERE ts_end='1970-01-01T00:00:00+00'::timestamptz
            """, back_as=dict)
            log("Picking up with an existing payday.")

        d['ts_start'] = d['ts_start'].replace(tzinfo=aspen.utils.utc)

        log("Payday started at %s." % d['ts_start'])

        payday = Payday()
        payday.__dict__.update(d)
        return payday


    def run(self):
        """This is the starting point for payday.

        This method runs every Thursday. It is structured such that it can be
        run again safely (with a newly-instantiated Payday object) if it
        crashes.

        """
        self.db.self_check()

        _start = aspen.utils.utcnow()
        log("Greetings, program! It's PAYDAY!!!!")

        if self.stage < 1:
            self.payin()
            self.mark_stage_done()
        if self.stage < 2:
            self.payout()
            self.mark_stage_done()
        if self.stage < 3:
            self.update_stats()
            self.update_cached_amounts()
            self.mark_stage_done()

        self.end()
        self.notify_participants()

        _end = aspen.utils.utcnow()
        _delta = _end - _start
        fmt_past = "Script ran for %%(age)s (%s)." % _delta
        log(aspen.utils.to_age(_start, fmt_past=fmt_past))


    def payin(self):
        """The first stage of payday where we charge credit cards and transfer
        money internally between participants.
        """
        with self.db.get_cursor() as cursor:
            self.prepare(cursor, self.ts_start)
            holds = self.create_card_holds(cursor)
            self.process_subscriptions(cursor)
            self.transfer_takes(cursor, self.ts_start)
            self.process_draws(cursor)
            transfers = cursor.all("""
                SELECT * FROM transfers WHERE "timestamp" > %s
            """, (self.ts_start,))
            try:
                self.settle_card_holds(cursor, holds)
                self.update_balances(cursor)
                check_db(cursor)
            except:
                # Dump transfers for debugging
                import csv
                from time import time
                with open('%s_transfers.csv' % time(), 'wb') as f:
                    csv.writer(f).writerows(transfers)
                raise
        self.take_over_balances()
        # Clean up leftover functions
        self.db.run("""
            DROP FUNCTION process_subscription();
            DROP FUNCTION process_take();
            DROP FUNCTION process_draw();
            DROP FUNCTION pay(text, text, numeric, payment_direction);
        """)


    @staticmethod
    def prepare(cursor, ts_start):
        """Prepare the DB: we need temporary tables with indexes and triggers.
        """
        cursor.run(PAYDAY, dict(ts_start=ts_start))
        log('Prepared the DB.')
<commit_msg>Keep this debug helper up to date<commit_after>"""This is Gratipay's payday algorithm.

Exchanges (moving money between Gratipay and the outside world) and transfers
(moving money amongst Gratipay users) happen within an isolated event called
payday. This event has duration (it's not punctiliar).

Payday is designed to be crash-resistant. Everything that can be rolled back
happens inside a single DB transaction. Exchanges cannot be rolled back, so they
immediately affect the participant's balance.

"""
from __future__ import unicode_literals

import itertools
from multiprocessing.dummy import Pool as ThreadPool

from balanced import CardHold

import aspen.utils
from aspen import log
from gratipay.billing.exchanges import (
    ach_credit, cancel_card_hold, capture_card_hold, create_card_hold, upcharge
)
from gratipay.exceptions import NegativeBalance
from gratipay.models import check_db
from psycopg2 import IntegrityError


with open('sql/payday.sql') as f:
    PAYDAY = f.read()

with open('sql/fake_payday.sql') as f:
    FAKE_PAYDAY = f.read()


class ExceptionWrapped(Exception): pass


def threaded_map(func, iterable, threads=5):
    pool = ThreadPool(threads)
    def g(*a, **kw):
        # Without this wrapper we get a traceback from inside multiprocessing.
        try:
            return func(*a, **kw)
        except Exception as e:
            import traceback
            raise ExceptionWrapped(e, traceback.format_exc())
    try:
        r = pool.map(g, iterable)
    except ExceptionWrapped as e:
        print(e.args[1])
        raise e.args[0]
    pool.close()
    pool.join()
    return r


class NoPayday(Exception):
    __str__ = lambda self: "No payday found where one was expected."


class Payday(object):
    """Represent an abstract event during which money is moved.

    On Payday, we want to use a participant's Gratipay balance to settle their
    tips due (pulling in more money via credit card as needed), but we only
    want to use their balance at the start of Payday. Balance changes should be
    atomic globally per-Payday.

    Here's the call structure of the Payday.run method:

        run
            payin
                prepare
                create_card_holds
                process_subscriptions
                transfer_takes
                process_draws
                settle_card_holds
                update_balances
                take_over_balances
            payout
            update_stats
            update_cached_amounts
            end

    """


    @classmethod
    def start(cls):
        """Try to start a new Payday.

        If there is a Payday that hasn't finished yet, then the UNIQUE
        constraint on ts_end will kick in and notify us of that. In that case
        we load the existing Payday and work on it some more. We use the start
        time of the current Payday to synchronize our work.

        """
        try:
            d = cls.db.one("""
                INSERT INTO paydays DEFAULT VALUES
                RETURNING id, (ts_start AT TIME ZONE 'UTC') AS ts_start, stage
            """, back_as=dict)
            log("Starting a new payday.")
        except IntegrityError:  # Collision, we have a Payday already.
            d = cls.db.one("""
                SELECT id, (ts_start AT TIME ZONE 'UTC') AS ts_start, stage
                  FROM paydays
                 WHERE ts_end='1970-01-01T00:00:00+00'::timestamptz
            """, back_as=dict)
            log("Picking up with an existing payday.")

        d['ts_start'] = d['ts_start'].replace(tzinfo=aspen.utils.utc)

        log("Payday started at %s." % d['ts_start'])

        payday = Payday()
        payday.__dict__.update(d)
        return payday


    def run(self):
        """This is the starting point for payday.

        This method runs every Thursday. It is structured such that it can be
        run again safely (with a newly-instantiated Payday object) if it
        crashes.

        """
        self.db.self_check()

        _start = aspen.utils.utcnow()
        log("Greetings, program! It's PAYDAY!!!!")

        if self.stage < 1:
            self.payin()
            self.mark_stage_done()
        if self.stage < 2:
            self.payout()
            self.mark_stage_done()
        if self.stage < 3:
            self.update_stats()
            self.update_cached_amounts()
            self.mark_stage_done()

        self.end()
        self.notify_participants()

        _end = aspen.utils.utcnow()
        _delta = _end - _start
        fmt_past = "Script ran for %%(age)s (%s)." % _delta
        log(aspen.utils.to_age(_start, fmt_past=fmt_past))


    def payin(self):
        """The first stage of payday where we charge credit cards and transfer
        money internally between participants.
        """
        with self.db.get_cursor() as cursor:
            self.prepare(cursor, self.ts_start)
            holds = self.create_card_holds(cursor)
            self.process_subscriptions(cursor)
            self.transfer_takes(cursor, self.ts_start)
            self.process_draws(cursor)
            payments = cursor.all("""
                SELECT * FROM payments WHERE "timestamp" > %s
            """, (self.ts_start,))
            try:
                self.settle_card_holds(cursor, holds)
                self.update_balances(cursor)
                check_db(cursor)
            except:
                # Dump payments for debugging
                import csv
                from time import time
                with open('%s_payments.csv' % time(), 'wb') as f:
                    csv.writer(f).writerows(payments)
                raise
        self.take_over_balances()
        # Clean up leftover functions
        self.db.run("""
            DROP FUNCTION process_subscription();
            DROP FUNCTION process_take();
            DROP FUNCTION process_draw();
            DROP FUNCTION pay(text, text, numeric, payment_direction);
        """)


    @staticmethod
    def prepare(cursor, ts_start):
        """Prepare the DB: we need temporary tables with indexes and triggers.
        """
        cursor.run(PAYDAY, dict(ts_start=ts_start))
        log('Prepared the DB.')

{'fraction_non_alphanumeric': 0.08001293138284975, 'fraction_numerical': 0.00371777256930413, 'mean_word_length': 2.0712335567138247, 'pattern_counts': {'":': 0, '<': 9, '<?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': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4409308', 'n_tokens_mistral': 3588, 'n_tokens_neox': 3388, 'n_words': 1126}