How to for YouTube Blogging: Lessons Learned

import os
import re
import json
import argparse
import requests
from datetime import datetime
from ollama import chat
import sys


def extract_markdown_metadata(markdown_path: str) -> dict:
    '''Extract frontmatter and body content from a markdown blog post.

    Args:
        markdown_path: Absolute path to the .md file

    Returns:
        Dict with 'title', 'tags', 'body' keys
    '''
    if not os.path.exists(markdown_path):
        raise FileNotFoundError(f'Markdown file not found at {markdown_path}')

    with open(markdown_path, 'r', encoding='utf-8') as f:
        content = f.read()

    # Extract YAML frontmatter between --- delimiters
    frontmatter_match = re.search(r'^---\n(.*?)\n---', content, re.DOTALL)
    if not frontmatter_match:
        raise ValueError('No YAML frontmatter found in markdown file')

    frontmatter_text = frontmatter_match.group(1)
    body = content[frontmatter_match.end():].strip()

    # Parse simple key-value frontmatter (extend for full YAML if needed)
    metadata = {}
    for line in frontmatter_text.split('\n'):
        if ':' in line:
            key, value = line.split(':', 1)
            metadata[key.strip()] = value.strip().strip('"')

    return {
        'title': metadata.get('title', 'Untitled Post'),
        'tags': [tag.strip() for tag in metadata.get('tags', '').split(',') if tag.strip()],
        'body': body
    }


def generate_youtube_metadata(blog_metadata: dict, llm_model: str = 'llama3:8b') -> dict:
    '''Generate YouTube-optimized title, description, tags using a local LLM.

    Args:
        blog_metadata: Output from extract_markdown_metadata
        llm_model: Ollama model name to use

    Returns:
        Dict with 'yt_title', 'yt_description', 'yt_tags' keys
    '''
    prompt = f'''You are a technical content SEO expert. Given the following blog post metadata, generate YouTube-optimized metadata:

Blog Title: {blog_metadata['title']}
Blog Tags: {', '.join(blog_metadata['tags'])}
Blog Body Excerpt (first 2000 chars): {blog_metadata['body'][:2000]}

Rules:
1. YouTube title must be under 60 characters, include 1-2 primary keywords, and be clickable but not clickbait
2. Description must be 150-200 words, include the blog link, 3-5 timestamps, and a call to action
3. Tags must be 15-20 relevant technical keywords, comma-separated
4. Return ONLY valid JSON with keys: "yt_title", "yt_description", "yt_tags" (string of comma-separated tags)
'''

    try:
        response = chat(
            model=llm_model,
            messages=[{'role': 'user', 'content': prompt}],
            format='json'
        )
    except Exception as e:
        raise ConnectionError(f'Failed to connect to Ollama LLM: {str(e)}')

    try:
        yt_metadata = json.loads(response['message']['content'])
    except json.JSONDecodeError:
        # Fallback: extract JSON from response if LLM added extra text
        json_match = re.search(r'\{.*\}', response['message']['content'], re.DOTALL)
        if json_match:
            yt_metadata = json.loads(json_match.group(0))
        else:
            raise ValueError('LLM returned invalid JSON')

    # Validate required fields
    required_keys = ['yt_title', 'yt_description', 'yt_tags']
    for key in required_keys:
        if key not in yt_metadata:
            raise KeyError(f'LLM response missing required key: {key}')

    return yt_metadata


def save_metadata_to_file(yt_metadata: dict, output_path: str) -> None:
    '''Save generated YouTube metadata to a JSON file for later upload.''' 
    with open(output_path, 'w', encoding='utf-8') as f:
        json.dump(yt_metadata, f, indent=2, ensure_ascii=False)
    print(f'Saved YouTube metadata to {output_path}')


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Generate YouTube metadata from markdown blog posts')
    parser.add_argument('--input', required=True, help='Path to input markdown file')
    parser.add_argument('--output', default='yt_metadata.json', help='Path to output JSON file')
    parser.add_argument('--model', default='llama3:8b', help='Ollama model to use')

    args = parser.parse_args()

    try:
        blog_meta = extract_markdown_metadata(args.input)
        print(f'Extracted blog metadata for: {blog_meta['title']}')

        yt_meta = generate_youtube_metadata(blog_meta, args.model)
        print(f'Generated YouTube title: {yt_meta['yt_title']}')

        save_metadata_to_file(yt_meta, args.output)
        sys.exit(0)
    except Exception as e:
        print(f'Error: {str(e)}', file=sys.stderr)
        sys.exit(1)

