data flow via google sheet

requirements.txt

@@ -15,6 +15,7 @@ google-cloud-speech==2.34.0
 google-api-python-client==2.184.0
 google-auth-oauthlib==1.2.3
 librosa==0.11.0
+gspread
 
 # aiosignal==1.4.0
 # annotated-types==0.7.0

src/automation.py

@@ -381,6 +381,7 @@ class ContentAutomation:
         if os.getenv("INFLOXA", "false").lower() == "true":
             from video_downloader import VideoDownloader
             download_path="testData/infloxa"
+            Path(download_path).mkdir(parents=True, exist_ok=True)
 
             allowed_videos = []
 

src/get_refresh_token.py

@@ -1,43 +1,24 @@
-"""
-get_refresh_token.py
----------------------
-Use this script to generate a YouTube OAuth refresh token.
-Run locally, log in with the YouTube account you want to upload to.
-Then copy the printed REFRESH_TOKEN into your GitHub Secrets.
-"""
-
 from google_auth_oauthlib.flow import InstalledAppFlow
+import os
+from dotenv import load_dotenv
+load_dotenv()
 
-SCOPES = ['https://www.googleapis.com/auth/youtube.upload']
+# ✅ Use YouTube scope (not Drive)
+SCOPES = ['https://www.googleapis.com/auth/drive.readonly']
 
 def main():
-    print("\n🔑 Starting YouTube OAuth flow...")
-    print("A browser window will open — log in with the YouTube account you want to upload to.\n")
+    print("🔑 Starting OAuth flow...")
 
