Python для автоматизации: 7 приемов, избавляющих от рутины

import os
import shutil
from pathlib import Path

def organize_files(directory):
# Определяем типы файлов и папки для них
file_types = {
'images': ['.jpg', '.jpeg', '.png', '.gif'],
'documents': ['.pdf', '.docx', '.txt', '.xlsx'],
'videos': ['.mp4', '.mov', '.avi'],
'audio': ['.mp3', '.wav', '.flac']
}

# Создаём папки, если они не существуют
for folder in file_types:
os.makedirs(os.path.join(directory, folder), exist_ok=True)

# Сортируем файлы
for file in os.listdir(directory):
file_path = os.path.join(directory, file)

# Пропускаем папки
if os.path.isdir(file_path):
continue

# Определяем расширение
file_ext = os.path.splitext(file)[1].lower()

# Перемещаем файл в соответствующую папку
for folder, extensions in file_types.items():
if file_ext in extensions:
shutil.move(file_path, os.path.join(directory, folder, file))
print(f"Moved {file} to {folder}")
break

# Пример использования
organize_files('/path/to/downloads')

import pandas as pd

# Загрузка данных
sales_data = pd.read_excel('sales_2023.xlsx')

# Анализ
monthly_sales = sales_data.groupby('month')['revenue'].sum()
top_products = sales_data.groupby('product')['quantity'].sum().sort_values(ascending=False).head(5)

# Создание отчёта
with pd.ExcelWriter('sales_report.xlsx') as writer:
monthly_sales.to_excel(writer, sheet_name='Monthly Summary')
top_products.to_excel(writer, sheet_name='Top Products')

print("Отчёт успешно создан!")

import requests
from bs4 import BeautifulSoup

def get_news_headlines(url):
# Отправляем запрос к сайту
response = requests.get(url)
if response.status_code != 200:
return "Ошибка при доступе к сайту"

# Парсим HTML-содержимое
soup = BeautifulSoup(response.content, 'html.parser')

# Извлекаем заголовки (предположим, что они в тегах h2 с классом 'news-title')
headlines = soup.find_all('h2', class_='news-title')

# Собираем результат
results = []
for headline in headlines:
results.append(headline.text.strip())

return results

# Пример использования
headlines = get_news_headlines('https://example-news-site.com')
for idx, headline in enumerate(headlines, 1):
print(f"{idx}. {headline}")

from selenium import webdriver
from selenium.webdriver.chrome.options import Options
import time

def scrape_dynamic_content(url):
# Настраиваем опции
chrome_options = Options()
chrome_options.add_argument('--headless') # Запуск браузера в фоновом режиме

# Инициализируем браузер
driver = webdriver.Chrome(options=chrome_options)

try:
# Переходим на страницу
driver.get(url)

# Ждем, пока JavaScript загрузит контент
time.sleep(3)

# Пример извлечения данных
elements = driver.find_elements_by_class_name('dynamic-content')

results = []
for element in elements:
results.append(element.text)

return results
finally:
# Закрываем браузер
driver.quit()

# Пример использования
data = scrape_dynamic_content('https://example-site-with-js.com')
print(data)

import requests
import json
from datetime import datetime

def create_task_from_event(event_data, api_key):
"""
Создаёт задачу в Trello на основе события из календаря
"""
# Параметры для Trello API
url = "https://api.trello.com/1/cards"
headers = {
"Accept": "application/json"
}
query = {
'key': api_key,
'token': 'your_trello_token',
'idList': 'your_list_id',
'name': f"Встреча: {event_data['title']}",
'desc': f"Дата: {event_data['date']}\nУчастники: {', '.join(event_data['participants'])}\nЗаметки: {event_data['notes']}"
}

# Отправляем запрос к API
response = requests.post(url, headers=headers, params=query)

if response.status_code == 200:
return f"Задача успешно создана: {json.loads(response.text)['shortUrl']}"
else:
return f"Ошибка при создании задачи: {response.text}"

# Пример использования
event = {
'title': 'Обсуждение квартального отчёта',
'date': datetime.now().strftime('%Y-%m-%d %H:%M'),
'participants': ['Анна', 'Иван', 'Сергей'],
'notes': 'Подготовить презентацию по финансовым показателям'
}

result = create_task_from_event(event, 'your_trello_api_key')
print(result)

from fastapi import FastAPI, Request, BackgroundTasks
import requests
import json

app = FastAPI()

