for Podcasting Camera vs Medium: Which Wins?

import logging
import os
import time
from pathlib import Path
from typing import Optional

# Import Podcasting Camera SDK v2.1.0 from https://github.com/podcastindex/podcasting-camera
from podcasting_camera import CameraClient, VideoEncoder, S3Uploader
from podcasting_camera.exceptions import CameraConnectionError, EncodeError, UploadError

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

class PodcastCameraProcessor:
    """Processes raw camera feeds for podcast episodes with benchmarked latency."""

    def __init__(self, camera_index: int = 0, s3_bucket: str = "podcast-media-prod"):
        self.camera_index = camera_index
        self.s3_bucket = s3_bucket
        self.client: Optional[CameraClient] = None
        self.encoder: Optional[VideoEncoder] = None
        self.uploader: Optional[S3Uploader] = None

    def initialize(self) -> bool:
        """Initialize all components with error handling."""
        try:
            # Initialize camera client with 1080p resolution, 30fps
            self.client = CameraClient(
                camera_index=self.camera_index,
                resolution=(1920, 1080),
                frame_rate=30,
                buffer_size=512  # MB
            )
            if not self.client.connect():
                logger.error(f"Failed to connect to camera {self.camera_index}")
                return False

            # Initialize H.264 encoder with CRF 23 (balanced quality/file size)
            self.encoder = VideoEncoder(
                codec="h264",
                crf=23,
                preset="fast",
                audio_bitrate=128  # kbps
            )

            # Initialize S3 uploader with retry logic
            self.uploader = S3Uploader(
                bucket_name=self.s3_bucket,
                aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
                aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"),
                region_name="us-east-1",
                max_retries=3
            )
            logger.info("All components initialized successfully")
            return True
        except CameraConnectionError as e:
            logger.error(f"Camera connection failed: {e}")
            return False
        except Exception as e:
            logger.error(f"Unexpected initialization error: {e}")
            return False

    def process_episode(self, duration_sec: int, output_path: Path) -> Optional[str]:
        """
        Capture, encode, and upload a podcast episode.

        Args:
            duration_sec: Episode duration in seconds
            output_path: Local path to save encoded file

        Returns:
            S3 URL of uploaded episode, or None if failed
        """
        start_time = time.perf_counter()
        try:
            # Capture raw frames from camera
            logger.info(f"Capturing {duration_sec}s of video from camera {self.camera_index}")
            raw_frames = self.client.capture_frames(duration_sec)
            capture_time = time.perf_counter() - start_time
            logger.info(f"Capture completed in {capture_time:.2f}s")

            # Encode raw frames to H.264
            encode_start = time.perf_counter()
            encoded_path = self.encoder.encode(
                frames=raw_frames,
                output_path=output_path,
                metadata={"episode_id": "ep-2024-001", "codec": "h264"}
            )
            encode_time = time.perf_counter() - encode_start
            logger.info(f"Encode completed in {encode_time:.2f}s")

            # Upload to S3
            upload_start = time.perf_counter()
            s3_url = self.uploader.upload(
                file_path=encoded_path,
                object_key=f"episodes/{output_path.name}"
            )
            upload_time = time.perf_counter() - upload_start
            logger.info(f"Upload completed in {upload_time:.2f}s")

            total_time = time.perf_counter() - start_time
            logger.info(f"Total processing time for {duration_sec}s episode: {total_time:.2f}s")
            return s3_url
        except EncodeError as e:
            logger.error(f"Encoding failed: {e}")
            return None
        except UploadError as e:
            logger.error(f"Upload failed: {e}")
            return None
        except Exception as e:
            logger.error(f"Unexpected processing error: {e}")
            return None
        finally:
            # Cleanup local encoded file
            if output_path.exists():
                output_path.unlink()
                logger.info(f"Cleaned up local file {output_path}")

if __name__ == "__main__":
    # Benchmark configuration (matches methodology: AWS t3.medium, Python 3.11.4)
    processor = PodcastCameraProcessor(camera_index=0, s3_bucket="podcast-media-prod")
    if not processor.initialize():
        logger.error("Failed to initialize processor. Exiting.")
        exit(1)

    # Process a 10-minute test episode (600 seconds)
    result = processor.process_episode(
        duration_sec=600,
        output_path=Path("/tmp/ep-2024-001.mp4")
    )

    if result:
        logger.info(f"Episode uploaded successfully: {result}")
    else:
        logger.error("Episode processing failed")

import logging
import os
import time
from typing import List, Optional

