Прошлая неделя — не последние семь дней: границы периодической выгрузки

В задаче еженедельной выгрузки постмортемов мне понадобился период от 00:00 прошлого понедельника до 00:00 текущего. Формулировка «за последние семь дней» для этого не подходит: она привязывает состав выгрузки к моменту запуска.

Ниже — самостоятельный пример расчёта периода и его проверок, а не описание устройства рабочего экспортёра.

Расписание запуска не определяет период данных

Допустим, задание должно выгрузить записи за неделю с 5 по 12 октября 2026 года. Правая граница — полночь 12 октября, она в период не входит.

Если задание запустилось в понедельник в 09:00 и вычислило now - 7 days, получится другой интервал: с 5 октября 09:00 до 12 октября 09:00. Первые девять часов нужной недели потеряны, девять часов следующей добавлены.

При повторе во вторник обе границы сдвинутся ещё на день. Технически задание повторилось, но набор запрашиваемых данных уже другой.

Поэтому расписание отвечает на вопрос «когда начать работу», а параметры выгрузки — «за какой период получить записи». Для повтора нужны прежние параметры, а не новый расчёт относительно текущего времени.

Сначала определить, к какой неделе относится запись

У одного документа может быть несколько дат. Время инцидента отвечает на вопрос «что произошло за неделю». Время первой публикации — «какие разборы стали доступны за неделю». Время изменения — «какие документы обновились за неделю».

Это три разных отчёта. Выбор часового пояса не исправит ошибку в выборе поля.

В примере будем выгружать впервые опубликованные документы. Условное поле first_published_at хранит неизменяемое время первой публикации. Его наличие и семантику нужно проверить в конкретном источнике: имя published_at само по себе ничего не гарантирует.

Контракт выборки получается таким:

поле: first_published_at
зона календаря: Europe/Berlin
начало: понедельник 00:00, включительно
конец: следующий понедельник 00:00, исключительно

Запись без выбранной даты нельзя молча отнести к дате инцидента или последнего изменения. Это отдельная ошибка данных, которую нужно показать в результате выгрузки.

Использовать интервал с включённым началом и исключённым концом

Условие отбора:

start <= first_published_at < end

Это полуоткрытый интервал [start, end). Запись ровно в start входит в выборку. Запись ровно в end относится уже к следующей неделе.

У соседних периодов правая граница одного совпадает с левой границей другого. Поэтому на общей границе нет ни пропуска, ни двойного включения. Это свойство интервалов, а не гарантия полноты ответа API.

Например, для учебной таблицы PostgreSQL с полем типа timestamptz условие выглядит так. $1 и $2 — передаваемые параметры, а не подстановка строк:

SELECT id, first_published_at
FROM documents
WHERE first_published_at >= $1
  AND first_published_at < $2
ORDER BY first_published_at, id;

BETWEEN здесь не подходит: в PostgreSQL он включает обе границы.¹ Вычитание секунды или микросекунды из end только добавляет зависимость от точности хранения времени.

Если API умеет только включительную верхнюю границу, можно получить более широкий ответ и отфильтровать его по строгому < end на своей стороне. Это не отменяет чтения всех страниц ответа и проверки того, какое именно поле фильтрует API.

Сначала календарные границы, потом UTC

Неделя в выбранной зоне — это две местные полуночи, а не обязательно 168 часов по UTC. При переводе часов смещения на её границах могут различаться. В Python ZoneInfo использует правила IANA, включая переходы между смещениями.²

Поэтому сначала определяем две календарные даты в нужной зоне. Затем независимо переводим обе полуночи в UTC. Не вычисляем левую UTC-границу как правую минус 168 часов.

Функция ниже принимает дату понедельника, который закрывает выгружаемую неделю. Она не читает часы и не использует системный часовой пояс. Для запуска нужен Python 3.9+ и база часовых поясов: системная либо пакет tzdata.²

from datetime import date, datetime, time, timedelta, timezone
from zoneinfo import ZoneInfo


def midnight_utc(day: date, zone: ZoneInfo) -> datetime:
    wall_time = datetime.combine(day, time.min)
    local = wall_time.replace(tzinfo=zone)
    utc = local.astimezone(timezone.utc)
    round_trip = utc.astimezone(zone).replace(tzinfo=None)

    # Не выбираем автоматически трактовку спорной местной полуночи.
    if (round_trip != wall_time
            or local.utcoffset() != local.replace(fold=1).utcoffset()):
        raise ValueError(f"Midnight is missing or ambiguous: {day} {zone.key}")
    return utc


def previous_week(end_monday: date, zone_name: str) -> tuple[datetime, datetime]:
    if type(end_monday) is not date:
        raise TypeError("end_monday must be a date, not a datetime")
    if end_monday.weekday() != 0:
        raise ValueError("end_monday must be a Monday")

    zone = ZoneInfo(zone_name)
    start_monday = end_monday - timedelta(days=7)
    return midnight_utc(start_monday, zone), midnight_utc(end_monday, zone)


