task-parser / src /google_calendar.py
Artem Nikolaev
PDF Task Parser with FastAPI and Google integration
ca8ebf7
from google.oauth2 import service_account
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
import datetime
import os
import sys
class GoogleCalendarExporter:
"""
Класс для экспорта задач в Google Calendar
"""
def __init__(self, credentials_path='credentials/google-credentials.json', calendar_id='primary'):
"""
Инициализация подключения к Google Calendar
Args:
credentials_path: путь к JSON-файлу с ключами сервисного аккаунта
calendar_id: ID календаря ('primary' для основного или конкретный ID)
"""
self.credentials_path = credentials_path
self.calendar_id = calendar_id
self.service = None
# Проверяем наличие файла с ключами
if not os.path.exists(credentials_path):
print(f"❌ Файл с ключами не найден: {credentials_path}")
print("💡 Убедитесь, что файл лежит в папке credentials/")
sys.exit(1)
self._authenticate()
def _authenticate(self):
"""Аутентификация в Google Calendar API через сервисный аккаунт"""
try:
# Определяем права доступа (нужны для записи)
SCOPES = ['https://www.googleapis.com/auth/calendar'] # Полный доступ к календарю [citation:6]
# Загружаем ключи сервисного аккаунта [citation:5]
credentials = service_account.Credentials.from_service_account_file(
self.credentials_path,
scopes=SCOPES
)
# Создаем сервис для работы с Calendar API [citation:6]
self.service = build('calendar', 'v3', credentials=credentials)
print("✅ Успешная аутентификация в Google Calendar")
except Exception as e:
print(f"❌ Ошибка аутентификации: {e}")
sys.exit(1)
def create_event_from_task(self, task):
"""
Создает событие в календаре из задачи
Args:
task: словарь с данными задачи (number, summary, full_description, responsible, due_date, due_date_str)
Returns:
ссылка на созданное событие или None при ошибке
"""
if not self.service:
print("❌ Сервис не инициализирован")
return None
# Проверяем наличие даты
if not task.get('due_date'):
print(f"⚠️ Задача #{task.get('number', '?')} пропущена: нет даты")
return None
try:
# Формируем событие
due_date = task['due_date']
# Создаем событие на целый день (если не указано время) [citation:6]
# Или можно задать конкретное время, например 10:00
event_date = due_date.strftime('%Y-%m-%d')
# Берем краткое описание или начало полного
summary = task.get('summary', '')
if not summary:
summary = task.get('full_description', '')[:50] + '...'
# Формируем описание события
description = f"""
📋 Задача #{task.get('number', '?')}
📝 Полное описание:
{task.get('full_description', '')}
👤 Ответственный: {task.get('responsible', 'Не указан')}
📅 Срок: {task.get('due_date_str', '')}
🔗 Создано автоматически парсером задач
""".strip()
# Создаем событие [citation:6]
event = {
'summary': f"Задача #{task['number']}: {summary}",
'description': description,
'start': {
'date': event_date, # Целый день
},
'end': {
'date': event_date, # Целый день
},
'reminders': {
'useDefault': True # Использовать стандартные напоминания
}
}
# Добавляем время, если нужно (например, сделать на 10:00)
# event['start']['dateTime'] = f"{event_date}T10:00:00+03:00"
# event['end']['dateTime'] = f"{event_date}T11:00:00+03:00"
# Создаем событие в календаре [citation:1]
created_event = self.service.events().insert(
calendarId=self.calendar_id,
body=event
).execute()
print(f"✅ Событие создано: {created_event.get('htmlLink')}")
return created_event.get('htmlLink')
except HttpError as e:
print(f"❌ Ошибка API при создании события для задачи #{task.get('number', '?')}: {e}")
return None
except Exception as e:
print(f"❌ Неожиданная ошибка для задачи #{task.get('number', '?')}: {e}")
return None
def create_events_from_tasks(self, tasks):
"""
Создает события для списка задач
Args:
tasks: список словарей с задачами
Returns:
список ссылок на созданные события
"""
results = []
print(f"\n📅 Создание событий в календаре (ID: {self.calendar_id})...")
for task in tasks:
event_link = self.create_event_from_task(task)
if event_link:
results.append({
'task_number': task.get('number'),
'task_summary': task.get('summary', '')[:30] + '...',
'event_link': event_link
})
print(f"\n✅ Создано {len(results)} событий из {len(tasks)} задач")
return results
def check_calendar_access(self):
"""Проверяет доступ к календарю (выводит список ближайших событий)"""
try:
now = datetime.datetime.utcnow().isoformat() + 'Z'
events_result = self.service.events().list(
calendarId=self.calendar_id,
timeMin=now,
maxResults=5,
singleEvents=True,
orderBy='startTime'
).execute()
events = events_result.get('items', [])
if not events:
print("📭 В календаре нет предстоящих событий")
else:
print(f"📅 Ближайшие события в календаре:")
for event in events:
start = event['start'].get('dateTime', event['start'].get('date'))
print(f" • {start}: {event.get('summary', 'Без названия')}")
return True
except HttpError as e:
print(f"❌ Ошибка доступа к календарю: {e}")
print("💡 Проверьте, что:")
print(" 1. Calendar API включен в Google Cloud Console")
print(" 2. Календарь расшарен на email сервисного аккаунта")
print(" 3. Calendar ID указан правильно")
return False