Представьте, что у вас есть следующая ситуация:

  • У вас есть конвейер обработки данных, в котором вы хотите обрабатывать свой DataFrame в определенные диапазоны рабочих дней месяца. Допустим, вы хотите сделать это с рабочего дня +1 до дня +15 и с рабочего дня -5 до последнего рабочего дня месяца. Как бы вы решили это?

Условие состоит в том, что мы должны написать это условие в скрипте Python. Если это условие выполняется, то обрабатывается остальная часть скрипта, в противном случае мы выходим из интерпретатора и останавливаем планировщик.

Сначала мы воспользуемся модулем calendar, чтобы решить эту конкретную задачу. Как определено на странице официальной документации:



calendar — общие функции, связанные с календарем
Исходный код: Lib/calendar.py Этот модуль позволяет вам выводить календари, как программа Unix cal, и предоставляет… docs.python.org



"This module allows you to output calendars like the Unix cal program, 
and provides additional useful functions related to the calendar."

Наша первоначальная стратегия будет следующей:

  • Шаг 1. Создайте список смежных рабочих дат, который будет представлен в виде кортежа: (день месяца, номер дня недели). По умолчанию модуль календаря начинается с 0 для понедельника.
  • Шаг 2: Установите границы и проверьте, попадает ли текущий день месяца в эти границы с помощью оператора IF. Если да, запустите скрипт.
import datetime
import calendar

# Generate a now object
current_timestamp = datetime.datetime.now()

# current_timestamp is now a datetime variable with year, month, minute, second

# Instantiate the calendar object

CalObject = calendar.Calendar() # by default it starts on Monday = 0

Допустим, сегодня 17 января. Мы хотим, чтобы рабочий процесс данных выполнялся с рабочего дня +1 до рабочего дня +15. Мы не хотим вручную проверять, приходится ли 1 января на период с понедельника по пятницу. Ради простоты мы считаем праздниками только субботу и воскресенье и игнорируем другие потенциальные праздники, такие как канун Нового года.

Январь 2023 года начинается в воскресенье. Поскольку наш объект календаря начинается с понедельника по умолчанию, ему нужно сначала пройти через понедельник, вторник, среду, четверг, пятницу и субботу (5 итераций), пока он не достигнет первого дня января 2023 года с помощью функции itermonthdays2(year, month). Эта функция, как сказано в документации, возвращает комбинацию номера дня месяца (0, если он не попадает в месяц) и номера дня недели (понедельник — 0, воскресенье — 6). Поэтому наши первые 7 кортежей будут для января 2023 года: (0, 0), (0, 1), (0, 2), (0, 3), (0, 4), (0, 5), (1). , 6). Что здесь произошло?

Наш объект календаря начинается в понедельник, но январь начинается в воскресенье. Он должен пройти все дни, пока не достигнет воскресенья, чтобы начать с номера 1 первого индекса кортежа. Точно так же январь заканчивается во вторник. Последний день января 2023 года в функции будет (1, 31) → 1 означает вторник, 31 означает 31 января. Остальные кортежи уйдут до следующего понедельника, то есть уже февраля. По умолчанию эти функции календаря будут выполняться с начального дня, когда они установлены (помните, что они настроены на запуск в понедельник по умолчанию) до последнего понедельника, даже если он выходит за рамки желаемой области.

Итак, если мы продолжим нашу работу с кодом, это будет:

# use code from above

business_days_tup = [x for x in CalObject.itermonthdays2(current_timestamp.year, current_timestamp.month) if (x[0] != 0 and x[1] < 5)]
business_day_numbers = [x[0] for x in business_days_tup]

b1 = business_day_numbers[0]
b15 = business_day_numbers[15]
bm5 = business_day_numbers[-5]
bm1 = business_day_numbers[-1]

today_num_day = current_timestamp.today().day

