detect_ads.py

"""
Detect advertisement segments in podcast transcripts using GPT-5 Nano.
Chunks large transcripts and merges results with confidence scores.
"""

import json
import os
import re
from pathlib import Path
from openai import OpenAI
import tiktoken
import config

def load_metadata():
    """Load metadata.json if it exists, create if it doesn't."""
    if config.METADATA_FILE.exists():
        try:
            with open(config.METADATA_FILE, 'r', encoding='utf-8') as f:
                data = json.load(f)
                # Ensure episodes key exists
                if 'episodes' not in data:
                    data['episodes'] = {}
                return data
        except json.JSONDecodeError:
            print("Warning: metadata.json is corrupted. Creating new one.")
            return {"episodes": {}}
        except Exception as e:
            print(f"Error loading metadata: {e}")
    
    return {"episodes": {}}

def save_metadata(metadata):
    """Save metadata to JSON file."""
    with open(config.METADATA_FILE, 'w', encoding='utf-8') as f:
        json.dump(metadata, f, indent=2, ensure_ascii=False)

def count_tokens(text, model="gpt-5-nano"):
    """Count tokens in text using tiktoken."""
    try:
        # Use cl100k_base encoding (used by GPT models)
        encoding = tiktoken.get_encoding("cl100k_base")
        return len(encoding.encode(text))
    except Exception as e:
        # Fallback: rough estimation (1 token ≈ 4 characters)
        return len(text) // 4

def chunk_transcript(transcript_data, max_tokens=350000, overlap_tokens=50000):
    """Chunk transcript into segments that fit within token limit."""
    chunks = []
    
    # Get full text and segments
    full_text = transcript_data.get('text', '')
    segments = transcript_data.get('segments', [])
    
    if not segments and full_text:
        # If no segments, treat entire text as one segment
        chunks.append({
            'start_time': 0.0,
            'end_time': 0.0,
            'text': full_text
        })
        return chunks
    
    # Build text with time markers
    current_chunk = {'start_time': None, 'end_time': None, 'text': '', 'tokens': 0}
    
    for segment in segments:
        segment_text = segment.get('text', '')
        segment_start = segment.get('start', 0.0)
        segment_end = segment.get('end', 0.0)
        
        # Estimate tokens for this segment
        segment_tokens = count_tokens(segment_text)
        
        # Check if adding this segment would exceed limit
        if current_chunk['tokens'] + segment_tokens > max_tokens and current_chunk['text']:
            # Save current chunk
            current_chunk['end_time'] = segment_start
            chunks.append(current_chunk)
            
            # Start new chunk with overlap
            overlap_text = current_chunk['text'][-5000:] if len(current_chunk['text']) > 5000 else current_chunk['text']
            current_chunk = {
                'start_time': max(0, segment_start - 10),  # Start slightly before
                'end_time': None,
                'text': overlap_text + '\n\n' + segment_text,
                'tokens': count_tokens(overlap_text + segment_text)
            }
        else:
            # Add to current chunk
            if current_chunk['start_time'] is None:
                current_chunk['start_time'] = segment_start
            
            current_chunk['end_time'] = segment_end
            current_chunk['text'] += segment_text + ' '
            current_chunk['tokens'] += segment_tokens
    
    # Add final chunk
    if current_chunk['text']:
        chunks.append(current_chunk)
    
    return chunks

def create_ad_detection_prompt(chunk_text, chunk_start_time):
    """Create prompt for GPT-5 Nano to detect ads."""
    prompt = f"""Analyze this podcast transcript segment and identify any advertisement segments. 

Return a JSON array of ad segments with the following structure:
[
  {{
    "start_time": <timestamp in seconds relative to chunk start, add {chunk_start_time} to get absolute time>,
    "end_time": <timestamp in seconds relative to chunk start, add {chunk_start_time} to get absolute time>,
    "text": "<the exact ad text from transcript>",
    "confidence": <0.0-1.0 confidence score>,
    "reasoning": "<brief explanation why this is identified as an ad>"
  }}
]

Ad indicators to look for:
- Promotional language (e.g., "sponsored by", "brought to you by", "try now", "visit us")
- Sponsor mentions and product names
- Calls to action (e.g., "use code", "visit website", "download now")
- Discount codes or special offers
- Website URLs or app names
- Interruption of main content flow
- Different voice/tone suggesting ad read
- Mentions of "advertisement", "commercial", "break"

Only identify clear advertisement segments. If unsure, use lower confidence scores.

Transcript segment:
{chunk_text}

Return only valid JSON, no additional text:"""
    
    return prompt

def detect_ads_in_chunk(chunk, client):
    """Use GPT-5 Nano to detect ads in a transcript chunk."""
    prompt = create_ad_detection_prompt(chunk['text'], chunk['start_time'])
    
    try:
        response = client.chat.completions.create(
            model=config.OPENAI_MODEL,
            messages=[
                {"role": "system", "content": "You are an expert at identifying advertisements in podcast transcripts. Always return valid JSON arrays."},
                {"role": "user", "content": prompt}
            ],
            temperature=0.3,  # Lower temperature for more consistent results
            max_tokens=4000  # Should be enough for ad segments
        )
        
        response_text = response.choices[0].message.content.strip()
        
        # Extract JSON from response (in case there's extra text)
        json_match = re.search(r'\[.*\]', response_text, re.DOTALL)
        if json_match:
            response_text = json_match.group(0)
        
        ad_segments = json.loads(response_text)
        
        # Adjust timestamps to absolute time
        for segment in ad_segments:
            if 'start_time' in segment:
                segment['start_time'] = chunk['start_time'] + segment.get('start_time', 0)
            if 'end_time' in segment:
                segment['end_time'] = chunk['start_time'] + segment.get('end_time', 0)
        
        return ad_segments
        
    except json.JSONDecodeError as e:
        print(f"  Error parsing JSON response: {e}")
        print(f"  Response: {response_text[:500]}")
        return []
    except Exception as e:
        print(f"  Error calling GPT-5 Nano: {e}")
        return []

