Автоматизация Excel с Python: методы интеграции для аналитиков

# Установка pandas для анализа данных
pip install pandas

# Установка openpyxl для работы с файлами Excel
pip install openpyxl

# Установка xlwings для интерактивной работы с Excel
pip install xlwings

# Установка matplotlib и seaborn для визуализации
pip install matplotlib seaborn

python -m venv excel_automation
source excel_automation/bin/activate # для Linux/MacOS
excel_automation\Scripts\activate # для Windows

pip install pandas==1.5.3 openpyxl==3.1.2 xlwings==0.30.10

pip freeze > requirements.txt

project/
├── src/
│ ├── __init__.py
│ ├── data_processing.py
│ ├── excel_formatter.py
│ └── report_generator.py
├── tests/
│ ├── __init__.py
│ └── test_processing.py
├── data/
│ ├── input/
│ └── output/
├── requirements.txt
└── main.py

import pandas as pd
import openpyxl
import xlwings as xw
import sys

print(f"Python версия: {sys.version}")
print(f"pandas версия: {pd.__version__}")
print(f"openpyxl версия: {openpyxl.__version__}")
print(f"xlwings версия: {xw.__version__}")

# Тестовое создание DataFrame и экспорт в Excel
df = pd.DataFrame({'A': [1, 2, 3], 'B': [4, 5, 6]})
df.to_excel('test_output.xlsx', index=False)
print("Файл test_output.xlsx успешно создан")

import pandas as pd

# Чтение всего листа
df = pd.read_excel('data.xlsx', sheet_name='Sheet1')
print(df.head())

# Чтение определённого диапазона ячеек
df = pd.read_excel('data.xlsx', sheet_name='Sheet1', usecols="A:C", skiprows=2, nrows=10)

# Чтение нескольких листов
excel_file = pd.ExcelFile('data.xlsx')
sheets = {sheet: pd.read_excel(excel_file, sheet) for sheet in excel_file.sheet_names}

# Чтение с обработкой пропущенных значений
df = pd.read_excel('data.xlsx', sheet_name='Sheet1', na_values=['NA', 'N/A', '-'])

from openpyxl import Workbook
from openpyxl.styles import Font, Alignment, PatternFill
import pandas as pd

# Создание нового файла Excel
wb = Workbook()
ws = wb.active
ws.title = "Отчет продаж"

# Заголовки с форматированием
headers = ["Продукт", "Регион", "Q1", "Q2", "Q3", "Q4", "Итого"]
for col_num, header in enumerate(headers, 1):
cell = ws.cell(row=1, column=col_num)
cell.value = header
cell.font = Font(bold=True)
cell.alignment = Alignment(horizontal='center')
cell.fill = PatternFill(start_color="FFFF00", end_color="FFFF00", fill_type="solid")

# Данные из pandas DataFrame
df = pd.DataFrame({
"Продукт": ["Товар A", "Товар B", "Товар C"],
"Регион": ["Север", "Юг", "Запад"],
"Q1": [1000, 2000, 3000],
"Q2": [1500, 2500, 3500],
"Q3": [2000, 3000, 4000],
"Q4": [2500, 3500, 4500]
})

# Добавление формул для подсчета итогов
for row_idx, row in enumerate(df.itertuples(), 2):
for col_idx, value in enumerate(row[1:], 1):
ws.cell(row=row_idx, column=col_idx, value=value)

# Формула для суммы квартальных значений
sum_formula = f"=SUM(C{row_idx}:F{row_idx})"
ws.cell(row=row_idx, column=7, value=sum_formula)

# Сохранение файла
wb.save("sales_report.xlsx")

import xlwings as xw

# Открытие существующего файла Excel
app = xw.App(visible=True) # Параметр visible=True позволяет видеть Excel
wb = app.books.open('data.xlsx')
sheet = wb.sheets['Sheet1']

# Чтение диапазона данных
data_range = sheet.range('A1:E10').value

