Автоматизация задач на Python: планировщики и их возможности

import schedule
import time

def job():
print("Задача выполняется...")

schedule.every(10).minutes.do(job)
schedule.every().hour.do(job)
schedule.every().day.at("10:30").do(job)

while True:
schedule.run_pending()
time.sleep(1)

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger

scheduler = BackgroundScheduler()

# Интервальный триггер
scheduler.add_job(job, 'interval', hours=2)

# Cron-триггер
scheduler.add_job(job, CronTrigger(hour=3, minute=30))

scheduler.start()

from celery import Celery
from celery.schedules import crontab

app = Celery('tasks', broker='redis://localhost:6379/0')

@app.task
def process_data():
# Обработка данных
pass

app.conf.beat_schedule = {
'process-every-morning': {
'task': 'tasks.process_data',
'schedule': crontab(hour=7, minute=30),
},
}

import schedule
import time

# Функция, которую нужно запускать периодически
def backup_database():
print("Выполняется резервное копирование...")
# Код для резервного копирования

# Ежедневное выполнение в определенное время
schedule.every().day.at("01:00").do(backup_database)

# Выполнение с заданным интервалом
schedule.every(4).hours.do(backup_database)

# Выполнение в определенные дни недели
schedule.every().monday.at("09:00").do(backup_database)
schedule.every().wednesday.at("13:15").do(backup_database)

# Цикл для проверки и запуска запланированных задач
while True:
schedule.run_pending()
time.sleep(1)

import schedule
import time
import threading

def run_threaded(job_func):
job_thread = threading.Thread(target=job_func)
job_thread.start()

schedule.every(10).minutes.do(run_threaded, backup_database)

while True:
schedule.run_pending()
time.sleep(1)

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.interval import IntervalTrigger
from apscheduler.triggers.cron import CronTrigger
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor
import datetime

# Настройка хранилища задач и исполнителей
jobstores = {
'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
}

executors = {
'default': ThreadPoolExecutor(20),
'processpool': ProcessPoolExecutor(5)
}

scheduler = BackgroundScheduler(jobstores=jobstores, executors=executors)

# Различные способы планирования задач

# 1. Запуск в конкретное время (один раз)
scheduler.add_job(
backup_database,
'date',
run_date=datetime.datetime(2023, 12, 31, 23, 59)
)

# 2. Интервальное выполнение
scheduler.add_job(
backup_database,
IntervalTrigger(hours=6),
id='backup_job',
replace_existing=True
)

# 3. Выполнение по cron-расписанию
scheduler.add_job(
backup_database,
CronTrigger(day_of_week='mon-fri', hour=3, minute=30),
id='workday_backup',
replace_existing=True
)

# Запуск планировщика
scheduler.start()

def job_with_error_handling():
try:
# Основной код задачи
print("Выполнение задачи...")
except Exception as e:
print(f"Произошла ошибка: {e}")
# Можно добавить логирование или оповещение
return False # Сигнализирует о необходимости повторного запуска
return True # Успешное выполнение

# Добавление задачи с повторными попытками
scheduler.add_job(
job_with_error_handling,
IntervalTrigger(minutes=30),
max_instances=1, # Предотвращает одновременное выполнение нескольких экземпляров
coalesce=True, # Объединяет пропущенные запуски
misfire_grace_time=300, # Допустимая задержка запуска (в секундах)
# Повторный запуск при неудаче (возврат False)
kwargs={'retry_on_failure': True}
)

#!/usr/bin/env python3
# data_processor.py

import sys
import logging

logging.basicConfig(
filename='/var/log/data_processor.log',
level=logging.INFO,
format='%(asctime)s – %(levelname)s – %(message)s'
)

try:
# Ваш код обработки данных
logging.info("Обработка данных успешно завершена")
except Exception as e:
logging.error(f"Ошибка при обработке данных: {e}")
sys.exit(1)

sys.exit(0)

chmod +x /path/to/data_processor.py

crontab -e

# Запуск каждый день в 2:30 утра
30 2 * * * /path/to/data_processor.py

# Запуск каждый понедельник в 9:00
0 9 * * 1 /path/to/data_processor.py

# Запуск каждые 15 минут
*/15 * * * * /path/to/data_processor.py

30 2 * * * /home/user/venv/bin/python /path/to/script.py

30 2 * * * /path/to/script.py > /path/to/output.log 2>&1