if b1 >= today_num_day <= b15 or bm5 >= today_num_day <= today_num_day and (today_num_day in business_day_numbers):
    print('Run workflow')
    # subprocess.run("""python workflow.py""", shell=True)
else:
    print('Do not run workflow')

В приведенном выше коде я распаковал рабочие дни как числа дней месяца, а затем сохранил четыре переменные, чтобы установить свои границы. С 1 по 15 рабочий день и с -5 до последнего.

Затем мы сохраняем номер дня текущего месяца, используя метод today().day, а затем выполняем оператор IF, чтобы проверить, попадает ли текущий день в диапазоны и действительно ли сегодня рабочий день.

Если да, то мы можем вызвать модуль подпроцесса, чтобы запустить наш скрипт Python в новом процессе.

Способ 2: использование панд

Pandas — отличная библиотека Python, которая помогает нам упростить анализ и обработку данных. Мы можем получить текущую дату без компонента времени, используя функцию pandas to_datetime, а затем соединив метод с normalize. Метод нормализации исключает компонент времени, поэтому мы можем сохранить наш рабочий процесс в зависимости только от дня.

После этого мы можем получить первый день месяца, получив метод today(), а затем используя метод replace(day=1), который принимает параметр дня, и мы передаем 1, чтобы начать с начала месяца.

Чтобы получить последний день месяца без пакета calendar, мы можем создать функцию, которая берет 28-е число текущего месяца (во всех месяцах не менее 28 дней) и добавляет 4 дня. Это приведет к переходу на следующий месяц. Например, 31 января + 4 дня = 4 февраля.

Затем мы берем результат (4 февраля) и вычитаем номер дня из результата (в данном случае 4), что будет 31 января (фактически последний день месяца).

Давайте возьмем другой пример, если это 29 января, и мы добавляем 4 = 2 февраля. Затем вычитаем 2 и снова получаем 31 января. Таким образом, мы всегда будем получать правильный результат.

Метод pandas pd.bdate_range(start, end, freq, holidays) принимает четыре основных параметра:

  • Начало: когда вы хотите начать свой индекс временного диапазона. Я ввожу переменную first_day_month .
  • Конец: вызвать переменную last_day_month .
  • Частота: мы можем ввести C для настройки. Причина, по которой я поставил C, а не B по умолчанию (обозначает рабочие дни), заключается в том, что я могу добавить настраиваемые праздники, выпадающие на субботу и воскресенье.
  • Я создал словарь под названием holidays, в котором каждому празднику присваивается имя, чтобы оно было более удобным для человека. В параметрах праздников я передаю список значений словаря, которые представляют собой временные метки в формате YYYY-MM-DD.
import pandas as pd

# Method 1: Last Day of the Month
# define a function: source Stack Overflow

def get_last_day_month(timestamp=datetime.datetime.today()):
    next_month = timestamp.replace(day=28) + datetime.timedelta(days=4)
    last_day = (next_month - datetime.timedelta(days=next_month.day))
    return last_day

today = pd.to_datetime('today').normalize()
first_day_month = datetime.datetime.today().replace(day=1)
last_day_month = get_last_day_month()
frequency = "C"

holidays = {"January First": "2023-01-01", "Lunar New Year": "2023-01-22"}

working_days = pd.bdate_range(start=first_day_month, end=last_day_month, freq=frequency,
holidays=list(holidays.values()))

if today in working_days:
    print('Run workflow...')
    # subprocess.run("""python myfile.py""")
else:
    print('Do not run workflow...')
    # subprocess.run("""python myfile.py""")

Возможно, вы захотите получить последний день месяца напрямую, не создавая для этого функцию. В этом случае вы можете использовать встроенный метод calendar, как показано ниже:

# Replace variable last_day_month for this

import calendar

# ... use previous code

last_day_month = calendar.monthrange(today.year, today.month)[1]

Я надеюсь, что вы нашли эту статью полезной. Пожалуйста, смотрите ниже библиографию и ссылки.

Ссылки: