Программирование и автоматизация
Это подробная статья о программировании и автоматизации, включающая контрольные вопросы, примеры скриптов, основные компоненты, инструменты мониторинга и многое другое. Смотрите заголовки!
In a Nutshell
Автоматизация через программирование облегчает IT-процессы, повышает эффективность и снижает количество ошибок. Для этого целенаправленно применяются скрипты, языки программирования и инструменты автоматизации.
Компактное описание
Повторяющиеся задачи вроде резервного копирования, управления пользователями или развертывания можно автоматизировать с помощью скриптов (например, Bash, PowerShell, Python). К этому добавляются планирование по расписанию (Cron, Task Scheduler), мониторинг, логирование и проверки безопасности.
Специализированные инструменты тоже используют собственные скриптовые компоненты (например, VMware PowerCLI). Часто требуется анализировать существующий код и целенаправленно его адаптировать для интеграции автоматизации в текущие системы.
Вам постоянно придется редактировать файлы, которые вы создаете или которые уже существуют. Иногда нужно заменить множество “форматеров” или других элементов переменной или найти и удалить ссылку. В таких случаях помогают небольшие простые Python-, Bash- или PowerShell-скрипты.
Контрольные вопросы
Целенаправленный выбор языка программирования
Выбор правильного языка программирования решающий для успеха автоматизации. Для простых системных задач под Linux подходит Bash, для сложной логики и кроссплатформенных решений предпочтительнее Python. PowerShell это выбор для Windows-окружения с глубокой интеграцией API.
Автоматизация снижает ручные операции и ошибки
Автоматизированные процессы минимизируют человеческие факторы ошибок и обеспечивают консистентные результаты. Повторяющиеся задачи вроде резервного копирования, создания пользователей или развертывания выполняются надежно и отслеживаемо.
Скриптовые языки для системных задач и логики процессов
Скриптовые языки позволяют автоматизировать системные задачи (управление файлами, управление процессами) и сложную логику процессов (условия, циклы, обработка ошибок). Они обеспечивают быструю разработку и простые возможности адаптации.
Платформенные инструменты: Bash (Linux), PowerShell (Windows)
Bash это стандарт под Linux для shell-scripting и системной автоматизации. PowerShell предоставляет объектно-ориентированную обработку и глубокую интеграцию в Windows-системы и API. Оба релевантны для системных администраторов на экзаменах.
Мониторинг через логи, коды выхода, триггеры
Автоматизированные процессы должны контролироваться. Логирование документирует все действия, коды выхода указывают на успех или неудачу, триггеры позволяют событийно-управляемую автоматизацию. Это обеспечивает прозрачность и надежность обработки ошибок.
Чувствительные процессы только с логированием и проверкой ролей
При чувствительной автоматизации (например, права пользователей, изменения системы) логирование и проверка ролей обязательны. Только авторизованные пользователи могут запускать критические скрипты, и все действия должны регистрироваться.
Меньше усилий при обслуживании и системном уходе
Автоматизация значительно снижает ручные усилия при обслуживании и системном уходе. Однажды разработанные скрипты можно использовать множество раз и легко адаптировать, что экономит время и ресурсы.
Документировать и версионировать изменения
Все изменения скриптов автоматизации должны быть задокументированы и версионированы. Это гарантирует отслеживаемость, позволяет откатываться и поддерживает командное сотрудничество.
Основные компоненты
1. Выбор языка программирования
Выбор правильного языка программирования зависит от задачи и платформы. Python подходит для сложной логики и кроссплатформенных решений, Bash для Linux-системных задач, PowerShell для Windows-окружения.
2. Платформенные инструменты для скриптирования
Каждая платформа предоставляет специфические инструменты автоматизации: Bash под Linux, PowerShell под Windows, Cron для планирования под Linux, Task Scheduler под Windows. Эти инструменты позволяют прямую интеграцию с системой.
3. Определение целей автоматизации
Перед реализацией нужно определить четкие цели: какие задачи должны быть автоматизированы? Какие выигрыши в эффективности ожидаются? Какие требования безопасности применяются?
4. Условное выполнение и циклы
Автоматизация часто требует условной логики (if/else) и циклов (for/while) для повторяющихся процессов. Эти управляющие структуры позволяют гибкую и адаптивную автоматизацию.
5. Обработка ошибок и логирование
Надежная автоматизация требует комплексной обработки ошибок (try/catch, коды выхода) и детального логирования. Это обеспечивает стабильность и отслеживаемость автоматизированных процессов.
6. Временные задачи (Cron, Task Scheduler)
Cron под Linux и Task Scheduler под Windows позволяют выполнять скрипты по расписанию. Это идеально для регулярных задач обслуживания вроде резервного копирования или очистки.
7. Интеграция в существующие системы
Скрипты автоматизации должны быть интегрированы в существующие системы. Это касается API, баз данных, файловых систем и других компонентов IT-инфраструктуры.
8. Контроль версий скриптов
Все скрипты автоматизации должны находиться под контролем версий (Git). Это позволяет отслеживать изменения, командное сотрудничество и безопасные откаты.
9. Проверки безопасности и концепции прав
Автоматизация требует строгих концепций безопасности: проверка прав, управление ролями, безопасная работа с паролями и audit-логирование для чувствительных операций.
10. Мониторинг автоматизированных процессов
Автоматизированные процессы должны контролироваться: проверка успеха, уведомление об ошибках, мониторинг производительности и алертинг при отклонениях.
Простой практический пример (Python-резервная копия)
import shutil
shutil.copy('/home/user/data.db', '/backup/data.db')Объяснение: скрипт автоматически копирует файл в каталог резервной копии и может запускаться регулярно через Cronjob.
Почему shutil вместо обычных команд копирования?
shutil означает “shell utilities” (не shadow utility) и это стандартный модуль Python для высокоуровневых файловых операций. Он предоставляет несколько преимуществ перед простыми командами копирования:
Преимущества shutil.copy():
- Кроссплатформенность: работает идентично на Windows, Linux и macOS
- Обработка ошибок: выбрасывает чистые исключения при проблемах (Permission, несуществующие пути)
- Метаданные: сохраняет права доступа файла и метаданные (при использовании
shutil.copy2()) - Безопасность: автоматически проверяет проблемы перезаписи
Альтернативы и почему они хуже:
# ❌ Плохо: зависит от ОС
import os
os.system('cp /home/user/data.db /backup/data.db') # Только Linux
os.system('copy C:\\data.db C:\\backup\\data.db') # Только Windows
# ❌ Плохо: нет обработки ошибок
f = open('/home/user/data.db', 'rb')
data = f.read()
f.close()
f = open('/backup/data.db', 'wb')
f.write(data)
f.close()
# ✅ Хорошо: shutil.copy()
import shutil
shutil.copy('/home/user/data.db', '/backup/data.db')shutil.copy() vs shutil.copy2():
copy(): копирует содержимое + базовые права доступаcopy2(): копирует содержимое + все метаданные (временные метки, права доступа) - обычно лучший выбор для резервных копий
Преимущества и недостатки
Преимущества
- Снижение ошибок за счёт автоматизации процессов
- Экономия времени на повторяющихся задачах
- Отслеживаемость благодаря логам
- Кроссплатформенность благодаря портативным языкам скриптов
Недостатки
- Ошибка в скрипте может привести к серьёзным последствиям
- Требует тестирования и правильного логирования
- Поведение может отличаться в зависимости от окружения
Типичные вопросы на экзамене (с кратким ответом)
- Когда использовать Bash вместо Python? Для простых системных задач под Linux (файлы, процессы, pipes).
- Типичные задачи автоматизации? Резервные копии, создание пользователей, архивирование логов, развёртывание.
- Как сделать автоматизацию безопасной? Логирование, проверка прав доступа, code reviews, обработка кодов выхода.
- Что такое Cronjob? Задача с расписанием под Linux.
- Преимущества PowerShell над Bash? Объектно-ориентирован и глубоко интегрирован с Windows и API.
Развёрнутый ответ
Тема объединяет выбор языка, зависимость от платформы и цели автоматизации. На экзаменах часто требуется прочитать скрипт, расширить его и правильно реализовать обработку ошибок и логирование.
Стратегия обучения по этой теме
1. Начало понимания: автоматизируй небольшую резервную копию локально
Решение:
# Bash скрипт резервной копии
#!/bin/bash
SOURCE="/home/user/documents"
TARGET="/backup/documents_$(date +%Y%m%d)"
mkdir -p "$TARGET"
cp -r "$SOURCE"/* "$TARGET/"
echo "Backup абgeschlossen: $TARGET" >> /var/log/backup.logНастрой Cron: 0 2 * * * /pfad/zum/backup.sh
2. Углубление: набросай процесс автоматизации (создание пользователя + права)
Решение:
# Python создание пользователя с проверкой прав
import subprocess
import logging
logging.basicConfig(filename='user_management.log', level=logging.INFO)
def create_user(username, groups):
try:
# Создание пользователя
subprocess.run(['useradd', '-m', username], check=True)
# Назначение групп
for group in groups:
subprocess.run(['usermod', '-aG', group, username], check=True)
logging.info(f"User {username} erstellt mit Gruppen: {groups}")
print(f"User {username} erfolgreich erstellt")
except subprocess.CalledProcessError as e:
logging.error(f"Fehler bei User-Erstellung {username}: {e}")
print(f"Fehler: {e}")
# Использование
create_user("newuser", ["sudo", "developers"])3. Экзаменационный тренинг: анализируй и дополняй скрипты
Решение:
- Анализ: проверь скрипт на обработку ошибок, логирование и безопасность
- Дополнение: обработай коды выхода, проверяй параметры, улучшай сообщения об ошибках
- Пример улучшения:
import sys
import os
def validate_path(path):
if not os.path.exists(path):
print(f"Fehler: Pfad {path} existiert nicht")
sys.exit(1)
return True
# Использование
if len(sys.argv) < 2:
print("Verwendung: skript.py <pfad>")
sys.exit(1)
validate_path(sys.argv[1])
# Rest des Skripts...4. Предотвращение ошибок: добавляй логирование и проверку прав
Решение:
import logging
import os
import sys
from pathlib import Path
# Настройка логирования
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('/var/log/automation.log'),
logging.StreamHandler()
]
)
def check_permissions(required_uid=0):
"""Проверяет, запущен ли скрипт с нужными правами"""
if os.geteuid() != required_uid:
logging.error(f"Skript muss als UID {required_uid} ausgeführt werden")
sys.exit(1)
def safe_operation(operation_func):
"""Декоратор для безопасных операций"""
def wrapper(*args, **kwargs):
try:
result = operation_func(*args, **kwargs)
logging.info(f"Operation {operation_func.__name__} erfolgreich")
return result
except Exception as e:
logging.error(f"Fehler in {operation_func.__name__}: {e}")
raise
return wrapper
@safe_operation
def critical_operation():
# Критическая операция здесь
passАнализ темы
- Технический стержень: Shell, Python, Scheduling, Logging
- Трудности при реализации: различия между платформами, предотвращение ошибок
- Безопасность: управление правами доступа, риск небезопасных скриптов
- Требования к документации: версионный контроль, журнал изменений
- Экономическая оценка: снижение затрат на ручной труд благодаря повторяемости
Современные программные продукты
Платформы автоматизации
- Ansible: Configuration Management и автоматизация (Open Source)
- Jenkins: CI/CD pipeline с богатой поддержкой плагинов
- GitHub Actions: облачный CI/CD прямо в репозитории
- GitLab CI/CD: интегрированное решение с Auto DevOps
Инструменты для скриптов
- Visual Studio Code: современный редактор с поддержкой Python и PowerShell
- PyCharm: Python IDE с отладкой и функциями тестирования
- Windows Terminal: современное терминальное окружение для PowerShell
- WSL (Windows Subsystem for Linux): Linux среда под Windows
Мониторинг и логирование
- Prometheus: сбор метрик и алертинг
- Grafana: визуализация данных мониторинга
- ELK Stack: Elasticsearch, Logstash, Kibana для анализа логов
- Nagios: классическая система мониторинга
Облачная автоматизация
- Azure Automation: PowerShell и Python Runbooks в Azure
- AWS Lambda: serverless функции для автоматизации
- Google Cloud Functions: автоматизация, управляемая событиями
- Terraform: Infrastructure as Code для multi-cloud
Вопросы и ответы
Вопрос 1: Какой язык для какой задачи автоматизации?
Ответ: Bash для простых системных задач Linux, PowerShell для интеграции с Windows, Python для сложной логики и кроссплатформенных решений.
Вопрос 2: Как защитить скрипты автоматизации?
Ответ: версионный контроль через Git, code reviews, комплексная обработка ошибок, логирование, проверка прав доступа и регулярные security-аудиты.
Вопрос 3: Какие типичные ошибки при автоматизации?
Ответ: отсутствие обработки ошибок, недостаточное логирование, жёсткое кодирование путей, отсутствие тестов, игнорирование кодов выхода.
Вопрос 4: Как тестировать скрипты автоматизации?
Ответ: unit-тесты функций, интеграционные тесты рабочих процессов, staging окружения для production тестов, мониторинг в реальной эксплуатации.
Вопрос 5: Какие альтернативы Cron существуют?
Ответ: systemd-Timer (современная альтернатива), Jenkins Scheduled Jobs, Kubernetes CronJobs, облачные планировщики (AWS EventBridge).
Дополнительная информация
- http://linuxcommand.org/lc3_learning_the_shell.php
- https://learn.microsoft.com/de-de/powershell/
- https://automatetheboringstuff.com/
- https://docs.github.com/en/actions
- https://crontab.guru/
Рекомендация по книгам
Углубляй знания об автоматизации на Python с этим практичным справочником для начинающих и опытных разработчиков. Книга предлагает конкретные решения для реальных задач автоматизации и помогает писать эффективные скрипты.
Routineaufgaben mit Python automatisieren: Praktische Programmierlösungen für Einsteiger*innenЭта книга идеальна для:
- Начинающих, которые хотят освоить Python для автоматизации
- Системных администраторов, которые хотят автоматизировать повторяющиеся задачи
- Разработчиков, которые оптимизируют свои повседневные рабочие процессы
- Всех, кто ищет практические Python решения для повседневных проблем
Небольшая коллекция скриптов для типичных автоматизаций на Windows, Linux и Mac
Вот 15 полезных Python-скриптов для решения распространённых задач автоматизации на разных платформах.
Linux-скрипты (5)
1. Сжатие и оптимизация изображений
from PIL import Image
import os
from pathlib import Path
def resize_images(folder_path, max_size=1920, quality=85):
"""Сжимает все изображения в папке"""
for file in Path(folder_path).glob('*'):
if file.suffix.lower() in ['.jpg', '.jpeg', '.png']:
with Image.open(file) as img:
img.thumbnail((max_size, max_size))
img.save(file, quality=quality, optimize=True)
print(f"Оптимизировано: {file.name}")
# Использование: resize_images('/home/user/Pictures')2. Поиск дубликатов файлов
import hashlib
from pathlib import Path
from collections import defaultdict
def find_duplicates(folder_path):
"""Ищет одинаковые файлы по MD5-хешу"""
hashes = defaultdict(list)
for file in Path(folder_path).rglob('*'):
if file.is_file():
file_hash = hashlib.md5(file.read_bytes()).hexdigest()
hashes[file_hash].append(file)
duplicates = {h: files for h, files in hashes.items() if len(files) > 1}
for hash_val, files in duplicates.items():
print(f"Найдено дубликатов ({len(files)} файлов):")
for file in files:
print(f" - {file}")
# Использование: find_duplicates('/home/user/Downloads')3. Автоматическая очистка логов
import os
from pathlib import Path
from datetime import datetime, timedelta
def clean_old_logs(log_folder, days=30):
"""Удаляет логи старше указанного количества дней"""
cutoff_date = datetime.now() - timedelta(days=days)
for file in Path(log_folder).glob('*.log'):
if datetime.fromtimestamp(file.stat().st_mtime) < cutoff_date:
file.unlink()
print(f"Удалено: {file.name}")
# Использование: clean_old_logs('/var/log/myapp', days=30)4. Объединение PDF-файлов
from PyPDF2 import PdfMerger
from pathlib import Path
def merge_pdfs(folder_path, output_name='combined.pdf'):
"""Объединяет все PDF в папке в один файл"""
merger = PdfMerger()
for file in sorted(Path(folder_path).glob('*.pdf')):
merger.append(file)
print(f"Добавлено: {file.name}")
merger.write(output_name)
merger.close()
print(f"Создано: {output_name}")
# Использование: merge_pdfs('/home/user/Documents/PDFs')5. Мониторинг системы и уведомления
import psutil
import smtplib
from email.mime.text import MIMEText
def check_disk_usage(threshold=90):
"""Проверяет заполненность диска и отправляет оповещение"""
for partition in psutil.disk_partitions():
usage = psutil.disk_usage(partition.mountpoint)
percent = usage.percent
if percent > threshold:
print(f"ВНИМАНИЕ: {partition.mountpoint} заполнен на {percent}%")
# Здесь можно добавить отправку письма
# send_alert_email(partition.mountpoint, percent)
# Использование: check_disk_usage(threshold=90)Windows-скрипты (5)
1. Очистка рабочего стола
import shutil
from pathlib import Path
from datetime import datetime
def clean_downloads():
"""Сортирует файлы в папке Downloads"""
downloads = Path.home() / 'Downloads'
# Создаёт папки для разных типов файлов
folders = {
'Pictures': ['.jpg', '.jpeg', '.png', '.gif'],
'Documents': ['.pdf', '.doc', '.docx', '.txt'],
'Music': ['.mp3', '.wav', '.flac'],
'Videos': ['.mp4', '.avi', '.mkv'],
'Archives': ['.zip', '.rar', '.7z']
}
for folder, extensions in folders.items():
target = downloads / folder
target.mkdir(exist_ok=True)
for file in downloads.glob('*'):
if file.suffix.lower() in extensions and file.is_file():
shutil.move(str(file), str(target / file.name))
print(f"Перемещено: {file.name} -> {folder}")
# Использование: clean_downloads()2. Удаление временных файлов
import shutil
import os
from pathlib import Path
def clean_temp_files():
"""Очищает временные папки и кэш"""
temp_folders = [
Path(os.environ.get('TEMP', '')),
Path(os.environ.get('TMP', '')),
Path.home() / 'AppData' / 'Local' / 'Temp',
]
for folder in temp_folders:
if folder.exists():
try:
for file in folder.glob('*'):
if file.is_file():
file.unlink()
print(f"Очищена: {folder}")
except PermissionError:
print(f"Нет доступа: {folder}")
# Использование: clean_temp_files()3. Сканирование Wi-Fi сетей
import subprocess
def scan_wifi_networks():
"""Показывает доступные Wi-Fi сети"""
try:
result = subprocess.run(
['netsh', 'wlan', 'show', 'networks'],
capture_output=True,
text=True
)
networks = []
for line in result.stdout.split('\n'):
if 'SSID' in line:
ssid = line.split(':')[1].strip()
networks.append(ssid)
print("Доступные сети:")
for network in networks:
print(f" - {network}")
except Exception as e:
print(f"Ошибка: {e}")
# Использование: scan_wifi_networks()4. Менеджер буфера обмена
import pyperclip
import time
from datetime import datetime
def clipboard_history(max_entries=10):
"""Отслеживает историю буфера обмена"""
history = []
last_content = ""
print("Мониторинг буфера обмена запущен (Ctrl+C для выхода)")
try:
while True:
current = pyperclip.paste()
if current != last_content and current:
timestamp = datetime.now().strftime("%H:%M:%S")
history.insert(0, f"[{timestamp}] {current[:50]}...")
if len(history) > max_entries:
history.pop()
print(f"\nНовая запись: {current[:30]}...")
for entry in history:
print(f" {entry}")
last_content = current
time.sleep(2)
except KeyboardInterrupt:
print("\nМониторинг буфера обмена завершён")
# Использование: clipboard_history()5. Скриншот по горячей клавише
import pyautogui
import keyboard
from datetime import datetime
from pathlib import Path
def screenshot_on_hotkey(hotkey='f9'):
"""Делает скриншот при нажатии горячей клавиши"""
screenshots = Path.home() / 'Desktop' / 'Screenshots'
screenshots.mkdir(exist_ok=True)
def take_screenshot():
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = screenshots / f"screenshot_{timestamp}.png"
pyautogui.screenshot(str(filename))
print(f"Скриншот сохранён: {filename}")
keyboard.add_hotkey(hotkey, take_screenshot)
print(f"Нажмите {hotkey.upper()} для создания скриншота")
keyboard.wait() # Ожидает нажатия клавиши
# Использование: screenshot_on_hotkey('f9')Скрипты для Mac (5)
1. Автоматизация сортировки Finder
import shutil
from pathlib import Path
def organize_downloads():
"""Сортирует папку Downloads на Mac"""
downloads = Path.home() / 'Downloads'
file_types = {
'Bilder': ['.jpg', '.jpeg', '.png', '.heic', '.gif'],
'Dokumente': ['.pdf', '.pages', '.doc', '.docx', '.txt'],
'Musik': ['.mp3', '.m4a', '.wav', '.aiff'],
'Videos': ['.mp4', '.mov', '.avi', '.mkv'],
'Archiv': ['.zip', '.rar', '.7z', '.dmg']
}
for folder, extensions in file_types.items():
target = downloads / folder
target.mkdir(exist_ok=True)
for file in downloads.glob('*'):
if file.suffix.lower() in extensions and file.is_file():
shutil.move(str(file), str(target / file.name))
print(f"Organisiert: {file.name}")
# Nutzung: organize_downloads()2. Отправка уведомлений macOS
import subprocess
def send_notification(title, message):
"""Отправляет системное уведомление macOS"""
cmd = f"""
osascript -e 'display notification "{message}" with title "{title}"'
"""
subprocess.run(cmd, shell=True)
# Nutzung: send_notification("Backup fertig", "Dein Backup wurde erfolgreich abgeschlossen")3. Управление Spotify
import subprocess
def spotify_control(action):
"""Управляет Spotify через AppleScript"""
commands = {
'play': 'tell application "Spotify" to play',
'pause': 'tell application "Spotify" to pause',
'next': 'tell application "Spotify" to next track',
'previous': 'tell application "Spotify" to previous track'
}
if action in commands:
cmd = f'osascript -e \'tell application "Spotify" to {commands[action]}\''
subprocess.run(cmd, shell=True)
print(f"Spotify: {action}")
else:
print(f"Unbekannter Befehl: {action}")
# Nutzung: spotify_control('play')4. Управление яркостью экрана
import subprocess
def set_brightness(level):
"""Устанавливает яркость экрана (0-100)"""
if 0 <= level <= 100:
cmd = f"""
osascript -e 'tell application "System Events" to set brightness of every display to {level/100}'
"""
subprocess.run(cmd, shell=True)
print(f"Helligkeit auf {level}% gesetzt")
else:
print("Helligkeit muss zwischen 0 und 100 liegen")
# Nutzung: set_brightness(50)5. Извлечение текста из PDF
import PyPDF2
from pathlib import Path
def extract_pdf_text(pdf_path):
"""Извлекает текст из PDF-файла"""
text = ""
with open(pdf_path, 'rb') as file:
reader = PyPDF2.PdfReader(file)
for page in reader.pages:
text += page.extract_text() + "\n"
return text
def save_pdf_text_to_file(pdf_path, output_path):
"""Сохраняет извлечённый текст в файл"""
text = extract_pdf_text(pdf_path)
with open(output_path, 'w', encoding='utf-8') as file:
file.write(text)
print(f"Text extrahiert: {output_path}")
# Nutzung: save_pdf_text_to_file('/Users/user/Dokument.pdf', '/Users/user/Dokument.txt')Рекомендации по использованию
-
Установите необходимые пакеты:
pip install pillow pyperclip pyautogui pypdf2 psutil -
Учитывайте права доступа. Некоторые скрипты требуют прав администратора.
-
Адаптируйте пути. Измените пути в соответствии с вашей локальной окружением.
-
Тестируйте сначала. Всегда проверяйте скрипты на тестовых данных.
-
Создавайте резервные копии. Перед выполнением операций удаления сделайте резервное копирование.
Коллекция скриптов Bash и PowerShell
На Mac по умолчанию установлен Bash, а в новых версиях macOS используется zsh. Вот 15 дополнительных скриптов на нативных языках оболочки.
Bash-скрипты для Linux (5)
1. Автоматизированное резервное копирование с ротацией
#!/bin/bash
# backup.sh - Automatisches Backup mit Rotation
SOURCE_DIR="/home/user/documents"
BACKUP_DIR="/backup"
MAX_BACKUPS=7
DATE=$(date +%Y%m%d_%H%M%S)
BACKUP_NAME="backup_$DATE.tar.gz"
# Backup erstellen
tar -czf "$BACKUP_DIR/$BACKUP_NAME" "$SOURCE_DIR"
echo "Backup erstellt: $BACKUP_NAME"
# Alte Backups löschen (Rotation)
cd "$BACKUP_DIR"
ls -t backup_*.tar.gz | tail -n +$((MAX_BACKUPS + 1)) | xargs -r rm
echo "Alte Backups bereinigt (max $MAX_BACKUPS behalten)"2. Проверка статуса сервера
#!/bin/bash
# server_check.sh - Prüft Server-Status und Services
echo "=== Server Status Check ==="
echo "Zeit: $(date)"
echo ""
# CPU-Last
echo "CPU-Last:"
top -bn1 | grep "Cpu(s)" | sed "s/.*, *\([0-9.]*\)%id.*/\1/" | awk '{print 100 - $1"%"}'
# Speicher
echo ""
echo "Speicher:"
free -h
# Festplatte
echo ""
echo "Festplatte:"
df -h
# Wichtige Services prüfen
echo ""
echo "Services:"
for service in nginx mysql docker; do
if systemctl is-active --quiet "$service"; then
echo "✓ $service läuft"
else
echo "✗ $service läuft nicht"
fi
done3. Анализ файлов журнала
#!/bin/bash
# log_analyzer.sh - Analysiert Log-Dateien auf Fehler
LOG_FILE="/var/log/syslog"
ERROR_COUNT=$(grep -i "error" "$LOG_FILE" | wc -l)
WARNING_COUNT=$(grep -i "warning" "$LOG_FILE" | wc -l)
echo "=== Log-Analyse für $LOG_FILE ==="
echo "Fehler: $ERROR_COUNT"
echo "Warnungen: $WARNING_COUNT"
echo ""
# Letzte 10 Fehler anzeigen
if [ $ERROR_COUNT -gt 0 ]; then
echo "Letzte 10 Fehler:"
grep -i "error" "$LOG_FILE" | tail -n 10
fi4. Поиск больших файлов
#!/bin/bash
# find_large_files.sh - Findet große Dateien
DIRECTORY="/home/user"
MIN_SIZE="100M" # Mindestgröße
echo "=== Dateien größer als $MIN_SIZE in $DIRECTORY ==="
find "$DIRECTORY" -type f -size +"$MIN_SIZE" -exec ls -lh {} \; | awk '{print $5, $9}'5. Мониторинг процессов
#!/bin/bash
# process_monitor.sh - Überwacht spezifische Prozesse
PROCESSES=("nginx" "mysql" "redis" "docker")
echo "=== Prozess-Monitoring ==="
for process in "${PROCESSES[@]}"; do
if pgrep -x "$process" > /dev/null; then
PID=$(pgrep -x "$process")
MEMORY=$(ps -p "$PID" -o %mem --no-headers)
CPU=$(ps -p "$PID" -o %cpu --no-headers)
echo "✓ $process (PID: $PID, RAM: $MEMORY%, CPU: $CPU%)"
else
echo "✗ $process läuft nicht"
fi
donePowerShell-скрипты для Windows (5)
1. Сбор информации о системе
# system_info.ps1 - Sammelt detaillierte System-Informationen
Write-Host "=== System-Informationen ===" -ForegroundColor Cyan
# System-Info
Write-Host "`nSystem:" -ForegroundColor Yellow
Get-CimInstance Win32_ComputerSystem | Select-Object Manufacturer, Model, TotalPhysicalMemory
# Betriebssystem
Write-Host "`nBetriebssystem:" -ForegroundColor Yellow
Get-CimInstance Win32_OperatingSystem | Select-Object Caption, Version, BuildNumber
# CPU
Write-Host "`nCPU:" -ForegroundColor Yellow
Get-CimInstance Win32_Processor | Select-Object Name, NumberOfCores, MaxClockSpeed
# Festplatten
Write-Host "`nFestplatten:" -ForegroundColor Yellow
Get-CimInstance Win32_LogicalDisk | Select-Object DeviceID, Size, FreeSpace | Format-Table2. Список учётных записей пользователей
# user_accounts.ps1 - Listet alle User-Accounts auf
Write-Host "=== User-Accounts ===" -ForegroundColor Cyan
Get-LocalUser | Select-Object Name, Enabled, LastLogon | Format-Table
# Admin-Accounts
Write-Host "`nAdmin-Accounts:" -ForegroundColor Yellow
Get-LocalGroupMember -Group "Administrators" | Select-Object Name, PrincipalSource3. Проверка обновлений Windows
# check_updates.ps1 - Prüft auf Windows-Updates
Write-Host "=== Windows-Updates prüfen ===" -ForegroundColor Cyan
try {
$UpdateSession = New-Object -ComObject Microsoft.Update.Session
$UpdateSearcher = $UpdateSession.CreateUpdateSearcher()
$Updates = $UpdateSearcher.Search("IsInstalled=0").Updates
if ($Updates.Count -eq 0) {
Write-Host "Keine ausstehenden Updates" -ForegroundColor Green
} else {
Write-Host "$($Updates.Count) ausstehende Updates:" -ForegroundColor Yellow
foreach ($Update in $Updates) {
Write-Host " - $($Update.Title)"
}
}
} catch {
Write-Host "Fehler beim Prüfen der Updates: $_" -ForegroundColor Red
}4. Мониторинг сетевых соединений
# network_monitor.ps1 - Überwacht aktive Netzwerk-Verbindungen
Write-Host "=== Aktive Netzwerk-Verbindungen ===" -ForegroundColor Cyan
Get-NetTCPConnection | Where-Object {$_.State -eq "Established"} |
Select-Object LocalAddress, LocalPort, RemoteAddress, RemotePort, State |
Format-Table
# Offene Ports
Write-Host "`nOffene Ports:" -ForegroundColor Yellow
Get-NetTCPConnection | Where-Object {$_.State -eq "Listening"} |
Select-Object LocalAddress, LocalPort, OwningProcess |
Format-Table5. Анализ журнала событий
# event_log.ps1 - Analysiert Windows Event-Logs
Write-Host "=== Event-Log Analyse (letzte 24h) ===" -ForegroundColor Cyan
$StartDate = (Get-Date).AddDays(-1)
# Fehler im System-Log
Write-Host "`nSystem-Fehler:" -ForegroundColor Yellow
Get-EventLog -LogName System -EntryType Error -After $StartDate |
Select-Object TimeGenerated, Source, Message |
Format-Table -Wrap
# Warnungen im Application-Log
Write-Host "`nApplication-Warnungen:" -ForegroundColor Yellow
Get-EventLog -LogName Application -EntryType Warning -After $StartDate |
Select-Object TimeGenerated, Source, Message |
Format-Table -WrapBash/zsh-скрипты для Mac (5)
1. Информация о системе Mac
#!/bin/bash
# mac_info.sh - Выводит информацию о системе Mac
echo "=== Mac-System-Информация ==="
echo ""
# Оборудование
echo "Оборудование:"
system_profiler SPHardwareDataType | grep -E "Model Name|Processor|Memory"
# Версия macOS
echo ""
echo "Версия macOS:"
sw_vers
# Диск
echo ""
echo "Диск:"
df -h
# Состояние батареи (ноутбуки)
echo ""
echo "Состояние батареи:"
system_profiler SPPowerDataType | grep -E "Charge|Capacity"2. Обновление пакетов Homebrew
#!/bin/bash
# brew_update.sh - Обновляет пакеты Homebrew
echo "=== Homebrew Update ==="
echo ""
# Обновление самого Homebrew
echo "Обновление Homebrew..."
brew update
# Обновление пакетов
echo ""
echo "Обновление пакетов..."
brew upgrade
# Очистка
echo ""
echo "Очистка..."
brew cleanup
echo "Обновление завершено!"3. Автоматизация сортировки Finder
#!/bin/bash
# finder_organize.sh - Сортирует папку Downloads
DOWNLOADS="$HOME/Downloads"
echo "=== Организация Downloads ==="
# Создание папок
mkdir -p "$DOWNLOADS"/{Bilder,Dokumente,Musik,Videos,Archiv}
# Перемещение файлов
for file in "$DOWNLOADS"/*; do
if [ -f "$file" ]; then
case "${file,,}" in
*.jpg|*.jpeg|*.png|*.gif|*.heic)
mv "$file" "$DOWNLOADS/Bilder/"
echo "Bilder: $(basename "$file")"
;;
*.pdf|*.doc|*.docx|*.txt|*.pages)
mv "$file" "$DOWNLOADS/Dokumente/"
echo "Dokumente: $(basename "$file")"
;;
*.mp3|*.m4a|*.wav|*.aiff)
mv "$file" "$DOWNLOADS/Musik/"
echo "Musik: $(basename "$file")"
;;
*.mp4|*.mov|*.avi|*.mkv)
mv "$file" "$DOWNLOADS/Videos/"
echo "Videos: $(basename "$file")"
;;
*.zip|*.rar|*.7z|*.dmg)
mv "$file" "$DOWNLOADS/Archiv/"
echo "Archiv: $(basename "$file")"
;;
esac
fi
done
echo "Организация завершена!"4. Управление SSH-ключами
#!/bin/bash
# ssh_keys.sh - Управляет SSH-ключами
SSH_DIR="$HOME/.ssh"
echo "=== Управление SSH-ключами ==="
echo ""
# Проверка SSH-директории
if [ ! -d "$SSH_DIR" ]; then
echo "Создание SSH-директории..."
mkdir -p "$SSH_DIR"
chmod 700 "$SSH_DIR"
fi
# Список существующих ключей
echo "Существующие ключи:"
ls -la "$SSH_DIR"/*.pub 2>/dev/null || echo "Публичные ключи не найдены"
# Создание нового ключа (опционально)
read -p "Создать новый SSH-ключ? (j/n): " create_key
if [ "$create_key" = "j" ]; then
read -p "Имя ключа (например, github): " key_name
ssh-keygen -t ed25519 -f "$SSH_DIR/$key_name" -C "$key_name"
echo "Ключ создан: $SSH_DIR/$key_name"
fi5. Резервная копия и восстановление Dock
#!/bin/bash
# dock_backup.sh - Создает резервную копию и восстанавливает конфигурацию Dock
DOCK_PLIST="$HOME/Library/Preferences/com.apple.dock.plist"
BACKUP_DIR="$HOME/DockBackups"
echo "=== Dock-Резервная копия ==="
# Создание резервной копии
backup_dock() {
mkdir -p "$BACKUP_DIR"
DATE=$(date +%Y%m%d_%H%M%S)
cp "$DOCK_PLIST" "$BACKUP_DIR/dock_backup_$DATE.plist"
echo "Dock сохранен: dock_backup_$DATE.plist"
}
# Восстановление Dock
restore_dock() {
echo "Доступные резервные копии:"
ls -lt "$BACKUP_DIR"/*.plist
read -p "Введите имя файла резервной копии: " backup_file
cp "$backup_file" "$DOCK_PLIST"
killall Dock
echo "Dock восстановлен и перезагружен"
}
# Меню
echo "1: Сохранить Dock"
echo "2: Восстановить Dock"
read -p "Выбор: " choice
case $choice in
1) backup_dock ;;
2) restore_dock ;;
*) echo "Неверный выбор" ;;
esacПримечания по использованию shell-скриптов
Bash-скрипты (Linux/Mac):
- Сделать исполняемым:
chmod +x skript.sh - Запустить:
./skript.shилиbash skript.sh - Проверить путь:
which bash(обычно/bin/bash)
PowerShell-скрипты (Windows):
- Проверить политику выполнения:
Get-ExecutionPolicy - При необходимости изменить:
Set-ExecutionPolicy RemoteSigned - Запустить:
.\skript.ps1илиpowershell -File skript.ps1
Специфика Mac:
- Стандартная оболочка с macOS Catalina - zsh
- Bash остается доступен:
/bin/bash - Для интеграции с AppleScript используйте:
osascript
Best Practices для скриптов автоматизации
1. Стандарты кодирования и соглашения об именовании
Python:
# ✅ Хорошие соглашения об именовании
def backup_database(source_path: str, backup_dir: str) -> bool:
"""Создает резервную копию базы данных"""
pass
# ❌ Плохие соглашения об именовании
def db(s, d):
passBash:
# ✅ Хорошие соглашения об именовании
backup_database() {
local source_path="$1"
local backup_dir="$2"
}
# ❌ Плохие соглашения об именовании
bd() {
s=$1
d=$2
}PowerShell:
# ✅ Хорошие соглашения об именовании (PascalCase, Verb-Noun)
function Backup-Database {
param(
[string]$SourcePath,
[string]$BackupDir
)
}
# ❌ Плохие соглашения об именовании
function bd {
param($s, $d)
}2. Документация и комментарии
Docstrings и комментарии:
def process_files(directory: str, pattern: str) -> list:
"""
Обрабатывает все файлы в директории, соответствующие шаблону.
Args:
directory: Путь к директории
pattern: Шаблон файла (например, '*.log')
Returns:
Список обработанных файлов
Raises:
FileNotFoundError: Если директория не существует
"""
if not os.path.exists(directory):
raise FileNotFoundError(f"Директория не найдена: {directory}")
# ... остальная часть функцииBash-комментарии:
#!/bin/bash
# backup.sh - Автоматическое резервное копирование с ротацией
# Создано: 2026-07-14
# Автор: Ваше имя
# Версия: 1.0
# Конфигурация
SOURCE_DIR="/home/user/documents" # Исходная директория
BACKUP_DIR="/backup" # Директория резервной копии
MAX_BACKUPS=7 # Максимальное количество резервных копий3. Модульность и повторное использование
Функции вместо монолита:
# ✅ Модульный подход
def validate_path(path: str) -> bool:
"""Проверяет существование пути"""
return os.path.exists(path)
def copy_file(source: str, target: str) -> bool:
"""Копирует файл с обработкой ошибок"""
try:
shutil.copy(source, target)
return True
except Exception as e:
log_error(f"Ошибка копирования: {e}")
return False
def backup_files(files: list, backup_dir: str) -> bool:
"""Основная функция для резервного копирования"""
if not validate_path(backup_dir):
return False
for file in files:
if not copy_file(file, backup_dir):
return False
return True4. Отделение конфигурации от кода
Использование файлов конфигурации:
# config.yaml
backup:
source_dir: "/home/user/documents"
backup_dir: "/backup"
max_backups: 7
compression: "gzip"
# script.py
import yaml
def load_config(config_file: str) -> dict:
"""Загружает конфигурацию из YAML-файла"""
with open(config_file, 'r') as f:
return yaml.safe_load(f)
config = load_config('config.yaml')
backup_config = config['backup']5. Логирование и мониторинг
Структурированное логирование:
import logging
from datetime import datetime
# Logging konfigurieren
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('automation.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
def process_data(data: str):
"""Verarbeitet Daten mit Logging"""
logger.info(f"Verarbeitung gestartet: {len(data)} Zeichen")
try:
result = transform_data(data)
logger.info(f"Verarbeitung erfolgreich")
return result
except Exception as e:
logger.error(f"Verarbeitung fehlgeschlagen: {e}")
raise6. Обработка ошибок и восстановление
Надежная обработка исключений:
import time
from functools import wraps
def retry(max_attempts: int = 3, delay: int = 1):
"""Decorator für Retry-Mechanismus"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt == max_attempts - 1:
raise
logger.warning(f"Versuch {attempt + 1} fehlgeschlagen: {e}")
time.sleep(delay)
return wrapper
return decorator
@retry(max_attempts=3, delay=2)
def download_file(url: str, target: str):
"""Lädt Datei mit Retry herunter"""
# Download-Logik
pass7. Тестирование и валидация
Простые тесты:
def test_validate_path():
"""Test für validate_path Funktion"""
# Test mit existierendem Pfad
assert validate_path('/tmp') == True
# Test mit nicht-existierendem Pfad
assert validate_path('/nicht/existent') == False
print("✓ validate_path Tests bestanden")
if __name__ == '__main__':
test_validate_path()8. Оптимизация производительности
Эффективная обработка данных:
# ❌ Ineffizient (für große Dateien)
def process_large_file_slow(filepath: str):
with open(filepath, 'r') as f:
content = f.read() # Lädt alles in RAM
return process_content(content)
# ✅ Effizient (streaming)
def process_large_file_fast(filepath: str):
with open(filepath, 'r') as f:
for line in f: # Zeilenweise Verarbeitung
yield process_line(line)9. Безопасность в скриптах
Не встраивайте секреты в код:
# ❌ Schlecht - Passworte im Code
password = "geheim123"
# ✅ Gut - Environment Variables
import os
password = os.getenv('DB_PASSWORD')
# ✅ Noch besser - Secrets Manager
from keyring import get_password
password = get_password('system', 'username')10. Версионирование и развертывание
Semantic Versioning:
# Versionierung im Skript
VERSION="1.2.3" # MAJOR.MINOR.PATCH
# Changelog
# 1.2.3 - Bugfix für Backup-Rotation
# 1.2.0 - Neue Logging-Funktion
# 1.0.0 - Erste VersionБезопасность в скриптах автоматизации
1. Управление секретами
Переменные окружения:
# .env Datei (nicht in Git committen!)
DB_PASSWORD=geheim123
API_KEY=sk-1234567890
# script.py
import os
from dotenv import load_dotenv
load_dotenv() # Lädt .env Datei
db_password = os.getenv('DB_PASSWORD')
api_key = os.getenv('API_KEY')Менеджеры секретов:
# AWS Secrets Manager
import boto3
def get_secret(secret_name: str) -> str:
"""Holt Secret aus AWS Secrets Manager"""
client = boto3.client('secretsmanager')
response = client.get_secret_value(SecretId=secret_name)
return response['SecretString']
# Azure Key Vault
from azure.keyvault.secrets import SecretClient
from azure.identity import DefaultAzureCredential
def get_azure_secret(vault_url: str, secret_name: str) -> str:
"""Holt Secret aus Azure Key Vault"""
credential = DefaultAzureCredential()
client = SecretClient(vault_url=vault_url, credential=credential)
return client.get_secret(secret_name).value2. Валидация входных данных
Проверка данных пользователя:
import re
from pathlib import Path
def validate_filename(filename: str) -> bool:
"""Validiert Dateinamen auf Sicherheit"""
# Erlaubt nur alphanumerische Zeichen, Unterstrich, Bindestrich
pattern = r'^[a-zA-Z0-9_-]+\.[a-zA-Z0-9]{3,4}$'
return bool(re.match(pattern, filename))
def validate_path(path: str) -> bool:
"""Validiert Pfad auf Path Traversal Angriffe"""
resolved_path = Path(path).resolve()
# Prüft ob Pfad außerhalb des erlaubten Bereichs liegt
return not any(part in ['..', '~'] for part in resolved_path.parts)
# Nutzung
filename = input("Dateiname: ")
if not validate_filename(filename):
raise ValueError("Ungültiger Dateiname")3. Права доступа и разрешения
Принцип минимальных привилегий:
#!/bin/bash
# Skript mit minimalen Rechten ausführen
# Prüfen ob Skript als root läuft
if [ "$EUID" -eq 0 ]; then
echo "Dieses Skript sollte nicht als root ausgeführt werden"
exit 1
fi
# Dateien mit sicheren Permissions erstellen
umask 077 # Nur Owner hat Rechte
touch sensitive_file.txtРазрешения в Python:
import os
import stat
def set_secure_permissions(filepath: str):
"""Setzt sichere Datei-Permissions (nur Owner)"""
os.chmod(filepath, stat.S_IRUSR | stat.S_IWUSR) # 0600
def set_executable_permissions(filepath: str):
"""Setzt ausführbare Permissions (Owner rwx)"""
os.chmod(filepath, stat.S_IRWXU) # 07004. Защита от SQL-инъекций
Параметризованные запросы:
# ❌ UNSICHER - SQL Injection möglich
query = f"SELECT * FROM users WHERE name = '{username}'"
cursor.execute(query)
# ✅ SICHER - Parameterized Query
query = "SELECT * FROM users WHERE name = %s"
cursor.execute(query, (username,))5. Защита от command injection
Безопасное выполнение команд:
import subprocess
# ❌ UNSICHER - Command Injection möglich
user_input = "user; rm -rf /"
os.system(f"echo {user_input}")
# ✅ SICHER - subprocess mit Liste
subprocess.run(['echo', user_input], check=True)
# ✅ Noch sicherer - shlex.quote
import shlex
safe_input = shlex.quote(user_input)
subprocess.run(f"echo {safe_input}", shell=True, check=True)6. Безопасность операций с файлами
Безопасная работа с файлами:
import os
import tempfile
def safe_file_write(filepath: str, content: str):
"""Sicheres Schreiben mit Atomic Write"""
# Temporäre Datei im gleichen Verzeichnis
dir_path = os.path.dirname(filepath)
with tempfile.NamedTemporaryFile(
mode='w',
dir=dir_path,
delete=False
) as tmp_file:
tmp_file.write(content)
tmp_path = tmp_file.name
# Atomic rename
os.rename(tmp_path, filepath)
def safe_file_delete(filepath: str):
"""Sicheres Löschen mit Bestätigung"""
if not os.path.exists(filepath):
return False
# Bestätigung einholen
response = input(f"Soll {filepath} wirklich gelöscht werden? (j/n): ")
if response.lower() == 'j':
os.remove(filepath)
return True
return False7. Логирование чувствительных данных
Маскирование секретов в логах:
import logging
from typing import Any
class SecureFormatter(logging.Formatter):
"""Formatter der sensitive Daten maskiert"""
SENSITIVE_KEYS = ['password', 'token', 'secret', 'key']
def format(self, record: logging.LogRecord) -> str:
msg = super().format(record)
for key in self.SENSITIVE_KEYS:
msg = msg.replace(f'{key}=', f'{key}=***')
return msg
# Nutzung
logger = logging.getLogger(__name__)
handler = logging.StreamHandler()
handler.setFormatter(SecureFormatter())
logger.addHandler(handler)
logger.info("Login mit password=geheim123")
# Ausgabe: Login mit password=***8. HTTPS и TLS
Защита сетевых соединений:
import requests
import urllib3
# Отключение SSL-предупреждений (не для Production!)
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
# ✅ БЕЗОПАСНО - HTTPS с проверкой сертификата
response = requests.get('https://api.example.com', verify=True)
# ❌ ОПАСНО - проверка SSL отключена
response = requests.get('https://api.example.com', verify=False)
# ✅ БЕЗОПАСНО - собственный CA-сертификат
response = requests.get(
'https://api.example.com',
verify='/path/to/ca-bundle.crt'
)9. Dependency Security
Сканирование уязвимостей:
# Проверка Python-зависимостей
pip install safety
safety check
# Проверка npm-зависимостей
npm audit
# Обновление всех зависимостей
pip install --upgrade pip
pip list --outdated10. Security Audits
Регулярные проверки безопасности:
import subprocess
import json
def run_security_audit():
"""Выполняет базовые проверки безопасности"""
checks = {
'python_version': check_python_version,
'dependencies': check_dependencies,
'file_permissions': check_file_permissions,
'hardcoded_secrets': check_hardcoded_secrets
}
results = {}
for check_name, check_func in checks.items():
results[check_name] = check_func()
return results
def check_hardcoded_secrets():
"""Ищет захардкодированные секреты в коде"""
import re
secret_patterns = [
r'password\s*=\s*["\'][^"\']+["\']',
r'api_key\s*=\s*["\'][^"\']+["\']',
r'secret\s*=\s*["\'][^"\']+["\']'
]
findings = []
for root, dirs, files in os.walk('.'):
for file in files:
if file.endswith('.py'):
filepath = os.path.join(root, file)
with open(filepath, 'r') as f:
content = f.read()
for pattern in secret_patterns:
if re.search(pattern, content):
findings.append(filepath)
return findingsТестирование скриптов автоматизации
1. Unit-тестирование Python-скриптов
Простые unit-тесты:
# test_backup.py
import unittest
import tempfile
import os
from backup_script import backup_files, validate_path
class TestBackupFunctions(unittest.TestCase):
def setUp(self):
"""Подготовка к тестам"""
self.temp_dir = tempfile.mkdtemp()
self.test_file = os.path.join(self.temp_dir, 'test.txt')
with open(self.test_file, 'w') as f:
f.write('Test content')
def tearDown(self):
"""Очистка после тестов"""
import shutil
shutil.rmtree(self.temp_dir)
def test_validate_path_existing(self):
"""Проверка существующего пути"""
self.assertTrue(validate_path(self.temp_dir))
def test_validate_path_non_existing(self):
"""Проверка несуществующего пути"""
self.assertFalse(validate_path('/nicht/existent'))
def test_backup_files_success(self):
"""Проверка успешного резервного копирования"""
backup_dir = os.path.join(self.temp_dir, 'backup')
os.makedirs(backup_dir)
result = backup_files([self.test_file], backup_dir)
self.assertTrue(result)
self.assertTrue(os.path.exists(os.path.join(backup_dir, 'test.txt')))
if __name__ == '__main__':
unittest.main()2. Mocking для внешних зависимостей
Имитация операций с файловой системой:
from unittest.mock import patch, MagicMock
import unittest
class TestBackupWithMocking(unittest.TestCase):
@patch('backup_script.shutil.copy')
def test_backup_with_mock(self, mock_copy):
"""Тест с имитацией функции shutil.copy"""
mock_copy.return_value = True
result = backup_files(['file1.txt'], '/backup')
self.assertTrue(result)
mock_copy.assert_called_once()
@patch('backup_script.os.path.exists')
def test_validate_path_mock(self, mock_exists):
"""Тест с имитацией os.path.exists"""
mock_exists.return_value = True
result = validate_path('/any/path')
self.assertTrue(result)
mock_exists.assert_called_once_with('/any/path')3. Integration-тестирование
Сквозные тесты:
# test_integration.py
import unittest
import tempfile
import os
import subprocess
class TestIntegration(unittest.TestCase):
def test_script_execution(self):
"""Проверка полного выполнения скрипта"""
script_path = 'backup.sh'
test_dir = tempfile.mkdtemp()
# Запуск скрипта
result = subprocess.run(
['bash', script_path, test_dir],
capture_output=True,
text=True
)
# Проверка результата
self.assertEqual(result.returncode, 0)
self.assertIn('Backup erstellt', result.stdout)
# Очистка
import shutil
shutil.rmtree(test_dir)4. Тестирование Bash-скриптов
Bash-фреймворк для unit-тестов:
#!/bin/bash
# test_backup.sh - тесты для backup.sh
# Test-функции
test_validate_path() {
source backup.sh
# Тест с существующим путём
if validate_path "/tmp"; then
echo "✓ validate_path: существующий путь"
else
echo "✗ validate_path: проверка существующего пути не прошла"
exit 1
fi
# Тест с несуществующим путём
if ! validate_path "/nicht/existent"; then
echo "✓ validate_path: несуществующий путь"
else
echo "✗ validate_path: проверка несуществующего пути не прошла"
exit 1
fi
}
test_backup_function() {
source backup.sh
local test_dir=$(mktemp -d)
local test_file="$test_dir/test.txt"
echo "Test content" > "$test_file"
# Проверка функции резервного копирования
backup_files "$test_file" "$test_dir/backup"
if [ -f "$test_dir/backup/test.txt" ]; then
echo "✓ backup_files: файл успешно скопирован"
else
echo "✗ backup_files: файл не скопирован"
exit 1
fi
# Очистка
rm -rf "$test_dir"
}
# Запуск всех тестов
echo "=== Bash-тесты ==="
test_validate_path
test_backup_function
echo "Все тесты прошли успешно!"5. PowerShell-тестирование
Pester фреймворк для PowerShell:
# test_backup.ps1 - Pester тесты для backup.ps1
Describe "Backup-функции" {
BeforeAll {
. .\backup.ps1
$testDir = Join-Path $env:TEMP "backup_test"
New-Item -Path $testDir -ItemType Directory -Force | Out-Null
}
AfterAll {
Remove-Item -Path $testDir -Recurse -Force -ErrorAction SilentlyContinue
}
Context "Validate-Path" {
It "должна вернуть существующий путь" {
$result = Validate-Path $testDir
$result | Should -Be $true
}
It "должна отклонить несуществующий путь" {
$result = Validate-Path "C:\nicht\existent"
$result | Should -Be $false
}
}
Context "Backup-Files" {
It "должна успешно скопировать файлы" {
$testFile = Join-Path $testDir "test.txt"
"Test content" | Out-File -FilePath $testFile
$backupDir = Join-Path $testDir "backup"
$result = Backup-Files $testFile $backupDir
$result | Should -Be $true
(Test-Path (Join-Path $backupDir "test.txt")) | Should -Be $true
}
}
}6. Test-Driven Development (TDD)
Workflow TDD:
# 1. Напишем тест (он будет падать)
def test_calculate_backup_size():
"""Тест для расчета размера резервной копии"""
result = calculate_backup_size('/tmp')
assert result > 0 # Будет падать, функция еще не существует
# 2. Реализуем функцию (чтобы тест прошел)
def calculate_backup_size(directory: str) -> int:
"""Вычисляет размер директории"""
total_size = 0
for dirpath, dirnames, filenames in os.walk(directory):
for filename in filenames:
filepath = os.path.join(dirpath, filename)
total_size += os.path.getsize(filepath)
return total_size
# 3. Запустим тест (теперь он должен пройти)
# 4. Рефакторинг (тесты продолжают проходить)7. Continuous Testing
Автоматические тесты при изменениях:
# watchmedo для автоматического запуска тестов
pip install watchdog watchmedo
# Запускать тесты при изменении файлов
watchmedo shell-command \
--patterns="*.py" \
--recursive \
--command='python -m pytest test_*.py -v'8. Test-Coverage
Анализ покрытия:
# Установим pytest-cov
pip install pytest-cov
# Запустим тесты с анализом покрытия
pytest --cov=backup_script --cov-report=html
# Откроем отчет
open htmlcov/index.html9. Property-Based Testing
Hypothesis для Python:
from hypothesis import given, strategies as st
import unittest
class TestBackupProperties(unittest.TestCase):
@given(st.lists(st.text()))
def test_backup_preserves_content(self, content_list):
"""Тест свойства: резервная копия сохраняет содержимое"""
# Тестируем с разными входными данными
for content in content_list:
with tempfile.NamedTemporaryFile(mode='w', delete=False) as f:
f.write(content)
temp_file = f.name
# Создаем резервную копию
backup_file = temp_file + '.backup'
shutil.copy(temp_file, backup_file)
# Проверяем содержимое
with open(backup_file, 'r') as f:
backup_content = f.read()
self.assertEqual(content, backup_content)
# Очистка
os.unlink(temp_file)
os.unlink(backup_file)10. Load-Testing для скриптов
Тесты производительности:
import time
import statistics
def test_backup_performance():
"""Тест производительности резервного копирования"""
test_times = []
for i in range(10):
start_time = time.time()
backup_files(['large_file.txt'], '/backup')
end_time = time.time()
test_times.append(end_time - start_time)
avg_time = statistics.mean(test_times)
max_time = max(test_times)
print(f"Среднее время: {avg_time:.2f}s")
print(f"Максимальное время: {max_time:.2f}s")
# Проверка производительности
assert avg_time < 5.0, "Резервная копия работает слишком медленно"
assert max_time < 10.0, "Превышено максимальное время резервного копирования"Контроль версий для скриптов автоматизации
1. Основы Git для скриптов
Инициализация и первый коммит:
# Инициализируем Git-репозиторий
git init
# Создаем .gitignore
cat > .gitignore << EOF
# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
env/
venv/
.venv/
# Secrets
.env
*.key
*.pem
secrets/
# Logs
*.log
# IDE
.vscode/
.idea/
*.swp
*.swo
# OS
.DS_Store
Thumbs.db
EOF
# Добавляем скрипты
git add backup.sh backup.py config.yaml
git commit -m "Initial commit: Backup-Скрипты добавлены"2. Стратегии ветвления
Git-Flow для скриптов:
# Main-ветка (продакшн)
git checkout -b main
# Develop-ветка (разработка)
git checkout -b develop
# Feature-ветка для новой функции
git checkout -b feature/add-logging
# Коммитим изменения
git add backup.py
git commit -m "Add logging functionality"
# Мержим обратно в develop
git checkout develop
git merge feature/add-logging
# Удаляем feature-ветку
git branch -d feature/add-logging3. Конвенциональные commit-сообщения
Conventional Commits:
# Формат: <type>(<scope>): <description>
# Примеры:
git commit -m "feat(backup): Add compression support"
git commit -m "fix(logging): Fix timestamp format"
git commit -m "docs(readme): Update installation instructions"
git commit -m "style(python): Apply PEP 8 formatting"
git commit -m "refactor(cleanup): Simplify error handling"
git commit -m "test(backup): Add unit tests for backup function"
git commit -m "chore(deps): Update dependencies"4. Code Review и Pull Request
Workflow Pull Request:
# Создаем feature-ветку
git checkout -b feature/backup-rotation
# Вносим изменения и коммитим
git add backup.sh
git commit -m "feat(backup): Add automatic backup rotation"
# Пушим в remote
git push origin feature/backup-rotation
# Создаем Pull Request (GitHub/GitLab/Bitbucket)
# Члены команды проверяют код
# После одобрения мержим5. Git-hooks для качества
Pre-commit hook:
#!/bin/bash
# .git/hooks/pre-commit
# Форматируем Python-код
autopep8 --in-place --aggressive *.py
# Линтируем Python-код
pylint *.py
# Запускаем тесты
python -m pytest test_*.py
# При успехе
exit 0Pre-push hook:
#!/bin/bash
# .git/hooks/pre-push
# Запускаем все тесты
python -m pytest
# Проверка безопасности
safety check
# При успехе
exit 06. Версионирование с помощью tags
Semantic Versioning:
# Основной релиз (несовместимые изменения)
git tag -a v2.0.0 -m "Major release: Breaking changes"
git push origin v2.0.0
# Минорный релиз (новые функции, совместимость сохранена)
git tag -a v1.2.0 -m "Minor release: New features"
git push origin v1.2.0
# Патч-релиз (исправления ошибок, совместимость сохранена)
git tag -a v1.1.1 -m "Patch release: Bug fixes"
git push origin v1.1.17. Ведение Changelog
Автоматический Changelog:
# С использованием conventional-changelog
npm install -g conventional-changelog-cli
# Генерируем Changelog
conventional-changelog -p angular -i CHANGELOG.md -s
# Структура CHANGELOG.md:
# ## [1.2.0] - 2026-07-14
# ### Added
# - Добавлена ротация резервных копий
# - Реализована функциональность логирования
#
# ### Fixed
# - Исправлена ошибка при валидации пути
#
# ## [1.1.0] - 2026-06-01
# ### Added
# - Первая версия скрипта резервного копирования8. Конфигурация Git
Git-конфигурация для скриптов:
# Настраиваем пользователя
git config user.name "Ваше имя"
git config user.email "your@email.com"
# Line-endings (важно для кроссплатформенности)
git config core.autocrlf input # Linux/Mac
git config core.autocrlf true # Windows
# Файловые атрибуты
git config core.filemode false # Игнорируем права доступа
# Настраиваем diff-tool
git config diff.tool vimdiff
git config merge.tool vimdiff9. Git-workflow для команд
Forking-workflow:
# 1. Форкируем репозиторий (GitHub/GitLab)
# 2. Клонируем fork
git clone https://github.com/your-username/repo.git
# 3. Добавляем upstream remote
git remote add upstream https://github.com/original/repo.git
# 4. Создаем feature-ветку
git checkout -b feature/neue-funktion
# 5. Коммитим изменения
git add script.py
git commit -m "feat: neue Funktion hinzugefügt"
# 6. Пушим в свой fork
git push origin feature/neue-funktion
# 7. Создаем Pull Request
# 8. После review мержим
# 9. Получаем изменения из upstream
git fetch upstream
git checkout main
git merge upstream/main10. Git для стратегий резервного копирования
Git как система резервного копирования:
#!/bin/bash
# git_backup.sh - Автоматическое резервное копирование Git
REPO_DIR="/path/to/repo"
BACKUP_REMOTE="git@github.com:user/backup-repo.git"
cd "$REPO_DIR"
# Добавить изменения
git add .
# Commit с временной меткой
git commit -m "Auto-backup: $(date '+%Y-%m-%d %H:%M:%S')"
# Отправить на Remote
git push "$BACKUP_REMOTE" main
echo "Резервная копия успешно отправлена в Git"CI/CD интеграция для скриптов автоматизации
1. GitHub Actions для Python-скриптов
Файл workflow:
# .github/workflows/automation.yml
name: Automation Scripts CI/CD
on:
push:
branches: [ main, develop ]
pull_request:
branches: [ main ]
jobs:
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.11'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements.txt
pip install pytest pytest-cov pylint safety
- name: Lint with pylint
run: pylint *.py
- name: Security check
run: safety check
- name: Run tests
run: pytest --cov=. --cov-report=xml
- name: Upload coverage
uses: codecov/codecov-action@v3
with:
file: ./coverage.xml
deploy:
needs: test
runs-on: ubuntu-latest
if: github.ref == 'refs/heads/main'
steps:
- uses: actions/checkout@v3
- name: Deploy to production
run: |
echo "Deploying automation scripts to production"
# Логика развёртывания здесь2. GitLab CI для Bash-скриптов
Файл GitLab CI:
# .gitlab-ci.yml
stages:
- test
- deploy
variables:
SCRIPT_DIR: "/opt/automation"
test_scripts:
stage: test
image: alpine:latest
before_script:
- apk add --no-cache bash shellcheck
script:
- shellcheck *.sh
- bash -n backup.sh
- bash test_backup.sh
artifacts:
reports:
junit: test-results.xml
deploy_production:
stage: deploy
image: alpine:latest
only:
- main
script:
- apk add --no-cache rsync openssh
- mkdir -p ~/.ssh
- echo "$SSH_PRIVATE_KEY" > ~/.ssh/id_rsa
- chmod 600 ~/.ssh/id_rsa
- rsync -avz --delete *.sh user@server:$SCRIPT_DIR/
- ssh user@server "chmod +x $SCRIPT_DIR/*.sh"3. Jenkins Pipeline
Jenkinsfile:
// Jenkinsfile
pipeline {
agent any
environment {
PYTHON_VERSION = '3.11'
DEPLOY_SERVER = 'user@server'
}
stages {
stage('Checkout') {
steps {
checkout scm
}
}
stage('Lint') {
steps {
sh 'pylint *.py || true'
sh 'shellcheck *.sh || true'
}
}
stage('Test') {
steps {
sh 'python -m pytest test_*.py --cov=. --cov-report=html'
}
}
stage('Security Scan') {
steps {
sh 'safety check'
}
}
stage('Deploy') {
when {
branch 'main'
}
steps {
sh '''
rsync -avz --delete *.sh ${DEPLOY_SERVER}:/opt/automation/
ssh ${DEPLOY_SERVER} "chmod +x /opt/automation/*.sh"
'''
}
}
}
post {
always {
archiveArtifacts artifacts: 'htmlcov/**', allowEmptyArchive: true
publishHTML target: [
reportDir: 'htmlcov',
reportFiles: 'index.html',
reportName: 'Coverage Report'
]
}
}
}4. Azure DevOps Pipeline
azure-pipelines.yml:
# azure-pipelines.yml
trigger:
branches:
include:
- main
- develop
pool:
vmImage: 'ubuntu-latest'
variables:
pythonVersion: '3.11'
stages:
- stage: Test
jobs:
- job: TestScripts
steps:
- task: UsePythonVersion@0
inputs:
versionSpec: '$(pythonVersion)'
- script: |
python -m pip install --upgrade pip
pip install pytest pytest-cov pylint safety
displayName: 'Install dependencies'
- script: |
pylint *.py
displayName: 'Lint Python'
- script: |
safety check
displayName: 'Security check'
- script: |
pytest --cov=. --cov-report=xml
displayName: 'Run tests'
- task: PublishCodeCoverageResults@1
inputs:
codeCoverageTool: 'Cobertura'
summaryFileLocation: '$(System.DefaultWorkingDirectory)/coverage.xml'
- stage: Deploy
dependsOn: Test
condition: and(succeeded(), eq(variables['Build.SourceBranch'], 'refs/heads/main'))
jobs:
- deployment: DeployScripts
environment: 'Production'
strategy:
runOnce:
deploy:
steps:
- script: |
rsync -avz --delete *.sh user@server:/opt/automation/
displayName: 'Deploy scripts'5. Docker-based CI/CD
Dockerfile для тестов:
# Dockerfile.test
FROM python:3.11-slim
# Системные зависимости
RUN apt-get update && apt-get install -y \
shellcheck \
bash \
&& rm -rf /var/lib/apt/lists/*
# Python-зависимости
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
RUN pip install --no-cache-dir pytest pytest-cov pylint safety
# Копировать скрипты
COPY *.py *.sh /app/
WORKDIR /app
# Команда для тестов
CMD ["pytest", "--cov=.", "--cov-report=html"]Docker Compose для CI:
# docker-compose.test.yml
version: '3.8'
services:
test:
build:
context: .
dockerfile: Dockerfile.test
volumes:
- .:/app
- ./htmlcov:/app/htmlcov
command: pytest --cov=. --cov-report=html6. Управление окружениями
Развёртывание в несколько окружений:
# .github/workflows/deploy.yml
name: Multi-Environment Deploy
on:
push:
branches: [ main ]
jobs:
deploy_staging:
runs-on: ubuntu-latest
environment: staging
steps:
- uses: actions/checkout@v3
- name: Deploy to Staging
run: |
rsync -avz --delete *.sh $STAGING_SERVER:/opt/automation/
env:
STAGING_SERVER: ${{ secrets.STAGING_SERVER }}
deploy_production:
runs-on: ubuntu-latest
environment: production
needs: deploy_staging
steps:
- uses: actions/checkout@v3
- name: Deploy to Production
run: |
rsync -avz --delete *.sh $PROD_SERVER:/opt/automation/
env:
PROD_SERVER: ${{ secrets.PROD_SERVER }}7. Стратегии отката
Автоматический откат:
# .github/workflows/rollback.yml
name: Rollback Deployment
on:
workflow_dispatch:
inputs:
version:
description: 'Version to rollback to'
required: true
jobs:
rollback:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
with:
ref: ${{ github.event.inputs.version }}
- name: Deploy previous version
run: |
rsync -avz --delete *.sh $PROD_SERVER:/opt/automation/
env:
PROD_SERVER: ${{ secrets.PROD_SERVER }}
- name: Notify team
run: |
echo "Rollback to version ${{ github.event.inputs.version }} completed"8. Мониторинг в CI/CD
Проверки здоровья конвейера:
# .github/workflows/monitoring.yml
name: Pipeline Monitoring
on:
schedule:
- cron: '0 * * * *' # Каждый час
jobs:
health_check:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Check script health
run: |
# Проверяем, работают ли скрипты
ssh $PROD_SERVER "ps aux | grep backup.sh"
# Проверяем, пишутся ли логи
ssh $PROD_SERVER "tail -n 10 /var/log/automation.log"
env:
PROD_SERVER: ${{ secrets.PROD_SERVER }}
- name: Alert on failure
if: failure()
run: |
# Отправляем алерт (Slack, Email и т. д.)
curl -X POST $SLACK_WEBHOOK -d '{"text":"Automation health check failed"}'
env:
SLACK_WEBHOOK: ${{ secrets.SLACK_WEBHOOK }}9. Автоматизированное тестирование в конвейере
Матрица тестов для разных версий Python:
# .github/workflows/test-matrix.yml
name: Test Matrix
on: [push, pull_request]
jobs:
test:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: ['3.9', '3.10', '3.11']
steps:
- uses: actions/checkout@v3
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v4
with:
python-version: ${{ matrix.python-version }}
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements.txt
pip install pytest
- name: Run tests
run: pytest test_*.py10. Управление секретами в CI/CD
Использование GitHub Secrets:
# .github/workflows/secure-deploy.yml
name: Secure Deployment
on:
push:
branches: [ main ]
jobs:
deploy:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Setup SSH key
run: |
mkdir -p ~/.ssh
echo "${{ secrets.SSH_PRIVATE_KEY }}" > ~/.ssh/id_rsa
chmod 600 ~/.ssh/id_rsa
ssh-keyscan -H ${{ secrets.DEPLOY_SERVER }} >> ~/.ssh/known_hosts
- name: Deploy with secrets
run: |
# Загружаем секреты из переменных окружения
export DB_PASSWORD="${{ secrets.DB_PASSWORD }}"
export API_KEY="${{ secrets.API_KEY }}"
# Запускаем скрипт
rsync -avz --delete *.sh ${{ secrets.DEPLOY_SERVER }}:/opt/automation/
ssh ${{ secrets.DEPLOY_SERVER }} "cd /opt/automation && ./deploy.sh"
env:
DEPLOY_SERVER: ${{ secrets.DEPLOY_SERVER }}Контейнеризация скриптов автоматизации
1. Dockerfile для Python-скриптов
Простой Dockerfile:
# Dockerfile
FROM python:3.11-slim
# Рабочая директория
WORKDIR /app
# Копируем зависимости
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Копируем скрипты
COPY backup.py .
COPY config.yaml .
# Скрипт как точка входа
ENTRYPOINT ["python", "backup.py"]
CMD ["--config", "config.yaml"]2. Dockerfile для Bash-скриптов
Bash в контейнере:
# Dockerfile.bash
FROM alpine:latest
# Устанавливаем Bash и инструменты
RUN apk add --no-cache bash rsync coreutils
# Копируем скрипты
COPY backup.sh /usr/local/bin/
RUN chmod +x /usr/local/bin/backup.sh
# Копируем конфигурацию
COPY config.yaml /etc/backup/
# Рабочая директория
WORKDIR /data
ENTRYPOINT ["/usr/local/bin/backup.sh"]3. Multi-Stage Dockerfile
Оптимизированная сборка в несколько этапов:
# Dockerfile.multi-stage
# Этап сборки
FROM python:3.11-slim AS builder
WORKDIR /app
COPY requirements.txt .
RUN pip install --user --no-cache-dir -r requirements.txt
# Этап выполнения
FROM python:3.11-slim
WORKDIR /app
COPY --from=builder /root/.local /root/.local
COPY backup.py .
COPY config.yaml .
ENV PATH=/root/.local/bin:$PATH
ENTRYPOINT ["python", "backup.py"]4. Docker Compose для полного окружения
docker-compose.yml:
version: '3.8'
services:
backup:
build:
context: .
dockerfile: Dockerfile
volumes:
- ./data:/data
- ./config:/config
- ./logs:/logs
environment:
- BACKUP_DIR=/data/backups
- SOURCE_DIR=/data/source
- LOG_LEVEL=INFO
restart: unless-stopped
monitoring:
image: prom/prometheus
ports:
- "9090:9090"
volumes:
- ./prometheus.yml:/etc/prometheus/prometheus.yml
grafana:
image: grafana/grafana
ports:
- "3000:3000"
volumes:
- grafana-data:/var/lib/grafana
volumes:
grafana-data:5. Развертывание на Kubernetes
YAML файл развертывания:
# k8s/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: automation-scripts
spec:
replicas: 2
selector:
matchLabels:
app: automation
template:
metadata:
labels:
app: automation
spec:
containers:
- name: backup
image: your-registry/backup:latest
env:
- name: BACKUP_DIR
value: "/data/backups"
- name: SOURCE_DIR
valueFrom:
configMapKeyRef:
name: backup-config
key: source_dir
volumeMounts:
- name: data
mountPath: /data
- name: config
mountPath: /config
volumes:
- name: data
persistentVolumeClaim:
claimName: backup-pvc
- name: config
configMap:
name: backup-config6. Health Check для Docker
Health Check в Dockerfile:
# Dockerfile с Health Check
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY backup.py health_check.py .
# Health Check скрипт
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
CMD python health_check.py || exit 1
ENTRYPOINT ["python", "backup.py"]Скрипт Health Check:
# health_check.py
import os
import sys
def check_health():
"""Проверяет здоровье backup-скрипта"""
# Проверяем наличие конфигурации
if not os.path.exists('config.yaml'):
print("Config file missing")
return False
# Проверяем доступность директории для записи
backup_dir = os.getenv('BACKUP_DIR', '/data/backups')
if not os.access(backup_dir, os.W_OK):
print("Backup directory not writable")
return False
print("Health check passed")
return True
if __name__ == '__main__':
sys.exit(0 if check_health() else 1)7. Оптимизация контейнеров
Лучшие практики для минимизации образов:
# Оптимизированный Dockerfile
FROM python:3.11-alpine AS builder
# Только необходимые зависимости
COPY requirements.txt .
RUN pip install --no-cache-dir --user --no-deps -r requirements.txt
FROM python:3.11-alpine
# Пользователь не от root
RUN addgroup -g 1000 backup && \
adduser -D -u 1000 -G backup backup
WORKDIR /app
COPY --from=builder /root/.local /root/.local
COPY --chown=backup:backup backup.py .
USER backup
ENV PATH=/root/.local/bin:$PATH
ENTRYPOINT ["python", "backup.py"]8. Безопасность контейнеров
Сканирование безопасности и укрепление:
# Dockerfile с усиленной безопасностью
FROM python:3.11-slim
# Обновления безопасности
RUN apt-get update && \
apt-get upgrade -y && \
apt-get install -y --no-install-recommends \
ca-certificates && \
rm -rf /var/lib/apt/lists/*
# Пользователь не от root
RUN groupadd -r backup && useradd -r -g backup backup
WORKDIR /app
COPY --chown=backup:backup requirements.txt .
RUN pip install --no-cache-dir --user -r requirements.txt
COPY --chown=backup:backup backup.py .
USER backup
ENTRYPOINT ["python", "backup.py"]Сканирование образа с помощью Trivy:
# Сканируем образ
trivy image your-registry/backup:latest
# Интегрируем в CI/CD
trivy image --exit-code 1 --severity HIGH,CRITICAL your-registry/backup:latest9. Container-Logging
Структурированное логирование:
# backup.py с Container-логированием
import logging
import json
import sys
class JSONFormatter(logging.Formatter):
"""JSON formatter для контейнерных логов"""
def format(self, record):
log_data = {
'timestamp': self.formatTime(record),
'level': record.levelname,
'message': record.getMessage(),
'logger': record.name
}
if hasattr(record, 'extra_data'):
log_data.update(record.extra_data)
return json.dumps(log_data)
# Настройка логирования
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JSONFormatter())
logger = logging.getLogger(__name__)
logger.addHandler(handler)
logger.setLevel(logging.INFO)10. Container-Orchestrierung
Docker Swarm сервис:
# docker-stack.yml
version: '3.8'
services:
backup:
image: your-registry/backup:latest
deploy:
replicas: 3
update_config:
parallelism: 1
delay: 10s
restart_policy:
condition: on-failure
delay: 5s
max_attempts: 3
environment:
- BACKUP_DIR=/data/backups
volumes:
- backup-data:/data
networks:
- automation-network
networks:
automation-network:
driver: overlay
volumes:
backup-data:Мониторинг и Alerting для скриптов автоматизации
1. Prometheus экспорт метрик
Python-метрики с Prometheus:
# metrics_exporter.py
from prometheus_client import start_http_server, Counter, Histogram, Gauge
import time
# Определение метрик
backup_counter = Counter('backup_total', 'Total number of backups', ['status'])
backup_duration = Histogram('backup_duration_seconds', 'Backup duration')
backup_size = Gauge('backup_size_bytes', 'Backup size in bytes')
def backup_with_metrics(source: str, target: str):
"""Backup с Prometheus-метриками"""
start_time = time.time()
try:
# Выполняем backup
size = perform_backup(source, target)
# Обновляем метрики
backup_counter.labels(status='success').inc()
backup_duration.observe(time.time() - start_time)
backup_size.set(size)
except Exception as e:
backup_counter.labels(status='failed').inc()
raise
# Запуск сервера метрик
if __name__ == '__main__':
start_http_server(8000)
print("Metrics server started on port 8000")2. Grafana-Dashboard
Конфигурация Grafana Dashboard:
{
"dashboard": {
"title": "Automation Scripts Monitoring",
"panels": [
{
"title": "Backup Success Rate",
"targets": [
{
"expr": "rate(backup_total{status='success'}[5m])"
}
]
},
{
"title": "Backup Duration",
"targets": [
{
"expr": "histogram_quantile(0.95, rate(backup_duration_seconds_bucket[5m]))"
}
]
},
{
"title": "Backup Size",
"targets": [
{
"expr": "backup_size_bytes"
}
]
}
]
}
}3. Анализ логов с ELK Stack
Конфигурация Filebeat:
# filebeat.yml
filebeat.inputs:
- type: log
enabled: true
paths:
- /var/log/automation/*.log
fields:
app: automation
fields_under_root: true
output.elasticsearch:
hosts: ["localhost:9200"]
index: "automation-logs-%{+yyyy.MM.dd}"
setup.kibana:
host: "localhost:5601"4. Alerting с Alertmanager
Конфигурация Alertmanager:
# alertmanager.yml
global:
resolve_timeout: 5m
route:
group_by: ['alertname']
group_wait: 10s
group_interval: 10s
repeat_interval: 1h
receiver: 'web.hook'
receivers:
- name: 'web.hook'
webhook_configs:
- url: 'http://localhost:5001/alerts'
- name: 'slack'
slack_configs:
- api_url: 'YOUR_SLACK_WEBHOOK_URL'
channel: '#automation-alerts'Правила Prometheus алертов:
# alerts.yml
groups:
- name: automation_alerts
rules:
- alert: BackupFailed
expr: rate(backup_total{status='failed'}[5m]) > 0
for: 1m
labels:
severity: critical
annotations:
summary: "Backup failed"
description: "Backup has been failing for the last 5 minutes"
- alert: BackupTooSlow
expr: histogram_quantile(0.95, rate(backup_duration_seconds_bucket[5m])) > 300
for: 5m
labels:
severity: warning
annotations:
summary: "Backup too slow"
description: "Backup duration exceeds 5 minutes"5. Health-Check Endpoints
HTTP Health-Check:
# health_check.py
from flask import Flask, jsonify
import os
import psutil
app = Flask(__name__)
@app.route('/health')
def health_check():
"""Health-Check endpoint"""
health_status = {
'status': 'healthy',
'checks': {}
}
# Проверяем свободное место на диске
disk_usage = psutil.disk_usage('/')
health_status['checks']['disk'] = {
'status': 'ok' if disk_usage.percent < 90 else 'warning',
'usage_percent': disk_usage.percent
}
# Проверяем память
memory = psutil.virtual_memory()
health_status['checks']['memory'] = {
'status': 'ok' if memory.percent < 90 else 'warning',
'usage_percent': memory.percent
}
# Проверяем файл конфигурации
config_exists = os.path.exists('config.yaml')
health_status['checks']['config'] = {
'status': 'ok' if config_exists else 'error',
'exists': config_exists
}
# Общий статус
if any(check['status'] == 'error' for check in health_status['checks'].values()):
health_status['status'] = 'error'
elif any(check['status'] == 'warning' for check in health_status['checks'].values()):
health_status['status'] = 'warning'
return jsonify(health_status)
if __name__ == '__main__':
app.run(host='0.0.0.0', port=8080)6. Slack интеграция
Отправка Slack алертов:
# slack_alerts.py
import requests
import json
def send_slack_alert(webhook_url: str, message: str, level: str = 'info'):
"""Отправляет алерт в Slack"""
colors = {
'info': '#36a64f',
'warning': '#ff9900',
'error': '#ff0000'
}
payload = {
'attachments': [{
'color': colors.get(level, '#36a64f'),
'title': f'Automation Alert: {level.upper()}',
'text': message,
'footer': 'Automation Scripts',
'ts': int(time.time())
}]
}
response = requests.post(webhook_url, json=payload)
return response.status_code == 200
# Использование
send_slack_alert(
webhook_url='YOUR_SLACK_WEBHOOK',
message='Backup failed for database',
level='error'
)7. Email алерты
Email алерты с Python:
# email_alerts.py
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
def send_email_alert(
subject: str,
message: str,
to_email: str,
from_email: str,
smtp_server: str,
smtp_port: int = 587
):
"""Отправляет email алерт"""
msg = MIMEMultipart()
msg['From'] = from_email
msg['To'] = to_email
msg['Subject'] = subject
msg.attach(MIMEText(message, 'plain'))
with smtplib.SMTP(smtp_server, smtp_port) as server:
server.starttls()
server.login(from_email, 'YOUR_PASSWORD')
server.send_message(msg)
# Использование
send_email_alert(
subject='Automation Alert: Backup Failed',
message='The backup script failed with error: ...',
to_email='admin@example.com',
from_email='automation@example.com',
smtp_server='smtp.example.com'
)8. Мониторинг системы с psutil
Отслеживание ресурсов системы:
# system_monitor.py
import psutil
import time
def monitor_system(interval: int = 60):
"""Отслеживает ресурсы системы"""
while True:
# CPU
cpu_percent = psutil.cpu_percent(interval=1)
# Memory
memory = psutil.virtual_memory()
# Disk
disk = psutil.disk_usage('/')
# Network
network = psutil.net_io_counters()
# Вывод
print(f"CPU: {cpu_percent}%")
print(f"Memory: {memory.percent}%")
print(f"Disk: {disk.percent}%")
print(f"Network Sent: {network.bytes_sent} bytes")
print(f"Network Recv: {network.bytes_recv} bytes")
print("-" * 40)
# Оповещение при высоких значениях
if cpu_percent > 90:
send_alert(f"High CPU usage: {cpu_percent}%")
if memory.percent > 90:
send_alert(f"High memory usage: {memory.percent}%")
if disk.percent > 90:
send_alert(f"High disk usage: {disk.percent}%")
time.sleep(interval)9. Ротация и управление логами
Конфигурация logrotate:
# /etc/logrotate.d/automation
/var/log/automation/*.log {
daily
rotate 14
compress
delaycompress
missingok
notifempty
create 0640 automation automation
sharedscripts
postrotate
systemctl reload automation-service > /dev/null 2>&1 || true
endscript
}10. Мониторинг доступности
Проверка доступности сервиса на Python:
# uptime_monitor.py
import requests
import time
from datetime import datetime
def check_uptime(url: str, interval: int = 60):
"""Отслеживает доступность сервиса"""
downtime_start = None
total_downtime = 0
while True:
try:
response = requests.get(url, timeout=10)
if response.status_code == 200:
if downtime_start:
downtime_duration = time.time() - downtime_start
total_downtime += downtime_duration
send_alert(f"Service is back up after {downtime_duration:.2f}s downtime")
downtime_start = None
print(f"{datetime.now()}: Service is UP")
else:
if not downtime_start:
downtime_start = time.time()
send_alert(f"Service returned status {response.status_code}")
print(f"{datetime.now()}: Service returned {response.status_code}")
except requests.RequestException as e:
if not downtime_start:
downtime_start = time.time()
send_alert(f"Service is down: {str(e)}")
print(f"{datetime.now()}: Service is DOWN - {str(e)}")
time.sleep(interval)Продвинутая обработка ошибок в скриптах автоматизации
1. Структурированная обработка исключений
Пользовательские классы исключений:
# exceptions.py
class AutomationError(Exception):
"""Базовое исключение для скриптов автоматизации"""
pass
class BackupError(AutomationError):
"""Ошибки, специфичные для резервного копирования"""
pass
class ConfigError(AutomationError):
"""Ошибки конфигурации"""
pass
class NetworkError(AutomationError):
"""Сетевые ошибки"""
pass
class ValidationError(AutomationError):
"""Ошибки валидации"""
pass2. Механизмы повтора с экспоненциальной задержкой
Умный повтор с backoff:
# retry_handler.py
import time
import random
from functools import wraps
from typing import Callable, Type
def retry_with_backoff(
max_retries: int = 3,
initial_delay: float = 1.0,
max_delay: float = 60.0,
exponential_base: float = 2,
jitter: bool = True,
exceptions: tuple = (Exception,)
):
"""Декоратор с экспоненциальной задержкой и jitter"""
def decorator(func: Callable) -> Callable:
@wraps(func)
def wrapper(*args, **kwargs):
delay = initial_delay
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except exceptions as e:
if attempt == max_retries - 1:
raise
# Добавляем jitter, чтобы избежать Thundering Herd
if jitter:
delay = delay * (1 + random.uniform(-0.1, 0.1))
print(f"Attempt {attempt + 1} failed: {e}. Retrying in {delay:.2f}s...")
time.sleep(min(delay, max_delay))
delay *= exponential_base
return wrapper
return decorator
# Использование
@retry_with_backoff(max_retries=5, initial_delay=2.0, exceptions=(ConnectionError, TimeoutError))
def download_file(url: str, target: str):
"""Загружает файл с повторными попытками"""
# Логика загрузки
pass3. Паттерн Circuit Breaker
Circuit Breaker для внешних сервисов:
# circuit_breaker.py
from enum import Enum
import time
class CircuitState(Enum):
CLOSED = "closed" # Нормальная работа
OPEN = "open" # Цепь разомкнута, запросы отклоняются немедленно
HALF_OPEN = "half_open" # Проверка восстановления сервиса
class CircuitBreaker:
def __init__(
self,
failure_threshold: int = 5,
recovery_timeout: float = 60.0,
expected_exception: Exception = Exception
):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.expected_exception = expected_exception
self.failure_count = 0
self.state = CircuitState.CLOSED
self.last_failure_time = None
def call(self, func: Callable, *args, **kwargs):
"""Выполняет функцию с защитой Circuit Breaker"""
if self.state == CircuitState.OPEN:
if self._should_attempt_reset():
self.state = CircuitState.HALF_OPEN
else:
raise Exception("Circuit breaker is OPEN")
try:
result = func(*args, **kwargs)
self._on_success()
return result
except self.expected_exception as e:
self._on_failure()
raise
def _should_attempt_reset(self) -> bool:
"""Проверяет, следует ли попытаться восстановить цепь"""
return (
self.last_failure_time and
time.time() - self.last_failure_time >= self.recovery_timeout
)
def _on_success(self):
"""При успешном выполнении"""
self.failure_count = 0
if self.state == CircuitState.HALF_OPEN:
self.state = CircuitState.CLOSED
def _on_failure(self):
"""При ошибке выполнения"""
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self.state = CircuitState.OPEN
# Использование
circuit_breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=30)
def call_external_api():
"""Вызов внешнего API с защитой Circuit Breaker"""
return circuit_breaker.call(requests.get, "https://api.example.com")4. Dead Letter Queue для неудачных задач
Реализация Dead Letter Queue:
# dead_letter_queue.py
import json
import os
from datetime import datetime
from typing import Any, Dict
class DeadLetterQueue:
def __init__(self, queue_file: str = "dead_letter_queue.json"):
self.queue_file = queue_file
self._load_queue()
def _load_queue(self):
"""Загружает очередь из файла"""
if os.path.exists(self.queue_file):
with open(self.queue_file, 'r') as f:
self.queue = json.load(f)
else:
self.queue = []
def _save_queue(self):
"""Сохраняет очередь в файл"""
with open(self.queue_file, 'w') as f:
json.dump(self.queue, f, indent=2)
def add(self, task: Dict[str, Any], error: Exception):
"""Добавляет неудачную задачу в очередь"""
dead_letter = {
'task': task,
'error': str(error),
'error_type': type(error).__name__,
'timestamp': datetime.now().isoformat(),
'retry_count': 0
}
self.queue.append(dead_letter)
self._save_queue()
def get_next(self) -> Dict[str, Any]:
"""Получает следующую задачу из очереди"""
if not self.queue:
return None
return self.queue.pop(0)
def retry(self, max_retries: int = 3):
"""Повторно выполняет задачи"""
remaining_tasks = []
for dead_letter in self.queue:
if dead_letter['retry_count'] < max_retries:
try:
# Повторное выполнение задачи
execute_task(dead_letter['task'])
print(f"Task {dead_letter['task']['id']} успешно переделана")
except Exception as e:
dead_letter['retry_count'] += 1
dead_letter['last_error'] = str(e)
remaining_tasks.append(dead_letter)
else:
print(f"Task {dead_letter['task']['id']} достигла максимума повторов")
self.queue = remaining_tasks
self._save_queue()
# Использование
dlq = DeadLetterQueue()
try:
execute_task(task)
except Exception as e:
dlq.add(task, e)5. Graceful Degradation
Стратегии fallback:
# graceful_degradation.py
from typing import Optional, Callable
class FallbackChain:
def __init__(self):
self.fallbacks = []
def add_fallback(self, func: Callable, priority: int = 0):
"""Добавляет функцию fallback"""
self.fallbacks.append((priority, func))
self.fallbacks.sort(key=lambda x: x[0], reverse=True)
def execute(self, *args, **kwargs) -> Optional[Any]:
"""Выполняет функции с fallback"""
for priority, func in self.fallbacks:
try:
result = func(*args, **kwargs)
if result is not None:
return result
except Exception as e:
print(f"Fallback {func.__name__} failed: {e}")
continue
return None
# Использование
fallback_chain = FallbackChain()
# Основной метод
def primary_backup(source: str, target: str):
"""Основной метод резервной копии (облако)"""
return upload_to_cloud(source, target)
# Fallback 1
def secondary_backup(source: str, target: str):
"""Вторичный метод резервной копии (FTP)"""
return upload_to_ftp(source, target)
# Fallback 2
def tertiary_backup(source: str, target: str):
"""Третичный метод резервной копии (локальный)"""
return copy_locally(source, target)
fallback_chain.add_fallback(primary_backup, priority=3)
fallback_chain.add_fallback(secondary_backup, priority=2)
fallback_chain.add_fallback(tertiary_backup, priority=1)
result = fallback_chain.execute(source, target)6. Comprehensive Logging для ошибок
Детальное логирование ошибок:
# error_logger.py
import logging
import traceback
import sys
from datetime import datetime
from typing import Optional
class ErrorLogger:
def __init__(self, log_file: str = "automation_errors.log"):
self.logger = logging.getLogger('error_logger')
self.logger.setLevel(logging.ERROR)
# File Handler
file_handler = logging.FileHandler(log_file)
file_handler.setLevel(logging.ERROR)
# Formatter
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
file_handler.setFormatter(formatter)
self.logger.addHandler(file_handler)
def log_error(
self,
error: Exception,
context: Optional[dict] = None,
critical: bool = False
):
"""Логирует ошибку с контекстом"""
error_info = {
'error_type': type(error).__name__,
'error_message': str(error),
'traceback': traceback.format_exc(),
'context': context or {},
'timestamp': datetime.now().isoformat()
}
log_message = f"Error: {error_info}"
if critical:
self.logger.critical(log_message)
else:
self.logger.error(log_message)
return error_info
def log_exception(self, func: Callable):
"""Декоратор для логирования исключений"""
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception as e:
self.log_error(e, context={'function': func.__name__})
raise
return wrapper
# Использование
error_logger = ErrorLogger()
@error_logger.log_exception
def risky_operation():
"""Рискованная операция с автоматическим логированием"""
pass7. Стратегии восстановления после ошибок
Автоматическое восстановление:
# recovery.py
from typing import Callable, Any
class RecoveryManager:
def __init__(self):
self.recovery_strategies = {}
def register_recovery(self, error_type: type, recovery_func: Callable):
"""Регистрирует стратегию восстановления для типа ошибки"""
self.recovery_strategies[error_type] = recovery_func
def execute_with_recovery(self, func: Callable, *args, **kwargs) -> Any:
"""Выполняет функцию с восстановлением"""
try:
return func(*args, **kwargs)
except Exception as e:
error_type = type(e)
if error_type in self.recovery_strategies:
print(f"Attempting recovery for {error_type.__name__}")
recovery_func = self.recovery_strategies[error_type]
return recovery_func(e, *args, **kwargs)
else:
raise
# Использование
recovery_manager = RecoveryManager()
def recover_from_connection_error(error: Exception, *args, **kwargs):
"""Восстановление после ошибки соединения"""
print("Reconnecting...")
time.sleep(5)
return func(*args, **kwargs)
def recover_from_disk_full(error: Exception, *args, **kwargs):
"""Восстановление после переполнения диска"""
print("Cleaning up old files...")
cleanup_old_files()
return func(*args, **kwargs)
recovery_manager.register_recovery(ConnectionError, recover_from_connection_error)
recovery_manager.register_recovery(OSError, recover_from_disk_full)8. Timeout-Handling
Timeout для долгоживущих операций:
# timeout_handler.py
import signal
from contextlib import contextmanager
import time
class TimeoutError(Exception):
"""Custom Timeout Exception"""
pass
@contextmanager
def timeout_handler(seconds: int):
"""Context Manager для timeout"""
def signal_handler(signum, frame):
raise TimeoutError(f"Operation timed out after {seconds} seconds")
# Регистрируем обработчик сигнала
old_handler = signal.signal(signal.SIGALRM, signal_handler)
signal.alarm(seconds)
try:
yield
finally:
# Восстанавливаем старый обработчик
signal.alarm(0)
signal.signal(signal.SIGALRM, old_handler)
# Использование
try:
with timeout_handler(30):
long_running_operation()
except TimeoutError:
print("Operation timed out, using fallback")
fallback_operation()9. Error Aggregation
Сбор нескольких ошибок:
# error_aggregator.py
from typing import List, Dict, Any
class ErrorAggregator:
def __init__(self):
self.errors: List[Dict[str, Any]] = []
def add_error(self, error: Exception, context: Dict[str, Any] = None):
"""Добавляет ошибку в коллекцию"""
self.errors.append({
'error': str(error),
'type': type(error).__name__,
'context': context or {},
'timestamp': time.time()
})
def has_errors(self) -> bool:
"""Проверяет наличие ошибок"""
return len(self.errors) > 0
def get_summary(self) -> Dict[str, Any]:
"""Возвращает сводку по ошибкам"""
error_types = {}
for error in self.errors:
error_type = error['type']
error_types[error_type] = error_types.get(error_type, 0) + 1
return {
'total_errors': len(self.errors),
'error_types': error_types,
'errors': self.errors
}
def clear(self):
"""Очищает все ошибки"""
self.errors = []
# Использование
aggregator = ErrorAggregator()
for item in items:
try:
process_item(item)
except Exception as e:
aggregator.add_error(e, context={'item': item})
if aggregator.has_errors():
print(f"Processing completed with {len(aggregator.errors)} errors")
print(aggregator.get_summary())10. Error Notification System
Уведомление о критических ошибках:
# error_notifier.py
from typing import List, Callable
class ErrorNotifier:
def __init__(self):
self.notifiers: List[Callable] = []
def add_notifier(self, notifier: Callable):
"""Добавляет функцию уведомления"""
self.notifiers.append(notifier)
def notify(self, error: Exception, context: dict = None):
"""Отправляет уведомление всем регистраторам"""
error_info = {
'error': str(error),
'type': type(error).__name__,
'context': context or {},
'timestamp': time.time()
}
for notifier in self.notifiers:
try:
notifier(error_info)
except Exception as e:
print(f"Notifier failed: {e}")
# Примеры уведомителей
def slack_notifier(error_info: dict):
"""Уведомление в Slack"""
send_slack_message(f"Error: {error_info['error']}")
def email_notifier(error_info: dict):
"""Уведомление по email"""
send_email(f"Error: {error_info['error']}", error_info)
# Использование
notifier = ErrorNotifier()
notifier.add_notifier(slack_notifier)
notifier.add_notifier(email_notifier)
try:
critical_operation()
except Exception as e:
notifier.notify(e, context={'operation': 'critical_operation'})