From Reddit to Insights: Building an AI-Powered Data Pipeline with Gemini (Cloud)¶

def fetch_and_save_posts(reddit, subreddit_name, db_utils, conn):
    """Fetches new posts from a subreddit since the last processed timestamp."""
    last_processed_utc = get_last_processed_timestamp(conn, subreddit_name)
    current_max_utc = 0

    subreddit = reddit.subreddit(subreddit_name)
    for submission in subreddit.hot(limit=20):
        # Skip already processed posts
        if submission.created_utc <= last_processed_utc:
            continue

        # Process post data
        post_data = {
            "subreddit": subreddit_name,
            "post_id": submission.id,
            "title": str(submission.title),
            "author": str(submission.author),
            "url": str(submission.url),
            "score": int(submission.score),
            "created_utc": submission.created_utc,
            "comments": []
        }

        # Handle comments if they exist
        if submission.num_comments > 0:
            try:
                submission.comments.replace_more(limit=0)  # Expand comment tree
                post_data["comments"] = [
                    {
                        "comment_id": comment.id,
                        "author": str(comment.author),
                        "body": str(comment.body),
                        "created_utc": comment.created_utc,
                    }
                    for comment in submission.comments.list()[:10]  # Top 10 comments
                    if comment.created_utc > last_processed_utc
                ]
            except Exception as e:
                logging.error(f"Error fetching comments for post {submission.id}: {e}")

        # Save to database
        db_utils.insert_raw_post_data(conn, post_data)
        current_max_utc = max(current_max_utc, submission.created_utc)

def save_to_gcs(storage_client: storage.Client, bucket_name: str, 
                subreddit_name: str, posts: List[Dict]) -> None:
    """Saves posts data to Google Cloud Storage."""
    try:
        # Get bucket
        bucket = storage_client.bucket(bucket_name)

        # Generate blob name with current timestamp
        batch_timestamp = datetime.utcnow()
        blob_name = get_gcs_blob_name(subreddit_name, batch_timestamp)
        blob = bucket.blob(blob_name)

        # Convert to newline-delimited JSON
        ndjson_data = '\n'.join(json.dumps(post) for post in posts)
        blob.upload_from_string(ndjson_data)

        logger.info(f"Saved {len(posts)} posts to GCS: {blob_name}")
    except Exception as e:
        logger.error(f"Error saving to GCS: {e}")
        raise

def preprocess_posts(df):
    """Preprocesses Reddit posts with content filtering and cleaning."""
    # Content filtering patterns
    profanity_pattern = r'(?i)\b(wordword2|word3|word4|word5)\b|' + \
                       r'(?i)\b(w\*\*d1|w\*rd2|w\*rd3|w\*\*d4)\b|' + \
                       r'(?i)w[^\w]?r[^\w]?d1|' + \
                       r'(?i)w[^\w]?r[^\w]?d2'

    # Filter and clean posts
    return df.filter(
        # Basic validation
        (F.col("title").isNotNull()) & 
        (F.col("title") != "") & 
        (F.col("title") != "[deleted]") &
        (F.col("author").isNotNull()) & 
        (F.col("author") != "[deleted]") &
        # Content filtering
        ~F.col("title").rlike(profanity_pattern)
    ).withColumn(
        # Clean and normalize text
        "title", F.regexp_replace("title", r'[\n\r\t]', ' ')
    ).withColumn(
        # Remove multiple spaces
        "title", F.regexp_replace("title", r'\s+', ' ')
    ).dropDuplicates(["post_id"])

def preprocess_comments(df):
    """Preprocesses Reddit comments with content filtering and cleaning."""
    # Parse and explode comments from JSONB
    comments_schema = ArrayType(StructType([
        StructField("body", StringType(), True),
        StructField("author", StringType(), True),
        StructField("comment_id", StringType(), True),
        StructField("created_utc", DoubleType(), True)
    ]))

    return df.withColumn(
        "comments_parsed",
        F.from_json(F.col("comments"), comments_schema)
    ).withColumn(
        "comment", F.explode("comments_parsed")
    ).select(
        "post_id",
        F.col("comment.body").alias("body"),
        F.col("comment.author").alias("author"),
        F.col("comment.comment_id").alias("comment_id"),
        F.to_timestamp(F.col("comment.created_utc")).alias("created_utc")
    ).filter(
        # Remove deleted/empty comments
        (F.col("body").isNotNull()) &
        (F.col("body") != "") &
        (F.col("body") != "[deleted]") &
        (F.col("author") != "[deleted]")
    ).withColumn(
        # Clean and normalize text
        "body", F.regexp_replace("body", r'[\n\r\t]', ' ')
    ).withColumn(
        # Remove multiple spaces
        "body", F.regexp_replace("body", r'\s+', ' ')
    ).dropDuplicates(["comment_id"])