start, end = previous_week(date(2026, 10, 12), "Europe/Berlin")
print(start.isoformat(), end.isoformat(), sep="\n")

Результат:

2026-10-04T22:00:00+00:00
2026-10-11T22:00:00+00:00

Сдвиг даты в UTC ожидаем: это те же местные полуночи 5 и 12 октября. astimezone() меняет представление момента, сохраняя сам момент времени.³

Проверка в midnight_utc() нужна для зон и дат, где местная полночь пропущена или встречается дважды. Такой период требует явного решения, а не молчаливого выбора одного смещения. Неизвестная зона тоже завершит расчёт ошибкой.

Сохранить период до первого запроса

Планировщик или оператор выбирает закрывающий понедельник один раз. Даже если выполнение отложилось на следующую неделю, исходная дата не меняется. Пропущенные недели оформляются отдельными периодами.

Перед запросами сохраняем параметры рядом с результатом. Для нашего примера:

time_field: first_published_at
timezone: Europe/Berlin
end_monday: "2026-10-12"
start_utc: "2026-10-04T22:00:00+00:00"
end_utc: "2026-10-11T22:00:00+00:00"
interval: "[start, end)"

Повтор читает сохранённые UTC-границы. Их не нужно заново вычислять по часам, зоне раннера или обновлённым правилам часовых поясов. Время фактического запуска записывается отдельно и служит для диагностики.

Одинаковые границы обеспечивают повторяемость запроса, но не неизменность данных. Документ могли исправить, удалить или сделать видимым позже. Для точного воспроизведения прежнего результата нужен сохранённый снимок исходных данных либо соответствующая возможность источника.

Поздние записи требуют отдельного решения

Представим учебный случай: документ имеет время первой публикации в воскресенье, но API начинает возвращать его только во вторник. Первая выгрузка прошлой недели его не увидит. Выгрузка следующей недели тоже не выберет его по first_published_at.

Для небольшой выгрузки можно повторно сверять несколько последних периодов. Глубина такого пересмотра должна следовать из допустимой задержки источника. Если верхняя граница задержки неизвестна, фиксированный пересмотр последних недель не гарантирует полноту: понадобится сверка более старых периодов или журнал изменений источника.

Повторно найденный документ сопоставляется с уже сохранённым по устойчивому ID, а не по заголовку, имени файла или номеру запуска. Новая версия должна обновить существующую запись либо сформировать явный diff, а не создать второй документ. Ручные правки в месте назначения нельзя молча перезаписывать.

Выборка по updated_at может помочь обнаруживать изменения старых документов, если источник гарантирует нужную семантику поля. Но она не заменяет дату первой публикации: обнаружение изменения и отнесение документа к отчётной неделе решают разные задачи.

Проверить границы, а не только обычный понедельник

Следующий блок можно добавить к коду выше и запустить обычным python3, без -O. Он проверяет конкретные ожидаемые моменты, а не повторяет алгоритм расчёта:

cases = [
    ("2026-10-12", "2026-10-04T22:00:00+00:00", "2026-10-11T22:00:00+00:00", 168),
    ("2026-03-02", "2026-02-22T23:00:00+00:00", "2026-03-01T23:00:00+00:00", 168),
    ("2026-01-05", "2025-12-28T23:00:00+00:00", "2026-01-04T23:00:00+00:00", 168),
    ("2025-03-31", "2025-03-23T23:00:00+00:00", "2025-03-30T22:00:00+00:00", 167),
    ("2025-10-27", "2025-10-19T22:00:00+00:00", "2025-10-26T23:00:00+00:00", 169),
]

for closing_day, expected_start, expected_end, hours in cases:
    window = previous_week(date.fromisoformat(closing_day), "Europe/Berlin")
    assert tuple(value.isoformat() for value in window) == (expected_start, expected_end)
    assert (window[1] - window[0]).total_seconds() == hours * 3600

start, end = previous_week(date(2026, 10, 12), "Europe/Berlin")
next_start, next_end = previous_week(date(2026, 10, 19), "Europe/Berlin")
assert end == next_start

for timestamp, belongs in [
    (start - timedelta(microseconds=1), False),
    (start, True),
    (end - timedelta(microseconds=1), True),
    (end, False),
]:
    assert (start <= timestamp < end) == belongs

assert next_start <= end < next_end
print("Window checks passed")

Длительность здесь вычисляется после перевода в UTC. Вычитание двух datetime с одним и тем же tzinfo в Python не учитывает изменение смещения так, как требуется для измерения прошедших часов.³

Для самого экспортёра нужны ещё проверки поведения:

Полуоткрытые интервалы устраняют пересечения и разрывы на границах соседних недель. Они не устраняют ошибки пагинации, задержки видимости и изменения исходных данных.

Вывод

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

Тогда повтор означает «обработать ту же неделю ещё раз», а не «выбрать другие семь дней». Полноту и обновление данных после этого можно проверять отдельно, не смешивая их с арифметикой календаря.