def merge_overlapping_ads(ad_segments_list):
    """Merge ad segments from multiple chunks, handling overlaps."""
    if not ad_segments_list:
        return []
    
    # Flatten list of ad segments
    all_ads = []
    for ads in ad_segments_list:
        all_ads.extend(ads)
    
    if not all_ads:
        return []
    
    # Sort by start time
    all_ads.sort(key=lambda x: x.get('start_time', 0))
    
    # Merge overlapping segments
    merged_ads = []
    current_ad = all_ads[0].copy()
    
    for ad in all_ads[1:]:
        current_end = current_ad.get('end_time', 0)
        next_start = ad.get('start_time', 0)
        
        # If segments overlap or are close (< 5 seconds apart), merge them
        if next_start <= current_end + 5:
            # Merge: extend end time and combine text
            current_ad['end_time'] = max(current_end, ad.get('end_time', 0))
            current_ad['text'] += ' ' + ad.get('text', '')
            # Use higher confidence
            current_ad['confidence'] = max(
                current_ad.get('confidence', 0.5),
                ad.get('confidence', 0.5)
            )
        else:
            # No overlap, save current and start new
            merged_ads.append(current_ad)
            current_ad = ad.copy()
    
    # Add final ad
    merged_ads.append(current_ad)
    
    return merged_ads

def detect_ads_in_episode(episode_id, transcript_path, client):
    """Detect ads in a single episode's transcript."""
    # Load transcript
    with open(transcript_path, 'r', encoding='utf-8') as f:
        transcript_data = json.load(f)
    
    # Chunk transcript if needed
    chunks = chunk_transcript(transcript_data, max_tokens=350000, overlap_tokens=50000)
    
    print(f"  Processing {len(chunks)} chunk(s)...")
    
    all_ad_segments = []
    
    for i, chunk in enumerate(chunks, 1):
        print(f"    Chunk {i}/{len(chunks)}...")
        ad_segments = detect_ads_in_chunk(chunk, client)
        if ad_segments:
            all_ad_segments.append(ad_segments)
    
    # Merge overlapping segments
    merged_ads = merge_overlapping_ads(all_ad_segments)
    
    return merged_ads

def detect_ads():
    """Main function to detect ads in all transcribed episodes."""
    if not config.OPENAI_API_KEY:
        print("Error: OPENAI_API_KEY not set in environment or config.py")
        return
    
    client = OpenAI(api_key=config.OPENAI_API_KEY)
    metadata = load_metadata()
    
    episodes_to_process = []
    
    # Find episodes that need ad detection
    for episode_key, episode_data in metadata.get('episodes', {}).items():
        if (episode_data.get('transcribed', False) and 
            not episode_data.get('ad_detected', False)):
            
            transcript_path = config.BASE_DIR / episode_data.get('transcript_path', '')
            if transcript_path.exists():
                episodes_to_process.append((episode_key, episode_data, transcript_path))
    
    if not episodes_to_process:
        print("No episodes to process for ad detection.")
        return
    
    print(f"\n=== Detecting ads in {len(episodes_to_process)} episodes ===\n")
    
    for i, (episode_key, episode_data, transcript_path) in enumerate(episodes_to_process, 1):
        episode_id = episode_key.split('_', 1)[1] if '_' in episode_key else episode_key
        print(f"[{i}/{len(episodes_to_process)}] {episode_data.get('title', 'Unknown')}")
        
        # Detect ads
        ad_segments = detect_ads_in_episode(episode_id, transcript_path, client)
        
        if ad_segments:
            # Save ad segments
            ad_file = config.ADS_DIR / f"{episode_id}.json"
            ad_data = {
                'episode_id': episode_id,
                'episode_title': episode_data.get('title', ''),
                'ad_segments': ad_segments,
                'total_ads': len(ad_segments),
                'total_ad_duration': sum(
                    seg.get('end_time', 0) - seg.get('start_time', 0)
                    for seg in ad_segments
                )
            }
            
            with open(ad_file, 'w', encoding='utf-8') as f:
                json.dump(ad_data, f, indent=2, ensure_ascii=False)
            
            # Update metadata
            episode_data['ad_detected'] = True
            episode_data['ad_path'] = str(ad_file.relative_to(config.BASE_DIR))
            episode_data['ad_count'] = len(ad_segments)
            metadata['episodes'][episode_key] = episode_data
            save_metadata(metadata)
            
            print(f"  ✓ Found {len(ad_segments)} ad segment(s)")
            print(f"  ✓ Saved to: {ad_file}")
        else:
            print(f"  ✓ No ads detected")
            episode_data['ad_detected'] = True
            episode_data['ad_count'] = 0
            metadata['episodes'][episode_key] = episode_data
            save_metadata(metadata)
    
    print("\n=== Ad detection complete ===")

if __name__ == "__main__":
    detect_ads()