390 lines
16 KiB
Python
390 lines
16 KiB
Python
"""
|
||
Data Fetcher Module - Абстракция доступа к данным
|
||
Repository Pattern: интерфейс DataFetcher с реализациями для разных источников.
|
||
|
||
Фаза 1: GoogleSheetsFetcher (текущая реализация)
|
||
Фаза 2: ApiFetcher (будущая реализация для ERP)
|
||
"""
|
||
|
||
from abc import ABC, abstractmethod
|
||
from typing import List, Dict, Any, Optional
|
||
import pandas as pd
|
||
from datetime import datetime
|
||
|
||
|
||
class DataFetcher(ABC):
|
||
"""
|
||
Интерфейс для получения данных.
|
||
Все реализации должны следовать этому контракту.
|
||
"""
|
||
|
||
@abstractmethod
|
||
def get_fleet_data(self) -> pd.DataFrame:
|
||
"""Получить данные по автопарку"""
|
||
pass
|
||
|
||
@abstractmethod
|
||
def get_requests_data(self) -> pd.DataFrame:
|
||
"""Получить данные по заявкам"""
|
||
pass
|
||
|
||
@abstractmethod
|
||
def get_accidents_data(self) -> pd.DataFrame:
|
||
"""Получить данные по ДТП и нарушениям"""
|
||
pass
|
||
|
||
@abstractmethod
|
||
def get_maintenance_data(self) -> pd.DataFrame:
|
||
"""Получить данные по ТО и ремонтам"""
|
||
pass
|
||
|
||
@abstractmethod
|
||
def get_fuel_data(self) -> pd.DataFrame:
|
||
"""Получить данные по ГСМ и финансам"""
|
||
pass
|
||
|
||
@abstractmethod
|
||
def get_employees_data(self) -> pd.DataFrame:
|
||
"""Получить данные по сотрудникам"""
|
||
pass
|
||
|
||
|
||
class GoogleSheetsFetcher(DataFetcher):
|
||
"""
|
||
Реализация получения данных из Google Sheets.
|
||
Требуется сервисный аккаунт с доступом к таблицам.
|
||
"""
|
||
|
||
def __init__(self, service_account_path: str, sheet_id: str):
|
||
"""
|
||
Инициализация коннектора к Google Sheets.
|
||
|
||
Args:
|
||
service_account_path: Путь к JSON-ключу сервисного аккаунта
|
||
sheet_id: ID Google таблицы (из URL)
|
||
"""
|
||
self.service_account_path = service_account_path
|
||
self.sheet_id = sheet_id
|
||
self._client = None
|
||
self._spreadsheet = None
|
||
|
||
def _get_client(self):
|
||
"""Ленивая инициализация клиента Google Sheets API"""
|
||
if self._client is None:
|
||
import gspread
|
||
from google.auth import service_account
|
||
|
||
credentials = service_account.Credentials.from_service_account_file(
|
||
self.service_account_path,
|
||
scopes=['https://www.googleapis.com/auth/spreadsheets']
|
||
)
|
||
self._client = gspread.authorize(credentials)
|
||
self._spreadsheet = self._client.open_by_key(self.sheet_id)
|
||
|
||
return self._spreadsheet
|
||
|
||
def _read_worksheet(self, worksheet_name: str) -> pd.DataFrame:
|
||
"""
|
||
Прочитать лист из Google Sheets в DataFrame.
|
||
|
||
Args:
|
||
worksheet_name: Название листа
|
||
|
||
Returns:
|
||
DataFrame с данными
|
||
"""
|
||
try:
|
||
spreadsheet = self._get_client()
|
||
worksheet = spreadsheet.worksheet(worksheet_name)
|
||
data = worksheet.get_all_values()
|
||
|
||
if not data or len(data) < 2:
|
||
return pd.DataFrame()
|
||
|
||
# Первая строка - заголовки
|
||
headers = data[0]
|
||
rows = data[1:]
|
||
|
||
df = pd.DataFrame(rows, columns=headers)
|
||
return df
|
||
|
||
except Exception as e:
|
||
print(f"Ошибка чтения листа '{worksheet_name}': {e}")
|
||
return pd.DataFrame()
|
||
|
||
def get_fleet_data(self) -> pd.DataFrame:
|
||
"""
|
||
Получить данные по автопарку.
|
||
Ожидаемые колонки: vehicle_id, vehicle_type, department, status, year, mileage
|
||
"""
|
||
df = self._read_worksheet("Автопарк")
|
||
return df
|
||
|
||
def get_requests_data(self) -> pd.DataFrame:
|
||
"""
|
||
Получить данные по заявкам.
|
||
Ожидаемые колонки: request_id, department, date, status, fulfilled, vehicle_id
|
||
"""
|
||
df = self._read_worksheet("Заявки")
|
||
return df
|
||
|
||
def get_accidents_data(self) -> pd.DataFrame:
|
||
"""
|
||
Получить данные по ДТП и нарушениям ПДД.
|
||
Ожидаемые колонки: incident_id, date, vehicle_id, driver_id, type, severity, fine_amount
|
||
"""
|
||
df = self._read_worksheet("ДТП_ПДД")
|
||
return df
|
||
|
||
def get_maintenance_data(self) -> pd.DataFrame:
|
||
"""
|
||
Получить данные по ТО и ремонтам.
|
||
Ожидаемые колонки: maintenance_id, vehicle_id, date, type, downtime_days, cost
|
||
"""
|
||
df = self._read_worksheet("ТО_Ремонты")
|
||
return df
|
||
|
||
def get_fuel_data(self) -> pd.DataFrame:
|
||
"""
|
||
Получить данные по ГСМ и финансам.
|
||
Ожидаемые колонки: vehicle_id, date, fuel_liters, fuel_cost, planned_cost, actual_cost
|
||
"""
|
||
df = self._read_worksheet("ГСМ_Финансы")
|
||
return df
|
||
|
||
def get_employees_data(self) -> pd.DataFrame:
|
||
"""
|
||
Получить данные по сотрудникам.
|
||
Ожидаемые колонки: employee_id, name, department, position, license_category
|
||
"""
|
||
df = self._read_worksheet("Сотрудники")
|
||
return df
|
||
|
||
|
||
class MockDataFetcher(DataFetcher):
|
||
"""
|
||
Тестовая реализация с mock-данными.
|
||
Используется для демонстрации и разработки без подключения к Google Sheets.
|
||
"""
|
||
|
||
def __init__(self):
|
||
"""Инициализация с тестовыми данными"""
|
||
self.departments = ["Автобаза №1", "Автобаза №2", "Автобаза №3", "Спецтехника"]
|
||
|
||
def get_fleet_data(self) -> pd.DataFrame:
|
||
"""Сгенерировать тестовые данные по автопарку"""
|
||
data = {
|
||
'vehicle_id': [f'V{i:03d}' for i in range(1, 51)],
|
||
'vehicle_type': ['Грузовой'] * 20 + ['Легковой'] * 15 + ['Спецтехника'] * 15,
|
||
'department': self.departments * 12 + self.departments[:2],
|
||
'status': ['В работе'] * 35 + ['В ремонте'] * 8 + ['Резерв'] * 7,
|
||
'year': [2020, 2021, 2022, 2023, 2024] * 10,
|
||
'mileage': [50000, 75000, 120000, 95000, 30000] * 10,
|
||
'purchase_cost': [2500000, 1800000, 3200000, 2100000, 4500000] * 10
|
||
}
|
||
df = pd.DataFrame(data)
|
||
df['mileage'] = df['mileage'] * (1 + pd.Series(range(50)) % 5 * 0.2)
|
||
return df
|
||
|
||
def get_requests_data(self) -> pd.DataFrame:
|
||
"""Сгенерировать тестовые данные по заявкам"""
|
||
import random
|
||
from datetime import datetime, timedelta
|
||
|
||
data = []
|
||
base_date = datetime(2024, 1, 1)
|
||
|
||
for i in range(500):
|
||
dept = random.choice(self.departments)
|
||
date = base_date + timedelta(days=random.randint(0, 365))
|
||
fulfilled = random.random() > 0.15 # 85% исполнения
|
||
|
||
data.append({
|
||
'request_id': f'R{i+1:04d}',
|
||
'department': dept,
|
||
'date': date.strftime('%Y-%m-%d'),
|
||
'status': 'Исполнено' if fulfilled else 'Отклонено',
|
||
'fulfilled': 1 if fulfilled else 0,
|
||
'vehicle_id': f'V{random.randint(1, 50):03d}'
|
||
})
|
||
|
||
return pd.DataFrame(data)
|
||
|
||
def get_accidents_data(self) -> pd.DataFrame:
|
||
"""Сгенерировать тестовые данные по ДТП"""
|
||
import random
|
||
from datetime import datetime, timedelta
|
||
|
||
data = []
|
||
base_date = datetime(2024, 1, 1)
|
||
incident_types = ['ДТП', 'Нарушение ПДД', 'Инцидент']
|
||
severities = ['Лёгкий', 'Средний', 'Тяжёлый']
|
||
|
||
for i in range(25):
|
||
date = base_date + timedelta(days=random.randint(0, 365))
|
||
severity = random.choice(severities)
|
||
fine = random.randint(500, 50000) if severity != 'Лёгкий' else random.randint(0, 5000)
|
||
|
||
data.append({
|
||
'incident_id': f'I{i+1:03d}',
|
||
'date': date.strftime('%Y-%m-%d'),
|
||
'vehicle_id': f'V{random.randint(1, 50):03d}',
|
||
'driver_id': f'D{random.randint(1, 30):03d}',
|
||
'type': random.choice(incident_types),
|
||
'severity': severity,
|
||
'fine_amount': fine
|
||
})
|
||
|
||
return pd.DataFrame(data)
|
||
|
||
def get_maintenance_data(self) -> pd.DataFrame:
|
||
"""Сгенерировать тестовые данные по ТО"""
|
||
import random
|
||
from datetime import datetime, timedelta
|
||
|
||
data = []
|
||
base_date = datetime(2024, 1, 1)
|
||
maintenance_types = ['Плановое ТО', 'Внеплановый ремонт', 'Диагностика']
|
||
|
||
for i in range(120):
|
||
date = base_date + timedelta(days=random.randint(0, 365))
|
||
mtype = random.choice(maintenance_types)
|
||
downtime = random.randint(1, 15) if mtype != 'Диагностика' else random.randint(0, 2)
|
||
cost = random.randint(5000, 150000)
|
||
|
||
data.append({
|
||
'maintenance_id': f'M{i+1:04d}',
|
||
'vehicle_id': f'V{random.randint(1, 50):03d}',
|
||
'date': date.strftime('%Y-%m-%d'),
|
||
'type': mtype,
|
||
'downtime_days': downtime,
|
||
'cost': cost
|
||
})
|
||
|
||
return pd.DataFrame(data)
|
||
|
||
def get_fuel_data(self) -> pd.DataFrame:
|
||
"""Сгенерировать тестовые данные по ГСМ"""
|
||
import random
|
||
from datetime import datetime, timedelta
|
||
|
||
data = []
|
||
base_date = datetime(2024, 1, 1)
|
||
|
||
for i in range(600):
|
||
date = base_date + timedelta(days=random.randint(0, 365))
|
||
vehicle_id = f'V{random.randint(1, 50):03d}'
|
||
planned = random.randint(8000, 15000)
|
||
actual = int(planned * (0.85 + random.random() * 0.3)) # +/- 15%
|
||
|
||
data.append({
|
||
'vehicle_id': vehicle_id,
|
||
'date': date.strftime('%Y-%m-%d'),
|
||
'fuel_liters': random.randint(100, 500),
|
||
'fuel_cost': actual,
|
||
'planned_cost': planned,
|
||
'actual_cost': actual
|
||
})
|
||
|
||
return pd.DataFrame(data)
|
||
|
||
def get_employees_data(self) -> pd.DataFrame:
|
||
"""Сгенерировать тестовые данные по сотрудникам"""
|
||
import random
|
||
|
||
first_names = ['Александр', 'Дмитрий', 'Сергей', 'Андрей', 'Михаил',
|
||
'Владимир', 'Николай', 'Евгений', 'Алексей', 'Игорь']
|
||
last_names = ['Иванов', 'Петров', 'Сидоров', 'Смирнов', 'Козлов',
|
||
'Новиков', 'Федоров', 'Морозов', 'Волков', 'Лебедев']
|
||
|
||
data = {
|
||
'employee_id': [f'D{i:03d}' for i in range(1, 31)],
|
||
'name': [f'{random.choice(last_names)} {random.choice(first_names)}'
|
||
for _ in range(30)],
|
||
'department': self.departments * 7 + self.departments[:2],
|
||
'position': ['Водитель'] * 25 + ['Механик'] * 3 + ['Начальник'] * 2,
|
||
'license_category': ['B', 'C', 'D', 'E'] * 7 + ['B', 'C']
|
||
}
|
||
|
||
return pd.DataFrame(data)
|
||
|
||
|
||
class ApiFetcher(DataFetcher):
|
||
"""
|
||
Заготовка для будущей реализации REST API коннектора к ERP системе.
|
||
Заполните методы согласно документации вашей ERP.
|
||
"""
|
||
|
||
def __init__(self, api_base_url: str, api_token: str):
|
||
"""
|
||
Инициализация API коннектора.
|
||
|
||
Args:
|
||
api_base_url: Базовый URL ERP API
|
||
api_token: Токен авторизации
|
||
"""
|
||
self.api_base_url = api_base_url
|
||
self.api_token = api_token
|
||
self._session = None
|
||
|
||
def _get_headers(self) -> Dict[str, str]:
|
||
"""Получить заголовки для API запросов"""
|
||
return {
|
||
'Authorization': f'Bearer {self.api_token}',
|
||
'Content-Type': 'application/json'
|
||
}
|
||
|
||
def _fetch_endpoint(self, endpoint: str, params: Optional[Dict] = None) -> List[Dict]:
|
||
"""
|
||
Выполнить GET запрос к API.
|
||
|
||
Args:
|
||
endpoint: Относительный путь endpoint
|
||
params: Параметры запроса
|
||
|
||
Returns:
|
||
Список записей
|
||
"""
|
||
import requests
|
||
|
||
if self._session is None:
|
||
self._session = requests.Session()
|
||
|
||
url = f"{self.api_base_url}/{endpoint}"
|
||
response = self._session.get(url, headers=self._get_headers(), params=params)
|
||
response.raise_for_status()
|
||
|
||
return response.json()
|
||
|
||
def get_fleet_data(self) -> pd.DataFrame:
|
||
"""Получить данные по автопарку из ERP"""
|
||
# TODO: Реализовать согласно документации ERP API
|
||
# data = self._fetch_endpoint('fleet/vehicles', {'year': 2024})
|
||
# return pd.DataFrame(data)
|
||
raise NotImplementedError("Метод get_fleet_data не реализован для API")
|
||
|
||
def get_requests_data(self) -> pd.DataFrame:
|
||
"""Получить данные по заявкам из ERP"""
|
||
# TODO: Реализовать согласно документации ERP API
|
||
raise NotImplementedError("Метод get_requests_data не реализован для API")
|
||
|
||
def get_accidents_data(self) -> pd.DataFrame:
|
||
"""Получить данные по ДТП из ERP"""
|
||
# TODO: Реализовать согласно документации ERP API
|
||
raise NotImplementedError("Метод get_accidents_data не реализован для API")
|
||
|
||
def get_maintenance_data(self) -> pd.DataFrame:
|
||
"""Получить данные по ТО из ERP"""
|
||
# TODO: Реализовать согласно документации ERP API
|
||
raise NotImplementedError("Метод get_maintenance_data не реализован для API")
|
||
|
||
def get_fuel_data(self) -> pd.DataFrame:
|
||
"""Получить данные по ГСМ из ERP"""
|
||
# TODO: Реализовать согласно документации ERP API
|
||
raise NotImplementedError("Метод get_fuel_data не реализован для API")
|
||
|
||
def get_employees_data(self) -> pd.DataFrame:
|
||
"""Получить данные по сотрудникам из ERP"""
|
||
# TODO: Реализовать согласно документации ERP API
|
||
raise NotImplementedError("Метод get_employees_data не реализован для API")
|