-    flow = InstalledAppFlow.from_client_config(
-        {
-            "installed": {
-                "client_id": input("Enter your CLIENT_ID: ").strip(),
-                "client_secret": input("Enter your CLIENT_SECRET: ").strip(),
-                "redirect_uris": ["http://localhost"],
-                "auth_uri": "https://accounts.google.com/o/oauth2/auth",
-                "token_uri": "https://oauth2.googleapis.com/token"
-            }
-        },
+    flow = InstalledAppFlow.from_client_secrets_file(
+        os.getenv("SERVER_GOOGLE_CLIENT_FILE"),
         SCOPES
     )
 
     creds = flow.run_local_server(port=0)
-    print("\n✅ Login successful!")
-    print("------------------------------------------------------")
-    print("📘 REFRESH TOKEN (save this in GitHub Secrets):")
+
+    print("\n✅ AUTH SUCCESS")
+    print("REFRESH TOKEN:\n")
     print(creds.refresh_token)
-    print("------------------------------------------------------")
-    print("\nExample GitHub Secrets setup:")
-    print("  YOUTUBE_CLIENT_ID=<your client id>")
-    print("  YOUTUBE_CLIENT_SECRET=<your client secret>")
-    print("  YOUTUBE_REFRESH_TOKEN=<this printed value>")
-    print("------------------------------------------------------")
 
 if __name__ == "__main__":
     main()

src/google_sheet_reader.py

@@ -0,0 +1,405 @@
+import os
+import csv
+import tempfile
+from pathlib import Path
+
+import gspread
+from google.auth import default
+from google.auth.credentials import Credentials as BaseCredentials
+
+from utils import logger
+
+
+class GoogleSheetReader:
+    """
+    Utility class to read & write Google Sheets using ADC (Local) / WIF (CI)
+    """
+
+    SCOPES = [
+        "https://www.googleapis.com/auth/spreadsheets",
+        "https://www.googleapis.com/auth/drive.file",
+    ]
+
+    def __init__(self):
+        logger.info("Initializing GoogleSheetReader")
+
+        # Support both name and ID
+        self.sheet_name = os.getenv("GSHEET_NAME")
+        self.sheet_id = os.getenv("GSHEET_ID")
+        
+        if not self.sheet_name and not self.sheet_id:
+            logger.error("Must provide either GSHEET_NAME or GSHEET_ID")
+            raise RuntimeError("Must provide either GSHEET_NAME or GSHEET_ID")
+        
+        self.worksheet_name = os.getenv("GSHEET_WORKSHEET")
+
+        logger.debug(
+            "Sheet config | name=%s | id=%s | worksheet=%s",
+            self.sheet_name or "N/A",
+            self.sheet_id or "N/A",
+            self.worksheet_name or "sheet1",
+        )
+
+        self.client = self._authorize()
+        self.sheet = self._open_sheet()
+
+        # -------- Temp directory handling --------
+        self.temp_dir = self._init_temp_dir()
+        logger.info("Using temp directory: %s", self.temp_dir)
+
+    # ------------------ Internal helpers ------------------
+
+    def _get_env(self, key: str) -> str:
+        value = os.getenv(key)
+        if not value:
+            logger.error("Missing required env variable: %s", key)
+            raise RuntimeError(f"Missing required env variable: {key}")
+        return value
+
+    def _authorize(self):
+        logger.info("Authorizing with Application Default Credentials (ADC)")
+        creds, _ = default()
+
+        # IMPORTANT: re-scope user ADC credentials
+        if isinstance(creds, BaseCredentials) and hasattr(creds, "with_scopes"):
+            creds = creds.with_scopes(self.SCOPES)
+
+        return gspread.authorize(creds)
+
+    def _open_sheet(self):
+        """Open spreadsheet by ID (preferred) or name"""
+        try:
+            if self.sheet_id:
+                logger.info("Opening spreadsheet by ID: %s", self.sheet_id)
+                spreadsheet = self.client.open_by_key(self.sheet_id)
+            else:
+                logger.info("Opening spreadsheet by name: %s", self.sheet_name)
+                spreadsheet = self.client.open(self.sheet_name)
+        except gspread.SpreadsheetNotFound:
+            logger.error("Spreadsheet not found!")
+            logger.error("Available spreadsheets:")
+            try:
+                for sheet in self.client.openall():
+                    logger.error("  - %s (ID: %s)", sheet.title, sheet.id)
+            except Exception as e:
+                logger.error("Could not list spreadsheets: %s", e)
+            raise
+
+        if self.worksheet_name:
+            logger.info("Opening worksheet: %s", self.worksheet_name)
+            return spreadsheet.worksheet(self.worksheet_name)
+
+        logger.info("Opening default worksheet (sheet1)")
+        return spreadsheet.sheet1
+
+    def _init_temp_dir(self) -> Path:
+        """
+        Creates a temp directory.
+        Uses fixed path during test automation if configured.
+        """
+        if os.getenv("TEST_AUTOMATION", "").lower() == "true":
+            base_dir = os.getenv("TEST_DATA_DIRECTORY")
+            if not base_dir:
+                raise RuntimeError("TEST_DATA_DIRECTORY must be set when TEST_AUTOMATION=true")
+
+            path = Path(base_dir) / "output"
+            path.mkdir(parents=True, exist_ok=True)
+            return path
+
+        return Path(tempfile.mkdtemp(prefix="gsheet_"))
+
+    # ------------------ Sheet creation helpers ------------------
+
+    def _get_or_create_spreadsheet(self, sheet_name: str):
+        try:
+            logger.info("Opening spreadsheet: %s", sheet_name)
+            return self.client.open(sheet_name)
+        except gspread.SpreadsheetNotFound:
+            logger.warning("Spreadsheet not found. Creating new one: %s", sheet_name)
+            return self.client.create(sheet_name)
+
+    def _get_or_create_worksheet(
+        self,
+        spreadsheet,
+        worksheet_name: str,
+        rows: int = 1000,
+        cols: int = 26,
+    ):
+        try:
+            logger.info("Opening worksheet: %s", worksheet_name)
+            return spreadsheet.worksheet(worksheet_name)
+        except gspread.WorksheetNotFound:
+            logger.warning("Worksheet not found. Creating: %s", worksheet_name)
+            return spreadsheet.add_worksheet(
+                title=worksheet_name,
+                rows=rows,
+                cols=cols,
+            )
+
+    def _ensure_header(self, worksheet, header: list[str]) -> list[str]:
+        """
+        Ensures header row exists and contains all required columns.
+        Returns final header order.
+        """
+        existing = worksheet.row_values(1)
+
+        # Case 1: Empty sheet
+        if not existing:
+            logger.info("Sheet empty. Writing header.")
+            worksheet.insert_row(header, index=1)
+            return header
+
+        # Case 2: Header exists, check for missing columns
+        updated = False
+        final_header = existing.copy()
+
+        for col in header:
+            if col not in final_header:
+                logger.info("Adding missing header column: %s", col)
+                final_header.append(col)
+                updated = True
+
+        if updated:
+            worksheet.update(values=[final_header], range_name="1:1")
+
+        return final_header
+
+
+    # ------------------ Core APIs ------------------
+
+    def get_all_values(self) -> list:
+        logger.debug("Fetching all values from sheet")
+        return self.sheet.get_all_values()
+
+    def get_dataframe(self):
+        logger.debug("Converting sheet to DataFrame")
+        import pandas as pd
+
+        data = self.get_all_values()
+        return pd.DataFrame(data[1:], columns=data[0])
+
+    # ------------------ Filtering APIs ------------------
+
+    def filter_rows(
+        self,
+        column_name: str = "STATUS",
+        match_value: str = "SELECTED",
+        case_insensitive: bool = True,
+    ) -> list:
+        logger.info(
+            "Filtering rows | column=%s | value=%s | case_insensitive=%s",
+            column_name,
+            match_value,
+            case_insensitive,
+        )
+
+        rows = self.get_all_values()
+        if not rows:
+            logger.warning("Sheet is empty")
+            return []
+
+        header = rows[0]
+
+        try:
+            col_idx = header.index(column_name)
+        except ValueError:
+            logger.error("Column '%s' not found in sheet", column_name)
+            raise RuntimeError(f"Column '{column_name}' not found in sheet")
+
+        def _match(val: str) -> bool:
+            if case_insensitive:
+                return val.strip().lower() == match_value.lower()
+            return val == match_value
+
+        filtered = [
+            row for row in rows[1:]
+            if len(row) > col_idx and _match(row[col_idx])
+        ]
+
+        logger.info(
+            "Filtered rows: %d / %d",
+            len(filtered),
+            max(len(rows) - 1, 0),
+        )
+
+        return [header] + filtered
+
+    def get_filtered_dataframe(
+        self,
+        column_name: str = "STATUS",
+        match_value: str = "SELECTED",
+        case_insensitive: bool = True,
+    ):
+        logger.debug("Creating filtered DataFrame")
+        import pandas as pd
+
+        rows = self.filter_rows(
+            column_name=column_name,
+            match_value=match_value,
+            case_insensitive=case_insensitive,
+        )
+
+        if len(rows) <= 1:
+            logger.warning("No matching rows found")
+            return pd.DataFrame(columns=rows[0] if rows else [])
+
+        return pd.DataFrame(rows[1:], columns=rows[0])
+
+    def create_or_update_sheet(
+        self,
+        sheet_name: str | None = None,
+        worksheet_name: str | None = None,
+        header: list[str] = None,
+        values: list[dict] = None,
+    ):
+        """
+        Create or update a sheet + worksheet.
+        Ensures headers exist and appends values.
+
+        If sheet_name or worksheet_name not provided, uses instance's sheet.
+        
+        values: List of dicts where keys match header names
+        """
+
+        if not values:
+            logger.warning("No values provided. Nothing to append.")
+            return
+
+        # Determine which spreadsheet to use
+        if sheet_name is None:
+            # No sheet_name provided - use instance's sheet
+            if self.sheet_id:
+                # Prefer ID over name (more reliable)
+                logger.info("Using instance sheet by ID: %s", self.sheet_id)
+                spreadsheet = self.client.open_by_key(self.sheet_id)
+            elif self.sheet_name:
+                logger.info("Using instance sheet name: %s", self.sheet_name)
+                spreadsheet = self._get_or_create_spreadsheet(self.sheet_name)
+            else:
+                raise ValueError("No sheet_name provided and no instance sheet configured")
+        else:
+            # sheet_name provided - check if it matches instance's sheet
+            if self.sheet_name and sheet_name == self.sheet_name and self.sheet_id:
+                # Same sheet name and we have an ID - use the ID for reliability
+                logger.info("Sheet name matches instance. Using ID: %s", self.sheet_id)
+                spreadsheet = self.client.open_by_key(self.sheet_id)
+            else:
+                # Different sheet or no ID - open/create by name
+                logger.info("Opening/creating sheet by name: %s", sheet_name)
+                spreadsheet = self._get_or_create_spreadsheet(sheet_name)
+        
+        # Use instance's worksheet if not specified
+        if worksheet_name is None:
+            worksheet_name = self.worksheet_name or "Sheet1"
+            logger.info("Using worksheet name: %s", worksheet_name)
+        
+        worksheet = self._get_or_create_worksheet(spreadsheet, worksheet_name)
+
+        final_header = self._ensure_header(worksheet, header)
+
+        # Convert dict rows -> ordered list rows
+        rows_to_append = []
+        for item in values:
+            row = [item.get(col, "") for col in final_header]
+            rows_to_append.append(row)
+
+        logger.info(
+            "Appending %d rows to %s / %s",
+            len(rows_to_append),
+            sheet_name or f"ID:{self.sheet_id}" if self.sheet_id else self.sheet_name,
+            worksheet_name,
+        )
+
+        worksheet.append_rows(
+            rows_to_append,
+            value_input_option="USER_ENTERED",
+        )
+
+
+    # ------------------ CSV Export ------------------
+
+    def export_csv(
+        self,
+        output_path: str | None = None,
+        column_name: str | None = None,
+        match_value: str = "SELECTED",
+        case_insensitive: bool = True,
+    ) -> Path:
+        """
+        Export sheet (or filtered rows) to CSV.
+        If output_path is None, file is written inside temp_dir.
+        """
+        if output_path:
+            output_path = Path(output_path)
+        else:
+            output_path = self.temp_dir / "sheet_export.csv"
+
+        logger.info("Exporting CSV to %s", output_path)
+
+        if column_name:
+            rows = self.filter_rows(
+                column_name=column_name,
+                match_value=match_value,
+                case_insensitive=case_insensitive,
+            )
+        else:
+            rows = self.get_all_values()
+
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+
+        with output_path.open("w", newline="", encoding="utf-8") as f:
+            writer = csv.writer(f)
+            writer.writerows(rows)
+
+        logger.info("CSV export completed | rows=%d", len(rows) - 1)
+        return output_path
+
+
+# ------------------ CLI entrypoint ------------------
+
+def main():
+    """
+    Examples are controlled via env flags:
+    - EXAMPLE_WRITE=true
+    - EXAMPLE_FILTER_EXPORT=true
+    """
+
+    try:
+        from dotenv import load_dotenv
+        load_dotenv()
+    except ImportError:
+        pass
+
+    reader = GoogleSheetReader()
+
+    # ------------------ EXAMPLE 1: CREATE / UPDATE SHEET ------------------
+    logger.info("Running EXAMPLE_WRITE")
+
+    reader.create_or_update_sheet(
+        # sheet_name="Elvoro Example Output",
+        # worksheet_name="Run_Example",
+        header=["ID", "STATUS", "MESSAGE", "DURATION"],
+        values=[
+            {"ID": "1", "STATUS": "SUCCESS", "MESSAGE": "Video done", "DURATION": "10s"},
+            {"ID": "2", "STATUS": "FAILED", "MESSAGE": "Render error"},
+        ],
+    )
+
+    # ------------------ EXAMPLE 2: FILTER + EXPORT CSV ------------------
+    # logger.info("Running EXAMPLE_FILTER_EXPORT")
+
+    # csv_path = reader.export_csv(
+    #     output_path=os.getenv("OUTPUT_CSV"),
+    #     column_name="STATUS",
+    #     match_value="SELECTED",
+    # )
+
+    # logger.info("Filtered CSV written to %s", csv_path)
+
+    # ------------------ DEFAULT: EXPORT FULL SHEET ------------------
+    # csv_path = reader.export_csv(os.getenv("OUTPUT_CSV"))
+    # logger.info("Full sheet exported to %s", csv_path)
+
+
+if __name__ == "__main__":
+    main()