`GoogleDriveService`

.env.template

@@ -3,15 +3,9 @@
 # APScheduler Configuration
 SCHEDULER_TIMEZONE=UTC
 
-# API Configuration
-API_HOST=0.0.0.0
-API_PORT=8000
-
 # Slack Configuration
 SLACK_BOT_TOKEN=🪙
-SLACK_SIGNING_SECRET=🔐
 SLACK_APP_TOKEN=🦥
-SLACK_USER_TOKEN=🦊
 
 # Vectorization Configuration
 EMBEDDING_MODEL=🌮
@@ -34,3 +28,10 @@ CHAT_MODEL=gpt-3.5-turbo
 MAX_TOKENS=150
 TEMPERATURE=0.8
 SYSTEM_PROMPT="You are a helpful teaching assistant for a data science class.\nBased on the students question, you will be given context retreived from class transcripts and materials to answer their question.\nYour responses should be:\n\n1. Accurate and based on the class content\n2. Clear and educational\n3. Concise but complete\nIf you're unsure about something, acknowledge it and suggest asking the professor."
+
+# Google Drive Configuration
+GOOGLE_PROJECT_ID=insufferable-slacker-123456
+GOOGLE_PRIVATE_KEY_ID=1a2b3c4d5e6f748891091d21304e506674829507
+GOOGLE_PRIVATE_KEY="-----BEGIN PRIVATE KEY-----\nMIIEvQIBADANBgkqhkiG9w0BAQEFAASC...\n-----END PRIVATE KEY-----\n"
+GOOGLE_CLIENT_EMAIL=botty-bot@insufferable-slacker-123456.iam.gserviceaccount.com
+GOOGLE_CLIENT_ID=123456789012345678901

pyproject.toml

@@ -36,7 +36,9 @@ dependencies = [
     "slack_bolt>=1.23.0",
     "motor>=3.7.0",
     "openai>=1.70.0"
-
+    "google-api-python-client>=2.167.0",
+    "google-auth>=2.39.0",
+    "google-auth-oauthlib>=1.2.1"
 ]
 
 [project.optional-dependencies]

src/ctp_slack_bot/core/config.py

@@ -39,7 +39,7 @@ class Settings(BaseSettings):
     SCORE_THRESHOLD: NonNegativeFloat
 
     # Hugging Face Configuration
-    HF_API_TOKEN: Optional[SecretStr] = None
+    HF_API_TOKEN: Optional[SecretStr] = None # TODO: Currently, this is unused.
 
     # OpenAI Configuration
     OPENAI_API_KEY: SecretStr
@@ -48,6 +48,18 @@ class Settings(BaseSettings):
     TEMPERATURE: NonNegativeFloat
     SYSTEM_PROMPT: str
 
