feat: Replace file-based progress with Google Sheets logging

- Add load_executed_from_gsheet() to fetch completed entries from GSheet
- Add log_progress_to_gsheet() to log progress by appending rows
- Remove old file-based progress tracking (get_progress_file, git_commit_progress)
- Update google_sheet_reader.py to create worksheet if not exists
- Add optional target_row param to create_or_update_sheet
- Support filtering by job_index and setup_type for parallel job tracking

src/google_sheet_reader.py

@@ -88,7 +88,15 @@ class GoogleSheetReader:
 
         if self.worksheet_name:
             logger.info("Opening worksheet: %s", self.worksheet_name)
-            return spreadsheet.worksheet(self.worksheet_name)
+            try:
+                return spreadsheet.worksheet(self.worksheet_name)
+            except gspread.WorksheetNotFound:
+                logger.warning("Worksheet not found. Creating: %s", self.worksheet_name)
+                return spreadsheet.add_worksheet(
+                    title=self.worksheet_name,
+                    rows=1000,
+                    cols=26,
+                )
 
         logger.info("Opening default worksheet (sheet1)")
         return spreadsheet.sheet1
@@ -250,14 +258,17 @@ class GoogleSheetReader:
         worksheet_name: str | None = None,
         header: list[str] = None,
         values: list[dict] = None,
+        target_row: int | None = None,
     ):
         """
         Create or update a sheet + worksheet.
-        Ensures headers exist and appends values.
+        Ensures headers exist and appends/inserts values.
 
         If sheet_name or worksheet_name not provided, uses instance's sheet.
         
         values: List of dicts where keys match header names
+        target_row: Optional 1-indexed row to write to. If row has data, inserts below it.
+                   If None, appends to end of sheet.
         """
 
         if not values:
@@ -297,22 +308,40 @@ class GoogleSheetReader:
         final_header = self._ensure_header(worksheet, header)
 
         # Convert dict rows -> ordered list rows
-        rows_to_append = []
+        rows_to_write = []
         for item in values:
             row = [item.get(col, "") for col in final_header]
-            rows_to_append.append(row)
+            rows_to_write.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,
-        )
+        if target_row is not None:
+            # Check if target row has data
+            try:
+                existing_row = worksheet.row_values(target_row)
+            except Exception:
+                existing_row = []
+            
+            if existing_row and any(cell.strip() for cell in existing_row):
+                # Row has data, insert new row below it
+                worksheet.insert_rows(rows_to_write, row=target_row + 1, value_input_option="USER_ENTERED")
+                logger.info(f"Inserted {len(rows_to_write)} rows below row {target_row}")
+            else:
+                # Row is empty, write directly to it
+                cell_range = f"A{target_row}"
+                worksheet.update(values=rows_to_write, range_name=cell_range, value_input_option="USER_ENTERED")
+                logger.info(f"Updated row {target_row}")
+        else:
+            # No target row, append to end
+            logger.info(
+                "Appending %d rows to %s / %s",
+                len(rows_to_write),
+                sheet_name or f"ID:{self.sheet_id}" if self.sheet_id else self.sheet_name,
+                worksheet_name,
+            )
+            worksheet.append_rows(
+                rows_to_write,
+                value_input_option="USER_ENTERED",
+            )
 
-        worksheet.append_rows(
-            rows_to_append,
-            value_input_option="USER_ENTERED",
-        )
 
 
     # ------------------ CSV Export ------------------

src/process_csv.py

@@ -1,8 +1,8 @@
 import asyncio
 import csv
-import subprocess
 import os, time
 from pathlib import Path