def process_order(order_data):
"""
Обрабатывает данные о заказе: добавляет клиента в CRM,
создает задачу на доставку и отправляет уведомление
"""
# Добавляем клиента в CRM
crm_response = requests.post(
'https://your-crm-api.com/customers',
json={
'name': order_data['customer_name'],
'email': order_data['email'],
'phone': order_data['phone']
},
headers={'Authorization': 'Bearer your_crm_api_key'}
)

# Создаём задачу на доставку
delivery_response = requests.post(
'https://your-delivery-api.com/tasks',
json={
'address': order_data['delivery_address'],
'items': order_data['items'],
'scheduled_time': order_data['delivery_time']
},
headers={'Authorization': 'Bearer your_delivery_api_key'}
)

# Отправляем SMS-уведомление
sms_response = requests.post(
'https://your-sms-provider.com/send',
json={
'to': order_data['phone'],
'message': f"Ваш заказ #{order_data['order_id']} принят. Ожидайте доставку {order_data['delivery_time']}."
},
headers={'Authorization': 'Bearer your_sms_api_key'}
)

# Логируем результаты для отладки
print(f"CRM: {crm_response.status_code}, Delivery: {delivery_response.status_code}, SMS: {sms_response.status_code}")

@app.post("/webhook/new-order")
async def new_order_webhook(request: Request, background_tasks: BackgroundTasks):
# Получаем данные из webhook
order_data = await request.json()

# Запускаем обработку в фоновом режиме
background_tasks.add_task(process_order, order_data)

return {"status": "processing"}

# Запуск сервера
# uvicorn main:app --reload

import os
import shutil
import datetime
import zipfile

def create_backup(source_dir, backup_dir):
# Создаём имя архива с текущей датой
date_str = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
zip_filename = f"backup_{date_str}.zip"
zip_path = os.path.join(backup_dir, zip_filename)

# Создаём архив
with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
for root, _, files in os.walk(source_dir):
for file in files:
file_path = os.path.join(root, file)
# Добавляем файл в архив, сохраняя структуру папок
zipf.write(
file_path, 
os.path.relpath(file_path, os.path.join(source_dir, '..'))
)

# Удаляем старые бэкапы, оставляя только 5 последних
all_backups = sorted([
os.path.join(backup_dir, f) for f in os.listdir(backup_dir)
if f.startswith("backup_") and f.endswith(".zip")
])

if len(all_backups) > 5:
for old_backup in all_backups[:-5]:
os.remove(old_backup)
print(f"Удален старый бэкап: {old_backup}")

return zip_path

# Пример использования
backup_file = create_backup(
"/path/to/project", 
"/path/to/backups"
)
print(f"Создан бэкап: {backup_file}")

import unittest
import os
from my_module import process_data

class TestDataProcessing(unittest.TestCase):

def setUp(self):
# Подготовка тестовых данных
self.test_data = [
{"name": "Test1", "value": 10},
{"name": "Test2", "value": 20}
]

def test_data_sum(self):
# Проверяем суммирование значений
result = process_data.calculate_sum(self.test_data)
self.assertEqual(result, 30)

def test_data_filter(self):
# Проверяем фильтрацию по значению
filtered = process_data.filter_by_value(self.test_data, min_value=15)
self.assertEqual(len(filtered), 1)
self.assertEqual(filtered[0]["name"], "Test2")

def tearDown(self):
# Очистка после тестов
pass

# Автоматически запускаем тесты при выполнении скрипта
if __name__ == "__main__":
unittest.main()

import psutil
import time
import smtplib
from email.mime.text import MIMEText

def monitor_resources(threshold=90, interval=60, email=None):
"""
Мониторит использование CPU, памяти и диска.
Отправляет уведомление при превышении порога.
"""
while True:
# Получаем метрики
cpu_percent = psutil.cpu_percent(interval=1)
memory_percent = psutil.virtual_memory().percent
disk_percent = psutil.disk_usage('/').percent

print(f"CPU: {cpu_percent}% | Memory: {memory_percent}% | Disk: {disk_percent}%")

# Проверяем превышение пороговых значений
if cpu_percent > threshold or memory_percent > threshold or disk_percent > threshold:
alert_message = f"ALERT: Resource usage exceeded {threshold}%!\n"
alert_message += f"CPU: {cpu_percent}%\n"
alert_message += f"Memory: {memory_percent}%\n"
alert_message += f"Disk: {disk_percent}%\n"

print(alert_message)

# Отправляем уведомление по email если указан
if email:
send_alert_email(email, "System Resource Alert", alert_message)

# Ждём до следующей проверки
time.sleep(interval)

def send_alert_email(to_email, subject, message):
from_email = "your_monitoring@example.com"