# Запись данных обратно с модификацией
for i, row in enumerate(data_range):
if isinstance(row, list):
if len(row) > 2 and row[2] is not None:
# Увеличиваем значение в третьей колонке на 10%
row[2] = row[2] * 1.1
sheet.range(f'C{i+1}').value = row[2]

# Добавление формулы
sheet.range('F1').value = '=SUM(C1:C10)'

# Форматирование ячеек
sheet.range('F1').color = (255, 255, 0) # Жёлтый фон

# Сохранение и закрытие
wb.save()
wb.close()
app.quit()

import pandas as pd
import numpy as np
from datetime import datetime

# Функция для валидации данных при чтении
def validate_amount(x):
try:
return float(x)
except (ValueError, TypeError):
return np.nan

def validate_date(x):
try:
if isinstance(x, datetime):
return x
return pd.to_datetime(x)
except:
return pd.NaT

# Чтение с применением функций валидации
converters = {
'Amount': validate_amount,
'TransactionDate': validate_date
}

df = pd.read_excel(
'transactions.xlsx', 
sheet_name='Data', 
converters=converters,
dtype={'CustomerId': str, 'ProductCode': str} # Принудительное определение типов
)

# Фильтрация невалидных данных
valid_df = df.dropna(subset=['Amount', 'TransactionDate'])
print(f"Прочитано записей: {len(df)}, из них валидных: {len(valid_df)}")

import os
import pandas as pd
from pathlib import Path

def process_excel_files(input_dir, output_file):
# Создаем пустой DataFrame для агрегации результатов
combined_df = pd.DataFrame()

# Получаем список всех Excel-файлов в директории
excel_files = list(Path(input_dir).glob("*.xlsx"))

for file_path in excel_files:
try:
print(f"Обрабатываем файл: {file_path}")
# Чтение данных из файла
df = pd.read_excel(file_path)

# Добавляем информацию о файле-источнике
df['Source_File'] = file_path.name

# Объединяем с общим DataFrame
combined_df = pd.concat([combined_df, df], ignore_index=True)

except Exception as e:
print(f"Ошибка при обработке файла {file_path}: {str(e)}")

# Сохраняем объединенный результат
combined_df.to_excel(output_file, index=False)
print(f"Обработка завершена. Результат сохранен в {output_file}")

# Пример использования
process_excel_files("./input_data", "./output/combined_result.xlsx")

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt

# Загружаем данные продаж
sales_df = pd.read_excel('sales_data.xlsx')

# Базовая статистика
print(sales_df.describe())

# Выявление аномалий (значений за пределами 3 сигм)
mean = sales_df['Revenue'].mean()
std = sales_df['Revenue'].std()
outliers = sales_df[abs(sales_df['Revenue'] – mean) > 3 * std]

# Группировка и агрегация
region_sales = sales_df.groupby(['Region', 'Product']).agg({
'Revenue': ['sum', 'mean', 'count'],
'Quantity': ['sum', 'mean'],
'Date': ['min', 'max'] # первая и последняя продажа
}).reset_index()

# Временной анализ
sales_df['Month'] = pd.DatetimeIndex(sales_df['Date']).month
sales_df['Year'] = pd.DatetimeIndex(sales_df['Date']).year
monthly_trend = sales_df.groupby(['Year', 'Month']).sum()['Revenue']

# Создание сводной таблицы (аналог PivotTable в Excel)
pivot = pd.pivot_table(
sales_df, 
values='Revenue', 
index=['Region'], 
columns=['Product'], 
aggfunc='sum', 
fill_value=0
)

# Сохранение результатов анализа в отдельные листы Excel
with pd.ExcelWriter('sales_analysis.xlsx') as writer:
sales_df.to_excel(writer, sheet_name='RawData', index=False)
outliers.to_excel(writer, sheet_name='Anomalies', index=False)
region_sales.to_excel(writer, sheet_name='RegionalSales')
monthly_trend.to_excel(writer, sheet_name='MonthlyTrend')
pivot.to_excel(writer, sheet_name='ProductByRegion')

import pandas as pd
import numpy as np
from scipy import stats
import statsmodels.api as sm

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