+    # Google Drive Configuration
+    GOOGLE_PROJECT_ID: str
+    GOOGLE_PRIVATE_KEY_ID: SecretStr
+    GOOGLE_PRIVATE_KEY: SecretStr
+    GOOGLE_CLIENT_ID: str
+    GOOGLE_CLIENT_EMAIL: str
+    GOOGLE_AUTH_URI: str = "https://accounts.google.com/o/oauth2/auth"
+    GOOGLE_TOKEN_URI: str = "https://oauth2.googleapis.com/token"
+    GOOGLE_AUTH_PROVIDER_CERT_URL: str = "https://www.googleapis.com/oauth2/v1/certs"
+    GOOGLE_CLIENT_CERT_URL: str = "https://www.googleapis.com/robot/v1/metadata/x509/ctp-slack-bot-714%40voltaic-reducer-294821.iam.gserviceaccount.com"
+    GOOGLE_UNIVERSE_DOMAIN: str = "googleapis.com"
+
     model_config = SettingsConfigDict(
         env_file=".env",
         env_file_encoding="utf-8",

src/ctp_slack_bot/services/google_drive_service.py

@@ -0,0 +1,144 @@
+from datetime import datetime
+from google.oauth2 import service_account
+from googleapiclient.discovery import build
+from googleapiclient.http import MediaIoBaseDownload
+from googleapiclient.errors import HttpError
+from io import BytesIO
+from loguru import logger
+from pydantic import BaseModel, PrivateAttr
+from typing import Collection, Dict, List, Optional, Self
+
+from ctp_slack_bot.core import Settings
+
+FOLDER_MIME_TYPE: str = "application/vnd.google-apps.folder"
+ROOT_FOLDER_NAME: str = "root"
+
+class GoogleDriveService(BaseModel):
+    """Service for interacting with Google Drive."""
+
+    settings: Settings
+    _google_drive_client: PrivateAttr = PrivateAttr()
+    _folder_cache: PrivateAttr = PrivateAttr(default_factory=dict)
+
+    class Config:
+        frozen=True
+
+    def __init__(self: Self, **data) -> None:
+        super().__init__(**data)
+        credentials = service_account.Credentials.from_service_account_info({
+            "type": "service_account",
+            "project_id": self.settings.GOOGLE_PROJECT_ID,
+            "private_key_id": self.settings.GOOGLE_PRIVATE_KEY_ID,
+            "private_key": self.settings.GOOGLE_PRIVATE_KEY,
+            "client_email": self.settings.GOOGLE_CLIENT_EMAIL,
+            "client_id": self.settings.GOOGLE_CLIENT_ID,
+            "token_uri": self.settings.GOOGLE_TOKEN_URI,
+        }, scopes=['https://www.googleapis.com/auth/drive'])
+        self._google_drive_client = build('drive', 'v3', credentials=credentials)
+        logger.debug("Created {}", self.__class__.__name__)
+
+    def _resolve_folder_id(self: Self, folder_path: str) -> Optional[str]:
+        """Resolve a folder path to a Google Drive ID."""
+
+        if not folder_path:
+            return ROOT_FOLDER_NAME
+            
+        if folder_path in self._folder_cache:
+            return self._folder_cache[folder_path]
+
+        current_id = ROOT_FOLDER_NAME
+        for part in folder_path.split('/'):
+            if not part:
+                continue
+            try:
+                results = self._google_drive_client.files().list(
+                    q=f"name='{part}' and mimeType='{FOLDER_MIME_TYPE}' and '{current_id}' in parents",
+                    fields='files(id,name)',
+                    supportsAllDrives=True,
+                    includeItemsFromAllDrives=True
+                ).execute()
+                if not results.get('files'):
+                    return None
+                current_id = results['files'][0]['id']
+            except HttpError as e:
+                logger.error("Error resolving folder path: {}", folder_path)
+                return None
+
+        self._folder_cache[folder_path] = current_id
+        return current_id
+
+    def list_directory(self: Self, folder_path: str) -> List[Dict]:
+        """List contents of a directory with basic metadata."""
+
+        folder_id = self._resolve_folder_id(folder_path)
+        if not folder_id:
+            return []
+        try:
+            results = self._google_drive_client.files().list(
+                q=f"'{folder_id}' in parents",
+                fields='files(id,name,mimeType,modifiedTime)',
+                supportsAllDrives=True,
+                includeItemsFromAllDrives=True,
+                pageSize=1000
+            ).execute()
+            
+            return [{
+                'id': f['id'],
+                'name': f['name'],
+                'modified': f['modifiedTime'],
+                'mime_type': f['mimeType']
+            } for f in results.get('files', [])]
+        except HttpError as e:
+            logger.error("Error listing folder by path, {}: {}", folder_path, e)
+            return []
+
+    def get_metadata(self: Self, item_path: str) -> Optional[Dict]:
+        """Get metadata for a specific file/folder by path."""
+
+        if '/' not in item_path:
+            folder_id = ROOT_FOLDER_NAME
+            item_name = item_path
+        else:
+            parts = item_path.split('/')
+            item_name = parts[-1]
+            folder_path = '/'.join(parts[:-1])
+            folder_id = self._resolve_folder_id(folder_path)
+            
+        if not folder_id:
+            return None
+            
+        try:
+            results = self._google_drive_client.files().list(
+                q=f"name='{item_name}' and '{folder_id}' in parents",
+                fields='files(id,name,mimeType,modifiedTime)',
+                supportsAllDrives=True,
+                includeItemsFromAllDrives=True,
+                pageSize=1
+            ).execute()
+            
+            if files := results.get('files'):
+                return {
+                    'id': files[0]['id'],
+                    'name': files[0]['name'],
+                    'modified': files[0]['modifiedTime'],
+                    'mime_type': files[0]['mimeType']
+                }
+        except HttpError as e:
+            logger.error("Error getting metadata for item by path, {}: {}", item_path, e)
+
+        return None
+
+    def read_file_by_id(self: Self, file_id: str) -> Optional[bytes]:
+        """Read contents of a file by its unique identifier."""
+
+        try:
+            request = self.service.files().get_media(fileId=file_id)
+            buffer = BytesIO()
+            downloader = MediaIoBaseDownload(buffer, request)
+            done = False
+            while not done:
+                _, done = downloader.next_chunk()
+            return buffer.getvalue()
+        except HttpError as e:
+            logger.error("Error reading file by ID, {}: {}", file_id, e)
+            return None