Автоматизируем добавление запланирванных стримов youtube каналов в google календарь
Проблематика
На днях столкнулся с такой ситуацией для себя, что мне необходимо добавлять запланированные стримы с каналов YouTube в свой календарь. Я ленивый человек и делать это вручную не особо желаю, поэтому решил это дело автоматизировать.
Для начала определился, можно ли вообще получить такую информацию и был очень удивлен, что у google можно получить очень большой спектр различной информации от разных сервисов, спасибо им большое за предоставленную возможность.
Дальше нашел уже готовую библиотеку по работе с сервисами google и подготовил необходимый алгоритм. Алгоритм состоял из таких шагов:
- получить список запланированных стримов
- записать стримы в основной календарь как запланированное мероприятие
Реализация
Итак, создаю необходимые переменные окружения:
# Настройки
YOUTUBE_API_KEY = os.getenv("YOUTUBE_API_KEY")
YOUTUBE_CHANNEL_ID = os.getenv("YOUTUBE_CHANNEL_ID")
GOOGLE_CALENDAR_ID = os.getenv("GOOGLE_CALENDAR_ID")
TELEGRAM_CHANNEL_ID = os.getenv("TELEGRAM_CHANNEL_ID")
TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")переменные окружения
GOOGLE_CALENDAR_ID - идентификатор канала с которого будут считываться запланированные стримы
Создаю сервис календаря с которым буду работать. По авторизации google приложений есть в документации, подробно останавливаться не буду.
# Авторизация в Google Calendar
creds = Credentials.from_authorized_user_file('м', ['https://www.googleapis.com/auth/calendar'])
service = build('calendar', 'v3', credentials=creds)авторизация
token.json - файл который появится после авторизации
Ошибки получаемые при запросах я буду отправлять в телеграм, мне так удобнее(все опционально)
def send_telegram_alert(message: str):
"""Отправляет сообщение об ошибке в Telegram канал."""
try:
url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
payload = {
"chat_id": TELEGRAM_CHANNEL_ID,
"text": message,
"disable_notification": False
}
response = requests.post(url, json=payload)
response.raise_for_status()
except Exception as e:
print(f"⚠️ Ошибка при отправке в Telegram: {str(e)}")функция отправки сообщений в канал
И теперь основные шаги алгоритма:
- получаю предстоящие стримы канала
def get_upcoming_streams():
"""Получает предстоящие стримы с YouTube API с обработкой ошибок."""
try:
# Формируем запрос
url = "https://www.googleapis.com/youtube/v3/search"
params = {
"part": "snippet",
"channelId": YOUTUBE_CHANNEL_ID,
"eventType": "upcoming",
"type": "video",
"key": YOUTUBE_API_KEY
}
# Выполняем запрос
response = requests.get(url, params=params)
response.raise_for_status() # Проверка HTTP ошибок
data = response.json()
# Проверка ошибок API
if "error" in data:
error_msg = data["error"].get("message", "Неизвестная ошибка API")
raise RuntimeError(f"Ошибка Youtube API: {error_msg}")
return data.get("items", [])
except (HTTPError, ConnectionError, Timeout) as e:
error_msg = f"🚨 Сетевая ошибка: {str(e)}"
print(error_msg)
send_telegram_alert(error_msg)
except Exception as e:
error_msg = f"⚠️ Неизвестная ошибка: {str(e)}"
print(error_msg)
send_telegram_alert(error_msg)
return [] # Возвращаем пустой список в случае ошибки
функция получения предстоящих стримов
- записываю стримы в календарь
def create_calendar_event(stream):
"""Записываем в календарь стрим"""
try:
snippet = stream["snippet"]
start_time = datetime.fromisoformat(snippet["publishedAt"].replace("Z", "+03:00"))
event = {
"summary": f"Стрим: {snippet['title']}",
"description": f"Ссылка: https://youtu.be/{stream['id']['videoId']}",
"start": {
"dateTime": start_time.isoformat(),
"timeZone": "UTC"
},
"end": {
"dateTime": (start_time + timedelta(hours=1)).isoformat(), # Предполагаемая длительность
"timeZone": "UTC"
},
"reminders": {
"useDefault": False,
"overrides": [
{"method": "popup", "minutes": 30} # Напоминание за 30 минут
]
}
}
service.events().insert(calendarId=GOOGLE_CALENDAR_ID, body=event).execute()
except HttpError as e:
error_msg = f"🚨 Ошибка Google Calendar API: {e.resp.status} {e.error_details}"
send_telegram_alert(error_msg)
except (ConnectionError, Timeout) as e:
error_msg = f"🚨 Сетевая ошибка: {str(e)}"
send_telegram_alert(error_msg)
except Exception as e:
error_msg = f"⚠️ Неизвестная ошибка: {str(e)}"
send_telegram_alert(error_msg)
функция записи стрима в календарь, создание события
Чтобы стримы не дублировались в календаре, я предусмотрел сохранение идентификаторов стрима в БД (sqlite). Можно использовать любой вариант для сохранения идентификаторов, выбирайте под себя.
Дополнительные функции для работы с БД
conn = sqlite3.connect("notifications.db")
cursor = conn.cursor()
cursor.execute("CREATE TABLE IF NOT EXISTS sent (video_id TEXT PRIMARY KEY)")
def is_notification_sent(video_id):
cursor.execute("SELECT 1 FROM sent WHERE video_id=?", (video_id,))
return cursor.fetchone() is not None
def mark_as_sent(video_id):
cursor.execute("INSERT INTO sent VALUES (?)", (video_id,))
conn.commit()инициализация БД и функции для работы
И сам запуск программы теперь:
# Запуск
streams = get_upcoming_streams()
for stream in streams:
if not is_notification_sent(stream["id"]["videoId"]):
create_calendar_event(stream)
mark_as_sent(stream["id"]["videoId"])основной цикл и проверка записей в БД
Дальше я использовал менеджер агентов Huginn для автоматического запуска этой программы по времени. Для этого я использую агента Shell для вызова скрипта python (делайте это на свой страх и риск, так как команда разработчиков huginn не рекомендует использовать этот тип агента), можно использовать механизм какой удобно. Настроил агента на запуск в 9 часов утра каждый день.
Дальше можно расширят функционал и задачи. Спасибо за просмотр!
Обсуждение участников