feat: add Google Drive upload fallback and cleanup support

- Add drive_utils.py with upload_file_to_drive() and delete_file_from_drive()
- Modify upload_file_to_gcs() to fallback to Drive on GCS failure (default: enabled)
- Update cleanup_manager.py to handle Drive URL deletions
- Rename 'Hook' column to 'Source Data' in Google Sheets logs
- Remove 'Success' column from logs (all logged entries are successful)
- Export drive utilities in google_src/__init__.py

src/cleanup_manager.py

@@ -3,6 +3,7 @@ import os
 from utils import logger
 from google_sheet_reader import GoogleSheetReader
 from google_src.gcs_utils import delete_gcs_file
+from google_src.drive_utils import delete_file_from_drive
 
 def extract_blob_name_from_url(url: str) -> str | None:
     """
@@ -19,10 +20,14 @@ def extract_blob_name_from_url(url: str) -> str | None:
         return "/".join(parts[2:]) # Skip domain and bucket
     return None
 
+def is_drive_url(url: str) -> bool:
+    """Check if URL is a Google Drive URL."""
+    return url and "drive.google.com" in url
+
 async def process_delete_entries():
     """
     Check Google Sheet logs for entries marked 'Delete Entry' = TRUE.
-    Delete corresponding video files from GCS and then remove the row from the sheet.
+    Delete corresponding video files from GCS or Drive and then remove the row from the sheet.
     """
     gsheet_name = os.getenv("GSHEET_WORKSHEET_LOGS")
     if not gsheet_name:
@@ -49,6 +54,7 @@ async def process_delete_entries():
             delete_flag = str(row["Delete Entry"]).strip().lower()
             if delete_flag == "true":
                 gcs_filename = ""
+                final_url = ""
                 
                 # Check GCS Filename column first
                 if has_filename:
@@ -57,16 +63,19 @@ async def process_delete_entries():
                     if "/" in val or "." in val:
                        gcs_filename = val
                 
-                # Fallback to URL if filename is empty or looks like just a hash (no slash/dot)
-                if (not gcs_filename) and has_url:
+                # Get URL for Drive detection
+                if has_url:
                     final_url = str(row["Final URL"]).strip()
+                
+                # Fallback to URL if filename is empty or looks like just a hash (no slash/dot)
+                if (not gcs_filename) and final_url:
                     extracted = extract_blob_name_from_url(final_url)
                     if extracted:
                         logger.info(f"Derived GCS filename from URL: {extracted}")
                         gcs_filename = extracted
                 
-                if gcs_filename:
-                    files_to_delete.append((idx + 2, gcs_filename)) 
+                if gcs_filename or final_url:
+                    files_to_delete.append((idx + 2, gcs_filename, final_url)) 
                 else:
                     # If no filename found, just mark row for deletion
                     to_delete_indices.append(idx + 2)
@@ -78,14 +87,23 @@ async def process_delete_entries():
         logger.info(f"Found {len(files_to_delete) + len(to_delete_indices)} entries to delete.")
         
         # Process file deletions
-        for sheet_row_idx, filename in files_to_delete:
-            # Check if filename is suspiciouly short (like just a hash) to avoid deleting buckets or wrong things
-            if len(filename) < 5 or "." not in filename:
+        for sheet_row_idx, filename, url in files_to_delete:
+            # Check if it's a Drive URL - delete from Drive
+            if is_drive_url(url):
+                success = delete_file_from_drive(url)
+                if success:
+                    to_delete_indices.append(sheet_row_idx)
+                else:
+                    logger.error(f"Skipping row deletion for row {sheet_row_idx} due to Drive file deletion failure.")
+                continue
+            
+            # Otherwise, delete from GCS
+            # Check if filename is suspiciously short (like just a hash) to avoid deleting buckets or wrong things
+            if not filename or len(filename) < 5 or "." not in filename:
                 logger.warning(f"⚠️ Filename '{filename}' looks invalid. Skipping file deletion, but will delete row.")
                 to_delete_indices.append(sheet_row_idx)
                 continue
 
-
             success = delete_gcs_file(filename)
             if success:
                 to_delete_indices.append(sheet_row_idx)

src/google_src/__init__.py

@@ -12,6 +12,8 @@ from .gcloud_wrapper import (
     get_default_wrapper,
 )
 
+from .drive_utils import upload_file_to_drive, delete_file_from_drive
+
 __all__ = [
     "GCloudWrapper",
     "GCloudAccount",
@@ -19,4 +21,7 @@ __all__ = [
     "create_wrapper_from_env",
     "create_default_wrapper",
     "get_default_wrapper",
+    "upload_file_to_drive",
+    "delete_file_from_drive",
 ]
+

src/google_src/drive_utils.py

@@ -0,0 +1,206 @@
+"""
+Google Drive utilities for centralized upload operations.
+Uses test_data credentials for Drive uploads as fallback when GCS fails.
+"""
+import os
+import uuid
+from pathlib import Path
+from typing import Optional
+from utils import logger
+from .gcloud_wrapper import get_default_wrapper
+
+# MIME type mappings for common file extensions
+MIME_TYPES = {
+    ".mp4": "video/mp4",
+    ".mov": "video/quicktime",
+    ".avi": "video/x-msvideo",
+    ".mkv": "video/x-matroska",
+    ".mp3": "audio/mpeg",
+    ".wav": "audio/wav",
+    ".aac": "audio/aac",
+    ".m4a": "audio/mp4",
+    ".png": "image/png",
+    ".jpg": "image/jpeg",
+    ".jpeg": "image/jpeg",
+    ".gif": "image/gif",
+    ".webp": "image/webp",
+    ".json": "application/json",
+    ".txt": "text/plain",
+    ".srt": "text/plain",
+}
+
+
+def get_drive_service(account_name: str = "test_data"):
+    """
+    Get a configured Google Drive service for the specified account.
+    Defaults to 'test_data' for uploads.
+    """
+    wrapper = get_default_wrapper()
+    creds = wrapper._get_credentials(account_name)
+    
+    from googleapiclient.discovery import build
+    return build("drive", "v3", credentials=creds)
+
+
+def upload_file_to_drive(
+    local_path: str,
+    folder_id: Optional[str] = None,
+    account_name: str = "test_data",
+) -> dict:
+    """
+    Upload a local file to Google Drive.
+    
+    Args:
+        local_path: Path to local file.
+        folder_id: Optional Drive folder ID to upload to. If None, uses DRIVE_UPLOAD_FOLDER_ID env var.
+        account_name: Account to use (default: test_data).
+        
+    Returns:
+        dict: {
+            "file_id": str,
+            "url": str (viewable link),
+            "download_url": str (direct download link)
+        }
+    """
+    from googleapiclient.http import MediaFileUpload
+    
+    try:
+        service = get_drive_service(account_name)
+        
+        # Determine folder ID
+        target_folder_id = folder_id or os.getenv("DRIVE_UPLOAD_FOLDER_ID")
+        
+        # Get file info
+        file_path = Path(local_path)
+        file_ext = file_path.suffix.lower()
+        
+        # Generate unique filename to avoid conflicts
+        unique_name = f"{uuid.uuid4().hex[:8]}_{file_path.name}"
+        
+        # Get MIME type
+        mime_type = MIME_TYPES.get(file_ext, "application/octet-stream")
+        
+        # Prepare file metadata
+        file_metadata = {
+            "name": unique_name,
+        }
+        
+        if target_folder_id:
+            file_metadata["parents"] = [target_folder_id]
+        
+        # Create media upload object
+        media = MediaFileUpload(
+            local_path,
+            mimetype=mime_type,
+            resumable=True
+        )
+        
+        logger.info(f"📤 Uploading {file_path.name} to Google Drive...")
+        
+        # Execute upload
+        file = service.files().create(
+            body=file_metadata,
+            media_body=media,
+            fields="id, name, webViewLink, webContentLink"
+        ).execute()
+        
+        file_id = file.get("id")
+        
+        # Make file publicly accessible (anyone with link can view)
+        try:
+            service.permissions().create(
+                fileId=file_id,
+                body={
+                    "type": "anyone",
+                    "role": "reader"
+                }
+            ).execute()
+            logger.info(f"✅ File shared publicly")
+        except Exception as e:
+            logger.warning(f"⚠️ Could not make file public: {e}")
+        
+        # Construct URLs
+        view_url = file.get("webViewLink", f"https://drive.google.com/file/d/{file_id}/view")
+        download_url = f"https://drive.google.com/uc?export=download&id={file_id}"
+        
+        result = {
+            "file_id": file_id,
+            "filename": unique_name,
+            "gcs_filename": f"drive/{unique_name}",  # For compatibility with GCS result format
+            "url": view_url,
+            "public_url": view_url,
+            "download_url": download_url,
+        }
+        
+        logger.info(f"✅ Uploaded to Drive: {view_url}")
+        return result
+        
+    except Exception as e:
+        logger.error(f"❌ Failed to upload to Drive: {e}")
+        raise e
+
+
+def extract_drive_file_id(url: str) -> str | None:
+    """
+    Extract file ID from various Google Drive URL formats:
+    - https://drive.google.com/file/d/FILE_ID/view
+    - https://drive.google.com/open?id=FILE_ID
+    - https://drive.google.com/uc?id=FILE_ID
+    - https://drive.google.com/uc?export=download&id=FILE_ID
+    """
+    import re
+    
+    if not url or "drive.google.com" not in url:
+        return None
+    
+    # Pattern 1: /file/d/FILE_ID/
+    match = re.search(r'/file/d/([a-zA-Z0-9_-]+)', url)
+    if match:
+        return match.group(1)
+
+    # Pattern 2: ?id=FILE_ID or &id=FILE_ID
+    match = re.search(r'[?&]id=([a-zA-Z0-9_-]+)', url)
+    if match:
+        return match.group(1)
+
+    return None
+
+
+def delete_file_from_drive(
+    url_or_file_id: str,
+    account_name: str = "test_data",
+) -> bool:
+    """
+    Delete a file from Google Drive.
+    
+    Args:
+        url_or_file_id: Either a Drive URL or direct file ID.
+        account_name: Account to use (default: test_data).
+        
+    Returns:
+        bool: True if deleted successfully, False otherwise.
+    """
+    try:
+        service = get_drive_service(account_name)
+        
+        # Extract file ID if URL was provided
+        if url_or_file_id.startswith("http"):
+            file_id = extract_drive_file_id(url_or_file_id)
+            if not file_id:
+                logger.error(f"❌ Could not extract file ID from URL: {url_or_file_id}")
+                return False
+        else:
+            file_id = url_or_file_id
+        
+        logger.info(f"🗑️ Deleting file from Drive: {file_id}")
+        service.files().delete(fileId=file_id).execute()
+        logger.info(f"✅ Deleted from Drive: {file_id}")
+        return True
+        
+    except Exception as e:
+        error_str = str(e)
+        if "404" in error_str or "not found" in error_str.lower():
+            logger.info(f"⚠️ File not found (already deleted?): {url_or_file_id}")
+            return True  # Consider it success if file doesn't exist
+        logger.error(f"❌ Failed to delete from Drive: {e}")
+        return False

src/google_src/gcs_utils.py

@@ -71,7 +71,8 @@ def delete_gcs_file(filename: str, bucket_name: Optional[str] = None, account_na
 def upload_file_to_gcs(
     local_path: str,
     account_name: str = "final_data",
-    generate_signed_url: bool = True
+    generate_signed_url: bool = True,
+    fallback_to_drive: bool = True,
 ) -> dict:
     """
     Upload a local file to GCS.