import os
import json
import argparse
import time
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from googleapiclient.http import MediaFileUpload
import sys

# YouTube Data API v3 scopes: https://developers.google.com/youtube/v3/guides/auth/scopes
SCOPES = ['https://www.googleapis.com/auth/youtube.upload']
API_SERVICE_NAME = 'youtube'
API_VERSION = 'v3'
QUOTA_COST_PER_UPLOAD = 1600  # YouTube API quota cost per video upload

def get_authenticated_service(client_secrets_path: str, token_path: str = 'token.json'):
    """Authenticate with YouTube Data API using OAuth 2.0.

    Args:
        client_secrets_path: Path to client_secrets.json from Google Cloud Console
        token_path: Path to save/load OAuth token

    Returns:
        Authenticated YouTube API service object
    """
    creds = None
    if os.path.exists(token_path):
        try:
            creds = Credentials.from_authorized_user_file(token_path, SCOPES)
        except Exception as e:
            print(f'Warning: Failed to load token from {token_path}: {str(e)}')

    if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token:
            try:
                creds.refresh(Request())
            except Exception as e:
                print(f'Warning: Failed to refresh token: {str(e)}')
                creds = None
        if not creds:
            if not os.path.exists(client_secrets_path):
                raise FileNotFoundError(f'Client secrets file not found at {client_secrets_path}')
            flow = InstalledAppFlow.from_client_secrets_file(client_secrets_path, SCOPES)
            creds = flow.run_local_server(port=8080)
        # Save credentials for next run
        with open(token_path, 'w') as token:
            token.write(creds.to_json())

    try:
        return build(API_SERVICE_NAME, API_VERSION, credentials=creds)
    except Exception as e:
        raise ConnectionError(f'Failed to build YouTube API service: {str(e)}')

def upload_video(youtube_service, video_path: str, metadata_path: str, privacy_status: str = 'private') -> str:
    """Upload a video to YouTube with pre-generated metadata.

    Args:
        youtube_service: Authenticated YouTube API service
        video_path: Path to video file (mp4, max 128GB)
        metadata_path: Path to yt_metadata.json generated earlier
        privacy_status: private, unlisted, or public

    Returns:
        YouTube video ID of uploaded video
    """
    if not os.path.exists(video_path):
        raise FileNotFoundError(f'Video file not found at {video_path}')
    if not os.path.exists(metadata_path):
        raise FileNotFoundError(f'Metadata file not found at {metadata_path}')

    # Load metadata
    with open(metadata_path, 'r', encoding='utf-8') as f:
        metadata = json.load(f)

    # Validate video file size (YouTube max 128GB, but we cap at 10GB for safety)
    video_size = os.path.getsize(video_path)
    if video_size > 10 * 1024 * 1024 * 1024:
        raise ValueError(f'Video file too large: {video_size / 1e9:.2f}GB (max 10GB)')

    # Prepare video metadata for API
    body = {
        'snippet': {
            'title': metadata['yt_title'][:60],  # YouTube title max 100 chars, but we cap at 60 for SEO
            'description': metadata['yt_description'],
            'tags': [tag.strip() for tag in metadata['yt_tags'].split(',') if tag.strip()],
            'categoryId': '28'  # Category 28 = Science & Technology
        },
        'status': {
            'privacyStatus': privacy_status,
            'selfDeclaredMadeForKids': False
        }
    }

    # Create media upload object
    media = MediaFileUpload(
        video_path,
        mimetype='video/mp4',
        chunksize=1024*1024*5,  # 5MB chunks
        resumable=True
    )

    # Insert video
    try:
        insert_request = youtube_service.videos().insert(
            part=','.join(body.keys()),
            body=body,
            media_body=media
        )
    except Exception as e:
        raise ValueError(f'Failed to create insert request: {str(e)}')

    # Execute resumable upload with progress tracking
    response = None
    error_count = 0
    max_errors = 3
    while response is None:
        try:
            status, response = insert_request.next_chunk()
            if status:
                print(f'Uploaded {int(status.progress() * 100)}%')
        except HttpError as e:
            if e.resp.status in [403, 500, 502, 503, 504]:
                # Retry on transient errors
                error_count += 1
                if error_count > max_errors:
                    raise Exception(f'Max retry errors exceeded: {str(e)}')
                print(f'Retrying after error: {str(e)}')
                time.sleep(2 ** error_count)  # Exponential backoff
            else:
                raise Exception(f'YouTube API error: {str(e)}')
        except Exception as e:
            raise Exception(f'Upload error: {str(e)}')

    if 'id' not in response:
        raise Exception(f'Upload failed: no video ID in response: {response}')

    print(f'Upload complete! Video ID: {response['id']}')
    print(f'Quota used: {QUOTA_COST_PER_UPLOAD} units (check Google Cloud Console for remaining quota)')
    return response['id']

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Upload video to YouTube with pre-generated metadata')
    parser.add_argument('--video', required=True, help='Path to video file (mp4)')
    parser.add_argument('--metadata', required=True, help='Path to yt_metadata.json')
    parser.add_argument('--secrets', default='client_secrets.json', help='Path to Google client secrets')
    parser.add_argument('--privacy', default='private', choices=['private', 'unlisted', 'public'], help='Video privacy status')

    args = parser.parse_args()

    try:
        youtube = get_authenticated_service(args.secrets)
        video_id = upload_video(youtube, args.video, args.metadata, args.privacy)
        sys.exit(0)
    except Exception as e:
        print(f'Error: {str(e)}', file=sys.stderr)
        sys.exit(1)

