Сводка, которая обновляется сама: конвейер целиком

Зачем это нужно

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

Программа, которую запускает планировщик, отличается от программы, которую запускает человек, ровно четырьмя вещами. Их и разбираем.

Сразу целиком

Файл konveyer.py. Четыре шага, три запуска и один упавший источник.

"""Урок 36: сводка, которая обновляется сама.

Четыре шага в одном main(), запись через временный файл, повтор без последствий
и отчёт, который переживает падение источника.
"""

import os
import sys
from pathlib import Path

HERE = Path(__file__).resolve().parent
REPORT = HERE / "svodka.txt"


def fetch(broken=False):
    """Шаг 1: взять данные. Здесь — вместо сети, чтобы урок был одинаковым."""
    if broken:
        raise RuntimeError("источник не ответил")
    return [("2024", 8.7), ("2025", 11.4)]


def clean(rows):
    """Шаг 2: привести к счёту. Строки без числа сюда не проходят."""
    return [(year, value) for year, value in rows if value is not None]


def count(rows):
    """Шаг 3: посчитать то, ради чего всё затевалось."""
    values = [value for _, value in rows]
    return {"лет": len(values), "среднее": round(sum(values) / len(values), 2)}


def render(rows, totals):
    """Шаг 4: собрать текст отчёта."""
    lines = [f"{year}: {value}" for year, value in rows]
    lines.append(f"среднее за {totals['лет']} года: {totals['среднее']}")
    return "\n".join(lines) + "\n"


def save(text, path):
    """Запись через временный файл: читатель не увидит половину отчёта."""
    temporary = path.with_suffix(path.suffix + ".tmp")
    temporary.write_text(text, encoding="utf-8")
    # os.replace меняет имя одним движением: файл либо старый, либо новый.
    os.replace(temporary, path)
    return path


def main(broken=False):
    """Конвейер целиком: видно, что происходит, и в каком порядке."""
    try:
        rows = fetch(broken=broken)
    except RuntimeError as error:
        print("  источник недоступен:", error, "— отчёт остаётся прежним")
        return 1
    rows = clean(rows)
    totals = count(rows)
    save(render(rows, totals), REPORT)
    print("  готово:", REPORT.name, "|", totals)
    return 0


print("== первый запуск")
code = main()
first = REPORT.read_text(encoding="utf-8")
print("  код возврата:", code)

print()
print("== второй запуск: ничего не должно измениться")
main()
print("  файл тот же:", REPORT.read_text(encoding="utf-8") == first)

print()
print("== запись не оставляет половины файла")
print("  временных файлов рядом:", len(list(HERE.glob("*.tmp"))))

print()
print("== источник упал")
code = main(broken=True)
print("  код возврата:", code)
print("  отчёт на месте и не пуст:", REPORT.exists() and REPORT.read_text(encoding="utf-8") == first)

print()
print("== что вернёт планировщику весь запуск")
print("  0 — сделано, 1 — не сделано; сейчас:", code)
sys.exit(0)

Выводит:

== первый запуск
  готово: svodka.txt | {'лет': 2, 'среднее': 10.05}
  код возврата: 0

== второй запуск: ничего не должно измениться
  готово: svodka.txt | {'лет': 2, 'среднее': 10.05}
  файл тот же: True

== запись не оставляет половины файла
  временных файлов рядом: 0

== источник упал
  источник недоступен: источник не ответил — отчёт остаётся прежним
  код возврата: 1
  отчёт на месте и не пуст: True

== что вернёт планировщику весь запуск
  0 — сделано, 1 — не сделано; сейчас: 1

Разбор

main() как оглавление

Главная функция должна читаться как оглавление: взять, привести, посчитать, показать. Ни одного цикла, ни одного if про формат данных — всё это внутри шагов.

Каждый шаг — функция, которая принимает данные и возвращает данные: clean(rows), count(rows), render(rows, totals). Ни один из них не знает, откуда пришли строки и куда уйдёт текст, поэтому любой можно позвать отдельно и проверить в одиночку — это та же мысль, что и в девятом уроке про функции с понятной границей.