+from datetime import datetime
 from load_config import load_configuration
 from main import (
     load_content_strategies,
@@ -12,120 +12,87 @@ from automation import ContentAutomation
 from utils import logger
 from data_holder import DataHolder
 from asset_selector import AssetSelector
+from google_sheet_reader import GoogleSheetReader
 import argparse
-import random
 import uuid
 
 DATA_DIR = Path("data")
 ALL_VIDEO_FILE_INFO = None
 
-def get_progress_file(job_index=None):
-    """Get the appropriate progress file for this job."""
-    if job_index is not None:
-        return DATA_DIR / f"executed_lines_job{job_index}.txt"
-    return DATA_DIR / "executed_lines.txt"
-
 
-def load_all_executed_lines():
-    """Load executed lines from all job-specific progress files.
-    Returns a list (not a set) to preserve duplicate counts.
-    """
-    executed = []  # Changed from set() to list
+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.
     
-    # Load from main progress file
-    main_progress = DATA_DIR / "executed_lines.txt"
-    if main_progress.exists():
-        with open(main_progress, "r") as f:
-            executed.extend(x.strip() for x in f if x.strip())
+    Args:
+        setup_type: Optional. If provided, only count rows where Safe Name contains this value.
+        job_index: Optional. If provided, only count rows for this specific job.
+    """
+    gsheet_name = os.getenv("GSHEET_WORKSHEET_LOGS")
     
-    # Load from all job-specific files
-    for job_file in DATA_DIR.glob("executed_lines_job*.txt"):
-        with open(job_file, "r") as f:
-            executed.extend(x.strip() for x in f if x.strip())
+    if not gsheet_name:
+        logger.warning("GSHEET_WORKSHEET_LOGS not set")
+        return []
     
-    return executed  # Returns list instead of set
-
-
-def git_commit_progress(job_index: int, commit=False):
-    """
-    Commit progress for a specific job. This flow is robust for parallel and CI/CD environments.
-    """
+    try:
+        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"]
+            
+            # Optional: filter by job_index
+            if job_index is not None and "Job" in df.columns:
+                successful = successful[successful["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)]
+            
+            executed = successful["Hook"].tolist()
+            logger.info(f"Loaded {len(executed)} executed entries from Google Sheets")
+            return executed
+        else:
+            logger.warning("Google Sheets missing required columns (Success, Hook)")
+            return []
+    except Exception as e:
+        logger.error(f"Failed to load from Google Sheets: {e}")
+        return []
+
+
+def log_progress_to_gsheet(tts_script: str, result: dict, job_index: int, commit=False):
+    """Log progress to Google Sheets by appending rows."""
     if os.getenv("DO_NOT_PUBLISH", "false").lower() == "true":
-        print("Testing do not publish.")
         return
 
     if not commit:
-        print("ℹ️ Skipping git commit (use --commit to enable).")
         return
 
-    progress_file = get_progress_file(job_index)
-    if not progress_file.exists():
-        logger.info(f"ℹ️ No progress file found for job {job_index}. Nothing to commit.")
+    gsheet_name = os.getenv("GSHEET_WORKSHEET_LOGS")
+    if not gsheet_name:
+        logger.warning("GSHEET_WORKSHEET_LOGS not set, skipping gsheet logging")
         return
 
     try:
-        branch = "feature/video-revamp"
-        max_retries = 3
-
-        # 1. Ensure we are on the correct branch and not in a detached HEAD state.
-        # This is critical for CI/CD environments like GitHub Actions.
-        logger.info(f"Git: Ensuring we are on branch '{branch}' for job {job_index}...")
-        subprocess.run(["git", "fetch", "origin", branch], check=True, capture_output=True)
-        subprocess.run(["git", "checkout", "-B", branch, f"origin/{branch}"], check=True, capture_output=True)
-
-        # 2. Stage and commit the local changes.
-        subprocess.run(["git", "add", str(progress_file)], check=True)
-        commit_result = subprocess.run(
-            ["git", "commit", "-m", f"✅ Job {job_index} progress update"],
-            capture_output=True, text=True # Capture output to check it
+        reader = GoogleSheetReader(worksheet_name=gsheet_name)
+
+        reader.create_or_update_sheet(
+            worksheet_name=gsheet_name,
+            header=["Timestamp", "Job", "Hook", "Success", "Final URL"],
+            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)),
+                "Final URL": result.get("final_url", ""),
+            }],
         )
 
-        # If the commit was successful, proceed. If not, check if it's just "nothing to commit".
-        if commit_result.returncode != 0:
-            if "nothing to commit" in commit_result.stdout or "nothing to commit" in commit_result.stderr:
-                logger.info(f"ℹ️ No new progress to commit for job {job_index}.")
-                return # This is a successful state, no need to push.
-            else:
-                # The commit failed for another reason. Raise an error.
-                raise subprocess.CalledProcessError(
-                    commit_result.returncode, cmd=commit_result.args,
-                    output=commit_result.stdout, stderr=commit_result.stderr
-                )
+        logger.info(f"✓ Logged progress to Google Sheet for job {job_index}")
 
-        # 3. Push with a retry loop to handle race conditions from other parallel jobs.
-        for attempt in range(max_retries):
-            try:
-                # Pull with rebase immediately before pushing to resolve any new remote commits.
-                logger.info(f"Git: Pulling with rebase (Attempt {attempt + 1}/{max_retries})...")
-                subprocess.run(["git", "pull", "--rebase", "origin", branch], check=True, capture_output=True)
-
-                # Attempt the push.
-                logger.info(f"Git: Pushing to remote (Attempt {attempt + 1}/{max_retries})...")
-                subprocess.run(
-                    ["git", "push", "origin", branch],
-                    check=True,
-                    timeout=45
-                )
-                
-                logger.info(f"✓ Committed progress successfully for job {job_index}")
-                return # Exit function on success
-
-            except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as e:
-                if attempt < max_retries - 1:
-                    # Use randomized exponential backoff to de-synchronize retries
-                    sleep_duration = random.uniform(2, 5) * (attempt + 1)
-                    logger.warning(
-                        f"Push failed for job {job_index}. Retrying in {sleep_duration:.2f} seconds..."
-                    )
-                    time.sleep(sleep_duration)
-                else:
-                    logger.error(f"❌ Failed to push progress for job {job_index} after {max_retries} attempts.")
-                    raise # Re-raise the final exception to signal a persistent failure
-
-    except subprocess.CalledProcessError as e:
-        error_message = e.stderr if e.stderr else e.stdout
-        logger.error(f"❌ A Git command failed for job {job_index}: {e.cmd}\nError: {error_message}")
-        subprocess.run(["git", "reset", "--hard"])
+    except Exception as e:
+        logger.error(f"❌ Failed to log to Google Sheet: {e}")
 
 async def process_row(row, config: dict):
     """Process one CSV row using the main pipeline."""
@@ -197,11 +164,8 @@ async def process_all_csvs(config, commit=False, job_index=None, total_jobs=None
         return
 
     # Load executed lines from ALL progress files (main + all jobs)
-    executed = load_all_executed_lines()
+    executed = load_executed_from_gsheet()
     logger.info(f"Skipping {len(executed)} already executed entries (from all jobs).")
-
-    # Get this job's progress file
-    progress_file = get_progress_file(job_index)
     
     # Load all rows from all CSVs with their global index
     all_rows = []
@@ -217,7 +181,6 @@ async def process_all_csvs(config, commit=False, job_index=None, total_jobs=None
             if idx % total_jobs == job_index
         ]
         logger.info(f"🔢 Job {job_index + 1}/{total_jobs}: Processing {len(rows_to_process)}/{len(all_rows)} rows")
-        logger.info(f"📝 Progress file: {progress_file}")
     else:
         rows_to_process = [(idx, csv_name, row) for idx, (csv_name, row) in enumerate(all_rows)]
         logger.info(f"Processing all {len(rows_to_process)} rows")
@@ -283,12 +246,7 @@ async def process_all_csvs(config, commit=False, job_index=None, total_jobs=None
                 success_count += 1
                 processed_scripts.append(tts_script)  # Track in session
                 
-                if os.getenv("DO_NOT_PUBLISH", "false").lower() != "true":
-                    # Write progress to THIS JOB's file only
-                    with progress_file.open("a") as pf:
-                        pf.write(f"{tts_script}\n")
-                
-                git_commit_progress(job_index, commit)
+                log_progress_to_gsheet(tts_script, result, job_index, commit)
                 
                 logger.info(
                     f"✅ {tts_script[:50]}... occurrence {occurrence}/{total_needed} "
@@ -296,8 +254,6 @@ async def process_all_csvs(config, commit=False, job_index=None, total_jobs=None
                 )
             else:
                 logger.warning(f"⚠️ {tts_script[:50]}... pipeline failed, NOT marking as complete")
-                if success_count % 2 == 1:
-                    git_commit_progress(job_index, commit)
 
             if os.getenv("DO_NOT_PUBLISH", "false").lower() == "true":
                 break
@@ -305,10 +261,6 @@ async def process_all_csvs(config, commit=False, job_index=None, total_jobs=None
             logger.error(f"❌ Error processing {tts_script[:50]}...: {e}", exc_info=True)
             continue
 
-    # Final commit for any remaining progress
-    if commit and success_count > 0:
-        git_commit_progress(job_index, commit)
-    
     logger.info(f"🏁 Job {job_index} finished: {success_count}/{processed_count} successful")
 
 async def create_plain_videos(config, commit=False, job_index=None, total_jobs=None):
@@ -320,16 +272,10 @@ async def create_plain_videos(config, commit=False, job_index=None, total_jobs=N
     n = int(os.getenv("PlAIN_VIDEO_COUNT", 100))
     logger.info(f"Creating {n} random videos for testing...")
 
-    progress_file = get_progress_file(job_index)
-
-    # How many videos already created across ALL jobs
-    completed = 0
-    if progress_file.exists():
-        with progress_file.open("r") as pf:
-            completed = sum(1 for line in pf if os.getenv("SETUP_TYPE") in line.strip())
-    else:
-        progress_file.touch()
-        completed = 0
+    # Load how many already completed from Google Sheets (filtered by SETUP_TYPE and job_index)
+    setup_type = os.getenv("SETUP_TYPE")
+    executed = load_executed_from_gsheet(setup_type=setup_type, job_index=job_index)
+    completed = len(executed)
 
     # --- ASSIGN SLICE TO THIS JOB ---
     per_job = n / total_jobs
@@ -350,12 +296,10 @@ async def create_plain_videos(config, commit=False, job_index=None, total_jobs=N
 
         config["current_audio_index"] = i
         result = await process_row(row, config)
+        print(result)
 
         if commit and result.get("success", False):
-            with progress_file.open("a") as pf:
-                print(result)
-                pf.write(f"{result.get('final_url').split('/')[-1]}\n")
-            git_commit_progress(job_index, commit)
+            log_progress_to_gsheet(row.get("TTS Script (AI Avatar)", ""), result, job_index, commit)
 
         if os.getenv("DO_NOT_PUBLISH", "false").lower() == "true":
             break