def generate_daily_summaries(spark: SparkSession, project_id: str, dataset_id: str, subreddits: list, bucket_name: str):
    """
    Generates daily summaries of the processed data using PySpark.
    Implements incremental loading by only processing data since the last processed timestamp.
    """
    current_timestamp = datetime.now()
    last_processed = get_last_processed_timestamp(spark, project_id, dataset_id)
    total_summaries_added = 0

    if last_processed:
        logger.info(f"Processing summaries from {last_processed} to {current_timestamp}")
    else:
        logger.info("No previous summaries found. Processing all available data.")

    for subreddit in subreddits:
        logger.info(f"Processing daily summary for subreddit: {subreddit}")

        # Load posts and comments tables from BigQuery
        posts_df = spark.read.format('bigquery') \
            .option('table', f"{project_id}.{dataset_id}.posts_{subreddit.lower()}") \
            .load()

        posts_count = posts_df.count()
        logger.info(f"Found {posts_count} posts for {subreddit}")

        comments_df = spark.read.format('bigquery') \
            .option('table', f"{project_id}.{dataset_id}.comments_{subreddit.lower()}") \
            .load()

        comments_count = comments_df.count()
        logger.info(f"Found {comments_count} comments for {subreddit}")

        # Filter posts and comments based on created_utc timestamp
        if last_processed:
            daily_posts_df = posts_df.filter(
                (F.col("created_utc") > last_processed) &
                (F.col("created_utc") <= current_timestamp)
            )

            daily_comments_df = comments_df.filter(
                (F.col("created_utc") > last_processed) &
                (F.col("created_utc") <= current_timestamp)
            )
        else:
            daily_posts_df = posts_df.filter(F.col("created_utc") <= current_timestamp)
            daily_comments_df = comments_df.filter(F.col("created_utc") <= current_timestamp)

        filtered_posts_count = daily_posts_df.count()
        logger.info(f"Filtered to {filtered_posts_count} posts for processing")

        if filtered_posts_count == 0:
            logger.info(f"No new posts to summarize for {subreddit}")
            continue

        # Join posts and comments on post_id
        joined_df = daily_posts_df.alias("posts").join(
            daily_comments_df.alias("comments"), 
            "post_id", 
            "right"
        )

        # Prepare the daily summary
        daily_summary_df = joined_df.select(
            F.monotonically_increasing_id().alias("id"),
            F.col("subreddit"),
            F.col("posts.post_id").alias("post_id"),
            F.col("posts.score").alias("post_score"),
            F.col("posts.url").alias("post_url"),
            F.col("comments.comment_id").alias("comment_id"),
            F.to_date(F.col("comments.created_utc")).alias("summary_date"),
            F.current_timestamp().alias("processed_date"),
            F.lit(True).alias("needs_processing"),
            F.col("posts.title").alias("post_content"),
            F.col("comments.body").alias("comment_body")
        )

        # Filter out rows where required fields are null
        daily_summary_df = daily_summary_df.filter(
            (F.col("comment_body").isNotNull()) & 
            (F.col("comment_id").isNotNull()) &
            (F.col("post_id").isNotNull())
        )

        filtered_summaries_count = daily_summary_df.count()
        logger.info(f"Generated {filtered_summaries_count} summaries before deduplication")

        # Deduplication Logic
        existing_summary_df = spark.read.format('bigquery') \
            .option('table', f"{project_id}.{dataset_id}.daily_summary_data") \
            .load()

        # Perform a left anti-join to get only new records
        new_daily_summary_df = daily_summary_df.join(
            existing_summary_df,
            ["post_id", "comment_id"],  # Using composite key for deduplication
            "left_anti"
        )

        new_summaries_count = new_daily_summary_df.count()

        if new_summaries_count > 0:
            # Write the new daily summaries to BigQuery
            new_daily_summary_df.write.format('bigquery') \
                .option('table', f"{project_id}.{dataset_id}.daily_summary_data") \
                .option('temporaryGcsBucket', bucket_name) \
                .mode('append') \
                .save()

            total_summaries_added += new_summaries_count
            logger.info(f"Successfully added {new_summaries_count} new summaries for {subreddit}")
        else:
            logger.info(f"No new unique summaries to add for {subreddit}")