# Создаём сообщение
msg = MIMEText(message)
msg['Subject'] = subject
msg['From'] = from_email
msg['To'] = to_email

# Отправляем через SMTP
try:
with smtplib.SMTP('smtp.example.com', 587) as server:
server.starttls()
server.login("username", "password")
server.send_message(msg)
print(f"Alert email sent to {to_email}")
except Exception as e:
print(f"Failed to send email: {e}")

# Пример использования
# monitor_resources(threshold=80, email="admin@example.com")

import subprocess
import os
import sys

def check_and_format_code(directory, autofix=True):
"""
Проверяет и форматирует Python-код в указанной директории.
Использует Black для форматирования и Flake8 для проверки.
"""
print(f"Processing code in {directory}...")

# Форматируем код с Black
if autofix:
print("Formatting with Black...")
black_result = subprocess.run(
["black", directory],
capture_output=True,
text=True
)

if black_result.returncode == 0:
print("Code formatting successful!")
print(black_result.stdout)
else:
print("Error during code formatting:")
print(black_result.stderr)

# Проверяем код с Flake8
print("\nRunning Flake8 checks...")
flake_result = subprocess.run(
["flake8", directory],
capture_output=True,
text=True
)

if flake_result.returncode == 0:
print("No issues found with Flake8!")
else:
print("Flake8 found code style issues:")
print(flake_result.stdout)

return flake_result.returncode == 0

# Пример использования
# if __name__ == "__main__":
# if len(sys.argv) < 2:
# print("Usage: python code_check.py <directory> [--no-autofix]")
# sys.exit(1)
# 
# directory = sys.argv[1]
# autofix = "--no-autofix" not in sys.argv
# 
# if not os.path.isdir(directory):
# print(f"Error: {directory} is not a valid directory")
# sys.exit(1)
# 
# success = check_and_format_code(directory, autofix)
# sys.exit(0 if success else 1)

import pandas as pd
import matplotlib.pyplot as plt
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.application import MIMEApplication
from datetime import datetime
import schedule
import time

def generate_sales_report(data_file):
"""
Создаёт отчёт о продажах на основе данных из CSV-файла
"""
# Загружаем данные
df = pd.read_csv(data_file)

# Анализируем продажи по категориям
category_sales = df.groupby('category')['amount'].sum().reset_index()

# Создаём график
plt.figure(figsize=(10, 6))
plt.bar(category_sales['category'], category_sales['amount'])
plt.title('Sales by Category')
plt.xlabel('Product Category')
plt.ylabel('Sales Amount ($)')
plt.xticks(rotation=45)
plt.tight_layout()

# Сохраняем график
report_date = datetime.now().strftime('%Y-%m-%d')
chart_file = f'sales_report_{report_date}.png'
plt.savefig(chart_file)
plt.close()

# Подготавливаем данные для отчёта
total_sales = df['amount'].sum()
best_category = category_sales.loc[category_sales['amount'].idxmax()]['category']
best_product = df.loc[df['amount'].idxmax()]['product']

report_html = f"""
<html>
<body>
<h1>Sales Report – {report_date}</h1>
<p>Total Sales: ${total_sales:,.2f}</p>
<p>Best Performing Category: {best_category}</p>
<p>Best Selling Product: {best_product}</p>
<img src="cid:sales_chart">
</body>
</html>
"""

return report_html, chart_file, report_date

def send_report_email(recipient, report_html, chart_file, report_date):
"""
Отправляет отчёт по email
"""
# Настройка сообщения
msg = MIMEMultipart()
msg['From'] = 'reports@example.com'
msg['To'] = recipient
msg['Subject'] = f'Sales Report – {report_date}'

# Добавляем HTML-содержимое
msg.attach(MIMEText(report_html, 'html'))

# Добавляем изображение
with open(chart_file, 'rb') as f:
img = MIMEApplication(f.read())
img.add_header('Content-ID', '<sales_chart>')
msg.attach(img)

# Отправляем сообщение
with smtplib.SMTP('smtp.example.com', 587) as server:
server.starttls()
server.login('username', 'password')
server.send_message(msg)

print(f"Report sent to {recipient}")

def scheduled_report_job():
"""
Функция для выполнения по расписанию
"""
try:
report_html, chart_file, report_date = generate_sales_report('sales_data.csv')
send_report_email('manager@example.com', report_html, chart_file, report_date)
except Exception as e:
print(f"Error generating report: {e}")

# Настройка расписания – каждый понедельник в 8 утра
schedule.every().monday.at("08:00").do(scheduled_report_job)

# Главный цикл для запуска по расписанию
# while True:
# schedule.run_pending()
# time.sleep(60)