Такой main() — ещё и единственное место, где видно порядок. Через полгода вы будете читать его, а не искать по файлам, что за чем идёт.

Запись через временный файл

save пишет не в отчёт, а рядом, а потом переименовывает:

temporary.write_text(text)
os.replace(temporary, path)

os.replace меняет имя одним движением: в файле либо целиком старый отчёт, либо целиком новый. Без этого читатель, открывший страницу в момент записи, увидит половину — а отчёт как раз и открывают утром, когда он обновляется.

Это стоит одной строки и снимает целый класс «иногда файл битый».

Повтор не должен ничего ломать

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

Проверка простая: запустить дважды и сравнить результат. Если он изменился — где-то в конвейере есть шаг, который дописывает вместо того, чтобы записывать заново.

Упавший источник не стирает отчёт

Главное правило урока. Когда fetch бросает исключение, программа не идёт дальше: она сообщает о сбое, возвращает 1 и оставляет вчерашний отчёт на месте.

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

В десятом уроке это было сформулировано как «не глотайте исключение молча». Здесь у того же правила есть цена в файлах.

Код возврата и журнал

Планировщик читает только число, которое вернула программа (двадцать третий урок): 0 — сделано, всё остальное — нет. Поэтому у main() всегда есть return, а запуск заканчивается raise SystemExit(main(sys.argv[1:])).

Журнал — по строке на шаг: сколько строк взяли, сколько осталось после уборки, что посчитали, куда записали. Наутро этого хватает, чтобы понять, что произошло, не запуская ничего заново.

--dry-run

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

Одно условие: сухой прогон должен быть сухим до конца. Если программа по дороге сохранила скачанные данные «просто чтобы не качать дважды» — это уже не сухой прогон, и однажды он перезапишет то, что не собирался.

Строка для cron

5 6 * * * cd /home/you/digest && .venv/bin/python main.py >> data/run.log 2>&1

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

Карта урока

Скажите своими словами

Не подглядывая, ответьте вслух или на бумаге. Ответы — в конце урока.

1. Зачем писать во временный файл, если можно записать сразу в отчёт?
2. Что должна сделать программа, если источник не ответил, и почему?
3. Чем сухой прогон отличается от обычного и что его портит?

Разминка

Три коротких шага перед заданием: предсказать, дописать, починить. Ответы — в конце урока, но сначала ответьте сами.

1. Предскажите. Что напечатает этот конвейер?

def fetch():
    return [1, 2, 3, 4]


def clean(rows):
    return [value for value in rows if value % 2 == 0]


def count(rows):
    return {"строк": len(rows), "сумма": sum(rows)}


print(count(clean(fetch())))
print("шагов в конвейере:", 3)

2. Заполните пропуск. Вместо ... поставьте то, что меняет имя одним движением.

# файл должен быть либо старым, либо новым, но не половиной
import os
from pathlib import Path

HERE = Path(__file__).resolve().parent
target = HERE / "otchet.txt"
target.write_text("старый отчёт\n", encoding="utf-8")

temporary = target.with_suffix(".txt.tmp")
temporary.write_text("новый отчёт\n", encoding="utf-8")
...

print("в файле:", target.read_text(encoding="utf-8").strip())
print("временный остался:", temporary.exists())

3. Почините. Источник упал, а отчёт стал пустым.

# «ну и ладно» стоило дневного отчёта
from pathlib import Path

HERE = Path(__file__).resolve().parent
report = HERE / "svodka2.txt"
report.write_text("2024: 8.7\n2025: 11.4\n", encoding="utf-8")


def fetch():
    raise RuntimeError("источник не ответил")


rows = []
try:
    rows = fetch()
except RuntimeError:
    pass                      # «ну и ладно»

report.write_text("\n".join(str(row) for row in rows) + "\n", encoding="utf-8")
lines = [line for line in report.read_text(encoding="utf-8").split("\n") if line]
print("строк в отчёте:", len(lines))

Задание

