1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
"""Cancel open orders older than N days.

:Copyright: 2019-2025 Jan Korneffel, Jochen Kupperschmidt
:License: Revised BSD (see `LICENSE` file for details)
"""

from datetime import timedelta

import click
from flask_babel import force_locale, gettext

from byceps.services.shop.order import (
    order_command_service,
    order_service,
    signals as shop_order_signals,
)
from byceps.services.shop.order.email import order_email_service
from byceps.services.shop.order.errors import (
    OrderActionFailedError,
    OrderAlreadyCanceledError,
)
from byceps.services.shop.order.models.order import Order
from byceps.util.result import Err, Ok

from _util import call_with_app_context
from _validators import validate_user_screen_name


@click.command()
@click.option('--shop-id', required=True)
@click.option('--minimum-age-in-days', required=True, type=click.INT)
@click.option('--canceler', required=True, callback=validate_user_screen_name)
@click.option('--locale', required=True, default='en')
@click.option('--reason')
@click.option('--notify/--no-notify', required=True)
@click.option('--limit', default=100)
def execute(
    shop_id,
    minimum_age_in_days: int,
    canceler,
    locale: str,
    reason: str | None,
    notify: bool,
    limit: int,
):
    overdue_orders = _collect_overdue_orders(
        shop_id, minimum_age_in_days, limit
    )
    for order in overdue_orders:
        with force_locale(locale):
            _cancel_order(order, canceler, reason, notify)


def _collect_overdue_orders(
    shop_id, minimum_age_in_days: int, limit: int
) -> list[Order]:
    older_than = timedelta(days=minimum_age_in_days)
    overdue_orders = order_service.get_overdue_orders(
        shop_id, older_than, limit=limit
    )
    click.secho(f'Found {len(overdue_orders)} overdue orders.', fg='yellow')
    return overdue_orders


def _cancel_order(order, canceler, reason: str | None, notify: bool) -> None:
    if not reason:
        reason = gettext(
            'The payment deadline has been exceed. '
            'Place a new order if you are still interested in attending.'
        )

    match order_command_service.cancel_order(order.id, canceler, reason):
        case Ok((canceled_order, canceled_event)):
            shop_order_signals.order_canceled.send(None, event=canceled_event)
            click.secho(
                f'Order {order.order_number} was successfully canceled.',
                fg='green',
            )

            if notify:
                _notify_orderer(canceled_order)
        case Err(e):
            if isinstance(e, OrderAlreadyCanceledError):
                message = (
                    f'Order {order.order_number} has already been canceled.'
                )
            elif isinstance(e, OrderActionFailedError):
                message = f'An error occurred on action execution for order {order.order_number}: {e.details}'
            else:
                message = f'Order {order.order_number} could not be canceled.'
            click.secho(message, fg='red')


def _notify_orderer(order) -> None:
    order_email_service.send_email_for_canceled_order_to_orderer(order)
    click.secho(
        f'Notified orderer of canceled order {order.order_number}.', fg='green'
    )


if __name__ == '__main__':
    call_with_app_context(execute)