Автоматизация Google Таблиц через Python: пишем код, экономим время

pip install gspread oauth2client google-auth google-auth-oauthlib google-auth-httplib2

import gspread
from google.oauth2.service_account import Credentials

# Определяем область доступа
scope = ['https://spreadsheets.google.com/feeds',
'https://www.googleapis.com/auth/drive']

# Подключаемся к сервисному аккаунту
credentials = Credentials.from_service_account_file('путь_к_вашему_файлу.json', scopes=scope)
client = gspread.authorize(credentials)

# Проверка подключения – открываем таблицу по URL или имени
spreadsheet = client.open('Название_вашей_таблицы')
print(f"Подключение успешно! Доступ к таблице: {spreadsheet.title}")

# Создание новой таблицы
new_spreadsheet = client.create('Аналитический отчет Q2 2023')

# Предоставление доступа другим пользователям
new_spreadsheet.share('colleague@example.com', perm_type='user', role='writer')

# Получение URL новой таблицы
print(f"Создана новая таблица: {new_spreadsheet.url}")

# Добавление нового листа
worksheet = new_spreadsheet.add_worksheet(title="Финансовые показатели", rows=100, cols=20)

# Переименование существующего листа (по умолчанию создается Sheet1)
default_sheet = new_spreadsheet.sheet1
default_sheet.update_title("Общие данные")

# Получение списка всех листов
all_worksheets = new_spreadsheet.worksheets()
print(f"Листы в таблице: {[ws.title for ws in all_worksheets]}")

# Удаление ненужного листа
worksheet_to_delete = new_spreadsheet.worksheet("Лишний лист")
new_spreadsheet.del_worksheet(worksheet_to_delete)

# Вместо последовательных обновлений отдельных ячеек
# worksheet.update_cell(1, 1, "Плохой метод")
# worksheet.update_cell(1, 2, "Очень медленно")
# worksheet.update_cell(1, 3, "Много API-вызовов")

# Используйте пакетное обновление
worksheet.update([
["Хороший метод", "Быстро", "Один API-вызов"],
["Еще данные", "Еще значения", "Эффективно"]
])

import time
from random import uniform

def update_with_retry(worksheet, values, max_retries=5):
retries = 0
while retries < max_retries:
try:
worksheet.update(values)
return True
except Exception as e:
if "Quota exceeded" in str(e):
wait_time = (2 ** retries) + uniform(0, 1)
print(f"Превышение квоты, ожидание {wait_time:.1f} секунд...")
time.sleep(wait_time)
retries += 1
else:
raise e
return False

# Добавление прав доступа с указанием роли
new_spreadsheet.share('team@company.com', perm_type='group', role='reader', notify=True, email_message="Новый отчет готов!")

# Просмотр текущих прав доступа
permissions = new_spreadsheet.list_permissions()
for p in permissions:
print(f"Email: {p.get('emailAddress')}, Role: {p.get('role')}")

# Получение всех значений с листа
all_values = worksheet.get_all_values()

# Получение конкретной строки или столбца
row_values = worksheet.row_values(1) # Первая строка
column_values = worksheet.col_values(3) # Третий столбец

# Получение диапазона ячеек
cell_range = worksheet.get_values('A1:C10')

# Получение конкретной ячейки
cell_value = worksheet.acell('B2').value
# или
cell_value = worksheet.cell(2, 2).value # row, col (нумерация с 1)

# Обновление одной ячейки
worksheet.update_cell(1, 1, "Новое значение") # row, col, value
# или
worksheet.update('A1', "Новое значение")

# Обновление диапазона ячеек
worksheet.update('A1:C3', [
["Имя", "Возраст", "Город"],
["Алексей", 28, "Москва"],
["Елена", 34, "Санкт-Петербург"]
])

# Добавление строк в конец таблицы
new_rows = [
["Дмитрий", 41, "Екатеринбург"],
["Ольга", 37, "Казань"]
]
worksheet.append_rows(new_rows)