Обязательное. Соберите конвейер из четырёх шагов с журналом, флагом --dry-run и кодом возврата. В данных нарочно есть строка без значения и повторённый год — уборка должна убрать первое и оставить последнее значение второго. Запустите три раза: обычно, сухим прогоном и со сломанным источником. Напечатайте журнал каждого запуска, код возврата и то, изменился ли файл.

Ожидаемый вывод:

== обычный запуск
  взято строк: 4
  после уборки: 3
  посчитано: {'лет': 3, 'среднее': 11.53}
  записано: svodka.txt
  код: 0

== сухой прогон
  взято строк: 4
  после уборки: 3
  посчитано: {'лет': 3, 'среднее': 11.53}
  сухой прогон: файл не тронут
  код: 0 | файл не изменился: True

== источник упал
  источник: источник не ответил
  код: 1 | вчерашний отчёт на месте: True

отчёт:
2023: 14.5
2024: 8.7
2025: 11.4
среднее за 3 года: 11.53

Готово, когда: вывод совпадает построчно; main возвращает код, а не печатает его; сухой прогон не пишет ничего; при сбое источника файл не тронут; запись идёт через временный файл.

На своих данных. Возьмите любую свою программу, которая что-то считает и куда-то пишет, и приведите её к этому виду: main() с кодом возврата, журнал по шагам, --dry-run, запись через временный файл. Потом поставьте её в расписание и через неделю прочитайте журнал — он покажет всё, о чём вы не подумали.

По желанию.

• Добавьте флаг --offline, который запрещает ходить в сеть и падает, если на диске ничего нет.
• Замерьте time.perf_counter() вокруг каждого шага и посмотрите, какой из них занимает всё время.
• Запустите две копии одновременно и убедитесь, что замок из двадцать третьего урока им нужен.

Куда это встанет в проекте

Шестнадцатый шаг, и им заканчивается модуль: сводка обновляется сама.

main() стал оглавлением, у него появились --dry-run и --offline, код возврата и журнал по строке на шаг. sholu/saqtau.py пишет всё через временный файл и os.replace. А источник, который не ответил, оставляет вчерашний отчёт на месте — и возвращает планировщику единицу.

В README шага лежит готовая строка для cron. После неё сводка живёт сама: раз в сутки берёт данные, считает, рисует, собирает страницу — и сообщает о себе только тогда, когда перестала возвращать ноль.

Долги. Журнал пишется в stdout и уходит в файл перенаправлением — этого хватает для одной программы на одной машине. Ротации у него нет, и рано или поздно он вырастет; это тот же долг, что был объявлен в уроке про расписание, и он всё ещё ждёт сервера.

Ответы

{'строк': 2, 'сумма': 6}
шагов в конвейере: 3

import os
from pathlib import Path

HERE = Path(__file__).resolve().parent
target = HERE / "otchet.txt"
target.write_text("старый отчёт\n", encoding="utf-8")

temporary = target.with_suffix(".txt.tmp")
temporary.write_text("новый отчёт\n", encoding="utf-8")
os.replace(temporary, target)

print("в файле:", target.read_text(encoding="utf-8").strip())
print("временный остался:", temporary.exists())

в файле: новый отчёт
временный остался: False

from pathlib import Path

HERE = Path(__file__).resolve().parent
report = HERE / "svodka2.txt"
report.write_text("2024: 8.7\n2025: 11.4\n", encoding="utf-8")


def fetch():
    raise RuntimeError("источник не ответил")


try:
    rows = fetch()
except RuntimeError as error:
    print("источник недоступен:", error, "— отчёт не трогаем")
else:
    report.write_text("\n".join(str(row) for row in rows) + "\n", encoding="utf-8")

print("строк в отчёте:", len([line for line in report.read_text(encoding="utf-8").split("\n") if line]))

источник недоступен: источник не ответил — отчёт не трогаем
строк в отчёте: 2

Источники

• os.replace — переименование, которое происходит целиком или не происходит.
• sys.exit и коды возврата — что программа сообщает тому, кто её запустил.
• Формат crontab — пять полей расписания и переменные окружения.