""" 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")