# Очистка ячеек
worksheet.clear() # Очистка всего листа
worksheet.batch_clear(['A1:B5', 'D1:E5']) # Очистка нескольких диапазонов

# Неэффективный подход (много запросов)
for i, name in enumerate(names, start=1):
worksheet.update_cell(i, 1, name)
worksheet.update_cell(i, 2, ages[i-1])
worksheet.update_cell(i, 3, cities[i-1])

# Эффективный подход (один запрос)
data = [[name, age, city] for name, age, city in zip(names, ages, cities)]
range_notation = f'A1:C{len(data)}'
worksheet.update(range_notation, data)

# Поиск ячейки по значению
cell = worksheet.find("Москва")
print(f"Найдено в ячейке R{cell.row}C{cell.col}")

# Поиск всех ячеек с заданным значением
cells = worksheet.findall("Москва")
locations = [f"R{cell.row}C{cell.col}" for cell in cells]
print(f"Все вхождения: {locations}")

# Более сложный поиск с регулярными выражениями
import re
pattern = re.compile(r'Мос\w+') # Найдет "Москва", "Московский" и т.д.
cell = worksheet.find(pattern)

# Получение данных в виде списка словарей
records = worksheet.get_all_records() # Первая строка считается заголовком

# Преобразование в pandas DataFrame для продвинутого анализа
import pandas as pd
df = pd.DataFrame(records)
print(df.describe()) # Статистический анализ

# Постепенная загрузка данных большой таблицы
all_data = []
batch_size = 1000
offset = 0

while True:
batch = worksheet.get_all_values(value_render_option='UNFORMATTED_VALUE', 
limit=batch_size, offset=offset)
if not batch:
break
all_data.extend(batch)
offset += len(batch)

print(f"Загружено {len(all_data)} строк данных")

def safe_update(worksheet, range_name, data):
max_attempts = 5
for attempt in range(max_attempts):
try:
return worksheet.update(range_name, data)
except Exception as e:
if attempt == max_attempts – 1:
raise
print(f"Ошибка при обновлении: {e}. Повторная попытка {attempt+1}/{max_attempts}")
time.sleep(2 ** attempt) # Экспоненциальная задержка

worksheet.format("A1:C1", {
"backgroundColor": {"red": 0.9, "green": 0.9, "blue": 0.9},
"horizontalAlignment": "CENTER",
"textFormat": {
"fontSize": 12,
"bold": True
}
})

formats = [
{
"range": "A1:F1",
"format": {
"backgroundColor": {"red": 0.2, "green": 0.4, "blue": 0.9},
"textFormat": {"foregroundColor": {"red": 1, "green": 1, "blue": 1}, "bold": True}
}
},
{
"range": "A2:F100",
"format": {
"backgroundColor": {"red": 0.95, "green": 0.95, "blue": 0.95},
"borders": {
"top": {"style": "SOLID"},
"bottom": {"style": "SOLID"}
}
}
}
]

worksheet.batch_format(formats)

# Выделить красным ячейки со значениями меньше 50
worksheet.conditional_format("B2:B100", {
"type": "NUMBER_LESS_THAN",
"values": [{"userEnteredValue": "50"}],
"format": {
"backgroundColor": {"red": 1, "green": 0.8, "blue": 0.8}
}
})

# Цветовая шкала от красного к зеленому
worksheet.conditional_format("C2:C100", {
"type": "GRADIENT",
"minpoint": {"type": "MIN", "color": {"red": 1}},
"midpoint": {"type": "PERCENTILE", "value": "50", "color": {"red": 1, "green": 1}},
"maxpoint": {"type": "MAX", "color": {"green": 1}}
})

# Добавление простой формулы суммы
worksheet.update("D2", "=SUM(A2:C2)")