def create_summarizer():
    """Initialize the summarization model."""
    model_name = "philschmid/bart-large-cnn-samsum"
    local_model_path = "/models/models--philschmid--bart-large-cnn-samsum/snapshots/e49b3d60d923f12db22bdd363356f1a4c68532ad"

    tokenizer = AutoTokenizer.from_pretrained(local_model_path)
    model = AutoModelForSeq2SeqLM.from_pretrained(
        local_model_path,
        torch_dtype=torch.float32,
        device_map="auto",
        low_cpu_mem_usage=True
    )

    return pipeline(
        "summarization",
        model=model,
        tokenizer=tokenizer,
        framework="pt"
    )

def generate_summary(comment_body, summarizer):
    """Generate summary for a given comment."""
    if len(comment_body.split()) <= 59:
        return comment_body 

    try:
        summary = summarizer(
            comment_body, 
            max_length=60, 
            min_length=15, 
            do_sample=False, 
            truncation=True
        )
        if summary and isinstance(summary, list) and len(summary) > 0:
            return summary[0]['summary_text']
    except Exception as e:
        logging.error(f"Summarization failed: {e}")

    return ""

def initialize_emotion_analyzer():
    """Initialize the emotion analysis model."""
    model_name = "SamLowe/roberta-base-go_emotions"
    local_model_path = "/models/models--SamLowe--roberta-base-go_emotions/snapshots/58b6c5b44a7a12093f782442969019c7e2982299"

    model = AutoModelForSequenceClassification.from_pretrained(local_model_path)
    tokenizer = AutoTokenizer.from_pretrained(local_model_path)

    return pipeline(
        "text-classification",
        model=model,
        tokenizer=tokenizer
    )

def perform_sentiment_analysis(text, emotion_analyzer):
    """Perform sentiment analysis on the given text."""
    try:
        if not text or text.strip() == "":
            return "Neutral", 0.0, "No content"

        truncated_text = truncate_text(text, max_length=500)
        emotion_output = emotion_analyzer(truncated_text)

        if not emotion_output:
            return "Neutral", 0.0, "Analysis failed"

        emotion_label = emotion_output[0]['label']
        emotion_score = emotion_output[0]['score']

        # Map emotions to sentiment categories
        sentiment_mapping = {
            'joy': 'Positive',
            'love': 'Positive',
            'admiration': 'Positive',
            'approval': 'Positive',
            'excitement': 'Positive',
            'gratitude': 'Positive',
            'optimism': 'Positive',
            'anger': 'Negative',
            'disappointment': 'Negative',
            'disgust': 'Negative',
            'fear': 'Negative',
            'sadness': 'Negative',
            'confusion': 'Neutral',
            'curiosity': 'Neutral',
            'surprise': 'Neutral'
        }

        sentiment_label = sentiment_mapping.get(emotion_label.lower(), 'Neutral')
        return sentiment_label, float(emotion_score), emotion_label

    except Exception as e:
        logging.error(f"Error during sentiment analysis: {str(e)}")
        return "Neutral", 0.0, "Error in analysis"

# MLflow configuration
mlflow.set_tracking_uri("http://mlflow:5000")
mlflow.set_experiment("reddit_sentiment_analysis_experiments")

with mlflow.start_run() as run:
    # Log parameters
    mlflow.log_param("model_name", "SamLowe/roberta-base-go_emotions")

def create_prompt_template():
    """Return the standard prompt template for Gemini analysis."""
    current_date = datetime.now().strftime('%Y-%m-%d')

    return f"""
    Analyze the provided text files, which contain Reddit posts and comments.
    **Instructions:**
    1.  **Content Filtering:** Check for harassment, hate speech, or explicit material
    2.  **Ranking:** Rank threads by total "Score"
    3.  **Summarization:** Utilize "Summary" fields for thread overviews
    4.  **Emotional Analysis:** Analyze "Emotion Label" and "Emotion Score"
    5.  **Point of View Extraction:** Extract top 3 viewpoints per thread
    6.  **Links:** Embed URLs in thread titles using Markdown

    **Output Format:**
    ---
    title: "{{subreddit_name}} subreddit"
    date: "{current_date}"
    description: "Analysis of top discussions and trends"
    tags: ["tag1", "tag2", "tag3"]
    ---

    # Overall Ranking and Top Discussions
    [Ranked list of threads with scores and summaries]

    # Detailed Analysis by Thread 
    [Thread-by-thread analysis with summaries, emotions, and viewpoints]
    """

