Как я автоматизировал 1000+ анализов в день с помощью Python: от рутины к автономной лаборатории
Когда рутина съедает 8 часов в день: цифры на бумаге против реальных пациентов
8 часов в день — не абстрактная цифра из курсов по менеджменту. Это конкретное время, которое я ежедневно тратил на превращение чисел из анализатора в диагнозы в карточках пациентов. С 9:00 до 17:00, без перерыва на кофе, я был оператором копирования. Автоматический биохимический анализатор выдавал строки результатов: PatientID=001, Glucose=5.6, Hb=142. И мой мир делился на две параллельные реальности: цифры на бумаге (или в Excel) и реальные пациенты, которые ждали этих цифр в своём электронном медицинском карте (ЭМК).
Вот как это выглядело в цифрах на 2021 год: - 500+ проб в день через анализатор Cobas c311 (версия ПО 2.1.1). - 15 параметров на каждую пробу: глюкоза, гемоглобин, креатинин и т.д. - Итого 7500+ числовых значений, которые нужно было перенести. - Система ЭМК — старая, на базе PostgreSQL 9.6, без API. Вход — только через веб-интерфейс. - Мой инструмент — Excel 2016 и мышь. Процесс: распечатать результаты из анализатора (PDF), вручную вбить ID пациента и каждое значение в таблицу, потом скопировать строку из Excel и вставить в форму ЭМК.
Время на одну пробу: около 1 минуты. 500 проб = 500 минут = 8 часов 20 минут чистого копирования. Это не включая время на проверку ошибок анализатора, повторные пробы, конфликты по ID. Реальная цифра была ближе к 9 часам.
Я был заведующим клинико-диагностической лабораторией. Моя должность предполагала контроль качества, обучение персонала, внедрение новых методов. Но я был самым дорогим и самым медленным оператором данных в учреждении. Каждый день я закрывал дверь кабинета и включал «режим машины»: монотонное движение рук, глаз, постепенно туман в голове. Ошибки начали появляться уже на сотой пробе.
# Пример того, что я видел в отчете анализатора (фрагмент CSV)
# Это реальный формат вывода Cobas c311, который я получал каждый день
Patient_ID, Sample_Type, Glucose_mmol/L, Hb_g/L, Creatinine_μmol/L
001, Serum, 5.6, 142, 78
002, Plasma, 6.1, 138, 82
003, Serum, 7.8, 155, 91
...
# И это нужно было превратить в вот такой запрос для ЭМК (через веб-форму)
# Вручную, для каждого пациента
UPDATE patient_results
SET glucose = 5.6, hemoglobin = 142, creatinine = 78
WHERE patient_id = '001' AND date = '2021-10-15';
Проблема была не только в времени. Она была в рисках. После 4-5 часов такой работы концентрация падает катастрофически. Я помню день, когда перепутал значения глюкозы и креатинина для пациента с диабетом. Вбил 7.8 (глюкоза) вместо 91 (креатинин) в поле креатинина. Система ЭМК не проверяла логические пределы. Результат попал в карту. К счастью, врач заметил несоответствие и перезвонил в лабораторию. Это был звонок, после которого я понял, что система, построенная на человеческом факторе после 8-часовой рутины, — это прямая угроза пациентам.
Но был и второй фронт — бумага. Любой автоматизации в медицине предшествует тонна бюрократии. Я должен был доказать, что: 1. Автоматизация не нарушает локальные нормативные акты по ведению ЭМК. 2. Данные будут передаваться без изменения (валидация). 3. Система будет иметь журнал аудита всех операций. 4. И самое главное — я не нарушу «принцип единого ввода», который требовал, чтобы результаты вносились только из официального интерфейса анализатора.
Первые попытки были комичными. Я попробовал использовать макросы Excel (VBA), чтобы автоматически заполнять форму ЭМК через браузер. Это выглядело так:
' Макрос, который я написал за ночь (и который сломался на третьем пациенте)
Sub FillEMKForm()
Dim ie As Object
Set ie = CreateObject("InternetExplorer.Application")
ie.navigate "http://internal-emk/patient_result_entry"
' Ждем загрузки (костыль)
Do While ie.Busy Or ie.readyState <> 4
Application.Wait (Now + TimeValue("0:00:01"))
Loop
' Находим поле ID и вбиваем значение из ячейки A2
ie.document.getElementById("patient_id").Value = Range("A2").Value
' ... и так для 15 полей
End Sub
Он не работал. Internet Explorer (единственный браузер, который поддерживала старая система ЭМК) блокировал автоматический ввод из макросов. Я потратил три дня, пытаясь обойти это через SendKeys. Результат — случайные вводы в другие поля, потеря данных, и окончательный запрет от IT-отдела на использование макросов для взаимодействия с ЭМК.
Именно тогда я понял, что Excel и VBA — это тупик. Мне нужен был инструмент, который может работать с данными, валидировать их, логировать каждое действие и, главное, взаимодействовать с системой на уровне, который не будет блокироваться как «автоматический скрипт». И я вспомнил, что в университете я немного писал на Python. Это была версия 2.7, и я делал простые скрипты для статистики. Но сейчас, в 2021, Python 3.9 выглядел как космический корабль compared to моему утлому Excel.
Цифры были настолько очевидными, что даже бюрократия не могла их игнорировать: - Текущее время обработки: 8+ часов. - Время при ручном вводе одной ошибки и её исправления: до 20 минут (поиск в бумажных журналах, переписка, изменение в ЭМК). - Риск ошибки после 4 часов работы: по моим грубым оценкам, 1 на 100 проб.
Я представил эти цифры на совете лаборатории. Не как абстрактное «нужно автоматизировать», а как конкретный отчет по времени и рискам. И получил принципиальное согласие на разработку «инструмента для ускорения и контроля ввода лабораторных данных» — так это было названо в официальном документе. Ключевая фраза: «без изменения текущего workflow и с сохранением полного аудита».
Так начался путь. Первая цель была не «полная автоматизация», а «сократить 8 часов до 4». И я уже знал, что Excel мне не поможет. Нужен был Python, база данных для промежуточного хранения и проверки, и какой-то способ общения с ЭМК без имитации браузера. Я открыл PyCharm Community Edition на своём домашнем компьютере и начал писать первый скрипт. Это была та точка, где цифры на бумаге начали превращаться в код, который должен был вернуть мне время для реальных пациентов.
От Excel макросов до первого Python скрипта: как я начал автоматизировать и сразу накосячил
Вот и пришёл момент, когда я понял: Excel — это не решение, а узаконенный садомазохизм. Макросы на VBA, которыми я гордился, как дипломом о высшем образовании, начали ломаться на каждом втором пациенте. Новый анализатор выдавал данные в .csv, а не в .txt, и моя «умная» книга с 50 листами и 2000 формулами просто не открывалась, весело показывая «Ошибка 1004».
Я сидел в 23:30 в пустой лаборатории, пялясь на ошибку Run-time error '9': Subscript out of range. Кофе остыл. Пациенты не остыли — их анализы ждали. И я принял судьбоносное решение: хватит это терпеть. Я погуглил «как прочитать csv файл». Первая ссылка вела на Stack Overflow с кодом на Python.
Первый скрипт родился через два часа борьбы. Установил Python 3.8 с официального сайта (помню, как галочку «Add Python to PATH» чуть не пропустил — это была бы первая катастрофа). Открыл IDLE. Это был мой космический корабль.
Вот он, этот уродец, который спас и тут же чуть не угробил мою карьеру в тот же день:
# analyze_v1.py - ПРОТОТИП АПОКАЛИПСИСА
# Python 3.8.5
import csv
def read_lab_results(file_path):
"""Читает CSV от анализатора Roche Cobas 6000."""
results = []
try:
with open(file_path, 'r', encoding='cp1251') as f: # У нас была русская кодировка
reader = csv.reader(f, delimiter=';')
for row in reader:
results.append(row)
return results
except Exception as e:
print(f"ОШИБКА: {e}")
return []
def calculate_deviations(results):
"""Считает отклонения от референсных значений. КРИВО."""
ref_values = {
'Глюкоза': (3.9, CBC 6.1),
'АЛТ': (0, 41),
'Креатинин': (44, 97)
}
for row in results:
if len(row) < 3:
continue
name, value, unit = row[0], float(row[1]), row[2]
if name in ref_values:
low, high = ref_values[name]
if value < low:
print(f"ВНИМАНИЕ! {name}: {value} — НИЖЕ НОРМЫ")
elif value > high:
print(f"ВНИМАНИЕ! {name}: {value} — ВЫШЕ НОРМЫ")
if __name__ == "__main__":
# Путь жестко забит. Первая ошибка.
data = read_lab_results("C:\\LabData\\today_results.csv")
calculate_deviations(data)
print("Обработано строк:", len(data))
Я запустил его. Он отработал. На экране появились строчки "ВНИМАНИЕ! Креатинин: 120 — ВЫШЕ НОРМЫ". Я почувствовал себя богом. Автоматизация! Магия! Я сэкономил минут 40 на одной пачке анализов.
А потом наступило утро. И пришла лаборант Анна Петровна.
— Доктор, что это за бумажка? — спросила она, протягивая мне распечатку из принтера.
Я посмотрел. Мой скрипт, гордо отработав ночью, не только вывел результаты в консоль, но и записал их обратно в файл. Криво. Я, в пылу творчества, добавил в конец функцию write_results_to_txt() для «удобства». Но забыл закрыть файловый дескриптор и использовал тот же путь, но с другим расширением. В итоге, исходный today_results.csv был благополучно испорчен. Вместо 150 строк с данными в нём теперь было 80 строк моих комментариев и часть заголовков.
Первый косяк: Отсутствие backup. Второй косяк: Прямая работа с продакшен: данными. Третий косяк: «Тихое» падение. Если файла нет, скрипт просто печатал "Обработано строк: 0" и завершался, как ни в чём не бывало.
Вот исправленная, уже более жизнеспособная версия 2.0, написанная той же ночью, но после литра крепкого чая и осознания своей ничтожности:
# analyze_v2.py - УЧЁТ ГОРЬКОГО ОПЫТА
# Python 3.8.5
import csv
import os
from datetime import datetime
import shutil
def safe_read_lab_results(file_path):
"""Читает CSV, создавая резервную копию ПРЕЖДЕ всего."""
backup_dir = "C:\\LabData\\backup\\"
os.makedirs(backup_dir, exist_ok=True)
# Создаём backup с timestamp
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_path = os.path.join(backup_dir, f"backup_{timestamp}.csv")
shutil.copy2(file_path, backup_path) # Копируем с метаданными
print(f"[INFO] Создан backup: {backup_path}")
results = []
try:
with open(file_path, 'r', encoding='cp1251') as f:
reader = csv.reader(f, delimiter=';')
for row in reader:
if row: # Игнорируем пустые строки
results.append(row)
except FileNotFoundError:
print(f"[CRITICAL] Файл не найден: {file_path}")
return None # Возвращаем None, а не пустой список!
except Exception as e:
print(f"[ERROR] Неожиданная ошибка: {e}")
return None
return results
def calculate_and_save(results, output_path):
"""Считает и сохраняет результат В ОТДЕЛЬНЫЙ файл."""
if results is None:
print("[ERROR] Нет данных для обработки. Выход.")
return
ref_values = {'Глюкоза': (3.9, 6.1), 'АЛТ': (0, 41), 'Креатинин': (44, 97)}
report_lines = ["Отчёт от " + datetime.now().strftime("%d.%m.%Y %H:%M"), "="*30]
for row in results:
if len(row) < 3:
continue
try:
name, value_str, unit = row[0], row[1], row[2]
value = float(value_str.replace(',', '.')) # Учёт русской десятичной запятой!
if name in ref_values:
low, high = ref_values[name]
status = "НОРМА"
if value < low:
status = "НИЖЕ НОРМЫ"
elif value > high:
status = "ВЫШЕ НОРМЫ"
report_lines.append(f"{name}: {value} {unit} [{status}]")
except ValueError:
print(f"[WARN] Не могу преобразовать '{value_str}' в число для '{name}'")
# Пишем в НОВЫЙ файл
try:
with open(output_path, 'w', encoding='utf-8') as f:
f.write("\n".join(report_lines))
print(f"[INFO] Отчёт сохранён: {output_path}")
except Exception as e:
print(f"[ERROR] Не удалось сохранить отчёт: {e}")
if __name__ == "__main__":
source = "C:\\LabData\\today_results.csv"
output = "C:\\LabData\\reports\\report_" + datetime.now().strftime("%Y%m%d_%H%M") + ".txt"
data = safe_read_lab_results(source)
calculate_and_save(data, output)
Что изменилось кардинально?
1. Backup-first подход. Первая строка рабочей логики — создание копии. Это святое.
2. Разделение входных и выходных данных. Скрипт больше не лезет в исходник.
3. Обработка ошибок. Файл не найден? Русская запятая в числе? Пустая строка? Теперь это не падает молча.
4. Логирование. Примитивное, но уже понятное, что происходит ([INFO], [WARN], [CRITICAL]).
С этой версией я прожил полгода. Она обрабатывала 200-300 анализов в день, сэкономив мне около 3 часов рутины ежедневно. Но это был тупик. Потому что дальше появились требования: «а можно чтобы результаты сразу в базу данных?», «а чтобы алерт на критические значения был на телефон?», «а чтобы отчёт в PDF?». И мой монолитный скрипт начал разрастаться в 1000 строк неподдерживаемого кода, в котором боялся разобраться даже я, его создатель.
Это был момент истины. Я понял, что автоматизация рутины — это только первый шаг. Дальше нужно строить систему. Но чтобы её построить, мне пришлось сначала сломать то, что я так гордо создал. Именно об этом — следующий шаг, где появились FastAPI, Redis и очередь задач Celery. Потому что когда скрипт падает в 2 ночи из-за кириллицы в данных, а ты на смене — это уже не смешно.
Core архитектура: FastAPI, Redis и Celery как три кита автоматизации (с реальным кодом)
Excel макросы сдохли героически, пытаясь спарсить CSV с новой аппаратуры. Настало время строить систему, а не латать костыли. Я понял главное: нужна не просто «автоматизация», а архитектура. Такая, чтобы в 3 часа ночи не пришлось объяснять главврачу, почему 500 анализов зависли из-за кривого форматирования даты.
Моя архитектура родилась из трёх принципов: 1. Всё — асинхронно. Лабораторный анализатор не должен ждать, пока я сохраню файл в базу. 2. Всё — в очереди. 1000 анализов в день — это не поток, это лавина. Её нужно дробить и контролировать. 3. Всё — по API. Интеграция с будущими системами (и с моим же фронтендом) должна быть простой, как HTTP-запрос.
Так родилась связка FastAPI + Redis + Celery. Это три кита, на которых держится вся автоматизация.
FastAPI: лицо системы (и её мозжечок)
Выбор пал на FastAPI, а не на Flask или Django, не из-за хайпа. Мне нужна была скорость разработки (время — кровь), автодокументация (чтобы не тратить часы на Swagger) и валидация на уровне типа (ошибки в данных — мой личный кошмар).
Вот как выглядит core-приложение. Версия FastAPI 0.104.1.
# core/app.py
from fastapi import FastAPI, BackgroundTasks, HTTPException
from pydantic import BaseModel, validator, Field
from typing import Optional, List
from datetime import datetime
import logging
from .tasks import process_analysis_batch
app = FastAPI(
title="LabAutoCore API",
description="Ядро автоматизации клинико-диагностической лаборатории",
version="2.1.0"
)
logger = logging.getLogger("lab_api")
class AnalysisSample(BaseModel):
sample_id: str = Field(..., min_length=8, max_length=12, regex=r'^LAB\d+$')
patient_id: str
test_code: str
value: float
unit: str
instrument_id: str = "BC-6800"
measured_at: datetime
@validator('value')
def value_must_be_positive(cls, v):
if v <= 0:
raise ValueError('Значение анализа должно быть положительным')
return v
@app.post("/api/v1/import/batch/", status_code=202)
async def import_batch(
samples: List[AnalysisSample],
background_tasks: BackgroundTasks
):
"""
Приём батча анализов. Валидация -> Постановка в очередь Celery -> Ответ 'Принято'.
"""
if not samples:
raise HTTPException(status_code=400, detail="Пустой список анализов")
batch_id = f"BATCH_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
logger.info(f"Принят батч {batch_id}, образцов: {len(samples)}")
# Асинхронно ставим задачу в очередь Celery
background_tasks.add_task(process_analysis_batch, samples, batch_id)
return {
"status": "accepted",
"batch_id": batch_id,
"message": f"Образцы ({len(samples)} шт.) поставлены в очередь на обработку"
}
@app.get("/health")
async def health_check():
"""Эндпоинт для проверки живости сервиса. Используется Prometheus и Docker."""
return {"status": "healthy", "timestamp": datetime.now().isoformat()}
Ключевой момент — status_code=202 в эндпоинте импорта. Система не обрабатывает анализы тут же. Она говорит: «Принял, иди дальше». Вся тяжёлая работа уходит в фоновые задачи.
Celery + Redis: мышечная память системы
Если FastAPI — это мозг, который принимает решения, то Celery — это руки, которые делают работу. А Redis — это почтальон между ними.
Конфигурация Celery (core/celery_app.py):
# core/celery_app.py
from celery import Celery
import os
redis_host = os.getenv("REDIS_HOST", "localhost")
redis_port = os.getenv("REDIS_PORT", "6379")
app = Celery(
'lab_tasks',
broker=f'redis://{redis_host}:{redis_port}/0',
backend=f'redis://{redis_host}:{redis_port}/1',
include=['core.tasks']
)
# Конфигурация
app.conf.update(
task_serializer='json',
accept_content=['json'],
result_serializer='json',
timezone='Europe/Moscow',
enable_utc=True,
task_acks_late=True, # Подтверждение задачи только после выполнения
worker_prefetch_multiplier=1, # По одной задаче на воркер для критичных данных
task_routes={
'core.tasks.process_analysis_batch': {'queue': 'high_priority'},
'core.tasks.generate_pdf_report': {'queue': 'low_priority'},
}
)
А вот сама задача-воркер, которая делает основную работу:
# core/tasks.py
from .celery_app import app
from .database import save_to_db, check_duplicate
from .business_logic import calculate_reference_range, flag_critical_values
import logging
from datetime import datetime
logger = logging.getLogger("lab_celery")
@app.task(bind=True, max_retries=3, default_retry_delay=60)
def process_analysis_batch(self, samples: list, batch_id: str):
"""
Фоновая задача Celery.
Обрабатывает батч анализов: сохраняет в БД, применяет бизнес-логику.
"""
logger.info(f"Начало обработки батча {batch_id}. Образцов: {len(samples)}")
processed = 0
errors = []
for sample in samples:
try:
# 1. Проверка дубля
if check_duplicate(sample.sample_id, sample.test_code):
logger.warning(f"Дубликат! {sample.sample_id} - {sample.test_code}")
continue
# 2. Бизнес-логика (нормы, флаги)
sample_dict = sample.dict()
sample_dict['ref_low'], sample_dict['ref_high'] = calculate_reference_range(
sample.test_code, sample.patient_id
)
sample_dict['flag'] = flag_critical_values(sample.value, sample_dict['ref_low'], sample_dict['ref_high'])
# 3. Сохранение в основную БД (PostgreSQL)
save_to_db("analysis_results", sample_dict)
processed += 1
except Exception as e:
errors.append({"sample": sample.sample_id, "error": str(e)})
logger.error(f"Ошибка обработки {sample.sample_id}: {e}")
# Логирование результата задачи
if errors:
logger.error(f"Батч {batch_id} обработан с ошибками. Успешно: {processed}, Ошибок: {len(errors)}")
raise self.retry(exc=Exception(f"Ошибки в батче: {errors}")) if len(errors) > 5 else None
else:
logger.info(f"Батч {batch_id} успешно обработан. Образцов: {processed}")
return {"processed": processed, "errors": errors}
Запускается это хозяйство одной командой:
# Запуск воркеров Celery для high_priority очереди (анализы)
celery -A core.celery_app worker --loglevel=info --concurrency=4 --queues=high_priority -n worker1@%h
Docker Compose: как собрать это в один клик
Всё это крутится в Docker. Мой docker-compose.prod.yml:
version: '3.8'
services:
redis:
image: redis:7-alpine
container_name: lab_redis
command: redis-server --appendonly yes
volumes:
- redis_data:/data
ports:
- "6379:6379"
restart: unless-stopped
postgres:
image: postgres:15-alpine
container_name: lab_postgres
environment:
POSTGRES_DB: lab_automation
POSTGRES_USER: lab_admin
POSTGRES_PASSWORD: ${DB_PASSWORD}
volumes:
- postgres_data:/var/lib/postgresql/data
- ./sql/init.sql:/docker-entrypoint-initdb.d/init.sql
ports:
- "5432:5432"
restart: unless-stopped
api:
build: ./backend
container_name: lab_api
command: uvicorn core.app:app --host 0.0.0.0 --port 8000 --workers 2
environment:
REDIS_HOST: redis
DATABASE_URL: postgresql://lab_admin:${DB_PASSWORD}@postgres/lab_automation
ports:
- "8000:8000"
depends_on:
- redis
- postgres
restart: unless-stopped
celery_worker:
build: ./backend
container_name: lab_celery_worker
command: celery -A core.celery_app worker --loglevel=info --concurrency=4 -Q high_priority -n worker@%h
environment:
REDIS_HOST: redis
DATABASE_URL: postgresql://lab_admin:${DB_PASSWORD}@postgres/lab_automation
depends_on:
- redis
- postgres
- api
restart: unless-stopped
volumes:
redis_data:
postgres_data:
Запуск всей инфраструктуры — одна команда:
DB_PASSWORD=supersecret docker-compose -f docker-compose.prod.yml up -d
Что это дало на практике?
Раньше импорт 500 анализов через Excel-макрос занимал 40-50 минут (с зависаниями). Теперь:
1. Лаборант загружает CSV в веб-интерфейс (об этом позже) → FastAPI валидирует за 2-3 секунды.
2. status_code=202 → Instant feedback, пользователь свободен.
3. Celery воркеры в фоне обрабатывают батч за 10-12 минут, не блокируя систему.
4. Все ошибки логируются, критические значения помечаются флагами.
Система перестала быть «скриптом». Она стала платформой, с чёткими границами ответственности: API принимает, очередь буферизирует, воркеры обрабатывают. И если ночью «умрёт» воркер, анализы просто накопятся в Redis, а не потеряются. Утром я перезапущу контейнер, и очередь начнёт разгребаться.
Следующий шаг был неизбежен: одна машина — это точка отказа. Пора было учиться на кластере и следить за метриками, чтобы не узнать о проблеме от главврача.
От одного сервера к кластеру: Docker, мониторинг Prometheus и алерт в 3 AM
Однажды мой шеф, главврач, посмотрел на график роста нагрузки и спросил: «А что будет, когда сервер накроется в 9 утра в понедельник?». Я промолчал, потому что ответ был очевиден: накроется всё. Мой уютный монолит на FastAPI, Celery и Redis, крутившийся на одном systemd-сервисе Ubuntu, уже дышал на ладан при 800 анализах. Пиковые нагрузки в 1100+ приводили к OutOfMemory, падению воркеров и моему личному инфаркту. Пришла пора взрослеть и оркестрировать.
Переход от одного сервиса к кластеру — это не про «добавим ещё серверов». Это смена парадигмы: от ручного управления процессами к декларативному описанию всей системы. Моим спасательным кругом стал Docker и чуть позже — docker-compose.
Первым делом я «законтейнеризировал» каждую часть системы. Не для красоты, а для воспроизводимости и изоляции. Больше не нужно гадать, почему на проде не работает, а на моём ноуте — да.
Вот базовый Dockerfile для моего FastAPI-приложения (я назвал его lab-core):
# Dockerfile.api
FROM python:3.10-slim
WORKDIR /app
# Копируем зависимости отдельно для кэширования
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Копируем весь код
COPY . .
# Запускаем через uvicorn с несколькими воркерами
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]
А вот сердце новой системы — docker-compose.yml. Это был мой первый раз, и я сделал классическую ошибку: положил всё в одну сеть без сегментации. Но для начала сойдёт.
# docker-compose.v1.yml (с ошибками, но рабочий)
version: '3.8'
services:
redis:
image: redis:7-alpine
command: redis-server --appendonly yes
volumes:
- redis_data:/data
restart: unless-stopped
postgres:
image: postgres:15-alpine
environment:
POSTGRES_DB: lab_db
POSTGRES_USER: lab_user
POSTGRES_PASSWORD: ${DB_PASSWORD} # Читаем из .env
volumes:
- postgres_data:/var/lib/postgresql/data
restart: unless-stopped
api:
build:
context: ./lab-core
dockerfile: Dockerfile.api
environment:
- REDIS_URL=redis://redis:6379/0
- DATABASE_URL=postgresql://lab_user:${DB_PASSWORD}@postgres:5432/lab_db
ports:
- "8000:8000"
depends_on:
- redis
- postgres
restart: unless-stopped
celery_worker:
build:
context: ./lab-core
dockerfile: Dockerfile.api # Тот же образ
command: celery -A tasks worker --loglevel=info --concurrency=4
environment:
- REDIS_URL=redis://redis:6379/0
- DATABASE_URL=postgresql://lab_user:${DB_PASSWORD}@postgres:5432/lab_db
depends_on:
- redis
- postgres
restart: unless-stopped
celery_beat:
build:
context: ./lab-core
dockerfile: Dockerfile.api
command: celery -A tasks beat --loglevel=info
environment:
- REDIS_URL=redis://redis:6379/0
depends_on:
- redis
restart: unless-stopped
volumes:
redis_data:
postgres_data:
Запуск одной командой: docker-compose -f docker-compose.v1.yml --env-file .env up -d. Магия. Система встала. Но магия быстро закончилась, когда я понял, что не вижу, что там внутри творится. При падении воркера я узнавал об этом только по нарастающей очереди в Redis. Нужен был мониторинг.
Я выбрал классический стек: Prometheus для сбора метрик и Grafana для визуализации. Alertmanager — для того, чтобы будить меня в 3 ночи, а не главврача в 9 утра.
Добавил в docker-compose новые сервисы и главное — экспортеры метрик для каждого компонента.
# Добавление в docker-compose.monitoring.yml
prometheus:
image: prom/prometheus:v2.45.0
volumes:
- ./prometheus.yml:/etc/prometheus/prometheus.yml
- prometheus_data:/prometheus
command:
- '--config.file=/etc/prometheus/prometheus.yml'
- '--storage.tsdb.path=/prometheus'
- '--web.console.libraries=/etc/prometheus/console_libraries'
- '--web.console.templates=/etc/prometheus/consoles'
- '--storage.tsdb.retention.time=200h'
- '--web.enable-lifecycle'
ports:
- "9090:9090"
restart: unless-stopped
grafana:
image: grafana/grafana:10.0.0
environment:
- GF_SECURITY_ADMIN_PASSWORD=${GRAFANA_PASSWORD}
volumes:
- grafana_data:/var/lib/grafana
ports:
- "3000:3000"
restart: unless-stopped
alertmanager:
image: prom/alertmanager:v0.25.0
volumes:
- ./alertmanager.yml:/etc/alertmanager/alertmanager.yml
ports:
- "9093:9093"
restart: unless-stopped
redis_exporter:
image: oliver006/redis_exporter:v1.54.0
environment:
- REDIS_ADDR=redis://redis:6379
restart: unless-stopped
postgres_exporter:
image: prometheuscommunity/postgres-exporter:v0.13.2
environment:
- DATA_SOURCE_NAME=postgresql://lab_user:${DB_PASSWORD}@postgres:5432/lab_db?sslmode=disable
restart: unless-stopped
node_exporter:
image: prom/node-exporter:v1.6.0
volumes:
- /proc:/host/proc:ro
- /sys:/host/sys:ro
- /:/rootfs:ro
command:
- '--path.procfs=/host/proc'
- '--path.rootfs=/rootfs'
- '--path.sysfs=/host/sys'
- '--collector.filesystem.mount-points-exclude=^/(sys|proc|dev|host|etc)($$|/)'
restart: unless-stopped
Но сами по себе метрики системы (CPU, память Redis) ничего не говорили о здоровье бизнес-процесса. Критическая метрика — длина очереди задач в Celery. Если она растёт быстрее, чем воркеры успевают справляться, — это красный флаг. Я добавил кастомные метрики Prometheus прямо в Celery.
Установил prometheus-client и модифицировал файл инициализации Celery (tasks.py):
# tasks.py (фрагмент)
from celery import Celery
from prometheus_client import start_http_server, Gauge, Counter
import os
# Запускаем HTTP-сервер для метрик на отдельном порту (в контейнере)
if os.environ.get('ENABLE_METRICS'):
start_http_server(8001)
# Кастомные метрики
CELERY_QUEUE_LENGTH = Gauge('celery_queue_length', 'Текущая длина очереди задач')
TASKS_PROCESSED = Counter('celery_tasks_processed_total', 'Общее количество обработанных задач', ['type'])
TASKS_FAILED = Counter('celery_tasks_failed_total', 'Общее количество упавших задач', ['type'])
app = Celery('lab_tasks',
broker=os.environ.get('REDIS_URL', 'redis://localhost:6379/0'),
backend=os.environ.get('REDIS_URL'))
# Обёртка для записи метрик
def task_with_metrics(task_func):
def wrapper(*args, **kwargs):
task_type = task_func.__name__
try:
result = task_func(*args, **kwargs)
TASKS_PROCESSED.labels(type=task_type).inc()
return result
except Exception as e:
TASKS_FAILED.labels(type=task_type).inc()
raise e
return wrapper
@app.task
@task_with_metrics
def process_blood_analysis(analysis_data: dict):
# ... основная логика обработки анализа ...
pass
И, наконец, мозг системы оповещений — alertmanager.yml. Правило простое: если очередь задач больше 50 в течение 5 минут — это предупреждение. Если больше 200 или воркеры неактивны — это критично и требует моего немедленного внимания, даже в 3 AM.
# alertmanager.yml
global:
smtp_smarthost: 'smtp.gmail.com:587'
smtp_from: 'lab.alerts@yourdomain.com'
smtp_auth_username: 'your-email@gmail.com'
smtp_auth_password: '${SMTP_PASSWORD}'
route:
group_by: ['alertname']
group_wait: 10s
group_interval: 10s
repeat_interval: 1h
receiver: 'critical-email'
receivers:
- name: 'critical-email'
email_configs:
- to: 'my-phone-alerts@operator.com' # Email для SMS
headers:
subject: '[CRITICAL] Лабораторная автоматизация'
- to: 'my-backup@email.com'
Первый алерт пришёл через неделю. В 3:17 на телефон упала SMS: «[CRITICAL] Лабораторная автоматизация: Celery queue length is 215». Я вскочил как ужаленный. Запустил Grafana, увидел график: один из четырёх воркеров Celery тихо умер час назад из-за утечки памяти в сторонней библиотеке для парсинга PDF. Очередь начала копиться. Перезапустил контейнер с воркером: docker-compose restart celery_worker_1. Через 10 минут очередь рассосалась. Главное — ни один анализ не был потерян, всё просто встало в очередь. А главврач продолжил спать.
Именно в этот момент я понял: мы перешли на новый уровень. Система перестала быть хрупким скриптом. Теперь это был живой организм, который мог болеть, но всегда звал на помощь того, кто его лечил. И этот «кто-то» был теперь я, а не испуганный лаборант с листком бумаги в 9 утра. Стоило оно того? Однозначно. Даже учитывая, что следующий алерт разбудил меня в 4 утра в субботу.
Провалы, которые стоили мне ночей: почему валидация данных — это святое
Всё работало. Я был гением. Мониторинг в Grafana показывал красивые зелёные графики, контейнеры в Docker Swarm переживали падения нод без единого сбоя, а Celery обрабатывал 1200 задач в час. Я спал как младенец. До того понедельника.
Первым звонком был тихий, но настойчивый звонок от терапевта из второго корпуса: «Андрей Викторович, у вас там с глюкозой всё в порядке? У пациентки Ивановой вчера было 5.2, сегодня — 18.7, но она натощак и чувствует себя нормально». Я отмахнулся: «Аппаратура калибрована, наверное, преаналитическая ошибка, пересдайте». Я был идиотом.
Проблема вскрылась через три дня, когда главный эндокринолог ворвался в лабораторию с распечаткой: «Что за бред? У семи пациентов с компенсированным диабетом сахар зашкаливает за 15! Вы что, калибровочный раствор вместо реагента залили?»
Мы отключили автоматическую выгрузку диагнозов и начали расследование. Оказалось, что новый лаборант, заполняя журнал ручного ввода для устаревшего гематологического анализатора (который пока не был подключен к системе), вносил данные в .csv файл. Но вместо точки в качестве десятичного разделителя использовал запятую. Мой парсер, написанный в лучших традициях «работает же», был слеп и беспощаден.
Вот он, «убийца», из моего сервиса parsers/legacy_haematology.py:
# parsers/legacy_haematology.py (версия 1.0 - "работает же")
import csv
from pydantic import BaseModel
from typing import Optional
class HaemAnalysis(BaseModel):
patient_id: str
wbc: Optional[float] # Лейкоциты
rbc: Optional[float] # Эритроциты
hgb: Optional[float] # Гемоглобин
def parse_haem_file(file_path: str) -> list[HaemAnalysis]:
results = []
with open(file_path, 'r') as f:
reader = csv.reader(f, delimiter=';')
next(reader) # Пропускаем заголовок
for row in reader:
try:
# Преобразуем ВСЁ, что похоже на число, во float
wbc = float(row[2]) if row[2] else None
rbc = float(row[3]) if row[3] else None
hgb = float(row[4]) if row[4] else None
results.append(
HaemAnalysis(
patient_id=row[0],
wbc=wbc,
rbc=rbc,
hgb=hgb
)
)
except ValueError as e:
# Логируем и пропускаем строку. ОШИБКА!
print(f"Ошибка в строке {row}: {e}")
continue
return results
Казалось бы, что тут такого? float("12,5") выбросит ValueError и мы пропустим строку, верно? Нет, не верно. В локали моего сервера (Ubuntu, en_US.UTF-8) float("12,5") не выбрасывает исключение. Он просто отбрасывает всё после запятой. float("12,5") равнялось 12.0. У пациента с истинным гемоглобином 12,5 г/дл в систему улетело значение 12,0. Но это был лучший сценарий.
Худший проявился с глюкозой. Данные приходили так: "5,2". float("5,2") → 5.0. Но лаборант, увидев, что числа «съедаются», начал исправлять файл… вручную в Блокноте. Он заменил запятые на точки, но сделал это криво. В одной строке появилось "15..7". float("15..7") — исключение, строка пропущена. Но в другой строке было "18,7", которую он исправил на "18.7". И вот это уже float("18.7") → 18.7. У здоровой пациентки появилась запись о гипергликемической коме. Автоматический алгоритм интерпретации, основанный на референтных пределах, тут же проставил флаг «КРИТИЧЕСКОЕ ЗНАЧЕНИЕ». И эта запись улетела в историю болезни.
Провал №1: Наивный парсинг. Я доверял float() как священной корове, не понимая его зависимости от локали и не обеспечивая строгую валидацию формата.
Мы залатали дыру за ночь, написав валидатор регулярками. Но через месяц случился Провал №2: Молчаливый отказ.
К нам поступила партия новых тест-полосок для анализа мочи. Их ридер выдавал JSON. Идеально, думал я. Мой новый, улучшенный парсер выглядел так:
# parsers/urine_analyzer.py (версия 2.0 - "теперь с валидацией!")
import json
import re
from pydantic import BaseModel, validator
from decimal import Decimal
class UrineAnalysis(BaseModel):
patient_id: str
leukocytes: Decimal # Лейкоциты в моче
protein: Decimal # Белок
@validator('patient_id')
def validate_id(cls, v):
if not re.match(r'^P\d{8}$', v):
raise ValueError('Invalid patient ID format')
return v
@validator('leukocytes', 'protein')
def validate_value_range(cls, v, field):
if field.name == 'leukocytes' and not (0 <= v <= 500):
raise ValueError(f'Leukocytes value {v} out of range (0-500)')
if field.name == 'protein' and not (0 <= v <= 1000):
raise ValueError(f'Protein value {v} out of range (0-1000)')
return v
def parse_urine_json(file_path: str) -> list[UrineAnalysis]:
with open(file_path, 'r') as f:
data = json.load(f)
results = []
errors = []
for item in data:
try:
# ПРЕОБРАЗОВАНИЕ! Где валидация?
item['leukocytes'] = Decimal(str(item['leukocytes']))
item['protein'] = Decimal(str(item['protein']))
results.append(UrineAnalysis(**item))
except (ValueError, KeyError, json.JSONDecodeError) as e:
errors.append({"raw_data": item, "error": str(e)})
# ЗАПИСЫВАЕМ ОШИБКУ В ЛОГ И... ПРОДОЛЖАЕМ.
logger.error(f"Failed to parse urine item: {item}. Error: {e}")
# ВОТ ОН - УБИЙЦА. Мы возвращаем только УСПЕШНЫЕ результаты.
return results
Кажется, всё продумано? Pydantic, Decimal, валидаторы диапазонов. Но что произошло на практике? Новый ридер, из-за глюка прошивки, в 5% случаев выдавал "leukocytes": "NEG" вместо числа. Мой код пытался сделать Decimal("NEG"), получал исключение, логировал ошибку и… выбрасывал эту запись из финального списка results. В итоге 5% пациентов не получили результатов анализа мочи вообще. Система не упала, задачи выполнились успешно, в логах были записи, но кто читает логи, когда всё зелёное? Отсутствие результата — это не ошибка для системы, это чёрная дыра для врача.
Мне пришлось вводить контрольные суммы по пачкам. Каждой пачке из 20 проб присваивался UUID. Если после парсинга количество успешных результатов не совпадало с ожидаемым, вся пачка маркировалась как NEEDS_MANUAL_REVIEW и не шла дальше по конвейеру.
Итоговый урок, стоивший мне десятков бессонных ночей и одного очень неприятного разбора полётов: Валидация в медицинской автоматизации — это не про «красивые ошибки». Это про полный идемпотентный контроль всего конвейера данных.
Мой текущий слой валидации теперь выглядит как параноидальный охранник с металлоискателем:
# core/validation.py (версия 3.0 - "после инцидентов")
from pydantic import BaseModel, Field, validator, root_validator
from decimal import Decimal, InvalidOperation
import re
from typing import Any, Dict, Optional, Tuple
from datetime import datetime
class StrictDecimal(Decimal):
@classmethod
def __get_validators__(cls):
yield cls.validate_strict
@classmethod
def validate_strict(cls, v: Any) -> Decimal:
# 1. Только строки или числа
if not isinstance(v, (str, int, float, Decimal)):
raise TypeError(f'Decimal expected, got {type(v)}')
# 2. Строка должна соответствовать жесткому regex
if isinstance(v, str):
if not re.match(r'^-?\d+(\.\d+)?$', v.strip()):
# СОБСТВЕННОЕ ИСКЛЮЧЕНИЕ с контекстом
raise StrictDecimalError(f"Invalid decimal format: '{v}'. Must be '123.456'")
# 3. Явное преобразование с отловом ошибок
try:
return Decimal(v.strip())
except InvalidOperation as e:
raise StrictDecimalError(f"Cannot convert '{v}' to Decimal: {e}")
# 4. Для чисел - тоже через строку для чистоты
try:
return Decimal(str(v))
except Exception as e:
raise StrictDecimalError(f"Cannot convert {v} to Decimal: {e}")
class StrictDecimalError(ValueError):
"""Кастомное исключение для точного отлова в конвейере."""
pass
class BaseLabResult(BaseModel):
batch_uuid: str = Field(..., min_length=36, max_length=36)
sample_id: str = Field(..., regex=r'^[A-Z0-9]{10}$')
analyzer_id: str
analyzed_at: datetime
# Валидатор НА ВСЕЙ МОДЕЛИ. Проверяет внутреннюю согласованность.
@root_validator(skip_on_failure=False)
def check_batch_sample_consistency(cls, values: Dict[str, Any]) -> Dict[str, Any]:
# Например, sample_id должен начинаться с префикса batch_uuid
batch_prefix = values.get('batch_uuid', '')[:8]
sample = values.get('sample_id', '')
if not sample.startswith(batch_prefix):
raise ValueError(f'Sample ID {sample} does not belong to batch {batch_prefix}')
return values
class Config:
extra = 'forbid' # ЗАПРЕТИТЬ ЛИШНИЕ ПОЛЯ. Никаких "unknown_field".
validate_assignment = True # Валидировать даже при присваивании.
# А главное - новый контракт для ВСЕХ парсеров:
ParserResult = Tuple[
list[BaseLabResult], # Успешные результаты
list[Dict], *СЫРЫЕ* данные с ошибками
list[Dict] Детали ошибок для КАЖДОГО сломанного элемента
]
Теперь каждый парсер обязан вернуть всю тройку. Если второй список (сырые данные с ошибками) не пуст — конвейерная задача process_batch переходит в состояние VALIDATION_FAILED. Она не удаляется, не считается успешной. Она висит в отдельной очереди manual_review, куда стекаются все такие инциденты. По ним срабатывает отдельный, тихий, но настойчивый алерт в Telegram: «Требуется ручная проверка пачки uuid. Количество ошибок: 5».
Автоматизация — это не про то, чтобы полностью заменить человека. Это про то, чтобы дать ему контроль над тысячами операций и безжалостно высветить те 0.1%, где система не уверена. После этих провалов я понял: самые важные алерты — не «сервер упал», а «данные странные». Молчаливый успех с потерей информации — самый страшный враг.
Выводы: 1000 анализов в день — это не магия, это системный подход и правильные инструменты
Выводы: 1000 анализов в день — это не магия, это системный подход и правильные инструменты
Прошел год с момента того звонка от хирурга, который чуть не стоил мне места. Система стабильно обрабатывает 1200-1400 анализов в сутки с пиковыми нагрузками до 300 в час. Я больше не сижу с калькулятором и таблицами. Моя роль сместилась с оператора на архитектора и смотрителя. И главный вывод, который я вынес, звучит банально, но стал для меня откровением: масштаб — это не про мощное железо, а про правильную декомпозицию процессов и параноидальную надёжность на каждом шаге.
Вот пять конкретных уроков, выжженных на моей подкорке бессонными ночами и критическими инцидентами. Не теория, а голая практика.
Урок 1. Автоматизируй не задачи, а потоки данных. С самого начала строй конвейер
Моя первая ошибка была в том, что я автоматизировал «задачу переноса из CSV в Excel». Это тупик. Нужно было думать шире: «Как данные рождаются в анализаторе, проходят валидацию, обогащаются, превращаются в диагноз и попадают к врачу?».
Что я делаю сейчас: Каждый этап — это отдельный микросервис (или Celery-задача) с чётким контрактом.
# Пример контракта (Pydantic model) для данных из анализатора.
# Это святое. Меняется только через Pull Request и миграцию.
from pydantic import BaseModel, Field, validator
from typing import Optional
from datetime import datetime
class RawAnalysisResult(BaseModel):
lab_id: str = Field(..., min_length=8, max_length=12)
patient_code: str
test_code: str
raw_value: float
unit: str
analyzer_id: str
measured_at: datetime
@validator('raw_value')
def value_make_sense(cls, v):
if v < 0:
raise ValueError('Отрицательное значение результата? Проверь калибровку анализатора!')
if v > 10000: # Для глюкозы, например, это явный артефакт
raise ValueError(f'Значение {v} за пределами физиологического диапазона. Возможна ошибка пробы.')
return v
Совет: Прежде чем писать код, нарисуй схему потока данных от источника до потребителя. Каждый прямоугольник на схеме — кандидат в отдельный сервис или очередь. Это спасёт тебя от монолитного кошмара.
Урок 2. Мониторинг — это не «поставить Grafana». Это система раннего оповещения о пожаре до того, как сгорит дом
Моя первая панель в Grafana была про «сколько задач обработано». Бесполезная метрика для спасения. Пожар в лаборатории — это не падение сервера, это тихая порча данных или накопление отставания в очередях.
Что я отслеживаю сейчас:
1. Скорость конвейера: Время от measured_at (анализатор) до delivered_at (врач). Цель — < 5 минут. Alert в Telegram, если среднее > 7 мин.
2. Процент отклонённых проб: Резкий скачок с 2% до 10% — не «статистика», а сломанный сканер штрих-кодов или новая медсестра, которой не объяснили процедуру. Alert.
3. Длина очереди Celery в Redis: Не общее число, а разбивка по критичным очередям (validation, critical_calc). Команда для срочной проверки:
bash
redis-cli -h lab-redis -p 6379 --stat
# Или конкретно
redis-cli -h lab-redis LLEN celery:queue:critical_calc
4. Health-check каждого сервиса: Не просто 200 OK, а проверка подключения к его БД, очередям, внешним API МИС (мед. информационной системы).
yaml
# docker-compose.override.yml для healthcheck API
services:
api:
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
interval: 30s
timeout: 10s
retries: 3
start_period: 40s
Совет: Настрой алерты не на «всё упало», а на «что-то идёт не так». 3 ночных ложных алерта лучше, чем одно тихое утро с сотней необработанных анализов онкобольных.
Урок 3. Валидация данных — это многослойная оборона. И первый рубеж — на аппаратуре
После того инцидента с пробиркой-призраком я стал параноиком. Валидация теперь на четырёх уровнях:
- Аппаратный (по возможности): Настройка анализатора на флаги «Липемия?», «Гемолиз?», «Мало материала?».
- Сырой данные (при приёме): Pydantic-модели, как выше. Плюс проверка контрольных проб (каждые 50 реальных анализов). Если контрольная проба с известным значением 5.0 ммоль/л пришла как 7.0 — конвейер останавливается, алерт максимального уровня.
- Бизнес-логика (в процессе): «У пациента с креатинином 900 мкмоль/л не может быть СКФ 90. Пересчитать!» «Гемоглобин 30 г/л? Срочный алерт врачу, даже если данные валидны».
- Финал (перед отправкой): Санити-чек: всем ли пробам за последний час посчитаны все показатели? Нет ли дублей?
Совет: Напиши сценарии «падения» для каждого источника данных (отключили анализатор, сломался сканер, слетела кодировка в файле CSV) и протестируй, как система себя ведёт. Данные должны либо валидироваться, либо изолироваться в карантинную очередь для ручного разбора. Никогда не теряться.
Урок 4. Деплой и инфраструктура должны быть идемпотентными и документированными командой
Больше никогда python main.py на продакшене. Никогда. Всё в контейнерах, всё через систему управления.
Мой стек и жёсткие правила:
- Docker + Docker Swarm (Kubernetes был overkill для 10 нод). Все конфиги в docker-compose.yml.
- Все секреты (пароли БД, токены API) — в Docker Swarm secrets или HashiCorp Vault. Никаких .env файлов в репозитории.
- Деплой — одна команда:
bash
# На manager-ноде Swarm
docker stack deploy -c docker-compose.yml -c docker-compose.prod.yml lab-stack
- Миграции БД — часть процесса деплоя. Запускаются отдельным контейнером migrator с тем же образом, что и API, но командой на применение Alembic-миграций.
dockerfile
# В Dockerfile API-сервиса
FROM python:3.11-slim
...
# Создаём entrypoint-скрипт
COPY entrypoint.sh .
RUN chmod +x entrypoint.sh
ENTRYPOINT ["./entrypoint.sh"]
bash
# entrypoint.sh
#!/bin/sh
# Если это контейнер для миграций — выполняем и выходим
if [ "$1" = "migrate" ]; then
alembic upgrade head
exit 0
fi
# Иначе — запускаем основной сервис
exec uvicorn app.main:app --host 0.0.0.0 --port 8000
Совет: Твой репозиторий инфраструктуры (infra/) должен содержать README.md, где любой (включая твоего будущего кофе-зависшего себя) за 5 минут сможет поднять полный стенд командой make up. Идеал — тестировать каждое изменение инфраструктуры на этом стенде.
Урок 5. Главный ресурс — не процессорное время, а доверие. Система должна объяснять свои решения
Это, пожалуй, самый важный урок. Когда врач видит в карте автоматически рассчитанный диагноз «ХБП 3 стадии», он должен иметь возможность в два клика увидеть: на основании чего?.
Что я реализовал:
- Каждому конечному результату (диагнозу, флагу) сопоставлен trace_id.
- По этому trace_id в логах (ELK Stack) можно найти всю историю: сырое значение, все этапы валидации, формулы расчёта, ссылки на нормативные документы (NKF-EPI, рекомендации Минздрава).
- В интерфейсе для врача — кнопка «Объяснить расчёт». Она не показывает код, а показывает человекочитаемую цепочку: «Креатинин = 150 мкмоль/л, возраст = 65, пол = жен. → СКФ (CKD-EPI) = 45 мл/мин/1.73м² → соответствует критериям ХБП 3 стадии».
Совет: Твоя система принимает решения, влияющие на здоровье. Она обязана быть не просто чёрным ящиком, а инструментом с обратной связью. Это снижает сопротивление врачей (ключевые пользователи!), облегчает аудит и, в конечном счёте, спасает от роковых ошибок.
Вместо эпилога: 1000 анализов в день — это что?
Это не магия. Это: - ~15 микросервисов в 40 Docker-контейнерах, жужжащих на 4 виртуалках. - 7 очередей в Redis, через которые проходит 1.5 миллиона задач в месяц. - 300+ автоматических алертов в месяц, из которых реально требуют моего вмешательства — 2-3. - Освобождённые 6-7 часов ежедневного времени врача-лаборанта (меня), которые теперь тратятся не на копирование, а на консультации клиницистов, разработку новых тестов и, прости господи, на написание статей на Хабре в три ночи.
Что бы я сделал иначе? Начал бы сразу с Docker и разделения на сервисы, минуя фазу «скрипта на коленке». Потратил бы больше времени на проектирование схемы данных и форматов обмена. И, чёрт возьми, сразу завёл бы чат в Telegram с алертами, а не ждал, пока меня разбудят звонком.
Если ты такой же сумасшедший, который хочет автоматизировать свою рутину — начинай. Но начинай с архитектуры, а не со скрипта. Думай о данных как о потоке, о надёжности как о главном KPI, а об объяснимости — как о единственном способе заслужить доверие.
Система работает. Я иду спать. Завтра в 7:30 новый рабочий день, 500+ проб и спокойная уверенность, что технологический конвейер, выстраданный за год, справится. А я буду делать то, для чего учился на врача — анализировать сложные случаи, а не переносить цифры из столбца А в столбец Б.