import gspread import logging from gspread.utils import rowcol_to_a1 # Настройка логирования logger = logging.getLogger(__name__) def log_exceptions(func): """Декоратор для стандартизированного логирования исключений в методах класса.""" def wrapper(self, *args, **kwargs): try: return func(self, *args, **kwargs) except gspread.exceptions.APIError as e: # Логируем специфичные ошибки API Google с деталями error_details = e.response.json().get('error', {}) status = error_details.get('status') message = error_details.get('message') logger.error( f"Ошибка Google API в методе {func.__name__}: {status} - {message}. " f"Проверьте права доступа сервисного аккаунта ({self.client_email}) к таблице." ) # Перевыбрасываем, чтобы вызывающий код мог ее обработать raise except Exception as e: logger.error(f"Непредвиденная ошибка в {func.__name__}: {e}", exc_info=True) raise return wrapper class GoogleSheets: def __init__(self, cred_file, sheet_url): """ Инициализация клиента для работы с Google Sheets. Args: cred_file (str): Путь к JSON-файлу с учетными данными сервисного аккаунта. sheet_url (str): URL Google Таблицы. """ try: # Используем service_account для аутентификации self.client = gspread.service_account(filename=cred_file) # Открываем таблицу по URL self.spreadsheet = self.client.open_by_url(sheet_url) logger.info(f"Успешное подключение к Google Sheet: '{self.spreadsheet.title}'") except gspread.exceptions.SpreadsheetNotFound: logger.error(f"Таблица по URL '{sheet_url}' не найдена. Проверьте URL и права доступа.") raise except FileNotFoundError: logger.error(f"Файл с учетными данными не найден по пути: {cred_file}") raise except Exception as e: logger.error(f"Ошибка инициализации клиента GoogleSheets или открытия таблицы {sheet_url}: {e}", exc_info=True) raise @log_exceptions def get_sheet(self, sheet_name): """Возвращает объект листа (worksheet) по его имени.""" try: sheet = self.spreadsheet.worksheet(sheet_name) logger.debug(f"Получен объект листа '{sheet_name}'") return sheet except gspread.exceptions.WorksheetNotFound: logger.error(f"Лист '{sheet_name}' не найден в таблице '{self.spreadsheet.title}'.") raise @log_exceptions def get_sheets(self): """Получает список всех листов в таблице.""" sheets = self.spreadsheet.worksheets() # Собираем информацию о листах: название и ID sheet_data = [{"title": sheet.title, "id": sheet.id} for sheet in sheets] logger.debug(f"Получено {len(sheet_data)} листов: {[s['title'] for s in sheet_data]}") return sheet_data @log_exceptions def clear_and_write_data(self, sheet_name, data): """ Очищает указанный лист и записывает на него новые данные. Args: sheet_name (str): Имя листа для записи. data (list): Список списков с данными для записи (первый список - заголовки). """ if not isinstance(data, list): raise TypeError("Данные для записи должны быть списком списков.") sheet = self.get_sheet(sheet_name) logger.info(f"Очистка листа '{sheet_name}'...") sheet.clear() logger.info(f"Лист '{sheet_name}' очищен.") # Проверяем, есть ли данные для записи if not data: logger.warning(f"Нет данных для записи на лист '{sheet_name}' после очистки.") return # Завершаем, если список данных пуст num_rows = len(data) num_cols = len(data[0]) if data and data[0] else 0 logger.info(f"Запись {num_rows} строк и {num_cols} колонок на лист '{sheet_name}'...") # Используем метод update для записи всего диапазона одним API-вызовом sheet.update(data, value_input_option='USER_ENTERED') logger.info(f"Данные успешно записаны на лист '{sheet_name}'.")