import os
import json
import argparse
import requests
import whisper
from datetime import datetime
import sys
import re

def transcribe_video(video_path: str, model_size: str = 'large-v3') -> dict:
    '''Transcribe video audio using local Whisper model, generate timestamps.

    Args:
        video_path: Path to video file (mp4)
        model_size: Whisper model size (tiny, base, small, medium, large-v3)

    Returns:
        Dict with 'transcription', 'timestamps', 'duration' keys
    '''
    if not os.path.exists(video_path):
        raise FileNotFoundError(f'Video file not found at {video_path}')

    # Check if CUDA is available for faster inference
    import torch
    device = 'cuda' if torch.cuda.is_available() else 'cpu'
    print(f'Using device: {device} for Whisper inference')

    try:
        model = whisper.load_model(model_size, device=device)
    except Exception as e:
        raise Exception(f'Failed to load Whisper model {model_size}: {str(e)}')

    try:
        result = model.transcribe(
            video_path,
            verbose=True,
            fp16=torch.cuda.is_available(),  # Use FP16 only on CUDA
            timestamp_granularities=['segment']  # Get per-segment timestamps
        )
    except Exception as e:
        raise Exception(f'Transcription failed: {str(e)}')

    # Parse segments into timestamps (format: 00:00:00 - 00:00:30: Segment text)
    timestamps = []
    for segment in result['segments']:
        start = str(datetime.timedelta(seconds=int(segment['start']))).zfill(8)
        end = str(datetime.timedelta(seconds=int(segment['end']))).zfill(8)
        text = segment['text'].strip()
        timestamps.append(f'{start} - {end}: {text}')

    return {
        'transcription': result['text'],
        'timestamps': timestamps,
        'duration': int(result['segments'][-1]['end']) if result['segments'] else 0
    }