# Проверка корреляций между затратами на маркетинг и продажами
correlation = df['Marketing_Spend'].corr(df['Sales'])
print(f"Корреляция между расходами на маркетинг и продажами: {correlation:.2f}")

# Построение регрессионной модели
X = sm.add_constant(df['Marketing_Spend']) # добавляем константу для пересечения
y = df['Sales']
model = sm.OLS(y, X).fit()
print(model.summary())

# Прогнозирование продаж на основе модели
df['Predicted_Sales'] = model.predict(X)
df['Residual'] = df['Sales'] – df['Predicted_Sales']

# Анализ эффективности маркетинговых каналов
channel_efficiency = df.groupby('Channel').apply(
lambda x: pd.Series({
'Total_Spend': x['Marketing_Spend'].sum(),
'Total_Sales': x['Sales'].sum(),
'ROI': (x['Sales'].sum() – x['Marketing_Spend'].sum()) / x['Marketing_Spend'].sum(),
'Conversion_Rate': x['Conversions'].sum() / x['Visits'].sum() if x['Visits'].sum() > 0 else 0
})
).sort_values('ROI', ascending=False)

# Сегментация клиентов по значимости
df['Customer_Value'] = df['Sales'] * df['Margin']
df['Customer_Segment'] = pd.qcut(
df['Customer_Value'], 
q=[0, 0.2, 0.5, 0.8, 1.0], 
labels=['Low', 'Medium-Low', 'Medium', 'High']
)

# Экспорт результатов с форматированием
with pd.ExcelWriter('marketing_analysis.xlsx', engine='openpyxl') as writer:
df.to_excel(writer, sheet_name='Detailed_Analysis', index=False)
channel_efficiency.to_excel(writer, sheet_name='Channel_Performance')

# Получение рабочей книги для форматирования
workbook = writer.book
worksheet = writer.sheets['Channel_Performance']

# Форматирование ROI как процентов
for row in range(2, len(channel_efficiency) + 2):
worksheet.cell(row=row, column=4).number_format = '0.00%'

import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
from openpyxl import load_workbook
from openpyxl.drawing.image import Image

# Загрузка и анализ данных
df = pd.read_excel('customer_data.xlsx')

# 1. Создание визуализаций
plt.figure(figsize=(10, 6))
sns.histplot(df['Purchase_Amount'], bins=30, kde=True)
plt.title('Распределение сумм покупок')
plt.savefig('purchase_distribution.png')
plt.close()

plt.figure(figsize=(12, 8))
sns.boxplot(x='Customer_Segment', y='Purchase_Amount', data=df)
plt.title('Суммы покупок по сегментам клиентов')
plt.savefig('segment_comparison.png')
plt.close()

# 2. Создание аналитических сводок
segment_analysis = df.groupby('Customer_Segment').agg({
'Customer_ID': 'count',
'Purchase_Amount': ['sum', 'mean', 'median', 'std'],
'Purchase_Frequency': ['mean', 'median']
})

recency_analysis = df.groupby('Last_Purchase_Days').agg({
'Customer_ID': 'count'
}).reset_index()
recency_analysis['Percent'] = recency_analysis['Customer_ID'] / recency_analysis['Customer_ID'].sum() * 100

# 3. Сохранение отчёта в Excel с интеграцией визуализаций
with pd.ExcelWriter('customer_analysis_report.xlsx', engine='openpyxl') as writer:
df.to_excel(writer, sheet_name='Raw_Data', index=False)
segment_analysis.to_excel(writer, sheet_name='Segment_Analysis')
recency_analysis.to_excel(writer, sheet_name='Recency_Analysis', index=False)

# 4. Добавление изображений в отчёт
workbook = load_workbook('customer_analysis_report.xlsx')
sheet = workbook.create_sheet('Visualizations')

img1 = Image('purchase_distribution.png')
sheet.add_image(img1, 'B2')

img2 = Image('segment_comparison.png')
sheet.add_image(img2, 'B20')

workbook.save('customer_analysis_report.xlsx')

