Python для автоматизации: 5 способов избавиться от рутины навсегда

# Пример автоматизации обработки файлов
import os
import shutil

def organize_files(directory):
for filename in os.listdir(directory):
if filename.endswith('.jpg') or filename.endswith('.png'):
shutil.move(os.path.join(directory, filename), 'images/')
elif filename.endswith('.pdf'):
shutil.move(os.path.join(directory, filename), 'documents/')

organize_files('downloads/')

import pandas as pd
import glob
import os

# Найти все Excel-файлы в директории
excel_files = glob.glob('sales_data/*.xlsx')

# Создать пустой DataFrame для объединения данных
all_data = pd.DataFrame()

# Обработать каждый файл и объединить данные
for file in excel_files:
# Извлечь название региона из имени файла
region = os.path.basename(file).split('_')[0]

# Прочитать данные
df = pd.read_excel(file)

# Добавить колонку с регионом
df['Region'] = region

# Объединить с основным DataFrame
all_data = pd.concat([all_data, df], ignore_index=True)

# Очистка данных
all_data.dropna(subset=['Sales'], inplace=True)

# Анализ данных
sales_by_region = all_data.groupby('Region')['Sales'].sum().reset_index()
best_products = all_data.groupby('Product')['Sales'].sum().nlargest(5).reset_index()

# Сохранение результатов
with pd.ExcelWriter('sales_report.xlsx') as writer:
all_data.to_excel(writer, sheet_name='Raw Data', index=False)
sales_by_region.to_excel(writer, sheet_name='Regional Sales', index=False)
best_products.to_excel(writer, sheet_name='Top Products', index=False)

import numpy as np
import pandas as pd

# Создание данных для анализа
data = pd.DataFrame({
'customer_id': np.random.randint(1, 1000, size=10000),
'purchase_amount': np.random.normal(100, 50, size=10000),
'date': pd.date_range(start='2023-01-01', periods=10000)
})

# Расчет метрик с использованием NumPy
average_purchase = np.mean(data['purchase_amount'])
median_purchase = np.median(data['purchase_amount'])
purchase_std = np.std(data['purchase_amount'])

# Выявление аномалий (покупки, отклоняющиеся более чем на 2 стандартных отклонения)
threshold = average_purchase + 2 * purchase_std
anomalies = data[data['purchase_amount'] > threshold]

print(f"Средняя покупка: {average_purchase:.2f}")
print(f"Медианная покупка: {median_purchase:.2f}")
print(f"Выявлено {len(anomalies)} аномальных транзакций")

import requests
from bs4 import BeautifulSoup

# Получение HTML-страницы
response = requests.get('https://news.ycombinator.com/')
soup = BeautifulSoup(response.text, 'html.parser')

# Извлечение заголовков новостей
headlines = soup.select('.titleline > a')

# Вывод результатов
for headline in headlines[:5]:
print(headline.text)

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.service import Service
from webdriver_manager.chrome import ChromeDriverManager

# Инициализация драйвера
driver = webdriver.Chrome(service=Service(ChromeDriverManager().install()))

# Открытие страницы
driver.get('https://www.python.org')

# Поиск по сайту
search_box = driver.find_element(By.NAME, 'q')
search_box.send_keys('automation')
search_box.submit()

# Извлечение результатов поиска
results = driver.find_elements(By.CSS_SELECTOR, '.list-recent-events li h3 a')
for result in results[:3]:
print(result.text)

# Закрытие браузера
driver.quit()

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.common.exceptions import TimeoutException, NoSuchElementException
import random
import time

driver = webdriver.Chrome()
driver.get('https://example.com/login')

try:
# Явное ожидание загрузки элемента
username_field = WebDriverWait(driver, 10).until(
EC.presence_of_element_located((By.ID, 'username'))
)

# Имитация человеческого ввода с задержками
for char in 'my_username':
username_field.send_keys(char)
time.sleep(random.uniform(0.05, 0.25))

password_field = driver.find_element(By.ID, 'password')
for char in 'secure_password':
password_field.send_keys(char)
time.sleep(random.uniform(0.05, 0.25))