import sqlite3
import pandas as pd
import csv
from datetime import datetime, timedelta

def database_maintenance(db_file):
"""
Выполняет типичные задачи обслуживания базы данных:
- Архивирует старые записи
- Оптимизирует индексы
- Создаёт отчёт о состоянии БД
"""
conn = sqlite3.connect(db_file)
cursor = conn.cursor()

print(f"Connected to database: {db_file}")

# 1. Архивирование старых записей
# Находим записи старше 1 года
archive_date = (datetime.now() – timedelta(days=365)).strftime('%Y-%m-%d')

# Экспортируем старые записи в CSV
old_records_query = f"""
SELECT * FROM orders 
WHERE order_date < '{archive_date}'
"""

old_records = pd.read_sql_query(old_records_query, conn)

if not old_records.empty:
archive_file = f"orders_archive_{datetime.now().strftime('%Y%m%d')}.csv"
old_records.to_csv(archive_file, index=False)
print(f"Archived {len(old_records)} old records to {archive_file}")

# Удаляем архивированные записи
cursor.execute(f"DELETE FROM orders WHERE order_date < '{archive_date}'")
print(f"Deleted {cursor.rowcount} old records from database")
else:
print("No old records to archive")

# 2. Оптимизация базы данных
cursor.execute("VACUUM")
cursor.execute("ANALYZE")
print("Database optimized")

# 3. Создание отчёта о состоянии БД
# Получаем размер таблиц
cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
tables = cursor.fetchall()

db_report = []
for table in tables:
table_name = table[0]
cursor.execute(f"SELECT COUNT(*) FROM {table_name}")
row_count = cursor.fetchone()[0]
db_report.append({
'Table': table_name,
'Rows': row_count
})

# Сохраняем отчёт
report_df = pd.DataFrame(db_report)
report_file = f"db_report_{datetime.now().strftime('%Y%m%d')}.csv"
report_df.to_csv(report_file, index=False)
print(f"Database report saved to {report_file}")

conn.commit()
conn.close()
print("Database maintenance completed successfully")

# Пример использования
# database_maintenance("company.db")

import os
import subprocess
import json
from datetime import datetime

def update_dependencies(project_dir, create_backup=True, requirements_file='requirements.txt'):
"""
Обновляет зависимости Python-проекта:
1. Создаёт резервную копию текущих зависимостей
2. Проверяет наличие устаревших пакетов
3. Обновляет их до последних версий
4. Фиксирует новые версии в requirements.txt
5. Создаёт отчёт об обновлениях
"""
os.chdir(project_dir)
print(f"Working in directory: {os.getcwd()}")

# Получаем список текущих зависимостей
try:
current_req = subprocess.check_output(['pip', 'freeze'], text=True)
print(f"Found {len(current_req.splitlines())} installed packages")

# Создаём резервную копию
if create_backup:
backup_file = f"requirements_backup_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt"
with open(backup_file, 'w') as f:
f.write(current_req)
print(f"Backup created: {backup_file}")
except Exception as e:
print(f"Error getting current dependencies: {e}")
return False

# Проверяем устаревшие пакеты
try:
outdated = subprocess.check_output(['pip', 'list', '--outdated', '--format=json'], text=True)
outdated_packages = json.loads(outdated)

if not outdated_packages:
print("All packages are up to date!")
return True

print(f"Found {len(outdated_packages)} outdated packages")

# Создаём отчёт об обновлениях
update_report = []

# Обновляем каждый пакет
for package in outdated_packages:
package_name = package['name']
current_version = package['version']
latest_version = package['latest_version']

print(f"Updating {package_name} from {current_version} to {latest_version}...")

try:
subprocess.check_call(['pip', 'install', '--upgrade', package_name])
update_report.append({
'package': package_name,
'from_version': current_version,
'to_version': latest_version,
'status': 'updated'
})
except Exception as e:
print(f"Failed to update {package_name}: {e}")
update_report.append({
'package': package_name,
'from_version': current_version,
'to_version': latest_version,
'status': f'failed: {str(e)}'
})

# Обновляем requirements.txt
subprocess.check_call(['pip', 'freeze', '>', requirements_file], shell=True)
print(f"Updated {requirements_file}")

# Сохраняем отчёт об обновлениях
report_file = f"update_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
with open(report_file, 'w') as f:
json.dump(update_report, f, indent=2)
print(f"Update report saved to {report_file}")

return True
except Exception as e:
print(f"Error during dependency update: {e}")
return False

# Пример использования
# update_dependencies("/path/to/your/project")