import pandas as pd
import numpy as np
import os
import logging
from datetime import datetime
from openpyxl import load_workbook
from openpyxl.styles import PatternFill, Font, Alignment, Border, Side
from openpyxl.chart import BarChart, Reference
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.base import MIMEBase
from email.mime.text import MIMEText
from email import encoders

# Настройка логирования
logging.basicConfig(
filename='report_automation.log',
level=logging.INFO,
format='%(asctime)s – %(levelname)s – %(message)s'
)

def generate_monthly_report(input_folder, output_path, email_recipients=None):
"""
Полностью автоматизированная функция для генерации и рассылки ежемесячного отчёта
"""
try:
# 1. Сбор и объединение данных из нескольких источников
logging.info("Начинаем сбор данных из исходных файлов")
all_data = pd.DataFrame()

for filename in os.listdir(input_folder):
if filename.endswith(('.xlsx', '.xls')):
file_path = os.path.join(input_folder, filename)
try:
df = pd.read_excel(file_path)
df['Source'] = filename
all_data = pd.concat([all_data, df], ignore_index=True)
logging.info(f"Успешно загружен файл: {filename}")
except Exception as e:
logging.error(f"Ошибка при чтении файла {filename}: {str(e)}")

if all_data.empty:
logging.error("Не удалось загрузить данные из исходных файлов")
return False

# 2. Обработка и анализ данных
logging.info("Начинаем анализ данных")

# Преобразование дат в единый формат
all_data['Date'] = pd.to_datetime(all_data['Date'], errors='coerce')

# Фильтрация за текущий месяц
current_month = datetime.now().month
current_year = datetime.now().year
monthly_data = all_data[
(all_data['Date'].dt.month == current_month) & 
(all_data['Date'].dt.year == current_year)
]

# Агрегация по различным разрезам
regional_summary = monthly_data.groupby('Region').agg({
'Sales': ['sum', 'mean', 'count'],
'Profit': ['sum', 'mean']
})

product_summary = monthly_data.groupby('Product').agg({
'Sales': ['sum', 'count'],
'Profit': ['sum']
})

# Расчёт дополнительных KPI
product_summary['Profit_Margin'] = (
product_summary[('Profit', 'sum')] / product_summary[('Sales', 'sum')]
)

# 3. Создание Excel-отчёта с форматированием
logging.info("Формируем отчёт в Excel")

# Создание файла с несколькими листами
with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
monthly_data.to_excel(writer, sheet_name='Raw_Data', index=False)
regional_summary.to_excel(writer, sheet_name='Regional_Summary')
product_summary.to_excel(writer, sheet_name='Product_Summary')

# Дополнительное форматирование с помощью openpyxl
wb = load_workbook(output_path)

# Форматирование листа с региональными данными
if 'Regional_Summary' in wb.sheetnames:
ws = wb['Regional_Summary']

# Определение стилей
header_fill = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid")
header_font = Font(color="FFFFFF", bold=True)
border = Border(
left=Side(style='thin'), right=Side(style='thin'),
top=Side(style='thin'), bottom=Side(style='thin')
)

# Применение стилей к заголовкам
for cell in ws[1]:
cell.fill = header_fill
cell.font = header_font
cell.border = border
cell.alignment = Alignment(horizontal='center')

# Добавление диаграммы
chart = BarChart()
chart.title = "Продажи по регионам"
chart.style = 10
chart.x_axis.title = "Регион"
chart.y_axis.title = "Сумма продаж"

data = Reference(ws, min_col=3, min_row=1, max_row=ws.max_row, max_col=3)
cats = Reference(ws, min_col=1, min_row=2, max_row=ws.max_row)
chart.add_data(data, titles_from_data=True)
chart.set_categories(cats)

ws.add_chart(chart, "H2")

# Сохранение форматированного файла
wb.save(output_path)
logging.info(f"Отчёт успешно сохранён: {output_path}")

# 4. Отправка отчёта по электронной почте
if email_recipients:
logging.info("Отправка отчёта по электронной почте")

try:
# Настройки почтового сервера
smtp_server = "smtp.company.com"
smtp_port = 587
sender_email = "reports@company.com"
password = os.environ.get("EMAIL_PASSWORD") # Безопасное хранение пароля