def post_to_dev_to(transcription: dict, metadata: dict, api_key: str, blog_url: str) -> str:
    '''Post a summary blog post to Dev.to with video timestamps.

    Args:
        transcription: Output from transcribe_video
        metadata: YouTube metadata from yt_metadata.json
        api_key: Dev.to API key (from https://dev.to/settings/extensions)
        blog_url: URL of the full blog post

    Returns:
        URL of the published Dev.to post
    '''
    # Generate post body with timestamps
    timestamp_section = '\n'.join([f'- {ts}' for ts in transcription['timestamps'][:15]])  # Top 15 timestamps
    post_body = f'''# {metadata['yt_title']}

{metadata['yt_description']}

## Video Timestamps
{timestamp_section}

## Full Transcription
{transcription['transcription'][:5000]}... (full transcription available in the video)

[Watch the full video on YouTube](https://youtube.com/watch?v=VIDEO_ID)
[Read the full blog post here]({blog_url})
'''

    headers = {
        'api-key': api_key,
        'Content-Type': 'application/json'
    }

    payload = {
        'article': {
            'title': metadata['yt_title'],
            'body_markdown': post_body,
            'tags': [tag.strip() for tag in metadata['yt_tags'].split(',') if tag.strip()][:5],
            'published': False  # Set to True to auto-publish
        }
    }

    try:
        response = requests.post(
            'https://dev.to/api/articles',
            headers=headers,
            json=payload,
            timeout=30
        )
        response.raise_for_status()
    except requests.exceptions.HTTPError as e:
        raise Exception(f'Dev.to API error: {response.status_code} - {response.text}')
    except Exception as e:
        raise Exception(f'Failed to post to Dev.to: {str(e)}')

    response_json = response.json()
    if 'url' not in response_json:
        raise Exception(f'Dev.to post failed: no URL in response: {response_json}')

    print(f'Posted to Dev.to: {response_json['url']}')
    return response_json['url']

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Transcribe video and post summary to Dev.to')
    parser.add_argument('--video', required=True, help='Path to video file')
    parser.add_argument('--metadata', required=True, help='Path to yt_metadata.json')
    parser.add_argument('--devto-key', required=True, help='Dev.to API key')
    parser.add_argument('--blog-url', required=True, help='URL of full blog post')
    parser.add_argument('--whisper-model', default='large-v3', help='Whisper model size')

    args = parser.parse_args()

    try:
        # Step 1: Transcribe video
        print('Starting transcription...')
        transcription = transcribe_video(args.video, args.whisper_model)
        print(f'Transcription complete! Duration: {transcription['duration']} seconds')

        # Step 2: Load metadata
        with open(args.metadata, 'r') as f:
            metadata = json.load(f)

        # Step 3: Post to Dev.to
        print('Posting to Dev.to...')
        post_url = post_to_dev_to(transcription, metadata, args.devto_key, args.blog_url)

        sys.exit(0)
    except Exception as e:
        print(f'Error: {str(e)}', file=sys.stderr)
        sys.exit(1)

import hashlib
import redis
import json

class LLMCache:
    def __init__(self, redis_url: str = 'redis://localhost:6379'):
        self.redis = redis.from_url(redis_url, decode_responses=True)
        self.cache_ttl = 30 * 24 * 60 * 60  # 30 days

    def get_cache_key(self, prompt: str) -> str:
        return f'llm_cache:{hashlib.sha256(prompt.encode()).hexdigest()}'

    def get(self, prompt: str) -> dict | None:
        key = self.get_cache_key(prompt)
        cached = self.redis.get(key)
        return json.loads(cached) if cached else None

    def set(self, prompt: str, response: dict) -> None:
        key = self.get_cache_key(prompt)
        self.redis.setex(key, self.cache_ttl, json.dumps(response))

import ffmpeg

def clip_shorts(video_path: str, output_dir: str, video_duration: int) -> list:
    """Generate 3 YouTube Shorts clips (60s each) from a long-form video.""" 
    clips = [
        (0, 60, 'intro'),
        (max(0, video_duration // 2 - 30), max(0, video_duration // 2 + 30), 'takeaway'),
        (max(0, video_duration - 60), video_duration, 'cta')
    ]
    output_paths = []
    for start, end, clip_type in clips:
        output_path = f'{output_dir}/short_{clip_type}.mp4'
        try:
            stream = ffmpeg.input(video_path, ss=start, t=end)
            stream = ffmpeg.output(stream, output_path, c='copy')  # No re-encode
            ffmpeg.run(stream, overwrite_output=True)
            output_paths.append(output_path)
        except Exception as e:
            print(f'Failed to clip {clip_type}: {str(e)}')
    return output_paths

from prometheus_client import Counter, start_http_server
import time

# Define Prometheus metrics
youtube_uploads = Counter('youtube_uploads_total', 'Total YouTube video uploads')
youtube_quota_used = Counter('youtube_quota_used_total', 'Total YouTube API quota used')

def track_upload_quota(video_id: str):
    """Track upload count and quota usage in Prometheus.""" 
    youtube_uploads.inc()
    youtube_quota_used.inc(1600)  # 1600 units per upload
    print(f'Tracked upload {video_id}: total uploads {youtube_uploads._value.get()}')

if __name__ == '__main__':
    start_http_server(8000)  # Expose metrics on port 8000
    print('Prometheus metrics server running on :8000')
    while True:
        time.sleep(60)  # Keep server running