def process_subreddit(model, cur, subreddit, output_dir):
    """Process a single subreddit's data."""
    # Fetch processed data
    cur.execute("""
        SELECT post_id, subreddit, post_score, post_url, comment_id,
               summary_date, post_content, comment_body, comment_summary,
               sentiment_score, sentiment_label
        FROM processed_data.joined_summary_analysis
        WHERE subreddit = %s
        ORDER BY post_score DESC
    """, (subreddit,))

    # Format data for analysis
    text_files = "".join(
        format_text_file(row) for row in cur.fetchall()
    )

    # Generate insights using Gemini
    response = model.generate_content(final_prompt + text_files)

    # Save formatted output
    output_file_path = os.path.join(output_dir, f"llm_{subreddit}.md")
    with open(output_file_path, "w", encoding="utf-8") as f:
        f.write(response.text)

def get_formatted_subreddit_name(subreddit: str) -> str:
    """Returns a properly formatted subreddit name for display."""
    subreddit_formats = {
        "claudeai": "ClaudeAI",
        "dataengineering": "Data Engineering",
        "datascience": "Data Science",
        "localllama": "LocalLLaMA",
        "machinelearning": "Machine Learning",
        "openai": "OpenAI",
        "singularity": "Singularity",
        "stablediffusion": "Stable Diffusion"
    }
    return f"{subreddit_formats.get(subreddit.lower(), subreddit)} Subreddit"

def analyze_data():
    """
    Main function to analyze Reddit data using Google's Gemini model.

    Pipeline steps:
    1. Set up logging and output directory with current date
    2. Initialize Gemini model
    3. Connect to BigQuery and GCS
    4. Process each subreddit
    5. Clean generated markdown files
    6. Upload results to GCS
    """
    try:
        # 1. Set up output directory with year/month/day structure
        current_date = datetime.now()
        year = current_date.strftime('%Y')
        month = current_date.strftime('%m')
        day = current_date.strftime('%d')

        output_dir = os.path.join(CLOUD_DIR, 'results', year, month, day)
        os.makedirs(output_dir, exist_ok=True)
        logger.info(f"Output directory set to {output_dir}")

        # 2. Initialize model
        genai.configure(api_key=GEMINI_CONFIG['GOOGLE_GEMINI_API_KEY'])
        model = genai.GenerativeModel('gemini-2.0-flash-exp')
        logger.info("Model loaded")

        # 3. Connect to BigQuery and GCS
        bq_client = bigquery.Client()
        storage_client = storage.Client()
        logger.info("BigQuery and GCS clients initialized")

        # 4. Process subreddits
        for subreddit in SUBREDDITS:
            process_subreddit(model, bq_client, subreddit, output_dir)

        # 5. Clean markdown files and upload to GCS
        for filename in os.listdir(output_dir):
            if filename.endswith('.md'):
                local_file_path = os.path.join(output_dir, filename)

                # Clean the markdown file
                clean_markdown_file(local_file_path)

                # Upload to GCS
                gcs_blob_path = get_gcs_path(year, month, day, filename)
                upload_to_gcs(local_file_path, gcs_blob_path, storage_client)

        logger.info("Analysis complete - all files processed and uploaded to GCS")

    except Exception as e:
        logger.error(f"Pipeline failed: {e}")
        raise

ingest_task >> ingest_metrics_task >> \
dbt_test_raw_sources >> dbt_test_raw_metrics_task >> \
dbt_staging_task >> dbt_test_staging_models >> \
summarize_task >> sentiment_task >> \
dbt_join_summary_analysis_task >> \
gemini_task >> push_gemini_results_task >> shutdown_vm

# Create Flask app
app = Flask(__name__)

def get_gcloud_path():
    """Get the full path to gcloud executable based on OS"""
    if platform.system() == "Windows":
        # Common installation paths for gcloud on Windows
        possible_paths = [
            r"C:\Program Files (x86)\Google\Cloud SDK\google-cloud-sdk\bin\gcloud.cmd",
            r"C:\Program Files\Google\Cloud SDK\google-cloud-sdk\bin\gcloud.cmd",
            os.path.expanduser("~") + r"\AppData\Local\Google\Cloud SDK\google-cloud-sdk\bin\gcloud.cmd"
        ]
        for path in possible_paths:
            if os.path.exists(path):
                return path
        raise Exception("gcloud not found. Please ensure Google Cloud SDK is installed and in PATH")
    return "gcloud"  # For non-Windows systems, assume it's in PATH