# Создание новой задачи
$action = New-ScheduledTaskAction -Execute "C:\Path\to\Python\python.exe" -Argument "C:\Path\to\your_script.py"
$trigger = New-ScheduledTaskTrigger -Daily -At "2:30AM"
$principal = New-ScheduledTaskPrincipal -UserId "SYSTEM" -LogonType ServiceAccount -RunLevel Highest
$settings = New-ScheduledTaskSettingsSet -StartWhenAvailable -DontStopOnIdleEnd
$task = New-ScheduledTask -Action $action -Trigger $trigger -Principal $principal -Settings $settings

# Регистрация задачи в системе
Register-ScheduledTask -TaskName "Python Data Processor" -InputObject $task

import win32com.client
from datetime import datetime, timedelta

# Создание объекта TaskScheduler
scheduler = win32com.client.Dispatch('Schedule.Service')
scheduler.Connect()
root_folder = scheduler.GetFolder('\\')

# Создание задачи
task_def = scheduler.NewTask(0)

# Установка общих настроек
task_def.RegistrationInfo.Description = 'Обработка данных Python-скриптом'
task_def.Settings.Enabled = True
task_def.Settings.Hidden = False

# Создание триггера (ежедневно в 2:30)
start_time = datetime.now() + timedelta(days=1)
start_time = start_time.replace(hour=2, minute=30, second=0, microsecond=0)

trigger = task_def.Triggers.Create(2) # 2 = TASK_TRIGGER_DAILY
trigger.StartBoundary = start_time.isoformat()
trigger.DaysInterval = 1
trigger.Enabled = True

# Создание действия
action = task_def.Actions.Create(0)
action.Path = 'C:\\Python\\python.exe'
action.Arguments = 'C:\\Scripts\\data_processor.py'

# Регистрация задачи
root_folder.RegisterTaskDefinition(
'PythonDataProcessor', # Имя задачи
task_def,
6, # TASK_CREATE_OR_UPDATE
None, # Имя пользователя (None = текущий пользователь)
None, # Пароль
3 # TASK_LOGON_INTERACTIVE_TOKEN
)

print("Задача успешно создана")

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
<key>Label</key>
<string>com.username.pythontask</string>
<key>ProgramArguments</key>
<array>
<string>/usr/bin/python3</string>
<string>/path/to/your_script.py</string>
</array>
<key>StartCalendarInterval</key>
<dict>
<key>Hour</key>
<integer>2</integer>
<key>Minute</key>
<integer>30</integer>
</dict>
<key>StandardOutPath</key>
<string>/tmp/pythontask.log</string>
<key>StandardErrorPath</key>
<string>/tmp/pythontask.err</string>
</dict>
</plist>

launchctl load ~/Library/LaunchAgents/com.username.pythontask.plist

import plistlib
import os
import subprocess
from pathlib import Path

def create_launchd_job(label, script_path, hour, minute):
"""Создает и загружает launchd задание для запуска Python-скрипта."""
home = Path.home()
agents_dir = home / "Library" / "LaunchAgents"
agents_dir.mkdir(parents=True, exist_ok=True)

plist_path = agents_dir / f"{label}.plist"

plist_content = {
"Label": label,
"ProgramArguments": [
"/usr/bin/python3",
script_path
],
"StartCalendarInterval": {
"Hour": hour,
"Minute": minute
},
"StandardOutPath": f"/tmp/{label}.log",
"StandardErrorPath": f"/tmp/{label}.err"
}

with open(plist_path, 'wb') as fp:
plistlib.dump(plist_content, fp)

# Загрузка задачи
subprocess.run(["launchctl", "unload", str(plist_path)], 
stderr=subprocess.DEVNULL, check=False)
subprocess.run(["launchctl", "load", str(plist_path)], check=True)

print(f"Задача {label} успешно создана и загружена")

# Пример использования
create_launchd_job(
"com.user.data_processor",
"/Users/username/scripts/data_processor.py",
2, # час
30 # минута
)

import logging
import sys
import time
import requests
from functools import wraps

# Настройка логирования
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s – %(name)s – %(levelname)s – %(message)s',
handlers=[
logging.FileHandler("automation.log"),
logging.StreamHandler(sys.stdout)
]
)
logger = logging.getLogger("automation")

# Декоратор для повторных попыток с экспоненциальной задержкой
def retry_with_backoff(retries=5, backoff_factor=0.5, exceptions=(Exception,)):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
max_retries = retries
retry_count = 0
delay = backoff_factor

