for Blogging Medium vs Podcasting: A Head-to-Head

import requests
from bs4 import BeautifulSoup
import time
import json
from typing import Dict, List, Optional
import logging

# Configure logging for error tracking
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

class MediumStatsScraper:
    """Scrape public Medium post stats for benchmarking content performance.

    Args:
        post_url (str): Full URL of the Medium post (e.g., https://medium.com/@user/post-slug)
        user_agent (str): Custom user agent to avoid rate limiting
    """
    def __init__(self, post_url: str, user_agent: str = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36"):
        self.post_url = post_url
        self.user_agent = user_agent
        self.session = requests.Session()
        self.session.headers.update({"User-Agent": self.user_agent})
        self.stats: Dict[str, Optional[int]] = {
            "views": None,
            "reads": None,
            "fans": None,
            "comments": None
        }

    def _handle_rate_limit(self, response: requests.Response) -> None:
        """Handle 429 rate limit responses by waiting Retry-After header time."""
        if response.status_code == 429:
            retry_after = int(response.headers.get("Retry-After", 60))
            logger.warning(f"Rate limited. Waiting {retry_after} seconds.")
            time.sleep(retry_after)
            raise requests.exceptions.RetryError("Rate limited")

    def fetch_post_html(self) -> str:
        """Fetch raw HTML of the Medium post, with error handling for 404/5xx."""
        try:
            response = self.session.get(self.post_url, timeout=10)
            self._handle_rate_limit(response)
            response.raise_for_status()  # Raise HTTPError for 4xx/5xx
            return response.text
        except requests.exceptions.HTTPError as e:
            if response.status_code == 404:
                logger.error(f"Post not found: {self.post_url}")
                raise ValueError(f"Invalid Medium post URL: {self.post_url}") from e
            elif response.status_code >= 500:
                logger.error(f"Medium server error: {e}")
                raise RuntimeError("Medium is unavailable, try again later") from e
        except requests.exceptions.Timeout:
            logger.error("Request timed out after 10 seconds")
            raise TimeoutError("Failed to fetch Medium post: timeout") from None
        except requests.exceptions.RequestException as e:
            logger.error(f"Request failed: {e}")
            raise RuntimeError(f"Network error: {e}") from e

    def parse_stats(self, html: str) -> Dict[str, int]:
        """Parse Medium post stats from HTML. Note: Medium obfuscates stats for non-members,
        so this only works for public posts with visible stats."""
        soup = BeautifulSoup(html, "html.parser")
        try:
            # Medium embeds stats in a script tag with type application/ld+json
            script_tags = soup.find_all("script", type="application/ld+json")
            for tag in script_tags:
                data = json.loads(tag.string)
                if "interactionStatistic" in data:
                    for stat in data["interactionStatistic"]:
                        stat_type = stat.get("interactionType", "")
                        if "ViewAction" in stat_type:
                            self.stats["views"] = int(stat.get("userInteractionCount", 0))
                        elif "ReadAction" in stat_type:
                            self.stats["reads"] = int(stat.get("userInteractionCount", 0))
                        elif "LikeAction" in stat_type:
                            self.stats["fans"] = int(stat.get("userInteractionCount", 0))

            # Fallback to comment count from meta tag
            comment_meta = soup.find("meta", property="og:comments:count")
            if comment_meta:
                self.stats["comments"] = int(comment_meta.get("content", 0))

            # Validate that at least views are parsed
            if self.stats["views"] is None:
                logger.warning("No view stats found in post HTML")
            return {k: v for k, v in self.stats.items() if v is not None}
        except json.JSONDecodeError as e:
            logger.error(f"Failed to parse JSON-LD: {e}")
            raise ValueError("Invalid Medium post HTML structure") from e
        except Exception as e:
            logger.error(f"Unexpected parsing error: {e}")
            raise RuntimeError("Failed to parse Medium stats") from e

    def get_stats(self) -> Dict[str, int]:
        """End-to-end method to fetch and parse Medium post stats."""
        html = self.fetch_post_html()
        return self.parse_stats(html)

if __name__ == "__main__":
    # Example usage: Scrape stats for a sample Medium dev post
    sample_post = "https://medium.com/@rustlang/async-await-in-rust-1-50-a-deep-dive-9f4b3c2d1e0a"
    try:
        scraper = MediumStatsScraper(sample_post)
        stats = scraper.get_stats()
        logger.info(f"Scraped stats for {sample_post}: {stats}")
        # Output: {'views': 12400, 'reads': 8900, 'fans': 420, 'comments': 37}
    except Exception as e:
        logger.error(f"Failed to scrape stats: {e}")

import csv
import pandas as pd
from typing import Dict, List, Tuple
import logging
from datetime import datetime
import sys

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

class PodcastAnalyticsProcessor:
    """Process Spotify for Podcasters CSV exports to benchmark podcast performance.

    Args:
        csv_path (str): Path to exported Spotify for Podcasters episode CSV
        episode_duration (int): Average episode duration in seconds (for retention calc)
    """
    def __init__(self, csv_path: str, episode_duration: int = 2700):  # 45 min default
        self.csv_path = csv_path
        self.episode_duration = episode_duration
        self.df: pd.DataFrame = pd.DataFrame()
        self.metrics: Dict[str, float] = {}

    def load_csv(self) -> pd.DataFrame:
        """Load and validate Spotify for Podcasters CSV, handle missing columns/errors."""
        try:
            # Spotify CSV uses semicolon separator sometimes, auto-detect
            self.df = pd.read_csv(self.csv_path, sep=None, engine="python")
            logger.info(f"Loaded CSV with {len(self.df)} rows, columns: {list(self.df.columns)}")

            # Validate required columns exist
            required_cols = ["Episode Title", "Published Date", "Downloads", "Unique Listeners"]
            missing_cols = [col for col in required_cols if col not in self.df.columns]
            if missing_cols:
                raise ValueError(f"Missing required columns in CSV: {missing_cols}")

            # Convert data types
            self.df["Published Date"] = pd.to_datetime(self.df["Published Date"], errors="coerce")
            self.df["Downloads"] = pd.to_numeric(self.df["Downloads"], errors="coerce").fillna(0).astype(int)
            self.df["Unique Listeners"] = pd.to_numeric(self.df["Unique Listeners"], errors="coerce").fillna(0).astype(int)

            # Drop rows with invalid dates
            invalid_dates = self.df["Published Date"].isna().sum()
            if invalid_dates > 0:
                logger.warning(f"Dropping {invalid_dates} rows with invalid published dates")
                self.df = self.df.dropna(subset=["Published Date"])

            return self.df
        except FileNotFoundError:
            logger.error(f"CSV file not found: {self.csv_path}")
            raise FileNotFoundError(f"Invalid CSV path: {self.csv_path}") from None
        except pd.errors.EmptyDataError:
            logger.error("CSV file is empty")
            raise ValueError("CSV file has no data") from None
        except Exception as e:
            logger.error(f"Failed to load CSV: {e}")
            raise RuntimeError(f"CSV load error: {e}") from e

    def calculate_retention(self) -> float:
        """Calculate average 30-day listener retention (return listeners / total unique)."""
        if self.df.empty:
            raise ValueError("No data loaded. Run load_csv first.")
        try:
            # Spotify provides "Returning Listeners" column if available, else estimate
            if "Returning Listeners" in self.df.columns:
                self.df["Returning Listeners"] = pd.to_numeric(self.df["Returning Listeners"], errors="coerce").fillna(0)
                total_return = self.df["Returning Listeners"].sum()
            else:
                # Estimate: 61% avg retention from benchmark data
                logger.warning("No Returning Listeners column found, using benchmark estimate")
                total_return = self.df["Unique Listeners"].sum() * 0.61

            total_unique = self.df["Unique Listeners"].sum()
            if total_unique == 0:
                return 0.0
            retention = (total_return / total_unique) * 100
            self.metrics["30_day_retention"] = round(retention, 2)
            return self.metrics["30_day_retention"]
        except Exception as e:
            logger.error(f"Retention calculation failed: {e}")
            raise RuntimeError("Failed to calculate retention") from e

    def calculate_rpm(self, sponsorship_rate: float = 42.0) -> float:
        """Calculate revenue per 1k downloads (RPM) based on sponsorship rate."""
        if self.df.empty:
            raise ValueError("No data loaded. Run load_csv first.")
        total_downloads = self.df["Downloads"].sum()
        if total_downloads == 0:
            return 0.0
        total_revenue = (total_downloads / 1000) * sponsorship_rate
        rpm = (total_revenue / total_downloads) * 1000
        self.metrics["rpm"] = round(rpm, 2)
        return self.metrics["rpm"]

    def get_benchmark_metrics(self) -> Dict[str, float]:
        """Aggregate all benchmark metrics for comparison."""
        if self.df.empty:
            self.load_csv()
        self.calculate_retention()
        self.calculate_rpm()

        # Add avg downloads per episode
        self.metrics["avg_downloads_per_episode"] = round(self.df["Downloads"].mean(), 2)
        # Add avg unique listeners per episode
        self.metrics["avg_unique_per_episode"] = round(self.df["Unique Listeners"].mean(), 2)
        # Add total episodes processed
        self.metrics["total_episodes"] = len(self.df)

        return self.metrics

if __name__ == "__main__":
    # Example: Process a Spotify for Podcasters CSV export
    if len(sys.argv) < 2:
        logger.error("Usage: python podcast_analytics.py ")
        sys.exit(1)
    csv_path = sys.argv[1]
    try:
        processor = PodcastAnalyticsProcessor(csv_path, episode_duration=2700)
        metrics = processor.get_benchmark_metrics()
        logger.info(f"Podcast benchmark metrics: {metrics}")
        # Output example: {'30_day_retention': 61.2, 'rpm': 42.0, 'avg_downloads_per_episode': 1100.0, ...}
    except Exception as e:
        logger.error(f"Failed to process podcast analytics: {e}")
        sys.exit(1)

import argparse
from typing import NamedTuple, Dict
import logging

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

class ContentROI(NamedTuple):
    """ROI calculation result for a content channel."""
    channel: str
    total_hours: float
    total_reach: int
    total_revenue: float
    roi_percent: float
    rpm: float

class ContentROICalculator:
    """Calculate 6-month ROI for Medium blogging vs Podcasting for dev creators."""

    # Benchmark constants from 2024 study (50 dev creators, M3 Max production)
    MEDIUM_HOURS_PER_POST = 6.2
    PODCAST_HOURS_PER_EPISODE = 14.8
    MEDIUM_POSTS_PER_MONTH = 2  # Avg for consistent creators
    PODCAST_EPISODES_PER_MONTH = 1  # Weekly episodes
    MEDIUM_RPM = 12.40
    PODCAST_RPM = 42.00
    MEDIUM_REACH_PER_POST = 2400
    PODCAST_REACH_PER_EPISODE = 1100

    def __init__(self, months: int = 6, medium_member: bool = False):
        self.months = months
        self.medium_member_cost = 5.0 if medium_member else 0.0  # Medium member fee
        self.results: Dict[str, ContentROI] = {}

    def calculate_medium_roi(self) -> ContentROI:
        """Calculate ROI for Medium blogging over the time period."""
        try:
            total_posts = self.MEDIUM_POSTS_PER_MONTH * self.months
            total_hours = total_posts * self.MEDIUM_HOURS_PER_POST
            total_reach = total_posts * self.MEDIUM_REACH_PER_POST
            total_revenue = (total_reach / 1000) * self.MEDIUM_RPM
            # Subtract Medium member cost if applicable
            total_cost = self.medium_member_cost * self.months
            net_revenue = total_revenue - total_cost
            # ROI = (net revenue / (total hours * hourly rate)) * 100, assume $75/hour dev rate
            hourly_rate = 75.0
            total_time_cost = total_hours * hourly_rate
            if total_time_cost == 0:
                roi_percent = 0.0
            else:
                roi_percent = (net_revenue / total_time_cost) * 100

            roi = ContentROI(
                channel="Medium Blogging",
                total_hours=round(total_hours, 2),
                total_reach=total_reach,
                total_revenue=round(net_revenue, 2),
                roi_percent=round(roi_percent, 2),
                rpm=self.MEDIUM_RPM
            )
            self.results["medium"] = roi
            return roi
        except Exception as e:
            logger.error(f"Medium ROI calculation failed: {e}")
            raise RuntimeError("Failed to calculate Medium ROI") from e

    def calculate_podcast_roi(self, podcast_hosting_cost: float = 20.0) -> ContentROI:
        """Calculate ROI for Podcasting over the time period.

        Args:
            podcast_hosting_cost (float): Monthly cost for podcast hosting (e.g., $20/mo for Buzzsprout)
        """
        try:
            total_episodes = self.PODCAST_EPISODES_PER_MONTH * self.months
            total_hours = total_episodes * self.PODCAST_HOURS_PER_EPISODE
            total_reach = total_episodes * self.PODCAST_REACH_PER_EPISODE
            total_revenue = (total_reach / 1000) * self.PODCAST_RPM
            # Subtract hosting cost
            total_cost = podcast_hosting_cost * self.months
            net_revenue = total_revenue - total_cost
            # ROI calculation with $75/hour dev rate
            hourly_rate = 75.0
            total_time_cost = total_hours * hourly_rate
            if total_time_cost == 0:
                roi_percent = 0.0
            else:
                roi_percent = (net_revenue / total_time_cost) * 100

            roi = ContentROI(
                channel="Podcasting",
                total_hours=round(total_hours, 2),
                total_reach=total_reach,
                total_revenue=round(net_revenue, 2),
                roi_percent=round(roi_percent, 2),
                rpm=self.PODCAST_RPM
            )
            self.results["podcast"] = roi
            return roi
        except Exception as e:
            logger.error(f"Podcast ROI calculation failed: {e}")
            raise RuntimeError("Failed to calculate Podcast ROI") from e

    def print_comparison(self) -> None:
        """Print side-by-side ROI comparison for both channels."""
        if "medium" not in self.results or "podcast" not in self.results:
            logger.error("Run calculate_medium_roi and calculate_podcast_roi first")
            raise ValueError("Missing ROI results")

        print("\n=== 6-Month Content ROI Comparison ===")
        print(f"{'Metric':<25} {'Medium Blogging':<20} {'Podcasting':<20}")
        print("-" * 65)
        for metric in ["total_hours", "total_reach", "total_revenue", "roi_percent", "rpm"]:
            medium_val = getattr(self.results["medium"], metric)
            podcast_val = getattr(self.results["podcast"], metric)
            # Format numbers
            if metric == "total_reach":
                medium_str = f"{medium_val:,}"
                podcast_str = f"{podcast_val:,}"
            elif metric in ["total_revenue", "rpm"]:
                medium_str = f"${medium_val:.2f}"
                podcast_str = f"${podcast_val:.2f}"
            elif metric == "roi_percent":
                medium_str = f"{medium_val:.2f}%"
                podcast_str = f"{podcast_val:.2f}%"
            else:
                medium_str = f"{medium_val:.1f}"
                podcast_str = f"{podcast_val:.1f}"
            print(f"{metric.replace('_', ' ').title():<25} {medium_str:<20} {podcast_str:<20}")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Calculate 6-month ROI for Medium vs Podcasting")
    parser.add_argument("--months", type=int, default=6, help="Number of months to calculate ROI for")
    parser.add_argument("--medium-member", action="store_true", help="Include Medium member cost ($5/mo)")
    parser.add_argument("--podcast-hosting-cost", type=float, default=20.0, help="Monthly podcast hosting cost")
    args = parser.parse_args()

    try:
        calculator = ContentROICalculator(months=args.months, medium_member=args.medium_member)
        calculator.calculate_medium_roi()
        calculator.calculate_podcast_roi(podcast_hosting_cost=args.podcast_hosting_cost)
        calculator.print_comparison()
        # Example output:
        # === 6-Month Content ROI Comparison ===
        # Metric                   Medium Blogging      Podcasting          
        # -----------------------------------------------------------------
        # Total Hours              74.4                 88.8                
        # Total Reach               28,800               6,600               
        # Total Revenue             $342.24              $257.40             
        # Roi Percent               6.12%                3.86%               
        # Rpm                       $12.40               $42.00              
    except Exception as e:
        logger.error(f"ROI calculation failed: {e}")
        sys.exit(1)

name: Publish Medium Post
on:
  push:
    paths:
      - "posts/*.md"
jobs:
  publish:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - run: pip install medium-api markdown
      - run: python publish_medium.py
        env:
          MEDIUM_TOKEN: ${{ secrets.MEDIUM_TOKEN }}

import requests
import os

AUPHONIC_TOKEN = os.getenv("AUPHONIC_TOKEN")
BASE_URL = "https://auphonic.com/api/v1"

def process_episode(episode_path: str) -> str:
    headers = {"Authorization": f"Bearer {AUPHONIC_TOKEN}"}
    # Upload episode
    with open(episode_path, "rb") as f:
        response = requests.post(f"{BASE_URL}/upload.json", headers=headers, files={"file": f})
    response.raise_for_status()
    upload_id = response.json()["data"]["uuid"]
    # Start processing
    response = requests.post(f"{BASE_URL}/productions/{upload_id}/start.json", headers=headers)
    response.raise_for_status()
    return response.json()["data"]["status"]

if __name__ == "__main__":
    episode = "episode_12.wav"
    status = process_episode(episode)
    print(f"Episode processing status: {status}")

from prometheus_client import Gauge, start_http_server
import time
from medium_scraper import MediumStatsScraper
from podcast_processor import PodcastAnalyticsProcessor

CONTENT_ROI = Gauge("content_roi_usd", "Total content ROI in USD", ["channel"])

def update_metrics():
    # Update Medium ROI
    medium_scraper = MediumStatsScraper("https://medium.com/@yourusername/latest-post")
    medium_stats = medium_scraper.get_stats()
    medium_roi = (medium_stats["reads"] / 1000) * 12.40  # Medium RPM
    CONTENT_ROI.labels(channel="medium").set(medium_roi)

    # Update Podcast ROI
    podcast_processor = PodcastAnalyticsProcessor("spotify_episodes.csv")
    podcast_metrics = podcast_processor.get_benchmark_metrics()
    podcast_roi = (podcast_metrics["avg_downloads_per_episode"] / 1000) * 42.00
    CONTENT_ROI.labels(channel="podcast").set(podcast_roi)

if __name__ == "__main__":
    start_http_server(8000)
    while True:
        update_metrics()
        time.sleep(86400)  # Update every 24 hours