# Добавление формул для диапазона ячеек
formulas = [["=SUM(A2:C2)"], ["=AVERAGE(A3:C3)"], ["=MAX(A4:C4)"]]
worksheet.update("D2:D4", formulas)

# Сложные формулы с условиями
worksheet.update("E2", '=IF(D2>1000, "Превышение", "Норма")')

# Создание столбчатой диаграммы
chart = {
"chartType": "COLUMN",
"source": {
"sheetId": worksheet.id,
"startRowIndex": 1,
"endRowIndex": 10,
"startColumnIndex": 0,
"endColumnIndex": 3
},
"title": "Анализ продаж по регионам"
}

# Для добавления диаграммы необходимо использовать spreadsheet.batch_update()
spreadsheet.batch_update({
"requests": [
{
"addChart": {
"chart": {
"spec": chart,
"position": {
"overlayPosition": {
"anchorCell": {
"sheetId": worksheet.id,
"rowIndex": 15,
"columnIndex": 0
}
}
}
}
}
}
]
})

import pandas as pd
import numpy as np
from scipy import stats

# Получение данных для анализа
data = worksheet.get_all_values()
headers = data[0]
values = data[1:]

# Преобразование в pandas DataFrame
df = pd.DataFrame(values, columns=headers)
df = df.apply(pd.to_numeric, errors='ignore') # Преобразование числовых столбцов

# Статистический анализ
stats_data = [
["Статистический показатель", "Значение 1", "Значение 2"],
["Среднее", df['Колонка1'].mean(), df['Колонка2'].mean()],
["Медиана", df['Колонка1'].median(), df['Колонка2'].median()],
["Стандартное отклонение", df['Колонка1'].std(), df['Колонка2'].std()],
["Корреляция", stats.pearsonr(df['Колонка1'], df['Колонка2'])[0], ""]
]

# Создание нового листа для статистики
stats_sheet = spreadsheet.add_worksheet(title="Статистика", rows=10, cols=3)
stats_sheet.update("A1:C5", stats_data)

# Форматирование листа со статистикой
stats_sheet.format("A1:C1", {"textFormat": {"bold": True}})

def consolidate_reports():
# Открываем исходные таблицы
sales_sheet = client.open('Продажи').sheet1
inventory_sheet = client.open('Склад').sheet1
marketing_sheet = client.open('Маркетинг').sheet1

# Получаем данные из каждой таблицы
sales_data = pd.DataFrame(sales_sheet.get_all_records())
inventory_data = pd.DataFrame(inventory_sheet.get_all_records())
marketing_data = pd.DataFrame(marketing_sheet.get_all_records())

# Обработка данных – например, объединение по коду продукта
result = pd.merge(sales_data, inventory_data, on='product_id', how='left')
result = pd.merge(result, marketing_data, on='campaign_id', how='left')

# Создаем новый отчет
summary_sheet = client.create(f'Сводный отчет {datetime.now().strftime("%Y-%m-%d")}')
worksheet = summary_sheet.sheet1

# Загружаем результаты
worksheet.update([result.columns.tolist()] + result.values.tolist())

# Форматирование и оформление
worksheet.format("A1:Z1", {"textFormat": {"bold": True}})

# Отправка уведомления об обновлении
summary_sheet.share('manager@company.com', perm_type='user', role='reader', 
notify=True, email_message="Обновленный сводный отчет готов!")

return summary_sheet.url

# Запуск задачи по расписанию
if __name__ == "__main__":
report_url = consolidate_reports()
print(f"Отчет доступен по ссылке: {report_url}")

import requests
from datetime import datetime

def update_exchange_rates():
# Получаем актуальные курсы валют из API
response = requests.get('https://api.exchangerate-api.com/v4/latest/USD')
rates = response.json()['rates']

# Открываем существующую таблицу с финансовыми показателями
finance_sheet = client.open('Финансовый мониторинг').worksheet('Курсы валют')