while True:
try:
return func(*args, **kwargs)
except exceptions as e:
retry_count += 1
if retry_count > max_retries:
logger.error(f"Превышено максимальное число попыток: {e}")
raise

wait_time = delay * (2 ** (retry_count – 1))
logger.warning(f"Попытка {retry_count} не удалась: {e}. "
f"Повторная попытка через {wait_time:.2f} секунд.")
time.sleep(wait_time)
return wrapper
return decorator

# Пример использования декоратора
@retry_with_backoff(retries=3, backoff_factor=1, exceptions=(requests.RequestException,))
def fetch_data_from_api(url):
response = requests.get(url, timeout=10)
response.raise_for_status()
return response.json()

# Обработка критических ошибок с оповещением
def critical_task():
try:
# Основной код задачи
data = fetch_data_from_api("https://api.example.com/data")
process_data(data)
except requests.RequestException as e:
logger.error(f"Ошибка API: {e}")
send_alert("API недоступен", str(e))
return False
except Exception as e:
logger.exception("Необработанное исключение:")
send_alert("Критическая ошибка", str(e))
return False
return True

def send_alert(subject, message):
"""Отправка оповещения через предпочтительный канал (email, SMS, Slack и т.д.)"""
# Реализация отправки оповещения
logger.info(f"Отправлено оповещение: {subject}")

import logging
import time
import contextlib
from dataclasses import dataclass
from datetime import datetime
from typing import Optional, Dict, Any

@dataclass
class TaskMetrics:
name: str
start_time: datetime
end_time: Optional[datetime] = None
success: Optional[bool] = None
execution_time: Optional[float] = None
metadata: Dict[str, Any] = None

def complete(self, success=True):
self.end_time = datetime.now()
self.success = success
self.execution_time = (self.end_time – self.start_time).total_seconds()

def to_dict(self):
return {
"task_name": self.name,
"start_time": self.start_time.isoformat(),
"end_time": self.end_time.isoformat() if self.end_time else None,
"success": self.success,
"execution_time": self.execution_time,
"metadata": self.metadata or {}
}

class TaskMonitor:
def __init__(self):
self.logger = logging.getLogger("task_monitor")
self.metrics = []

@contextlib.contextmanager
def track_task(self, task_name, metadata=None):
metrics = TaskMetrics(
name=task_name,
start_time=datetime.now(),
metadata=metadata
)
self.logger.info(f"Начало выполнения задачи: {task_name}")

try:
yield metrics
metrics.complete(success=True)
self.logger.info(
f"Задача {task_name} успешна завершена за "
f"{metrics.execution_time:.2f} секунд"
)
except Exception as e:
metrics.complete(success=False)
self.logger.error(
f"Ошибка при выполнении задачи {task_name}: {e}. "
f"Время выполнения: {metrics.execution_time:.2f} секунд"
)
raise
finally:
self.metrics.append(metrics)
# Можно отправлять метрики в систему мониторинга
self._store_metrics(metrics)

def _store_metrics(self, metrics):
"""Сохранение метрик в базу данных или отправка в систему мониторинга"""
# Здесь может быть интеграция с Prometheus, InfluxDB и т.д.
pass

# Пример использования
monitor = TaskMonitor()

def process_data_batch(batch_id):
with monitor.track_task(f"process_batch_{batch_id}", {"batch_id": batch_id}):
# Имитация обработки
time.sleep(2)

# Выполнение подзадачи с отдельным отслеживанием
with monitor.track_task(f"validate_batch_{batch_id}"):
# Валидация данных
time.sleep(1)

# Продолжение основной задачи
time.sleep(1)

# Для демонстрации ошибки раскомментируйте следующую строку
# if batch_id == 3: raise ValueError("Демонстрационная ошибка")

return f"Результат обработки партии {batch_id}"

# Обработка нескольких партий
for i in range(1, 6):
try:
result = process_data_batch(i)
print(result)
except Exception as e:
print(f"Ошибка при обработке партии {i}: {e}")

import concurrent.futures
from functools import partial
import time

def process_item(item, config=None):
"""Обрабатывает отдельный элемент данных."""
# Имитация обработки с разным временем
time.sleep(item / 10)
return {"item_id": item, "result": item * item}

def process_batch_parallel(items, max_workers=4, config=None):
"""Параллельная обработка партии элементов."""
results = []
errors = []

# Частичное применение функции для передачи конфигурации
process_func = partial(process_item, config=config)

with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
# Запуск задач и создание маппинга future -> item_id
future_to_item = {executor.submit(process_func, item): item for item in items}