# Клик с вероятностным смещением (для имитации человека)
login_button = driver.find_element(By.CSS_SELECTOR, '.login-button')
driver.execute_script("""
arguments[0].scrollIntoView();
arguments[0].click();
""", login_button)

# Проверка успешного входа
success_element = WebDriverWait(driver, 15).until(
EC.presence_of_element_located((By.CSS_SELECTOR, '.welcome-message'))
)
print("Успешный вход в систему!")

except TimeoutException:
print("Тайм-аут при ожидании элемента")
except NoSuchElementException:
print("Элемент не найден – возможно структура сайта изменилась")
finally:
driver.quit()

import os
import shutil
from datetime import datetime
from pathlib import Path

# Определим структуру для сортировки по типам файлов
EXTENSIONS = {
'documents': ['.pdf', '.docx', '.txt', '.xlsx', '.csv'],
'images': ['.jpg', '.jpeg', '.png', '.gif', '.svg'],
'videos': ['.mp4', '.mov', '.avi', '.mkv'],
'audio': ['.mp3', '.wav', '.flac', '.aac'],
'archives': ['.zip', '.rar', '.7z', '.tar.gz']
}

def organize_downloads(download_dir):
download_path = Path(download_dir)

# Создание директорий для каждого типа файлов, если их нет
for folder in EXTENSIONS:
(download_path / folder).mkdir(exist_ok=True)

# Создание папки для файлов старше 30 дней
(download_path / "old_files").mkdir(exist_ok=True)

# Обработка каждого файла
for item in download_path.iterdir():
if item.is_file():
# Проверка возраста файла
file_age = datetime.now() – datetime.fromtimestamp(item.stat().st_mtime)
if file_age.days > 30:
shutil.move(str(item), str(download_path / "old_files" / item.name))
continue

# Сортировка по расширению
moved = False
for folder, extensions in EXTENSIONS.items():
if item.suffix.lower() in extensions:
shutil.move(str(item), str(download_path / folder / item.name))
moved = True
break

# Если расширение не определено, оставляем файл на месте
if not moved:
print(f"Не удалось определить тип файла: {item.name}")

# Пример использования
organize_downloads("C:/Users/Username/Downloads")

import pandas as pd
from docx import Document
from docx.shared import Inches
import matplotlib.pyplot as plt
from datetime import datetime

def generate_sales_report(sales_data, output_folder):
# Анализ данных продаж с помощью pandas
df = pd.read_csv(sales_data)

# Создаем визуализацию
plt.figure(figsize=(10, 6))
df.groupby('product')['revenue'].sum().plot(kind='bar')
plt.title('Доходы по продуктам')
chart_path = f"{output_folder}/sales_chart.png"
plt.savefig(chart_path)

# Создаем документ Word с отчетом
doc = Document()
doc.add_heading('Отчет о продажах', 0)
doc.add_paragraph(f'Сгенерировано: {datetime.now().strftime("%d.%m.%Y %H:%M")}')

# Добавляем таблицу с топ-5 продуктами
top_products = df.groupby('product')['revenue'].sum().nlargest(5)
doc.add_heading('Топ-5 продуктов по выручке', level=1)
table = doc.add_table(rows=1, cols=2)
table.style = 'Table Grid'
header_cells = table.rows[0].cells
header_cells[0].text = 'Продукт'
header_cells[1].text = 'Выручка'

for product, revenue in top_products.items():
row_cells = table.add_row().cells
row_cells[0].text = product
row_cells[1].text = f"${revenue:,.2f}"

# Добавляем график
doc.add_heading('Визуализация продаж', level=1)
doc.add_picture(chart_path, width=Inches(6))

# Сохраняем отчет
report_path = f"{output_folder}/sales_report_{datetime.now().strftime('%Y%m%d')}.docx"
doc.save(report_path)

return report_path

import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.base import MIMEBase
from email import encoders
import ssl
import schedule
import time