# Подготовка данных для загрузки
date_now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
currencies = ['EUR', 'GBP', 'JPY', 'CNY', 'RUB']
rates_data = [[date_now] + [rates[currency] for currency in currencies]]

# Добавляем новые данные поверх старых
existing_data = finance_sheet.get_all_values()
if len(existing_data) > 1000: # Ограничиваем историю
finance_sheet.delete_rows(2, len(existing_data) – 999) # Оставляем заголовок и 999 строк

# Вставляем новые данные после заголовка
finance_sheet.insert_rows(rates_data, 2)

# Обновляем графики курсов валют
update_currency_charts(finance_sheet)

return date_now

def system_health_check():
# Проверяем доступность различных сервисов
services = [
{'name': 'Website', 'url': 'https://example.com', 'expected_status': 200},
{'name': 'API', 'url': 'https://api.example.com/status', 'expected_status': 200},
{'name': 'Admin Panel', 'url': 'https://admin.example.com', 'expected_status': 200},
]

results = []
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')

for service in services:
try:
response = requests.get(service['url'], timeout=5)
status = 'OK' if response.status_code == service['expected_status'] else 'FAIL'
response_time = response.elapsed.total_seconds()
except Exception as e:
status = 'ERROR'
response_time = 0

results.append([timestamp, service['name'], status, response_time])

# Обновляем таблицу мониторинга
monitoring_sheet = client.open('System Monitoring').sheet1
monitoring_sheet.append_rows(results)

# Помечаем проблемные сервисы
highlight_errors(monitoring_sheet)

# Отправляем оповещение при обнаружении проблем
if any(row[2] != 'OK' for row in results):
send_alert(results)

return results

def process_new_orders():
# Получаем новые заказы из таблицы
orders_sheet = client.open('Заказы').worksheet('Новые')
new_orders = orders_sheet.get_all_records()

# Обрабатываем каждый заказ
for order in new_orders:
# Проверяем наличие на складе через API складской системы
inventory_status = check_inventory(order['product_id'], order['quantity'])

if inventory_status['available']:
# Резервируем товар
reserve_result = reserve_product(order['product_id'], order['quantity'])

# Создаем накладную в системе доставки
delivery_result = create_delivery(order['customer_address'], order['product_id'], 
order['quantity'])

# Обновляем статус заказа
update_order_status(orders_sheet, order['order_id'], 'PROCESSING')

# Добавляем информацию в таблицу "В обработке"
processing_sheet = client.open('Заказы').worksheet('В обработке')
processing_sheet.append_row([
order['order_id'], 
order['customer_name'], 
order['product_id'],
order['quantity'],
delivery_result['tracking_number'],
datetime.now().strftime('%Y-%m-%d')
])
else:
# Обновляем статус заказа
update_order_status(orders_sheet, order['order_id'], 'WAITING')

# Отправляем уведомление о задержке
send_delay_notification(order['customer_email'], inventory_status['estimated_restock'])

# Очищаем обработанные заказы из таблицы "Новые"
clear_processed_orders(orders_sheet)

import schedule
import time

# Настраиваем расписание выполнения задач
schedule.every().day.at("08:00").do(update_exchange_rates)
schedule.every(15).minutes.do(system_health_check)
schedule.every().monday.do(consolidate_reports)

# Запускаем бесконечный цикл для выполнения задач
while True:
schedule.run_pending()
time.sleep(60) # Проверяем расписание каждую минуту

import logging

logging.basicConfig(filename='spreadsheet_automation.log', level=logging.INFO,
format='%(asctime)s – %(levelname)s – %(message)s')

def safe_automation_task(task_func):
try:
result = task_func()
logging.info(f"Задача {task_func.__name__} выполнена успешно: {result}")
return result
except Exception as e:
logging.error(f"Ошибка при выполнении задачи {task_func.__name__}: {str(e)}")
send_admin_alert(task_func.__name__, str(e))
return None