# Import Medium SDK v1.3.2 from https://github.com/Medium/medium-sdk-python
from medium import MediumClient, Post, MediumAPIError, RateLimitError
from medium.models import Publication

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

class PodcastMediumPublisher:
    """Publishes podcast show notes and audio embeds to Medium with rate limit handling."""

    def __init__(self, publication_id: Optional[str] = None):
        self.publication_id = publication_id
        self.client: Optional[MediumClient] = None

    def initialize(self) -> bool:
        """Initialize Medium client with API token from env."""
        try:
            api_token = os.getenv("MEDIUM_API_TOKEN")
            if not api_token:
                logger.error("MEDIUM_API_TOKEN environment variable not set")
                return False

            self.client = MediumClient(api_token=api_token)
            # Verify token by fetching user details
            user = self.client.get_current_user()
            logger.info(f"Authenticated as Medium user: {user.username} (ID: {user.id})")
            return True
        except MediumAPIError as e:
            logger.error(f"Medium API authentication failed: {e}")
            return False
        except Exception as e:
            logger.error(f"Unexpected initialization error: {e}")
            return False

    def publish_show_notes(
        self,
        title: str,
        episode_id: str,
        audio_url: str,
        show_notes: str,
        tags: List[str]
    ) -> Optional[str]:
        """
        Publish show notes post with embedded audio.

        Args:
            title: Post title
            episode_id: Podcast episode ID
            audio_url: Public URL of episode audio/video
            show_notes: Markdown show notes content
            tags: List of tags for the post

        Returns:
            Medium post URL, or None if failed
        """
        start_time = time.perf_counter()
        try:
            # Construct post content with audio embed and show notes
            content = f"""## Episode {episode_id} Show Notes



  Your browser does not support the audio element.


{show_notes}

*Subscribe to the podcast on [Apple Podcasts](https://apple.com/podcasts) or [Spotify](https://spotify.com).*
"""

            # Create post object
            post = Post(
                title=title,
                content=content,
                content_format="markdown",
                tags=tags,
                publish_status="public"  # Options: public, draft, unlisted
            )

            # Publish to publication if specified, else to user's profile
            if self.publication_id:
                publication = Publication(id=self.publication_id)
                result = self.client.create_post_in_publication(publication, post)
                logger.info(f"Published to publication {self.publication_id}")
            else:
                result = self.client.create_post(post)
                logger.info("Published to user profile")

            post_url = f"https://medium.com/p/{result.id}"
            total_time = time.perf_counter() - start_time
            logger.info(f"Post published in {total_time:.2f}s: {post_url}")
            return post_url
        except RateLimitError as e:
            logger.error(f"Rate limit exceeded: {e}. Retry after {e.retry_after}s")
            time.sleep(e.retry_after)
            return self.publish_show_notes(title, episode_id, audio_url, show_notes, tags)
        except MediumAPIError as e:
            logger.error(f"Medium API error: {e}")
            return None
        except Exception as e:
            logger.error(f"Unexpected publishing error: {e}")
            return None

    def get_post_analytics(self, post_id: str) -> dict:
        """Fetch analytics for a published post (requires publication admin access)."""
        try:
            analytics = self.client.get_post_analytics(post_id)
            return {
                "views": analytics.views,
                "reads": analytics.reads,
                "claps": analytics.claps
            }
        except MediumAPIError as e:
            logger.error(f"Failed to fetch analytics: {e}")
            return {}

if __name__ == "__main__":
    # Initialize publisher
    publisher = PodcastMediumPublisher(publication_id=os.getenv("MEDIUM_PUB_ID"))
    if not publisher.initialize():
        logger.error("Failed to initialize Medium publisher. Exiting.")
        exit(1)

    # Publish test show notes for ep-2024-001
    post_url = publisher.publish_show_notes(
        title="Podcasting Camera vs Medium: Benchmark Review",
        episode_id="ep-2024-001",
        audio_url="https://s3.amazonaws.com/podcast-media-prod/episodes/ep-2024-001.mp4",
        show_notes="""### Key Topics
- Benchmark methodology for podcast tools
- Cost comparison of media processing vs publishing
- Workflow integration tips

### Links
- Podcasting Camera Repo: https://github.com/podcastindex/podcasting-camera
- Medium SDK Repo: https://github.com/Medium/medium-sdk-python""",
        tags=["podcasting", "developer-tools", "benchmarks"]
    )

    if post_url:
        logger.info(f"Published show notes: {post_url}")
        # Fetch analytics after 1 minute (simulate delay)
        time.sleep(60)
        analytics = publisher.get_post_analytics(post_url.split("/")[-1])
        logger.info(f"Post analytics: {analytics}")
    else:
        logger.error("Failed to publish show notes")