def send_email_report(subject, body, recipient, attachment_path=None):
# Настройки SMTP
smtp_server = "smtp.gmail.com"
port = 587
sender_email = "your_email@gmail.com"
password = "your_app_password" # Рекомендуется использовать переменные окружения

# Создание сообщения
message = MIMEMultipart()
message["From"] = sender_email
message["To"] = recipient
message["Subject"] = subject

# Добавление тела письма
message.attach(MIMEText(body, "plain"))

# Добавление вложения при наличии
if attachment_path:
with open(attachment_path, "rb") as attachment:
part = MIMEBase("application", "octet-stream")
part.set_payload(attachment.read())
encoders.encode_base64(part)

filename = attachment_path.split("/")[-1]
part.add_header(
"Content-Disposition",
f"attachment; filename= {filename}",
)
message.attach(part)

# Отправка письма
context = ssl.create_default_context()
with smtplib.SMTP(smtp_server, port) as server:
server.starttls(context=context)
server.login(sender_email, password)
server.send_message(message)

print(f"Email sent to {recipient} successfully!")

# Планирование регулярной отправки отчета
def scheduled_report():
report_path = generate_sales_report("sales_data.csv", "reports")
send_email_report(
"Еженедельный отчет о продажах",
"Пожалуйста, ознакомьтесь с отчетом о продажах за прошедшую неделю.",
"manager@example.com",
report_path
)

# Запуск отчета каждый понедельник в 8:30
schedule.every().monday.at("08:30").do(scheduled_report)

# Запуск планировщика
while True:
schedule.run_pending()
time.sleep(60)

# Пример использования Fabric для автоматизации деплоя
from fabric import Connection, task

@task
def deploy(ctx, branch="main"):
"""
Автоматизированный деплой приложения на production-сервер
"""
# Подключение к серверу
with Connection('user@production-server') as conn:
# Обновление кодовой базы
conn.run(f"cd /var/www/app && git pull origin {branch}")

# Установка зависимостей
conn.run("cd /var/www/app && pip install -r requirements.txt")

# Применение миграций базы данных
conn.run("cd /var/www/app && python manage.py migrate")

# Сборка статических файлов
conn.run("cd /var/www/app && python manage.py collectstatic --noinput")

# Перезапуск сервисов
conn.sudo("systemctl restart gunicorn")
conn.sudo("systemctl restart nginx")

# Проверка статуса
conn.run("systemctl status gunicorn | grep active")

print("Деплой успешно завершен!")

# Пример автоматизации работы с AWS с использованием boto3
import boto3
import time

def create_auto_scaling_environment():
"""
Создание автомасштабируемой инфраструктуры в AWS
"""
# Инициализация клиентов
ec2 = boto3.client('ec2')
elb = boto3.client('elbv2')
autoscaling = boto3.client('autoscaling')

# Создание Security Group
sg_response = ec2.create_security_group(
GroupName='WebAppSG',
Description='Security group for web application'
)
sg_id = sg_response['GroupId']

# Настройка правил безопасности
ec2.authorize_security_group_ingress(
GroupId=sg_id,
IpPermissions=[
{
'IpProtocol': 'tcp',
'FromPort': 80,
'ToPort': 80,
'IpRanges': [{'CidrIp': '0.0.0.0/0'}]
},
{
'IpProtocol': 'tcp',
'FromPort': 443,
'ToPort': 443,
'IpRanges': [{'CidrIp': '0.0.0.0/0'}]
}
]
)

# Создание конфигурации запуска
autoscaling.create_launch_configuration(
LaunchConfigurationName='WebAppLC',
ImageId='ami-0c55b159cbfafe1f0', # Amazon Linux 2 AMI
InstanceType='t2.micro',
SecurityGroups=[sg_id],
UserData='''#!/bin/bash
yum update -y
yum install -y httpd
systemctl start httpd
systemctl enable httpd
echo "Hello World from $(hostname)" > /var/www/html/index.html
'''
)

# Создание Auto Scaling Group
autoscaling.create_auto_scaling_group(
AutoScalingGroupName='WebAppASG',
LaunchConfigurationName='WebAppLC',
MinSize=2,
MaxSize=5,
DesiredCapacity=2,
HealthCheckType='ELB',
HealthCheckGracePeriod=300,
VPCZoneIdentifier='subnet-12345678,subnet-87654321' # Подсети VPC
)

