#4648 Folder-Based Innovation Workflow

#4648 Folder-Based Innovation Workflow

#4648 Assess and recommend an improved implementation of the following application description: """

0. Python pplication steps

1. Read a prompt.txt file from a user specified folder. The file contains both a prompt and a destination response folder name.

2. Loop through all other text files in the user specified folder.

2.a Apply the prompt to each file individually and submit the prompt to a Gemini LLM

2.b Save the response to a text file in the destination response folder.

2.c Delete the original text file from the original user specified folder."""

Here is an assessment of your proposed process along with a production-grade, improved implementation strategy.

1. Architectural & Reliability Assessment

While the underlying logic is straightforward, the raw batch processing flow presented in your specification contains several operational single-points-of-failure:

Feature / Issue	Risk Level	Mitigation Strategy
Parsing prompt.txt	High	Coupling prompt text and path metadata in a single unformatted text file leads to brittle string manipulation. Use structured input (e.g., YAML/JSON frontmatter or distinct CLI flags).
Immediate File Deletion (2.c)	Critical	Deleting source files immediately after receiving an LLM response means data is irrecoverable if the file write fails, permissions error occurs, or the output is corrupted/truncated.
Synchronous Processing	Medium	Processing files sequentially via blocking API calls is bottlenecked by API network latency.
API Failure / Rate Limits	High	Unhandled API exceptions (429 Rate Limits, 5xx Server Errors, or safety filter blocks) will crash the script mid-execution, leaving input data half-processed.

2. Key Architectural Improvements

1. Move-over-Delete Strategy (Safety Staging): Instead of deleting source files outright, move processed input files into an .archived/ or .processed/ subdirectory within the source path only after confirming the output file write succeeded.

2. Explicit Metadata Management: Separate configuration from prompt text, or parse structured file headers using standard separators (e.g., Markdown YAML frontmatter).

3. Atomic Writes & Error Handling: Write responses to a temporary .tmp extension before renaming to destination paths. Wrap Gemini API calls with robust retry logic (google-genai / tenacity).

4. Structured Progress Logging: Replace silent operations with standard logging outputs so file operations, API metrics, and errors are traceable.

3. Recommended Implementation

This refactored Python script uses standard libraries alongside google-genai and robust exception handling.

Prerequisites

pip install google-genai pydantic

Improved Python Script (batch_processor.py)

#!/usr/bin/env python3
"""
Batch Text Processor utilizing Google Gemini API.
Reads input files, applies a common prompt template, outputs responses,
and safely archives processed source files.
"""

import os
import sys
import shutil
import logging
from pathlib import Path
from google import genai
from google.genai import types
from google.genai.errors import APIError

# Setup logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)]
)

def parse_prompt_file(prompt_path: Path) -> tuple[str, str]:
    """
    Parses prompt.txt assuming the first line is the destination folder path,
    and the remaining text is the system/user prompt.
    
    Format expected in prompt.txt:
        DESTINATION: /path/to/output_folder
        ---
        Your actual prompt text goes here...
    """
    content = prompt_path.read_text(encoding="utf-8").strip()
    
    if not content:
        raise ValueError(f"Prompt file at '{prompt_path}' is empty.")
    
    lines = content.splitlines()
    dest_dir = None
    prompt_body = []
    
    in_body = False
    for line in lines:
        if line.startswith("DESTINATION:") and not in_body:
            dest_dir = line.split("DESTINATION:", 1)[1].strip()
        elif line.strip() == "---" and not in_body:
            in_body = True
        else:
            prompt_body.append(line)
            
    if not dest_dir:
        raise ValueError("Missing 'DESTINATION: <folder_path>' header in prompt.txt")
        
    return dest_dir, "\n".join(prompt_body).strip()

def process_batch(source_dir_path: str, model_name: str = "gemini-2.5-flash"):
    source_dir = Path(source_dir_path).resolve()
    prompt_file = source_dir / "prompt.txt"

    if not prompt_file.exists():
        logging.error(f"Configuration file 'prompt.txt' not found in {source_dir}")
        return

    # 1. Parse configuration & prompt
    try:
        dest_dir_str, system_prompt = parse_prompt_file(prompt_file)
    except Exception as e:
        logging.error(f"Failed to parse prompt.txt: {e}")
        return

    dest_dir = Path(dest_dir_str).resolve()
    dest_dir.mkdir(parents=True, exist_ok=True)
    
    # Optional: Archive folder for input files instead of outright deletion
    archive_dir = source_dir / ".processed_archive"
    archive_dir.mkdir(exist_ok=True)

    # Initialize Gemini Client
    client = genai.Client()

    # 2. Loop through eligible text files
    input_files = [
        f for f in source_dir.glob("*.txt") 
        if f.name != "prompt.txt" and f.is_file()
    ]

    logging.info(f"Found {len(input_files)} file(s) to process in {source_dir}")

    for file_path in input_files:
        logging.info(f"Processing: {file_path.name}")
        
        try:
            input_text = file_path.read_text(encoding="utf-8")
            
            # Combine prompt template with individual file contents
            full_contents = f"{system_prompt}\n\n--- Input Content ---\n{input_text}"

            # 2.a Submit to Gemini LLM
            response = client.models.generate_content(
                model=model_name,
                contents=full_contents,
                config=types.GenerateContentConfig(
                    temperature=0.2,
                )
            )

            # 2.b Atomic Save to destination
            out_file_path = dest_dir / f"response_{file_path.name}"
            temp_out_path = dest_dir / f".tmp_{file_path.name}"

            temp_out_path.write_text(response.text, encoding="utf-8")
            temp_out_path.replace(out_file_path)  # Atomic rename
            logging.info(f"Saved response -> {out_file_path}")

            # 2.c Safe handling of source file
            # Moving to archive is safer than raw deletion:
            shutil.move(str(file_path), str(archive_dir / file_path.name))
            logging.info(f"Archived original -> {archive_dir / file_path.name}")

        except APIError as e:
            logging.error(f"Gemini API Error processing {file_path.name}: {e}")
            continue
        except Exception as e:
            logging.error(f"Unexpected failure processing {file_path.name}: {e}")
            continue

if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser(description="Batch process text files using Gemini API.")
    parser.add_argument("folder", type=str, help="Path to folder containing prompt.txt and targets")
    parser.add_argument("--model", type=str, default="gemini-2.5-flash", help="Gemini model ID")
    
    args = parser.parse_args()
    process_batch(args.folder, model_name=args.model)

4. Expected prompt.txt Structure

To support the script cleanly without parsing bugs, structure prompt.txt like this:

DESTINATION: /path/to/your/output_directory
---
Summarize the key action items from the following text file. 
Present the results as a clean Markdown bulleted list.

From <https://gemini.google.com/app/094b5a7541c40864>  Google Gemini (3.6 Flash)

Notes to Self:

Upper level folder per project

Initial sub-folder has

    Initial Prompt (w/destination folder name)

    Initial Inputs (one per text file)