# Создание сообщения
msg = MIMEMultipart()
msg['From'] = sender_email
msg['To'] = ", ".join(email_recipients)
msg['Subject'] = f"Ежемесячный отчёт по продажам – {datetime.now().strftime('%B %Y')}"

body = """
Уважаемые коллеги,

Во вложении ежемесячный отчёт по продажам.

С уважением,
Система автоматизированной отчётности
"""
msg.attach(MIMEText(body, 'plain'))

# Добавление вложения
attachment = open(output_path, "rb")
part = MIMEBase('application', 'octet-stream')
part.set_payload(attachment.read())
encoders.encode_base64(part)
part.add_header(
'Content-Disposition',
f'attachment; filename="{os.path.basename(output_path)}"'
)
msg.attach(part)
attachment.close()

# Отправка сообщения
server = smtplib.SMTP(smtp_server, smtp_port)
server.starttls()
server.login(sender_email, password)
server.send_message(msg)
server.quit()

logging.info("Отчёт успешно отправлен по электронной почте")

except Exception as e:
logging.error(f"Ошибка при отправке отчёта по email: {str(e)}")

return True

except Exception as e:
logging.error(f"Критическая ошибка при формировании отчёта: {str(e)}")
return False

# Пример использования
if __name__ == "__main__":
result = generate_monthly_report(
input_folder="./monthly_sales_data",
output_path="./reports/monthly_sales_report.xlsx",
email_recipients=["team_lead@company.com", "cfo@company.com"]
)

print(f"Формирование отчёта {'успешно' if result else 'не удалось'}")

import pandas as pd
import os
import hashlib
import json
from datetime import datetime
import shutil

def create_version_controlled_report(data_df, template_path, output_dir):
"""
Создает отчёт с контролем версий и отслеживанием изменений
"""
# 1. Создание уникального хеша данных для отслеживания изменений
data_hash = hashlib.md5(pd.util.hash_pandas_object(data_df).values).hexdigest()

# 2. Создание метаданных отчёта
metadata = {
"creation_time": datetime.now().isoformat(),
"data_hash": data_hash,
"creator": os.getlogin(),
"rows_count": len(data_df),
"columns": list(data_df.columns)
}

# 3. Определение версии отчёта
version_file = os.path.join(output_dir, "version_history.json")

if os.path.exists(version_file):
with open(version_file, 'r') as f:
version_history = json.load(f)
current_version = version_history[-1]["version"] + 1
else:
version_history = []
current_version = 1

# 4. Создание файла отчёта с версией в имени
timestamp = datetime.now().strftime("%Y%m%d")
report_filename = f"report_v{current_version}_{timestamp}.xlsx"
report_path = os.path.join(output_dir, report_filename)

# 5. Копирование шаблона, если он указан
if template_path and os.path.exists(template_path):
shutil.copy2(template_path, report_path)

# Загрузка книги и запись данных
with pd.ExcelWriter(report_path, mode='a', engine='openpyxl', if_sheet_exists='replace') as writer:
data_df.to_excel(writer, sheet_name='Data', index=False)
else:
# Создание нового файла
with pd.ExcelWriter(report_path, engine='openpyxl') as writer:
data_df.to_excel(writer, sheet_name='Data', index=False)

# Создание листа с метаданными
pd.DataFrame([metadata]).to_excel(writer, sheet_name='Metadata', index=False)

# 6. Обновление истории версий
version_entry = {
"version": current_version,
"filename": report_filename,
"timestamp": datetime.now().isoformat(),
"data_hash": data_hash,
"creator": os.getlogin()
}

version_history.append(version_entry)

with open(version_file, 'w') as f:
json.dump(version_history, f, indent=4)

return report_path, current_version

# Пример использования
sales_data = pd.read_excel('monthly_sales.xlsx')
report_path, version = create_version_controlled_report(
sales_data, 
template_path='report_template.xlsx',
output_dir='./secured_reports'
)
print(f"Создан отчёт версии {version}: {report_path}")