print("Автомасштабируемая инфраструктура успешно создана!")

# Пример автоматизации работы с Docker с использованием docker-py
import docker
import time

def deploy_microservices():
"""
Автоматизированное развертывание микросервисов в Docker
"""
client = docker.from_env()

# Создание сети для микросервисов
network = client.networks.create("microservices_network", driver="bridge")

# Развертывание базы данных
db_container = client.containers.run(
"postgres:13",
detach=True,
name="database",
environment=["POSTGRES_PASSWORD=secret", "POSTGRES_DB=app"],
network="microservices_network",
volumes={"/data/postgres": {"bind": "/var/lib/postgresql/data", "mode": "rw"}},
restart_policy={"Name": "always"}
)

# Ждем инициализации базы данных
print("Ожидание инициализации базы данных...")
time.sleep(15)

# Развертывание API-сервиса
api_container = client.containers.run(
"my-api-service:latest",
detach=True,
name="api_service",
environment=["DB_HOST=database", "DB_PASSWORD=secret"],
network="microservices_network",
ports={"8000/tcp": 8000},
restart_policy={"Name": "always"}
)

# Развертывание фронтенд-сервиса
frontend_container = client.containers.run(
"my-frontend:latest",
detach=True,
name="frontend",
environment=["API_URL=http://api_service:8000"],
network="microservices_network",
ports={"80/tcp": 80},
restart_policy={"Name": "always"}
)

print("Микросервисы успешно развернуты!")
print(f"API доступен по адресу: http://localhost:8000")
print(f"Фронтенд доступен по адресу: http://localhost:80")

# Автоматизация сканирования безопасности кода
import subprocess
import json
import requests
import os
import sys

def security_scan():
"""
Выполняет сканирование кода на наличие уязвимостей
и отправляет результаты в систему мониторинга
"""
# Запуск Bandit для анализа уязвимостей в Python-коде
print("Запуск сканирования безопасности кода...")
bandit_result = subprocess.run(
["bandit", "-r", ".", "-f", "json", "-o", "security-report.json"],
capture_output=True,
text=True
)

# Проверка результата сканирования
if os.path.exists("security-report.json"):
with open("security-report.json", "r") as report_file:
report_data = json.load(report_file)

# Подсчет уязвимостей по уровням
high_severity = len([i for i in report_data["results"] if i["issue_severity"] == "HIGH"])
medium_severity = len([i for i in report_data["results"] if i["issue_severity"] == "MEDIUM"])
low_severity = len([i for i in report_data["results"] if i["issue_severity"] == "LOW"])

print(f"Найдено уязвимостей: {high_severity} высоких, {medium_severity} средних, {low_severity} низких")

# Отправка результатов в систему мониторинга
if "MONITORING_API_URL" in os.environ and "MONITORING_API_KEY" in os.environ:
response = requests.post(
os.environ["MONITORING_API_URL"],
headers={"Authorization": f"Bearer {os.environ['MONITORING_API_KEY']}"},
json={
"project": os.environ.get("CI_PROJECT_NAME", "unknown"),
"branch": os.environ.get("CI_COMMIT_BRANCH", "unknown"),
"security_scan": {
"high": high_severity,
"medium": medium_severity,
"low": low_severity,
"timestamp": report_data["generated_at"]
}
}
)
print(f"Результаты отправлены в систему мониторинга: {response.status_code}")

# Проверка политики безопасности
if high_severity > 0:
print("Критическая ошибка: обнаружены уязвимости высокой степени тяжести")
sys.exit(1)
elif medium_severity > 3:
print("Критическая ошибка: обнаружено слишком много уязвимостей средней степени тяжести")
sys.exit(1)
else:
print("Сканирование безопасности пройдено успешно!")
else:
print("Ошибка: отчет о сканировании безопасности не создан")
sys.exit(1)

if __name__ == "__main__":
security_scan()