Кирилл Феткулин
Автоматизируй рутину в
разработке и
4	"	*
администрировании
Python-
инструментарий:
Автоматизируй
рутину в разработке и
администрировании
Автор: Феткулин Кирилл Ренатович
Оглавление
Введение
Часть I: Фундамент автоматизации
Глава 1: Настройка окружения и первый автоматический скрипт
Глава 2: Мастер работы с файлами и директориями
Глава 3: Общение с системой: процессы и командная строка
Часть II: Автоматизация в разработке (Dev)
Глава 4: Магия регулярных выражений (regex)
Глава 5: Парсим данные: от CSV до JSON и Excel
Глава 6: Веб-скрейпиш: добыча данных из интернета
Часть III: Автоматизация в администрировании (Ops)
Глава 7: Мониторинг системы и отправка уведомлений
Глава 8: Автоматизация по сети: SSH и управление серверами
Глава 9: Взаимодействие с облачными провайдерами (на примере AWS)
Глава 10: Работа с Docker через Python
Часть IV: Профессиональные техники
Глава И: Создание профессиональных CLI-утилит
Глава 12: Дотирование и обработка ошибок
Глава 13: Параллельное и асинхронное выполнение
Глава 14: Упаковка скрипта в Docker-контейнер
Заключение
Приложения
Приложение А: Шпаргалка по основным модулям Python
Приложение Б: Best Practices
Python-инструментарий: Автоматизируй
рутину в разработке и администрировании
Введение
Python прочно вошел в число самых популярных языков
программирования благодаря своей простоте, универсальности и богатой
экосистеме библиотек. Его возможности выходят далеко за рамки веб-
разработки и data science — Python является идеальным инструментом
для автоматизации рутинных задач в разработке и системном
админис!рировании.
Для кого эта книга? Эта книга предназначена для разработчиков,
системных администраторов, DevOps-инженеров и всех специалистов,
которые хотят повысить свою эффективность за счет автоматизации
повторяющихся задач. Мы будем писать скрипты для мониторинга,
деплоя, пестирования и работы с данными.
Что внутри? Книга разделена на четыре логические части:
1	Фундамент автоматизации — настройка окружения, работа с файлами
и командной строкой
2.	Автоматизация в разработке — регулярные выражения, парсинг
данных, веб-скрейпинг
3.	Автоматизация в администрировании — мониторинг, управление
серверами,облачные провайдеры
4.	Профессиональные техники — создание CLI-утилит, легирование,
параллельное выполнение
Как работать с книгой? Для максимальной пользы рекомендуем:
1.	Установить Python 3.3 или новее
2.	Создать виртуальное окружение для экспериментов
3.	Запускать все примеры кода на своей системе
4.	Модифицировать примеры под свои задачи
5.	Изучать документацию к библиотекам, упомянутым в книге
Часть I: Фундамент автоматизации
Глава 1: Настройка окружения и первый автоматический
скрипт
Правильная настройка окружения — критически важный первый шаг для
эффективной работы с Python. Начнем с установки интерпретатора и
создания виртуального окружения, которое изолирует зависимости
нашего проекта.
Установка Python и создание виртуального окружения
python
#	check_python_version.py
import sys
def chcck_python_version()'
..Проверяет версию Python....
required_version = (3, 8)
current_version = sys versionjnfo
if current_version < requircd_version
print(f"Tpe6ye гея Python {required..veision[P]} {required. version[l]} или выше")
рпп^Г'Текущая версия {current version.major} {current_version. minor)
{current version micro} )
sys exit(l)
else:
print(f"BepcHfl Python подходит {current_version major} {current_version minor}
{current version micro}")
if _ name__ == "_main__
check_python_version()
После проверки версии Python создаем виртуальное окружение:
python
#	creaLe_venv.py
import os
import sys
import subprocess
import venv
def create_virtual_environment(venv_name=".venv"):
..Создает виртуальное окружение
try:
#	Создаем виртуальное окружение
builder - venv Env6uilder(with_pip=True)
builder create(venv_name^
рпп1(ГВиртуальное окружение создано: {venvname}")
#	Определяем путь к pip в зависимости от ОС
if os.name == 'nt': # Windows
pip path = os path join(venv_name, "Scripts , "pip.exe')
else: # Linux/Mac
pip path = os path join(venv_name "bin", pip' )
#	Устанавливаем базовые пакеты
packages = ["requests", "psutil", "beautifulsoup4", "pandas"]
subprocess run(Fpip_pathJ "install ] + packages. check=True)
print("Sa3OBbie пакеты установлены')
except Exception as c
рпп1(ГОшибка при создании виртуального окружения: {е}")
sys exit(l)
if_name___== "___mam___":
crcate_virtual_environment()
Первый скрипт автоматизации - резервное копирование
Теперь напишем наш первый полезный скрипт — систему резервного
копирования:
python
#	backup script.py
import os
import shutil
import datetime
import argparse
import zipfile
import logging
#	Настройка вотирования
logging basicConfig(
level = logging INFO,
format='%(asctime)s - %(levelname)s - %(message)s'.
handlers-[
logging FileHandler('backu]j.log ),
logging Stream Handlers)
]
)
def create_backup(source_dir, oackup_dir, compress=True):
Создает резервную копию директории
Args:
source dir (str): Директория для резервного копирования
backup_dir (str): Директория для хранения бэкапов
compress (bool): Сжимать ли бэкап в zip
try:
#	Проверяем существование исходной директории
if not os.path.exists(source_dir).
raise ЕПеМо1ЕоипбЕггог(Г'Директория {source_dir} не существует")
#	Создаем директорию для бэкапов если её нет
os.makedirs(backup, dir, existok-True)
#	Создаем имя файла с timestamp
timestamp = datetime datetime now().strftimc( '%Y%m%d_%H%M%S")
backup name = f"backup_{os.path basename(source_dir)}_{timestamp}
backup_path = os path joim,backup_dir, backup_name)
if compress:
#	Создаем zip-архив
backup_path += ".zip"
with zipfile ZipFile(backup path, 'w', zipfile ZIP_DEFLA'rED) as zipf-
for root, dirs files in os.walk(scurce_dir):
for file in files:
file_path = os path join(root, file)
arcname = os path relpath(fiie_path start-source_dir)
zipf write(file_path, arcname)
logging info(f"Co3flaH zip-бэкап {backuppath} )
else:
#	Копируем директорию
shutil.copytree(source_dir. backup_path)
logging info(f"Co3flaHa копия директории: {backup_path} )
#	Удаляем старые бэкапы (больше 30 дней)
cleanup_old_backups(backup_dir, days=30)
return backup_path
except Exception as e:
logging еггог(Р'Ошибка при создании бэкапа: {е} )
raise
def cleanup old backups(backup_dir days=30):
..Удаляет старые бэкапы.....1
now = datetime datetime nowi)
for item in os listdir[backup_dir):
item_path = os path join(backup_dir item)
if os.path isfilefitem path):
#	Для файлов
modtime = datetime datetime fromtimestamplos path getmtime(item_path))
else:
#	Для директорий
mod_time = datetime.datetime fromtimestamplos.path getctimelitem path))
if (now mod_time).days > days:
try:
it os.path isfile(itcm_path)'
os remove(item_ path)
else:
shutil.rrntree(itempath)
logging тМГ'Удален старый бэкап: {item_path} )
except Exception as e
logging.error(f"Ouin6Ka при удалении {item_path} -fe} )
def main():
parser = argparse.ArgumentParser(descnption="CKpnnT для создания резервных копий")
parser.add_argument("source", Ье1р="Директория для резервного копирования")
parser add argument -d", destination". default="./backups",
Ье1р="Директория для хранения бэкагсв')
parser add_argument( -z", "-no-zip", action="store_false , dest="compress",
help="He сжимать бэкап в zip ')
args = parser parse_args()
try:
backup_path = create_backuplargs source args.destination, args.compress)
рпп1(Г'Резервная копия успешно создана {backup_path} )
except Exception as e
рппЦТОшибка: {е}")
sys exit(l)
if__name__== "__main__
main()
Этот скрипт демонстрирует несколько важных концепций:
Работу с аргументами командной строки
Обработку ошибок и легирование
Манипуляции с файловой системой
Создание zip-архивов
Организацию кода в функции
Глава 2: Мастер работы с файлами и директориями
В згой главе изучим модули os, shutil и pathlib для эффективной работы с
файловой системой,
Использование модуля os для базовых операций
python
# fHe_operatiorrs.py
import os
import shutil
def expkre_directory(path=' ):
..Исследует содержимое директории....
try:
items = os listdir(path)
рппЦТСодержимое директории {path} )
for item in items
item_path = os.path.join(path, item)
if os.path isfile(item_path):
size = os path getsize(iterr_path)
print(f" □ {item} ({size} байт)")
elif os.path isdir(item_path):
print(f" □ {item}/")
elif os path.islink(item_path):
print(f" □ {item} (ссылка)")
except FileNotFoundError:
рпп1(ГДиректория '{path}1 не существует")
except PermissionError
print(f"HeT доступа к директории {path} )
def organize_directory(path= ):
..Организует файлы в директории по типам.........
#	Создаем директории для разных типов файлов
categories = {
'images': ['.jpg , 'Jpeg', '.png , '.gif', '.bmp'],
'documents': [ .pdf', '.docx', '.txt', '.xlsx , .pptx'],
'archives': ['.zip', '.rar', .tar', '.gz'],
'code': [ .py , '.js , .html', '.css', ' java']
}
for category in categories
os rnakedirs^os path jointpath category) exist_ok=True)
#	Перемещаем файлы в coo гветствующие директории
for item in os listdir (path):
item_path = os.path jointpath, item)
it os path.isfiletitem path)'
ext = os.path.splitcxt(itcm)[l] lower()
for category, extensions in categories.itemst).
if ext in extensions.
dest_dir = os path.jointpath, category)
try:
shutil move(item_path os.path.join(dest_dir, item))
рппиР'Перемещен: {item} > {category}/")
except Exception as e
рпп1(Г0шибка при перемещении {item}: {e}")
break
it___name__== "____main__":
explore_directory()
organize_directory()
рпп1("Организация файлов завершена- )
explore_directory()
Современный подход с pathlib
Модуль pathlib предоставляет объектно-ориентированный интерфейс для
работы с путями:
python
# pathlib_examples.py
from pathlib import Path
import datetime
def demonstrate_pathlib():
..Демонстрирует возможности pathlib
#	Создание путей
current_dir = Path cwd()
home dir = Path homeO
config path = home_dir/ .config' I 'myapp' / settings ini'
print(f"TeKyinaq директория {current_dir} )
рпп1(ГДомашняя директория. {home_dir} )
print(f"nyTb к конфигу: {config_path} )
#	Создание директорий
config_path parent mkdinparents=True, exist_ol<=True)
#	Работа с файлами
hg_file = Path( logs') I 'app.log'
log_ file, parent, mkdir (exist. ok=True)
#	Запись в файл
timestamp = datetime datetime now() isoformatf)
lc>g_file write text(f"[{timestamp}] Application started\n'. encoding = 'utf-8')
#	Чтение из файла
if log file existsO:
content = log file read_textfencoding='utl-3')
рппДГ'Содержимое лога: {content} )
#	Поиск файлов
print("\nPython файлы в текущей директории )
for py_file in PathC ).globi'*.py ):
print(f" {py_file} ({py_file stat().st_size} байт)")
#	Рекурсивный поиск
print("\nBce .txt файлы (рекурсивно):")
for txt_file in Path( ).rglob('*.txt):
print(f" {txt_file} )
if__name__== "___main__
demonstratepathiibO
Глава 3: Общение с системой: процессы и командная
строка
Автоматизация часто involves взаимодействие с системными утилитами и
другими программами. Модуль subprocess поедоставляет мощные
инструменты для этого.
Базовое использование subprocess
python
# subprocess_examples.py
import subprocess
import sys
def run command(command, capturc_output=Falsc):
....Выполняет команду и возвращает результат.....
try:
if capture_output
result = subprocess run(
command
shell=True,
capture_output=True,
text-True,
timeout=30
)
return result returncode, result stdout, result.stderr
else:
result = subprocess run(command. shell=True, timeout=30)
return result returncode, None, None
except subprocess.TimeoutExpired
рппЦГ'Команда '{command}' выполняется слишком долго")
return -1, None, None
except Exception as e
рппЦР'Ошибка при выполнении команды {е}")
return -1, None, None
def system_info():
..Собирает информацию о системе......
commands = [
("Дата и время", "date"),
("Имя пользователя", "whoami"),
("Дисковое пространство", "df-h" if sys.platform != "Win32" else "wmic logicaldisk get
size,freespace,caption'),
("Свободная память", "free -h" if sys.platform != "Win32" else "wmic OS get
Free Physical Memory"),
]
for description, command in commands
print(f \n{'='*50}")
print(f (description} )
рппНР{'='*50} )
returncode, stdout stderr = run_command(command. capture_output=True)
if returncode == 0:
print(stdout)
else:
print(f"0uJH6Ka (stderr} )
def monitor_processes().
..Мониторит запущенные процессы......
if sys.platform == "Win32":
command = "tasklist"
else:
command = "ps aux"
rcturncode. stdout, stderr = run_command(command, capture output=True)
if returncode == 0:
рпЩС'Запущенные процессы: )
print(stdout)
# Простой анализ (для Linux/Мас)
if sys platform != "Win32":
lines = stdout split('\n')
user_processes = {}
for line in lines[l:]: # Пропускаем заголовок
if line strip():
parts = line split()
if len(parts) >= 11:
user = parts[O]
user_processes[user] = user_processes.get(user 0) + 1
print( ЛпПроцессы по пользователям:' )
for user, count in user_processes.items():
print(f" {user ] {count} процессе a'1)
else:
print(f"OujM6Ka: {stderr} )
if_name____== ' _main____":
print("C6op информации о системе.. )
system_info()
рпп!("\п\пМониторинг процессов... )
rnonitor_processes()
# Пример с обработкой вывода
рмп1("\п\пПровсрка сетевых соединений.. )
if sys.platform == "Win32":
command = "netstat -an"
else:
command = "netstat tuln'
returncode, stdout, stderr = run_command(command, capture_output=True)
if returncode == 0:
# Анализ сетевых соединений
lines = stdout split('\n )
listening_ports = []
for line in lines:
if "LISTEN" in line or "ПРОСЛУШИВАЕТ" in line
listening_ports append (line. stripO)
print(f"HanfleHO {len(listening_ports)} прослушивающих портов:")
for port in listening_pcrts[:10]:	- ем первые 10
print(f" {port} )
else:
print(f"OLLiH6Ka: {stderr} )
Автоматизация работы с Git
python
# gitautamatian.py
import subprocess
import os
from pathlib import Path
class GitAutomator:
def init (self, repopath):
self repo_path = Path(repo_path)
os.chdir(sclf repo_pathj
def iuri git_command(self. command)
..Выполняет git команду"""
full_command = f"git {command}
result - subprocess run(full_command, shell=True, capture_output=True, text=True)
return result.returncode, result.stdout, result stderr
def status(self):
..Показывает статус репозитория.....
return self.run git command! 'status")
def commit_ali(self. message)
..Добавляет все файлы и создает ксммит......
# Добавляем все файлы
returncode, stdout stderr = self run_grt_command( add . )
if returncode 1 = 0:
return returncode, stdout, stderr
# Создаем коммит
return self.run_git_command(f'commit -m {message) ’")
def push(self, remote="origin", branch= ’main"):
..Пушит изменения в удаленный репозиторий.......
return self.run_git_command(f"pjsh {remote} {branch} )
def auto_commit_push(self, message):
..Автоматически добавляет все файлы, коммитит и пушит.........
рппг("Проверка статуса репозитория...1)
returncode, stdout, stderr = self statusO
if returncode == 0:
if "nothing to commit" in stdout
print("HeT изменений для коммита')
return 0, stdout, stderr
else:
print("EcTb изменения, создаем коммит.. )
returncode, stdout, stderr = self.commit aii;message)
if returncode == 0:
print("KoMHT успешно создан, пушим.. )
return self push()
else:
return returncode, stdout, stderr
else:
return returncode, stdout, stderr
# Использование
if__name___== main_______":
#	Укажите путь к вашему репозиторию
automator = GitAutomator("/path/to/your/repo')
#	Автоматический коммит и пуш
returncode. stdout, stderr = automator auto_commit_push( Автоматический коммит из
Python скрипта1)
if returncode == 0:
рнпЦ"Успешно! )
print(stdout)
else:
рппг("Ошибка:")
print(stderr)
Часть II: Автоматизация в разработке (Dev)
Глава 4: Магия регулярных выражений (regex)
Регулярные выражения — это мощный инструмент для поиска и
обработки текста. В Python они реализованы в модуле ге.
Основы регулярных выражений
python
# regex_examples.py
import re
def demonstrate_regex():
..Демонстрирует различные возможности regex........
text = ..
Контакты:
Email: john.doc@cxample.com, janc_ smith! 23@tcst.co.uk
Телефоны: +7 (123) 456-78-90 8 900-123-45-67, 495-123-45-67
Даты: 2023-12 31, 31.12.2023, 01/01/2024
IP адреса: 192.168.! 1, 8 8.8 8, 2001:0db8 85аЗ:0000 0000 8a2c:0370 7334
#	Поиск email-адресов
email_pattern = rl\b[A-Za-zO-9._%+-]+@[A-Za-zO-9.-]+\.[A-Z|a-z]{2,}\bl
emails = re findall(email_pattern. text)
printC'Email адреса:")
for email in emails
print(f" {email}")
#	Поиск телефонных номеров (российский формат)
phone_pattern = r'(\+7|8)[\s\-]?\(?\d{3}\)?[\s\-]?\d{3}[\s\-]?\d{2}[\s\-]?\d{2}'
phones = re.findall(phone_pattern text)
рпп!("\пТелефонные номера: )
for phone in phones
print(f" [phone} )
#	Поиск дат
date_pattern = r'\b\d{4}[-./]\d{2}[-./]\d{2}\b|\b\d{2}[-./]\d{2}[-./]\d{4}\b'
dates = re findall(date_pattern, text)
print("\nflaTbi:")
for date in dates
printff" {date} )
#	Поиск IP-адресов
ip_pattern = r'\b(?:\d{ 1,3 }\. ){3 }\d{ l,3}\b|\b(?:[A-F0-9]{ 1,4}:){ 7} [A-F0-9]{ l,4}\b'
ips = re findall(ip_pattern, text, re IGNORECASE)
print("\nlP-aflpeca: )
for ip .n ips
print(f" {ip} )
#	Замена с использованием regex
anonymized_text = re.sub(r'\b\d{l,3}\.\d{l,3}\.\d{l,3}\.\d{l,3}\b', 1XXX. XXX. XXX. XXX',
text)
anonymized_text = re.sub(r'\b[A-Za-zO-9._%+-]+@[A-Za-zO-9.-]+\.[A-Z|a-z]{2,}\b',
'EMAILigJHIi^DEN1, anonymized text)
рг1п1("\пАнонимизировэнный текст: )
print(anonymized text)
def validate, input():
..Валидация пользоватсльско! о ввода с помощью regex...........
patterns = {
'email': rIZ4[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$',
'phone1: r'~(\+7|8)[\s\ ]?\(?\d{3}\)?[\s\-]?\d{3}[\s\-]?\d{2}[\s\ ]?\d{2}$',
'date': r'~\d{4} \d{2} \d{2}$',
'ip': r'~\d{l,3}\.\d{l,3}\.\d{l,3}\.\d{l,3}$'
}
test_cases = [
('email', 'test<g)example.com'),
('email', 'invalid.email),
('phone', '+7 (123) 456-78-90'),
('phone', '123-456-789'),
('date', '2023-12-31'),
('date', '31.12.2023'),
('ip', '192.168.1.1'),
('ip', '999.999.999.999')
]
for field, value in test_cases
pattern = patternsffield]
is_valid = re.match(pattern, value) is not None
print(f {field} {value} {'/'if is_valid else X'}")
if name
main
рпп1("Демонстрация возможностей регулярных выражений:")
demonstrate_regex()
рпп1("\п\пВалидация пользовательского ввода:")
validate_input()
Парсинг лог-файлов с помощью regex
python
# log_parser.py
import re
from pathlib import Path
from collections import defaultdict
class LogParser:
def iii!t_(self):
self patterns = {
'error': r'ERROR|Error|error',
'warning': r'WARNING|Warning|warning ,
'http': r'HTTP/\d\.\d"\s\d{3}',
'ip': r'\d{l,3}\.\d{l,3}\.\d{l,3}\.\d{l,3}',
'timestamp': r\d{4} \d{2}-\d{2} \d{2}:\d{2}:\d{2}'
}
def parse. Iog_file(self. file_path)'
...Анализирует rwi файл и извлекает информацию"""
log_file = Path(file_path)
if not log file.exists():
raise РПсМо1ЕиигФЕггог(ГФайл {fllepath} не найден')
results = {
'error_count': 0,
'warning_count': 0,
'http_statuses': defaultdict(int),
'uniquejps': set(),
'timestamps': []
}
with open(log_file, r' encoding = 'utf-8 ) as file:
for line in file:
# Поиск ошибок
if re searchfself patterns['error'], line):
results?error_count'] += 1
#	Поиск предупреждений
if re.search(self patternsf warning'], liner
results' warning count'] += 1
#	Поиск НГГР статусов
http_match = re search(self.patternsf http'], line)
if http_match
status = httpjnatch.group)).split) )[-l]
results) http_statuses'][status] += 1
#	Поиск IP-адресов
ip_match = re search(se!f patterns) .p ], line)
if ipmatch
results) uniquejps ] add(ip match.group)))
	# Поиск временных меток
tirne_match = re search(self.patterns[ timestamp'], line)
if time_match
resultsl timestamps'].append(time match.group)))
return results
def gcnerate_report(sclf. results)
..Генерирует отчет no анализу лог-файла........
report = []
report.append) '=" * 50)
report append) 'АНАЛИЗ ЛОГ-ФАЙЛА ')
report append) '=" * 50)
report.append(f" Количество ошибок: {results['error_count']}")
report.append(f" Количество предупреждений: {results['warning_count']}")
report.append("\nHTTP статусы:')
for status, count in resultsf'httpstatuses ].items)):
report append(f" {status) {count}")
report.append(f"\nУникaльныe IP-адреса: {len(results['unique_ips'])}")
if resultst'uniquejps']:
report append) 1 + jcin(list(results['unique_iDs'])[:10]))
if len(results['unique_ips']) > 10:
report append(f" ... и еще {len(results[:unique_ips']) - 10} )
report.append(f"\nBpeMeHHbie метки: {len(results['timestamps'])} записей")
if results['timestamps ]:
report append(f Первая {resultsf timestamps'][C]} )
report.append(f" Последняя: {results['timestamps'][-l]}")
return "\n‘ jom(report)
# Использование
if__name_ == " nain_____
parser = LogParsen)
#	Пример с искусственным лог-файло.м
samplejog = ....
2023-12-31 23:59:01 INFO: Application started
2023-12-31 23:59:02 WARNING: Conjuration file missing, using defaults
2023 12-31 23:59:03 ERROR Database connection failed
2023-12-31 23:59:04 INFO: Request from 192.168.1.1: GET/api/vl/users HTTP/1.1" 200
2023-12-31 23:59:05 INFO: Request from 192.168.1.2: POST/api/vl/login HTTP/1.1" 401
2023-12-31 23:59:06 WARNING: High memory usage detected
2023-12-31 23:59:07 INFO: Request from 192.168.1.3: GET/api/vl/products HTTP/1.1" 500
#	Создаем временный файл для демонстрации
with open('sample.log‘, 'w'. encoding='utf-8') as f:
f.write(sample_ log)
try:
results = parser parsejogJi'lef sample.log')
report = parser.generate_report(results)
print(report)
finally:
#	Удаляем временный файл
Path (sample, log1) unlinkfmissingok-True)
Глава 5: Парсим данные: от CSV до JSON и Excel
Работа с данными в различных форматах — важная часть автоматизации.
Python предоставляет отличные инструменты для работы с CSV, JSON,
Excel и другими форматами.
Работа с CSV и JSON
python
# data -processing.ру
import csv
import json
import pandas as pd
from pathlib import Path
def process csv_rile(input_file. output_file):
...Обрабатывает CSV файл и преобразует его в JSON"1"1
data = []
with open(input_tile, 'r', encoding='utf-8 ) as csvfile
reader = csv DictReader(csvfilc)
for r-'W in reader
# Пример обработки данных
row!'processed'] = True
row|'full_name'] = f {row get( firstjiame', ")} {row.get('last_name', )} stripO
data.append(row)
# Сохраняем в JSON
with open(output_file, 'w', encoding= utf-8') as jsonhle:
json dump'data, jsonfile, ensure_ascii=False, indent=2)
return data
def analyzejson_file(json_file):
..Анализирует JSON файл и возвоащает статистику........
with open(json_file, 'r'. enccding = 'utf-8') as file:
data = json.load(file)
if not data
return {}
# Собираем статистику
stats = {
'total_records': ien(data),
'fields': list(data[O].keys()) if data else [],
'field_stats': {}
}
# Для каждого поля собираем статистику
for field in stats['fields ]:
values = [item get(field) for item in data if field .n item]
if values
# Проверяем тип данных
firsi_value = values[0]
if isinstance(first_value, (int, float)):
stats['neld_stats'][field] = {
'type': 'numeric',
'min': n in(values),
'max': max(values),
'avg': sum(values) I len(values)
}
elif isinstance(first_value. str):
stats['field stab'][field] = {
'type': 'string',
'unique_values': len(set(values)),
'maxjength': max(lenlv) foi v in values if v)
}
else:
stats['field_stats'][fieldl = {
'type': 'other',
'count': len(values)
}
return stats
def excel operations(input_file. output_file):
'" Выполняет различные операции c Excel файлами с помощью pandas""
# Чтение Excel файла
df = pd read_excel(input_file)
print(f"l/lcxoflHbie данные: {df.shape[0]} строк, {df.shape[l]} колонок")
print(' \nnepBbie 5 строк )
printfdf. head ())
# Базовая очистка данных
# Удаляем полное гью пустые строки
df.dropna(how='all', inpiace=True)
#	Заполняем частично пустые значения
for col in df columns
if df[col] dtype == 'object':
df [col].fi II na('Unknown', inplace=True)
else:
dfrcol].fillna(0, inplace=True)
#	Добавляем вычисляемые поля
numeric_cols = df select_dtypes(include = [ number1]).columns
if len(numeric_cols) > 0:
dff'total] = df[numeric_cois].sum(axis=l)
df['average'l = df[numeric_cols].mean(axis=l)
#	Сохраняем обработанные данные
with pd ExcelWriter(output file) as writer
df to_excel(writer, sheet_name= Processed lata , index=False)
#	Создаем сводную таблицу
if len(numeric_cols) > 0:
pivot = df gruupbyidf.columns[0] if len(df.columns) > 0 else None)[numeric_cols].sum()
pivot.to_cxcel( writer. shcct_namc='Sjmmary')
рпп1(Г\пОбработанные данные сохранены в: {output_file}")
return df
# Создаем демонстрационные данные
def create_sample_data():
..Создает демонстрационные данные для примеров "1
#	CSV данные
csv_data = [
['id', 'first_name', 'last_name', 'age', 'city'],
[1, 'John', 'Doe1, 30, 'New York ],
[2, 'Jane', 'Smith', 25, London'],
[3, 'Bob', 'Johnson', 35, 'Paris'],
[4, 'Alice', 'Brown , 28, 'Berlin']
]
with open('sample_data.csv', 'w', newline=", encoding = 'utf-8') as csvfile:
writer = csv writer(csvfile)
writer writerows(csv_data)
#	Excel данные
df = pd DataFrame({
'Product1: ['Apple', 'Banana', 'Cherry', 'Cate'],
'Price': [1.20, 0.80, 2.50, 1.50],
'Quantity': [100, 150, 80, 120],
'Category': ['Fruit', 'Fruit', 'Fruit', 'Fruit']
})
df co excel('sample_data.xlsx', index=False)
print( Демонстрационные данные созданы")
if__name__== " nain_____":
#	Создаем демонстрационные данные
create_sample_data()
#	Обрабатываем CSV
рг1пг("\пОбработка CSV файла. )
csv_data = process_csv_file('sample_data.csv', 'processed_data.json')
print(f"O6pa6oTaHo {len(csv_data)} записей')
#	Анализируем JSON
рппЦ"\пАнализ JSON файла... )
stats = analyze json fileCprocessed data । son')
рг1п1(С'Статистика: {stats} )
#	Работаем c Excel
print("\nO6pa6oTKa Excel файла... )
df = excel_operations('sample_data.xlsx', 'processed_data.xlsx j
#	Очищаем демонстрационные данные
for file in ['sample_data.csv', 'sample_data.xlsx', 'processed_data.json',
'processed_data.xlsx'J:
Path (file).unlink(missing ok=True)
Продвинутая обработка данных c Pandas
python
# advanced_data-processing.py
import pandas as pd
import numpy as np
from datetime import datetime
def advanced_data_analysis():
..Продвинутый анализ данных с помощью pandas........
#	Создаем демонстрационный dataset
np random.seed(42)
dates = pd date_range( 20230101' periods=100)
data = {
'date': dates,
'sales': np.random randint(100, 1000, size=100),
'customers': np random randint(10, 100, size=100),
'temperature': np random normal(29, 5, stze=100),
'category': np random choice(['A , 'В', 'C'], size=100)
}
df = pd DataFrame(data)
dffrevenue'] = dff'sales'] * np.random.uniform(5, 20, size=100)
print( 'Исходные данные: )
pnnt(dt headO)
print(f"\nPa3Mep данны> {df shape} )
#	1. Агрегация данных
print( '\nl. Агрегация no категориям:")
catcyory_stats = df.groupby('category') agg({
'sales': ['sum1, 'mean', 'std J,
'customers': ['sum', 'mean ],
'revenue': 'sum'
}).round(2)
print(catcgory_stats)
#	2. Временной анализ
print( '\n2. Временной анализ )
df['month'] = dff'date'] dt month
monthly sales = df groupby('month').agg'{
'sales': 'sum',
'revenue': 'sum',
'customers': 'sum'
})
print(monthly_sales)
#	3. Корреляционный анализ
print("\n3. Корреляция между показателями:")
correlation = df[['sales', 'customers', 'temperature', 'revenue']].corr()
print(correlation)
#	4. Обработка пропущенных значений и выбоосов
print("\n4. Обработка пропущенных значений и выбросов:")
#	Добавляем некоторые пропущенные значения и выбросы для демонстрации
df_with_mis5ing = df соруО
df with_missing loc[::10, 'sales'] = пр пап
df_with_missing loc[::15, 'revenue J = пр.пап
#	Заполняем пропущенные значения
df cleaned = df_with_missing copyQ
df_cleaned['sales ].fillna(df_cleaned['sales].median]), inplace=True)
df_cleaned['revenue ].fillnatdf_cleanec[ revenue'].mean]), inplace=True)
#	Обрабатываем выбросы с помощью iQR метода
QI = dfcleanedl'sales'].quantile(0.25)
Q3 = df-Cleanedfsales'Lquantilefe 75)
IQR = Q3 QI
lower_bound = QI 1.5 * IQR
upper_bound = Q3 + 1 5 * IOR
outliers = df_cleanedHdf_cleaned[ sales'] < lower bound) | (df_cleanedl sales'] >
upper bound)]
рппЦГ'Найдено {len(outliers)} выбросов в данных о продажах")
#	5. Сохранение результатов
with pd.ExcelWriter('sales_analysis.xlsx') as writer:
df .to_excel(writer, sheet_name= Raw Data', irdex^ False)
category stats toexceKwriter sheet_name='Category Stats )
monthly sales to_excel(writer, sheet_name='Monthly Sales')
correlation.to_exceKwriter sheet_name= Correlat on )
рг1п1("\пАнализ завершен. Результаты сохранены в 'sales_analysis.xlsx'")
return df. category stats. monthly_sales, correlation
if__name___== "___main__":
advanced_data_analysis()
Глава 6: Веб-скрейпинг: добыча данных из интернета
Веб-скрейпинг позволяет автоматизировать сбор данных с веб-сайтов. В
этой главе мы изучим библиотеки requests и BeautifulSoup для извлечения
данных из веб-страниц.
Основы веб-скрейпинга с BeautifuiSoup
python
# webscrapingbasics.py
import requests
from bs4 import BeaucifulSoup
import pandas as pd
from urllib parse import urljoin urlparse
import time
class WebScraper:
def _init_(self, delay=l):
seif session = requests SessionO
self delay = delay
self headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML,
like Gecko) Chrome/91.0.4472 124 Safan/537.36'
}
def fetch page(self, url):
..Загружает веб страницу.......
try:
time.sleep(self.delay) # Уважаем robots.txt и избегаем блокировок
response = self session.get(url, headers=-self.hcadcrs, timeout-10)
response raise_for_status()
return response.text
except requests RequestException as e
рпп1(ГОшибка при загрузке страницы {url} {e} )
return None
def extract_links(self, html, base_url, frlter_pattern=None):
"" 'Извлекает все ссылки co страницы.....
soup = BeautifulSouprtitml, 'html.parser')
links = []
for link in soup find_all('a', href=True):
href = linkfhref ]
full_url = urljoin(base_url, href)
if filter_pattern
if filter_pattern in full_url
links.append ((full_url, link.get_text() stripf)))
else:
links.append((full_urL link get_text().strip()))
return links
def extract_tables(self, html):
..Извлекает таблицы co страницы.........
soup = BeautifulSoup(html, 'html oarser')
tables = []
for table in soup find_all('table ):
rows = []
for row in table find_all('tr):
cells = [cell.get_text() strip() for cell in row find_all([ th1, 'td'])]
it cells
rows.append(cells)
if rows:
tables. append( rows)
return tables
def scrape_ website data(self. url, selectors):
..Извлекает конкретные данные с веб-страницы с помощью CSS селекторов""
html = self fctchpage(url)
if not html
return None
soup = BeautifulSoup(html, 'htmlparser)
results = {}
for key, selector in selectors items():
elements = soup selccttselector)
if elements
results[key] = [elem gct_text() stripO fcrelem in elements]
else:
results[key] = []
return results
def demo_scraping():
...Демонстрация возможностей веб-скрейпинга...........
scraper = WebScraper(delay-2)
#	Пример: получение заголовков новостей с hypothetical news site
url = "https://httpbin.org/html" # Используем тестовый URL
#	Для реального примера можно использовать:
#	url = "https://news.ycombinator.com"
#	selectors = {
#	'titles': '.titleline > a',
#	'scores': '.score',
#	'links': '.titleline > a'
#}
#	Для тестового примера:
selectors = {
'heading1: 'hl',
'paragraphs': 'p'
}
data = scraper scrape_website_data(url, selectors)
if data-
рпгД("Извлеченные данные: )
for key. values in data items()
print(f"\n {key} )
for i, value in enumerate(values[:3]): # Покажем первые 3 элемента
print(f" {i Fl} {value} )
if len(values) > 3:
print(f" ... и еще {lcn(valucs) - 3} элементов )
#	Пример извлечения таблиц
tables_url = "https://httpbin.org/html" # Замените на URL с таблицами
html content = scraper fetch pageltabtes url)
if html_content
tables = scraper extract_tables(html_content)
print(f"\nHai/ifleH5 таблиц: {len(tables)} )
for i, table in enumerate(tables[:2]): # Покажем первые 2 таблицы
рпп1(Р'\пТаблица {i+1}: )
for row in table[:5]:
print( " + " | ’ join(row))
if len(table) > 5:
print(f" ... и еще {len(table) - 5} строк")
def ethical_scraping_guidelines():
"" ’Руководство по этичному веб-скрейпингу.......
guidelines = [
"Всегда проверяйте файл robots.txt сайта ,
"Уважайте настройки Crawl-Delay в robots.txt",
"Используйте задержки между запросами (не менее 1-2 секунд)",
"Кэшируйте загруженные страницы для избежания повторных запросов",
"Идентифицируйте себя с помощью правильного User-Agent",
"Не скрейпите сайты, которые явно запрещают это в своих условиях
использования",
"Рассмотрите использование официального API вместо скрейпинга",
"Ограничьте обьем запрашиваемых данных до необходимого минимума",
"Не перенагружайте серверы сайта большим количеством запросов"
]
print("PyKOBOflCTBO по этичному веб-скрейпингу:")
tor I, guideline in enumerate(guidclines 1):
print(f"{i} {guideline} )
|f_ name _ == "__main
print( Демонстрация веб-скрейпинга )
demo scrapingO
print("\n" + " = "*50)
ethical_scraping guidelinesO
Продвинутый скрейпинг: обработка JavaScript и пагинации
python
# advanced scraping, ру
import requests
from bs4 import BeautifuiSoup
import pandas as pd
import time
from selenium import webdriver
from selenium webdriver.common.by import By
from selenium webdriver.support ui import WebDriverWait
from selenium webdriver support import expectedconditions as EC
class AdvancedScraper:
def _init (self, use selenium=False):
self.use_selenium = useselenium
if use_ selenium
self driver = webdriver.Chromel)	Требуется ChromeDriver
else:
self session = requests SessionO
self.headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
}
def scrape_dynamic_contenttself, url. wait_for=None):
..Скрапинг javaScript-генерируемого контента с помощью Selenium........
if not self.use_seleniurn
print("Selenium не инициализирован")
return None
try:
self driver get(url)
#	Ожидаем загрузки конкретного элемента
if wait_for
WebDrivei Wait(self.driver, 10) until(
EC presence_of_elen?ent_located((By CSS_SELEC10K wait_for))
)
#	Даем время для полной загрузки JavaScript
time.sleep(2)
page_source = self driver.page_sourcc
return page_source
except Exception as e
print(f"0uJH6Ka при загрузке динамического контента: {е}")
return None
def handle_pagination(self base url. page_param='page', max_pages=10):
..Обработка пагинации на сайте'""
all_data = []
for page m range(l, max pages + 1):
print(f"06pa6oTKa страницы (page} )
#	Формируем URL страницы
if ?' in base_uri
url = f"{base_url}&(page_param}=-{page}
else:
url = f {base_url}?{page_param} = {page}"
if self use_selenium
html = self scrape_dynamic_content(url wait_for='.product)	електора
else:
html = self fetch_page(url)
if not html
print(f"He удалось загрузить страницу {page}")
break
# Извлекаем данные cc страницы (зависит от структуры сайта)
page data = self extract_page_data(html)
if not page data
print(f"HeT данных на странице {page}, прекращаем пагинацию")
break
all_data.extend(page_data)
time.sleep(l) # Задержка между запросами
return all data
def fetch page(self. url):
...Загружает веб-страницу (для статических сайтов)........
try:
response = self.session.get(url, headers-self headers. timeout=10)
response. raise_for_status()
return response.text
except requests RcqucstException as e
рпп1(ГОшибка при загрузке страницы {url} {e} )
return None
def extract_page_data(self, html)
...Извлекает данные co страницы (шаблонный метод)'1""
#	Эта функция должна быть адаптирована под конкретный сайт
soup = BeautifulSoup(html 'htm parser')
data = []
#	Пример: извлечение данных продуктов
products = soup.selectC.product') # Пример селектора
for product in products.
try:
name = product select_one( .product-name ).get_text().strip()
price = product.select_one(‘.price ) get_text().strip()
data.append({ name:: name, 'price': price})
except AttributeErrcr:
continue
return data
def close(self):
..Закрывает ресурсы.....
if self use_selenium
self driver quitO
def ecommerce_scraping_example():
..Пример скрейпинга интернет-магазина
#	Note: Это демонстрационный код, который нужно адаптировать под конкретный
сайт
scraper = AdvancedScraper(use_selenium=True) # Используем Selenium для JavaScript
try:
#	Пример URL (замените на реальный)
base_url = "https://httpbin.org/html" # Тестовый URL
#	Для реального примера:
#	base_url = "https://example.com/products"
#	data = scraper handle pagmation(oase_url, max pages=5)
#	Демонстрация с тестовыми данными
рмпЦ"Загрузка динамического контента...")
html = scraper.scrape dynamic_content(basc_url wait_for= hl)
if html:
soup = BeautifulSoupihtml 'html.parser')
heading = soup find('hl')
print(f"3ar-' ловок страницы {heading get_text() if heading else 'He найден'} )
#	Сохранение данных в CSV
#	df= pd.DataFrame(data)
#	df.to_csv(products.csv', index=Faise, encoding='utf-8‘)
#	print(f"CoxpaHCHO {len(data)} продуктов в products.csv")
finally:
scraper closet)
if__name___== "___main__":
print( Пример продвинутого веб-скрейпинга: )
ecommerce_scraping_example()
Заключение
В этой книге мы рассмотрели множество аспектов автоматизации с
помощью Python — от базовых операций с файлами до веб-скрейпинга и
работы с данными. Каждая глава содержала практические примеры кода,
которые вы можете адаптировать под свои задачи.
Ключевые takeaways:
1.	Python предоставляет богатую экосистему библиотек для автоматизации
практически любых задач
2.	Правильная организация кода и обработка ошибок критически важны для
надежных скриптов
3.	Этичный подход к веб-скрейпингу помогает поддерживать хорошие
отношения с владельцами сайтов
4.	Автоматизация рутинных задач высвобождает время для более важной
работы
Дальнейшие шаги:
1.	Изучите официальную документацию библиотек, упомянутых в книге
2.	Присоединитесь к сообществам Python для обмена опытом
3.	Постепенно адаптируйте примеры из книги под свои конкретные задачи
4.	Изучайте исходный код популярных библиотек для лучшего понимания
лучших практик
Помните, что лучший способ обучения — это практика. Начните с
автоматизации небольших задач и постепенно переходите к более
сложным проектам.
Приложения
Приложение А: Шпаргалка по основным модулям Python
Для работы с файловой системой:
•	os — базовые операции с файловой системой
•	shutil — высокоуровневые файловые операции
•	pathlib — объектно-ориентированная работа с путями
Для работы с данными:
csv — чтение и запись CSV файлов
json — работа cJSON данными
pandas — мощная обработка и анализ данных
Для веб-скрейпинга:
requests — HTTP запросы
BeautifulSoup — парсинг HTML
Selenium — авгоматизация браузера
Для системного администрирования:
subprocess — управление процессами
psutil — мониторинг системы
paramiko — SSH соединения
Приложение Б: Best Practices
Организация кода:
1.	Используйте виртуальные окружения для изоляции зависимостей
2.	Разбивайте большие скрипты на модули и функции
3.	Добавляйте комментарии и docstrings для документирования кода
4.	Используйте систему контроля версий (Git)
Обработка ошибок:
1.	Всегда используйте try-except для обработки исключений
2.	Добавляйте информативные сообщения об ошибках
3.	Реализуйте легирование для отладки и мониторинга
Производительное ть:
1.	Используйте генераторы для работы с большими объемами данных
2.	Кэшируйте результаты дорогостоящих операций
3.	Используйте асинхронное программирование для I/O операций
Безопасность:
1.	Никогда не храните credentials в коде
2.	Используйте env ronment variables для конфиденциальных данных
3.	Проверяйте и санируйте пользовательский ввод
Этичный скрейпинг:
1.	Уважайте robots.txt
2.	Используйте задержки между запросами
3.	Кэшируйте загруженные страницы
4.	Используйте официальные API когда это возможно
Эта книга дала вам основы автоматизации на Python, но настоящие
знания приходят с практикой. Начните автоматизировать свои
повседневные задачи сегодня, и вы быстро увидите, насколько Python
может повысить вашу продуктивность.