import logging
import os
import time
from pathlib import Path
from typing import Optional

# Import both SDKs
from podcasting_camera import CameraClient, VideoEncoder
from medium import MediumClient, Post

from podcasting_camera.exceptions import CameraError
from medium.exceptions import MediumError

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class HybridPodcastPipeline:
    """Automated pipeline combining media processing and publishing."""

    def __init__(self):
        self.camera_client: Optional[CameraClient] = None
        self.encoder: Optional[VideoEncoder] = None
        self.medium_client: Optional[MediumClient] = None

    def initialize(self) -> bool:
        """Initialize all components."""
        try:
            # Init Podcasting Camera
            self.camera_client = CameraClient(
                camera_index=0,
                resolution=(1920, 1080),
                frame_rate=30
            )
            if not self.camera_client.connect():
                raise CameraError("Failed to connect to camera")
            self.encoder = VideoEncoder(codec="h264", crf=23)

            # Init Medium
            self.medium_client = MediumClient(api_token=os.getenv("MEDIUM_API_TOKEN"))
            self.medium_client.get_current_user()  # Verify auth

            logger.info("Hybrid pipeline initialized")
            return True
        except CameraError as e:
            logger.error(f"Camera init failed: {e}")
            return False
        except MediumError as e:
            logger.error(f"Medium init failed: {e}")
            return False

    def run_pipeline(
        self,
        episode_duration: int,
        episode_id: str,
        show_notes: str,
        s3_bucket: str
    ) -> bool:
        """Run full pipeline: capture -> encode -> upload -> publish."""
        try:
            # Step 1: Capture and encode
            logger.info(f"Starting media processing for {episode_id}")
            raw_frames = self.camera_client.capture_frames(episode_duration)
            encoded_path = Path(f"/tmp/{episode_id}.mp4")
            self.encoder.encode(raw_frames, encoded_path)

            # Step 2: Upload to S3 (simplified for example)
            s3_url = f"https://{s3_bucket}.s3.amazonaws.com/episodes/{encoded_path.name}"
            logger.info(f"Media uploaded to {s3_url}")

            # Step 3: Publish to Medium
            post = Post(
                title=f"Episode {episode_id} Show Notes",
                content=f"\n\n{show_notes}",
                content_format="html",
                tags=["podcasting"]
            )
            result = self.medium_client.create_post(post)
            logger.info(f"Published to Medium: https://medium.com/p/{result.id}")

            # Cleanup
            encoded_path.unlink()
            return True
        except Exception as e:
            logger.error(f"Pipeline failed: {e}")
            return False

if __name__ == "__main__":
    pipeline = HybridPodcastPipeline()
    if pipeline.initialize():
        pipeline.run_pipeline(
            episode_duration=600,
            episode_id="ep-2024-002",
            show_notes="Test show notes for hybrid pipeline.",
            s3_bucket="podcast-media-prod"
        )

from podcasting_camera import VideoEncoder
from threading import Lock

class EncoderPool:
    def __init__(self, pool_size: int = 4):
        self.pool = [VideoEncoder(codec="h264", crf=23) for _ in range(pool_size)]
        self.lock = Lock()

    def get_encoder(self) -> VideoEncoder:
        with self.lock:
            return self.pool.pop(0)

    def return_encoder(self, encoder: VideoEncoder):
        with self.lock:
            self.pool.append(encoder)

from medium import MediumClient, Post

client = MediumClient(api_token="your_token")
posts = [
    Post(title=f"Episode {i} Show Notes", content="...", content_format="markdown")
    for i in range(25)
]

# Batch publish to user profile
results = client.batch_create_posts(posts)
print(f"Published {len(results)} posts in 1 request")

import time
from medium import MediumClient, RateLimitError

def retry_medium_request(func, *args, max_retries=5, **kwargs):
    for attempt in range(max_retries):
        try:
            return func(*args, **kwargs)
        except RateLimitError as e:
            sleep_time = e.retry_after + (2 ** attempt) + (time.random() * 0.1)
            time.sleep(sleep_time)
    raise Exception(f"Max retries exceeded for {func.__name__}")

client = MediumClient(api_token="your_token")
post = Post(title="Test", content="...", content_format="markdown")
# Use retry wrapper instead of direct call
result = retry_medium_request(client.create_post, post)