Building an AI-Powered Podcast Generator with Gemini and Cloud Run

RSS_FEEDS = [
    "https://deepmind.google/blog/rss.xml",
    "https://raw.githubusercontent.com/Olshansk/rss-feeds/main/feeds/feed_anthropic_news.xml",
    "https://openai.com/blog/rss.xml",
    "https://simonwillison.net/atom/everything/"
]

def get_new_rss_entries(feed_urls, last_times):
    """
    Fetches new entries from a list of RSS feeds, ensuring a balanced selection.
    """
    all_new_entries = []
    latest_times = last_times.copy()
    # ... (headers and other setup)

    for url in feed_urls:
        # ... (fetches and parses the feed)

        # Sort entries for this feed by date and take the most recent 3
        feed_entries.sort(key=lambda e: e.published_parsed, reverse=True)
        all_new_entries.extend(feed_entries[:3])

    # Sort all collected entries by date to ensure a chronological podcast
    all_new_entries.sort(key=lambda e: e.published_parsed)
    return all_new_entries, latest_times

def summarize_entries(entries, api_key):
    """Summarizes each RSS entry individually in English."""
    client = genai.Client(api_key=api_key)
    individual_summaries = []

    for entry in entries:
        content = entry.get('content', [{}])[0].get('value', entry.get('summary', ''))
        prompt = f"""
        Your role is a professional podcast host writing a script for an English-language audio briefing on Artificial Intelligence.
        Your task is to summarize the following article.

        Guidelines for the summary:
        - Write in a natural, conversational, and engaging podcast style.
        - The output must be a clean paragraph of plain text.
        - It must be suitable for direct text-to-speech conversion.

        **CRITICAL INSTRUCTIONS:**
        - **DO NOT** use any Markdown formatting.
        - **DO NOT** begin with conversational filler like "Of course, here is a summary...".
        - **DO NOT** announce what you are doing. Just provide the summary directly.

        Article to summarize:
        Title: {entry.title}
        Content: {content}
        """
        response = client.models.generate_content(
            model='gemini-2.5-pro',
            contents=prompt
        )
        individual_summaries.append({'title': entry.title, 'summary': response.text.strip()})

    return individual_summaries

def generate_and_upload_stitched_audio(summaries, bucket_name):
    """
    Generates audio for each summary, stitches them together, and uploads to GCS.
    """
    try:
        tts_client = texttospeech.TextToSpeechClient()
        voice = texttospeech.VoiceSelectionParams(language_code="en-US", name="en-US-Studio-O")
        audio_config = texttospeech.AudioConfig(audio_encoding=texttospeech.AudioEncoding.MP3)

        all_audio_segments = []

        # Generate and add the intro
        intro_text = "Good morning, and welcome to your AI briefing. Here is the latest news."
        intro_segment = text_to_audio_segment(intro_text, tts_client, ...)
        if intro_segment:
            all_audio_segments.append(intro_segment)

        # Loop through summaries, creating audio for the title and content
        for summary in summaries:
            title_text = f"The next story is titled: {summary['title']}."
            title_segment = text_to_audio_segment(title_text, tts_client, ...)
            if title_segment:
                all_audio_segments.append(title_segment)

            summary_segment = text_to_audio_segment(summary['summary'], tts_client, ...)
            if summary_segment:
                all_audio_segments.append(summary_segment)
        # Generate and add the outro
        outro_text = "And that's all for your briefing today. Thanks for listening."
        outro_segment = text_to_audio_segment(outro_text, tts_client, ...)
        if outro_segment:
            all_audio_segments.append(outro_segment)
        # Stitch everything together using pydub and upload
        final_audio = sum(all_audio_segments)

        # ... export and upload to GCS ...

        return "gs://<your-bucket-name>/summary-YYYY-MM-DD.mp3"

    except Exception as e:
        # ... error handling ...
        return None

def generate_signed_url(bucket_name, object_name, expiration_minutes=1440):
    """Generate a signed URL using impersonated credentials."""
    try:
        # Get default credentials from Cloud Run metadata server
        source_credentials, project = default()

        # Create impersonated credentials for the target service account
        target_credentials = impersonated_credentials.Credentials(
            source_credentials=source_credentials,
            target_principal=SERVICE_ACCOUNT_EMAIL,
            target_scopes=['https://www.googleapis.com/auth/cloud-platform'],
        )

        # Create storage client with impersonated credentials
        storage_client = storage.Client(credentials=target_credentials)
        bucket = storage_client.bucket(bucket_name)
        blob = bucket.blob(object_name)

        # Generate signed URL (24 hours expiration)
        signed_url = blob.generate_signed_url(
            version="v4",
            expiration=timedelta(minutes=expiration_minutes),
            method="GET"
        )

        return signed_url

    except Exception as e:
        # ... error handling ...
        return None

@app.route('/')
def index():
    """Main page showing available audio episodes."""
    try:
        storage_client = storage.Client()
        bucket = storage_client.bucket(GCS_BUCKET_NAME)

        # List all MP3 files in the bucket
        blobs = bucket.list_blobs(prefix="summary-", delimiter=".mp3")
        episodes = []

        for blob in blobs:
            if blob.name.endswith('.mp3'):
                # Extract date from filename (summary-YYYY-MM-DD.mp3)
                date_str = blob.name.replace('summary-', '').replace('.mp3', '')

                # Generate secure signed URL for audio access
                signed_url = generate_signed_url(GCS_BUCKET_NAME, blob.name)

                if signed_url:
                    episodes.append({
                        'filename': blob.name,
                        'date': date_str,
                        'signed_url': signed_url,
                        'created': blob.time_created
                    })

        # Sort episodes by date (newest first)
        episodes.sort(key=lambda x: x['created'], reverse=True)

        return render_template('index.html', episodes=episodes)

    except Exception as e:
        # ... error handling ...
        return render_template('error.html')