103 lines
5.3 KiB
Python
103 lines
5.3 KiB
Python
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}'.") |