@@ -82,12 +83,14 @@ def upload_file_to_gcs(
         bucket_name: Target bucket name.
         account_name: Account to use.
         generate_signed_url: Whether to generate a signed URL.
+        fallback_to_drive: If True, fallback to Google Drive upload on GCS failure.
         
     Returns:
         dict: {
             "gcs_filename": str,
             "url": str (signed or public),
-            "public_url": str
+            "public_url": str,
+            "storage_type": str ("gcs" or "drive")
         }
     """
     import os
@@ -135,7 +138,8 @@ def upload_file_to_gcs(
         result = {
             "gcs_filename": blob_name,
             "public_url": public_url,
-            "url": public_url
+            "url": public_url,
+            "storage_type": "gcs"
         }
         
         if generate_signed_url:
@@ -156,6 +160,20 @@ def upload_file_to_gcs(
 
     except Exception as e:
         logger.error(f"❌ Failed to upload to GCS: {e}")
+        
+        # Fallback to Google Drive if enabled
+        if fallback_to_drive:
+            logger.info("🔄 Falling back to Google Drive upload...")
+            try:
+                from .drive_utils import upload_file_to_drive
+                drive_result = upload_file_to_drive(local_path, account_name="test_data")
+                drive_result["storage_type"] = "drive"
+                logger.info(f"✅ Fallback to Drive successful: {drive_result['url']}")
+                return drive_result
+            except Exception as drive_error:
+                logger.error(f"❌ Drive fallback also failed: {drive_error}")
+                raise e  # Re-raise original GCS error
+        
         # Re-raise to let caller handle critical failure
         raise e
 

src/process_csv.py

@@ -27,10 +27,10 @@ SHARED_API_CLIENTS = None  # Shared instance to avoid redundant GCS/TTS client i
 
 def load_executed_from_gsheet(setup_type=None, job_index=None):
     """Load executed scripts from Google Sheets logs.
-    Returns a list of TTS Scripts that have Success=True.
+    Returns a list of Source Data values (all entries are considered executed).
     
     Args:
-        setup_type: Optional. If provided, only count rows where Safe Name contains this value.
+        setup_type: Optional. If provided, only count rows where Final URL contains this value.
         job_index: Optional. If provided, only count rows for this specific job.
     """
     gsheet_name = os.getenv("GSHEET_WORKSHEET_LOGS")
@@ -43,23 +43,22 @@ def load_executed_from_gsheet(setup_type=None, job_index=None):
         reader = GoogleSheetReader(worksheet_name=gsheet_name)
         df = reader.get_dataframe()
         
-        # Filter for successful entries only
-        if "Success" in df.columns and "Hook" in df.columns:
-            successful = df[df["Success"].str.lower() == "true"]
+        if "Source Data" in df.columns:
+            result_df = df
             
             # Optional: filter by job_index
             if job_index is not None and "Job" in df.columns:
-                successful = successful[successful["Job"] == str(job_index)]
+                result_df = result_df[result_df["Job"] == str(job_index)]
             
             # Optional: filter by setup_type in Final URL
             if setup_type and "Final URL" in df.columns:
-                successful = successful[successful["Final URL"].str.contains(setup_type, na=False)]
+                result_df = result_df[result_df["Final URL"].str.contains(setup_type, na=False)]
             
-            executed = successful["Hook"].tolist()
+            executed = result_df["Source Data"].tolist()
             logger.info(f"Loaded {len(executed)} executed entries from Google Sheets")
             return executed
         else:
-            logger.warning("Google Sheets missing required columns (Success, Hook)")
+            logger.warning("Google Sheets missing required column (Source Data)")
             return []
     except Exception as e:
         logger.error(f"Failed to load from Google Sheets: {e}")
@@ -81,12 +80,11 @@ def log_progress_to_gsheet(tts_script: str, result: dict, job_index: int, commit
 
         reader.create_or_update_sheet(
             worksheet_name=gsheet_name,
-            header=["Timestamp", "Job", "Hook", "Success", "Final URL", "GCS Filename", "Delete Entry"],
+            header=["Timestamp", "Job", "Source Data", "Final URL", "GCS Filename", "Delete Entry"],
             values=[{
                 "Timestamp": datetime.now().isoformat(),
                 "Job": str(job_index if job_index is not None else 0),
-                "Hook": tts_script,
-                "Success": str(result.get("success", False)),
+                "Source Data": tts_script,
                 "Final URL": result.get("final_url", ""),
                 "GCS Filename": result.get("gcs_filename", ""),
                 "Delete Entry": "False",