for future in concurrent.futures.as_completed(future_to_item):
item = future_to_item[future]
try:
result = future.result()
results.append(result)
except Exception as e:
errors.append({"item_id": item, "error": str(e)})

return {
"successful": results,
"failed": errors,
"success_rate": len(results) / len(items) if items else 0
}

# Для CPU-bound задач лучше использовать ProcessPoolExecutor
def cpu_intensive_batch_processing(items, max_workers=None):
"""Параллельная обработка CPU-интенсивных задач."""
with concurrent.futures.ProcessPoolExecutor(max_workers=max_workers) as executor:
results = list(executor.map(process_item, items))
return results

# Пример использования
items_to_process = list(range(1, 11))
config = {"option1": "value1", "option2": "value2"}

print("Выполнение с ThreadPoolExecutor:")
start = time.time()
result = process_batch_parallel(items_to_process, max_workers=4, config=config)
end = time.time()
print(f"Обработано {len(result['successful'])} элементов за {end-start:.2f} секунд")
print(f"Успешность: {result['success_rate']:.2%}")

print("\nВыполнение с ProcessPoolExecutor:")
start = time.time()
result = cpu_intensive_batch_processing(items_to_process)
end = time.time()
print(f"Обработано {len(result)} элементов за {end-start:.2f} секунд")

import json
import os
from datetime import datetime
import signal
import sys

class TaskState:
def __init__(self, state_file, auto_restore=True):
self.state_file = state_file
self.state = {
"last_run": None,
"last_completed_id": None,
"processed_items": [],
"status": "idle"
}

# Регистрация обработчиков сигналов для корректного завершения
signal.signal(signal.SIGINT, self._handle_exit)
signal.signal(signal.SIGTERM, self._handle_exit)

if auto_restore and os.path.exists(state_file):
self.restore()

def update(self, **kwargs):
"""Обновляет состояние задачи."""
self.state.update(kwargs)
self.save()

def mark_started(self):
"""Отмечает начало выполнения задачи."""
self.state["status"] = "running"
self.state["last_run"] = datetime.now().isoformat()
self.save()

def mark_completed(self, item_id=None):
"""Отмечает успешное завершение задачи или подзадачи."""
if item_id is not None:
self.state["last_completed_id"] = item_id
self.state["processed_items"].append(item_id)
else:
self.state["status"] = "completed"
self.save()

def mark_failed(self, error=None):
"""Отмечает неудачное завершение задачи."""
self.state["status"] = "failed"
if error:
self.state["last_error"] = str(error)
self.save()

def save(self):
"""Сохраняет текущее состояние в файл."""
with open(self.state_file, 'w') as f:
json.dump(self.state, f, indent=2)

def restore(self):
"""Восстанавливает состояние из файла."""
try:
with open(self.state_file, 'r') as f:
saved_state = json.load(f)
self.state.update(saved_state)
print(f"Состояние восстановлено: {self.state['status']}")
except (json.JSONDecodeError, FileNotFoundError) as e:
print(f"Ошибка восстановления состояния: {e}")

def _handle_exit(self, signum, frame):
"""Обрабатывает сигналы завершения процесса."""
if self.state["status"] == "running":
self.state["status"] = "interrupted"
print("\nПроцесс прерван. Сохранение состояния...")
self.save()
sys.exit(0)

# Пример использования для долго выполняющейся задачи
def process_large_dataset(items, state_file="task_state.json"):
state = TaskState(state_file)

# Определение, с какого места начинать/продолжать обработку
last_completed = state.state["last_completed_id"]
start_index = 0

if last_completed is not None:
try:
start_index = items.index(last_completed) + 1
print(f"Продолжение с элемента {start_index}")
except ValueError:
print("Последний обработанный элемент не найден, начинаем сначала")

state.mark_started()

try:
for i in range(start_index, len(items)):
# Обработка элемента
result = process_item(items[i])
print(f"Обработан элемент {items[i]}: {result}")

# Сохранение прогресса
state.mark_completed(items[i])

# Успешное завершение всей задачи
state.mark_completed()

except Exception as e:
state.mark_failed(error=e)
raise

# Демонстрация работы
if __name__ == "__main__":
items_to_process = list(range(1, 20))

try:
process_large_dataset(items_to_process)
except KeyboardInterrupt:
print("\nПрограмма остановлена пользователем")
except Exception as e:
print(f"\nПроизошла ошибка: {e}")