@app.route("/", methods=["POST"])
def start_vm_http():
    """
    HTTP endpoint for Cloud Run to start a Compute Engine instance.
    Accepts POST requests with optional JSON body:
    {'trigger_source': 'scheduler'|'manual'|'dag'}
    """
    return start_vm(request)

def start_vm(request):
    """
    Function to start a Compute Engine instance and execute startup script.
    """
    project = os.environ.get('PROJECT_ID')
    zone = os.environ.get('ZONE')
    instance = os.environ.get('INSTANCE')
    gh_repo = os.environ.get('GH_REPO')

    # Setup logging
    logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
    logger = logging.getLogger(__name__)
    logger.info(f"Environment: PROJECT_ID={project}, ZONE={zone}, INSTANCE={instance}, GH_REPO={gh_repo}")

    # Parse request data
    trigger_source = "scheduler"
    if request and hasattr(request, 'is_json') and request.is_json:
        data = request.get_json()
        if data and 'trigger_source' in data:
            trigger_source = data.get('trigger_source')

    logger.info(f"Start triggered by: {trigger_source}")

    try:
        # Get gcloud path
        gcloud_path = get_gcloud_path()
        logger.info(f"Using gcloud path: {gcloud_path}")

        # Check if instance is running
        status_cmd = [gcloud_path, "compute", "instances", "describe", instance,
                     f"--zone={zone}", f"--project={project}",
                     "--format=get(status)"]
        status = subprocess.run(status_cmd, text=True, capture_output=True)

        is_running = status.stdout.strip().upper() == "RUNNING"

        if not is_running:
            # Start the instance
            start_cmd = [gcloud_path, "compute", "instances", "start", instance,
                        f"--zone={zone}", f"--project={project}", "--quiet"]
            logger.info(f"Starting instance with command: {' '.join(start_cmd)}")
            process = subprocess.run(start_cmd, text=True, capture_output=True)

            if process.returncode != 0:
                error_msg = f"Failed to start instance: {process.stderr}"
                logger.error(error_msg)
                return jsonify({'status': 'error', 'message': error_msg}), 500

            # Wait for VM to be fully started
            logger.info("Waiting 90 seconds for VM to be fully ready...")
            time.sleep(90)

        # Execute startup script
        logger.info("Executing startup script...")
        startup_cmd = [gcloud_path, "compute", "ssh", "--quiet", f"airflow@{instance}",
                      f"--zone={zone}", f"--project={project}", 
                      "--command", f"bash /opt/airflow/{gh_repo}/Cloud/infrastructure/terraform/vm_scripts/start_dag.sh"]

        process = subprocess.run(startup_cmd, text=True, capture_output=True)

        if process.returncode == 0:
            logger.info("Startup script executed successfully")
            logger.info(f"Startup stdout: {process.stdout}")
            msg = f'Successfully started instance and executed startup script'
            return jsonify({'status': 'success', 'message': msg})
        else:
            error_msg = f"Startup script failed: {process.stderr}"
            logger.error(error_msg)
            return jsonify({'status': 'error', 'message': error_msg}), 500

    except Exception as e:
        error_msg = f'Error during startup process: {str(e)}'
        logger.error(error_msg)
        logger.error(f"Full stack trace: {traceback.format_exc()}")
        return jsonify({'status': 'error', 'message': error_msg}), 500

import os
import json
import logging
import subprocess
from flask import Flask, request, jsonify
import platform
import traceback

# Create Flask app
app = Flask(__name__)

def get_gcloud_path():
    """Get the full path to gcloud executable based on OS"""
    if platform.system() == "Windows":
        # Common installation paths for gcloud on Windows
        possible_paths = [
            r"C:\Program Files (x86)\Google\Cloud SDK\google-cloud-sdk\bin\gcloud.cmd",
            r"C:\Program Files\Google\Cloud SDK\google-cloud-sdk\bin\gcloud.cmd",
            os.path.expanduser("~") + r"\AppData\Local\Google\Cloud SDK\google-cloud-sdk\bin\gcloud.cmd"
        ]
        for path in possible_paths:
            if os.path.exists(path):
                return path
        raise Exception("gcloud not found. Please ensure Google Cloud SDK is installed and in PATH")
    return "gcloud"  # For non-Windows systems, assume it's in PATH

