Модуль 8: Автоматизація та звіти
УРОК 8
Час 90-120 хвилин Потрібно Уроки 1-7 Результат Автоматизований пайплайн зі звітами
Задача
Написати скрипт — половина справи. Друга половина — зробити так, щоб він запускався автоматично та видавав готовий звіт.
Автоматичний запуск по розкладу
Генерація звітів у різних форматах
Логування та обробка помилок
Сповіщення про результати
Нові терміни
Пайплайн Послідовність кроків обробки. Дані проходять через етапи: збір → очищення → аналіз → звіт. Cron Планувальник задач у Linux. Запускає скрипти по розкладу. Логування Запис подій у файл. Що сталося, коли, чи були помилки. Шаблон Заготовка документа з місцями для даних. Jinja2 — популярний шаблонізатор.
Частина 1: Структура проєкту
Організований проєкт легше підтримувати та розширювати:
osint_project/
├── config.py # Налаштування
├── main.py # Головний скрипт
├── collectors/ # Збір даних
│ ├── __init__.py
│ └── web_collector.py
├── analyzers/ # Аналіз
│ ├── __init__.py
│ └── text_analyzer.py
├── reports/ # Генерація звітів
│ ├── __init__.py
│ └── html_report.py
├── data/ # Вхідні дані
├── output/ # Результати
└── logs/ # Логи
Файл конфігурації
# config.py
from pathlib import Path
# Шляхи
BASE_DIR = Path(__file__).parent
DATA_DIR = BASE_DIR / 'data'
OUTPUT_DIR = BASE_DIR / 'output'
LOG_DIR = BASE_DIR / 'logs'
# Створюємо папки якщо не існують
for d in [DATA_DIR, OUTPUT_DIR, LOG_DIR]:
d.mkdir(exist_ok=True)
# Налаштування
SETTINGS = {
'max_retries': 3,
'timeout': 30,
'log_level': 'INFO'
}
Частина 2: Логування
Логи — історія роботи скрипта. Критично для діагностики проблем.
import logging
from datetime import datetime
from config import LOG_DIR
# Налаштування логера
def setup_logger(name='osint'):
logger = logging.getLogger(name)
logger.setLevel(logging.INFO)
# Формат повідомлень
formatter = logging.Formatter(
'%(asctime)s | %(levelname)s | %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
# Запис у файл
log_file = LOG_DIR / f"{name}_{datetime.now().strftime('%Y%m%d')}.log"
file_handler = logging.FileHandler(log_file, encoding='utf-8')
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
# Виведення в консоль
console_handler = logging.StreamHandler()
console_handler.setFormatter(formatter)
logger.addHandler(console_handler)
return logger
# Використання
logger = setup_logger()
logger.info("Скрипт запущено")
logger.warning("Увага: файл не знайдено")
logger.error("Помилка підключення")
Рівні логування
DEBUG Детальна інформація для діагностики INFO Підтвердження нормальної роботи WARNING Щось неочікуване, але скрипт працює ERROR Серйозна проблема, частина функцій не працює CRITICAL Критична помилка, скрипт не може продовжити
Частина 3: Обробка помилок
import logging
logger = logging.getLogger('osint')
def safe_process(func):
"""Декоратор для безпечного виконання функції"""
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except FileNotFoundError as e:
logger.error(f"Файл не знайдено: {e}")
return None
except Exception as e:
logger.error(f"Помилка в {func.__name__}: {e}")
return None
return wrapper
@safe_process
def load_data(filepath):
"""Завантажує дані з файлу"""
logger.info(f"Завантаження: {filepath}")
with open(filepath, 'r', encoding='utf-8') as f:
return f.read()
# Приклад з retry
import time
def with_retry(func, max_attempts=3, delay=1):
"""Повторює виконання при помилці"""
for attempt in range(max_attempts):
try:
return func()
except Exception as e:
logger.warning(f"Спроба {attempt+1}/{max_attempts} не вдалась: {e}")
if attempt < max_attempts - 1:
time.sleep(delay)
logger.error(f"Всі {max_attempts} спроб не вдались")
return None
Частина 4: Генерація звітів
HTML-звіт
from datetime import datetime
def generate_html_report(data, output_path):
"""Генерує HTML-звіт з даними"""
html = f"""<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<title>Аналітичний звіт</title>
<style>
body {{ font-family: Arial, sans-serif; margin: 40px; }}
h1 {{ color: #2c5282; }}
table {{ border-collapse: collapse; width: 100%; }}
th, td {{ border: 1px solid #ddd; padding: 8px; text-align: left; }}
th {{ background-color: #4a5568; color: white; }}
tr:nth-child(even) {{ background-color: #f2f2f2; }}
.summary {{ background: #ebf8ff; padding: 15px; margin: 20px 0; }}
</style>
</head>
<body>
<h1>Аналітичний звіт</h1>
<p>Згенеровано: {datetime.now().strftime('%Y-%m-%d %H:%M')}</p>
<div class="summary">
<h2>Загальна статистика</h2>
<p>Усього подій: {data['total_events']}</p>
<p>Період: {data['period']}</p>
</div>
<h2>Деталі</h2>
<table>
<tr><th>Категорія</th><th>Кількість</th><th>%</th></tr>
"""
for cat, count in data['categories'].items():
pct = count / data['total_events'] * 100
html += f" <tr><td>{cat}</td><td>{count}</td><td>{pct:.1f}%</td></tr>\n"
html += """ </table>
</body>
</html>"""
with open(output_path, 'w', encoding='utf-8') as f:
f.write(html)
return output_path
Excel-звіт
import pandas as pd
from datetime import datetime
from config import OUTPUT_DIR
def generate_excel_report(data, filename='report.xlsx'):
"""Генерує Excel-звіт з кількома аркушами"""
output_path = OUTPUT_DIR / filename
with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
# Аркуш 1: Загальна статистика
summary = pd.DataFrame([
{'Метрика': 'Усього подій', 'Значення': data['total_events']},
{'Метрика': 'Період', 'Значення': data['period']},
{'Метрика': 'Згенеровано', 'Значення': datetime.now().strftime('%Y-%m-%d %H:%M')}
])
summary.to_excel(writer, sheet_name='Статистика', index=False)
# Аркуш 2: Деталі по категоріях
categories = pd.DataFrame([
{'Категорія': k, 'Кількість': v}
for k, v in data['categories'].items()
])
categories.to_excel(writer, sheet_name='Категорії', index=False)
# Аркуш 3: Сирі дані (якщо є)
if 'raw_data' in data:
raw_df = pd.DataFrame(data['raw_data'])
raw_df.to_excel(writer, sheet_name='Дані', index=False)
return output_path
Частина 5: Пайплайн обробки
Об'єднуємо все в єдиний процес:
# main.py
import logging
from datetime import datetime
from config import DATA_DIR, OUTPUT_DIR
# Налаштування
logger = setup_logger()
def run_pipeline():
"""Головний пайплайн обробки"""
logger.info("=" * 50)
logger.info("ЗАПУСК ПАЙПЛАЙНУ")
logger.info("=" * 50)
results = {
'start_time': datetime.now(),
'steps': [],
'errors': []
}
# Крок 1: Збір даних
logger.info("Крок 1: Збір даних")
try:
data = collect_data()
results['steps'].append({'name': 'collect', 'status': 'ok', 'count': len(data)})
logger.info(f" Зібрано {len(data)} записів")
except Exception as e:
logger.error(f" Помилка збору: {e}")
results['errors'].append(str(e))
return results
# Крок 2: Очищення
logger.info("Крок 2: Очищення даних")
try:
clean_data = clean(data)
results['steps'].append({'name': 'clean', 'status': 'ok', 'count': len(clean_data)})
logger.info(f" Після очищення: {len(clean_data)} записів")
except Exception as e:
logger.error(f" Помилка очищення: {e}")
results['errors'].append(str(e))
clean_data = data # Продовжуємо з неочищеними
# Крок 3: Аналіз
logger.info("Крок 3: Аналіз")
try:
analysis = analyze(clean_data)
results['steps'].append({'name': 'analyze', 'status': 'ok'})
logger.info(" Аналіз завершено")
except Exception as e:
logger.error(f" Помилка аналізу: {e}")
results['errors'].append(str(e))
return results
# Крок 4: Звіт
logger.info("Крок 4: Генерація звіту")
try:
report_path = generate_report(analysis)
results['steps'].append({'name': 'report', 'status': 'ok', 'path': str(report_path)})
logger.info(f" Звіт: {report_path}")
except Exception as e:
logger.error(f" Помилка звіту: {e}")
results['errors'].append(str(e))
# Підсумок
results['end_time'] = datetime.now()
results['duration'] = (results['end_time'] - results['start_time']).seconds
logger.info("=" * 50)
logger.info(f"ЗАВЕРШЕНО за {results['duration']} сек")
logger.info(f"Помилок: {len(results['errors'])}")
logger.info("=" * 50)
return results
if __name__ == '__main__':
run_pipeline()
Частина 6: Автозапуск
Linux (cron)
# Відкрити редактор cron
crontab -e
# Формат: хвилина година день місяць день_тижня команда
# Щодня о 06:00
0 6 * * * /usr/bin/python3 /home/user/osint_project/main.py
# Кожні 4 години
0 */4 * * * /usr/bin/python3 /home/user/osint_project/main.py
# Кожного понеділка о 09:00
0 9 * * 1 /usr/bin/python3 /home/user/osint_project/main.py
Windows (Task Scheduler)
Через графічний інтерфейс або PowerShell:
# PowerShell: створити задачу
$action = New-ScheduledTaskAction -Execute "python" -Argument "C:\osint\main.py"
$trigger = New-ScheduledTaskTrigger -Daily -At 6am
Register-ScheduledTask -TaskName "OSINT_Daily" -Action $action -Trigger $trigger
💡 Для тестування використовуй короткі інтервали (кожні 5 хвилин), потім зміни на робочий розклад.
Практична задача
Повний пайплайн аналізу. Створи структуру проєкту та main.py:
# main.py — повний приклад
import pandas as pd
import logging
from datetime import datetime
from pathlib import Path
from collections import Counter
# --- КОНФІГУРАЦІЯ ---
BASE_DIR = Path(__file__).parent
DATA_DIR = BASE_DIR / 'data'
OUTPUT_DIR = BASE_DIR / 'output'
LOG_DIR = BASE_DIR / 'logs'
for d in [DATA_DIR, OUTPUT_DIR, LOG_DIR]:
d.mkdir(exist_ok=True)
# --- ЛОГЕР ---
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s | %(levelname)s | %(message)s',
handlers=[
logging.FileHandler(LOG_DIR / f"run_{datetime.now():%Y%m%d_%H%M}.log"),
logging.StreamHandler()
]
)
logger = logging.getLogger()
# --- ФУНКЦІЇ ---
def load_events():
"""Завантажує події з CSV"""
filepath = DATA_DIR / 'events.csv'
if not filepath.exists():
logger.warning(f"Файл {filepath} не знайдено, створюю тестові дані")
# Тестові дані
df = pd.DataFrame({
'date': ['2025-01-15'] * 6,
'time': ['08:23', '09:12', '10:45', '11:30', '14:00', '15:30'],
'event_type': ['movement', 'fire', 'contact', 'movement', 'fire', 'contact'],
'sector': ['A', 'B', 'A', 'C', 'B', 'A']
})
df.to_csv(filepath, index=False)
return pd.read_csv(filepath)
def analyze_events(df):
"""Аналізує події"""
return {
'total': len(df),
'by_type': df['event_type'].value_counts().to_dict(),
'by_sector': df['sector'].value_counts().to_dict(),
'period': f"{df['date'].min()} — {df['date'].max()}"
}
def generate_report(analysis):
"""Генерує HTML-звіт"""
timestamp = datetime.now().strftime('%Y%m%d_%H%M')
output_path = OUTPUT_DIR / f'report_{timestamp}.html'
html = f"""<!DOCTYPE html>
<html><head><meta charset="UTF-8"><title>Звіт {timestamp}</title>
<style>
body {{ font-family: Arial; margin: 40px; }}
h1 {{ color: #2c5282; }}
.box {{ background: #f7fafc; padding: 15px; margin: 10px 0; border-radius: 5px; }}
</style></head><body>
<h1>Аналітичний звіт</h1>
<p>Згенеровано: {datetime.now():%Y-%m-%d %H:%M}</p>
<div class="box">
<h2>Статистика</h2>
<p>Усього подій: {analysis['total']}</p>
<p>Період: {analysis['period']}</p>
</div>
<div class="box">
<h2>По типах</h2>
<ul>{''.join(f"<li>{k}: {v}</li>" for k,v in analysis['by_type'].items())}</ul>
</div>
<div class="box">
<h2>По секторах</h2>
<ul>{''.join(f"<li>{k}: {v}</li>" for k,v in analysis['by_sector'].items())}</ul>
</div>
</body></html>"""
with open(output_path, 'w', encoding='utf-8') as f:
f.write(html)
return output_path
# --- ГОЛОВНИЙ ПРОЦЕС ---
def main():
logger.info("=" * 40)
logger.info("ЗАПУСК")
try:
# Крок 1
logger.info("Завантаження даних...")
df = load_events()
logger.info(f" Завантажено: {len(df)} записів")
# Крок 2
logger.info("Аналіз...")
analysis = analyze_events(df)
# Крок 3
logger.info("Генерація звіту...")
report = generate_report(analysis)
logger.info(f" Звіт: {report}")
logger.info("УСПІШНО ЗАВЕРШЕНО")
except Exception as e:
logger.error(f"ПОМИЛКА: {e}")
raise
if __name__ == '__main__':
main()
Якщо не працює
Помилка Рішення PermissionError Немає прав на запис. Перевір права на папку output/logs cron не запускає Перевір шлях до python: which python3. Використовуй повний шлях Логи порожні Перевір, що logger має handlers. Додай StreamHandler для дебагу FileNotFoundError Використовуй Path та перевіряй .exists() перед читанням
Робота з AI
Напиши пайплайн для OSINT-аналізу:
Вхід: папка з CSV-файлами (events_*.csv)
Етапи:
1. Зібрати всі CSV в один DataFrame
2. Очистити дані (видалити дублікати, заповнити пропуски)
3. Аналіз: топ-10 слів, розподіл по годинах, кластери координат
4. Згенерувати HTML-звіт з графіками
5. Відправити на email (опційно)
Вимоги:
- Логування кожного етапу
- Обробка помилок з retry
- Конфігурація в окремому файлі
Чек-лист
☐
Структура проєкту створена
☐
Логування працює (файл + консоль)
☐
Помилки обробляються коректно
☐
HTML-звіт генерується
☐
main.py запускається без помилок
☐
Автозапуск налаштовано (cron/Task Scheduler)
Самостійна практика
Email-сповіщення. Додай відправку звіту на email (smtplib)
Telegram-бот. Відправляй сповіщення в Telegram
PDF-звіт. Використай reportlab для PDF
Наступний урок
Урок 9: Робота з базами даних
SQLite, збереження та запити, індексація для швидкого пошуку.
← Модуль 7
Зміст
Модуль 9 →