@app.route("/", methods=["POST"])
def stop_vm_http():
    """
    HTTP endpoint for Cloud Run to stop a Compute Engine instance.
    Accepts POST requests with optional JSON body:
    {'trigger_source': 'scheduler'|'manual'|'dag'}
    """
    return stop_vm(request)

def stop_vm(request):
    """
    Function to stop a Compute Engine instance.
    Only runs shutdown script if VM is running.

    Args:
        request (flask.Request): The request object
        - If called from manual trigger: expects JSON with {'trigger_source': 'manual'}
        - If called from Scheduler: no specific payload needed
    Returns:
        JSON response with status message
    """
    project = os.environ.get('PROJECT_ID')
    zone = os.environ.get('ZONE')
    instance = os.environ.get('INSTANCE')
    gh_repo = os.environ.get('GH_REPO')

    # Setup detailed logging
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(levelname)s - %(message)s'
    )
    logger = logging.getLogger(__name__)

    # Log environment variables (excluding any sensitive data)
    logger.info(f"Environment: PROJECT_ID={project}, ZONE={zone}, INSTANCE={instance}, GH_REPO={gh_repo}")

    # Parse request data if present
    trigger_source = "scheduler"
    if request and hasattr(request, 'is_json') and request.is_json:
        data = request.get_json()
        if data and 'trigger_source' in data:
            trigger_source = data.get('trigger_source')

    logger.info(f"Shutdown triggered by: {trigger_source}")

    try:
        # Get gcloud path
        gcloud_path = get_gcloud_path()
        logger.info(f"Using gcloud path: {gcloud_path}")

        # Check if instance is running
        status_cmd = [gcloud_path, "compute", "instances", "describe", instance, 
                     f"--zone={zone}", f"--project={project}", 
                     "--format=get(status)"]
        status = subprocess.run(status_cmd, text=True, capture_output=True)

        is_running = status.stdout.strip().upper() == "RUNNING"

        if not is_running:
            msg = f'Instance {instance} is already stopped (Triggered by: {trigger_source})'
            logger.info(msg)
            return jsonify({'status': 'success', 'message': msg})

        # Execute shutdown script via SSH since VM is running
        logger.info("Executing shutdown script via SSH...")
        shutdown_cmd = f"bash /opt/airflow/{gh_repo}/Cloud/infrastructure/terraform/vm_scripts/shutdown.sh"
        ssh_cmd = [gcloud_path, "compute", "ssh", "--quiet", f"airflow@{instance}", 
                  f"--zone={zone}", f"--project={project}", "--command", shutdown_cmd]

        logger.info(f"Executing command: {' '.join(ssh_cmd)}")
        process = subprocess.run(ssh_cmd, text=True, capture_output=True)

        if process.returncode != 0:
            logger.error(f"SSH command failed with return code {process.returncode}")
            logger.error(f"SSH stderr: {process.stderr}")
            logger.error(f"SSH stdout: {process.stdout}")
            raise Exception(f"Failed to execute shutdown script: {process.stderr}")

        logger.info("Shutdown script executed successfully. Stopping instance...")
        logger.info(f"SSH stdout: {process.stdout}")

        # Stop the instance
        stop_cmd = [gcloud_path, "compute", "instances", "stop", instance, 
                   f"--zone={zone}", f"--project={project}", "--quiet"]
        logger.info(f"Executing stop command: {' '.join(stop_cmd)}")
        process = subprocess.run(stop_cmd, text=True, capture_output=True)

        if process.returncode != 0:
            logger.error(f"Stop command failed with return code {process.returncode}")
            logger.error(f"Stop stderr: {process.stderr}")
            logger.error(f"Stop stdout: {process.stdout}")
            raise Exception(f"Failed to stop instance: {process.stderr}")

        logger.info(f"Stop command stdout: {process.stdout}")
        msg = f'Successfully executed shutdown script and stopped instance {instance} (Triggered by: {trigger_source})'
        logger.info(msg)
        return jsonify({'status': 'success', 'message': msg})

    except Exception as e:
        error_msg = f'Error during shutdown process: {str(e)}'
        logger.error(error_msg)
        logger.error(f"Full stack trace: {traceback.format_exc()}")
        return jsonify({'status': 'error', 'message': error_msg}), 500

if __name__ == "__main__":
    # For local testing with Flask
    app.run(host="0.0.0.0", port=int(os.environ.get("PORT", 9099)), debug=False)