Update app.py

app.py

@@ -576,8 +576,18 @@ def audio_transcription_tool(file_path: str) -> str:
         return f"Error: Audio file not found: '{file_path}'"
     
     try:
-        transcription = asr_pipeline(str(audio_path))
+        transcription = asr_pipeline(
+            str(audio_path),
+            return_timestamps=True,  # ← Add this!
+            chunk_length_s=30,       # ← Process in 30-second chunks
+            stride_length_s=5        # ← 5-second overlap between chunks
+        )
+        
+        # Extract just the text (ignore timestamps)
         result_text = transcription.get("text", "")
+        # OR if you want to see the chunks:
+        # chunks = transcription.get("chunks", [])
+        # result_text = " ".join([chunk["text"] for chunk in chunks])
         
         if not result_text:
             return "Error: Transcription empty."
@@ -672,110 +682,94 @@ class YoutubeInput(BaseModel):
 @tool(args_schema=YoutubeInput)
 def get_youtube_transcript(video_url: str) -> str:
     """
-    Fetches YouTube video transcript with retry logic.
-    Returns N/A if video is inaccessible.
+    Fetches YouTube video transcript/captions using YouTube Data API v3.
+    Much more reliable than yt-dlp on cloud environments.
     """
     if not video_url:
         return "Error: Invalid URL."
     
-    print(f"📺 YouTube transcript: {video_url}")
+    print(f"📺 YouTube transcript (API v3): {video_url}")
     
-    max_retries = 3
-    for attempt in range(max_retries):
-        try:
-            # Extract video ID
-            video_id = None
-            if "watch?v=" in video_url:
-                video_id = video_url.split("v=")[1].split("&")[0]
-            elif "youtu.be/" in video_url:
-                video_id = video_url.split("youtu.be/")[1].split("?")[0]
-            
-            if not video_id:
-                return f"Error: Could not extract video ID from URL."
-            
-            cmd = [
-                'yt-dlp',
-                '--skip-download',
-                '--write-auto-subs',
-                '--write-subs',
-                '--sub-lang', 'en',
-                '--sub-format', 'vtt',
-                '--output', video_id,
-                video_url
-            ]
-            
-            print(f"🔧 Running yt-dlp (attempt {attempt + 1}/{max_retries})...")
-            result = subprocess.run(cmd, capture_output=True, text=True, timeout=45)
-            
-            if result.returncode != 0:
-                stderr = result.stderr
-                
-                # Check for network errors
-                if 'Failed to resolve' in stderr or 'No address associated' in stderr:
-                    if attempt < max_retries - 1:
-                        print(f"⚠️ Network error, retrying...")
-                        time.sleep(2 ** attempt)
-                        continue
-                    return "N/A - YouTube is inaccessible due to network issues."
-                
-                return f"Error: Could not fetch subtitles - {stderr[:200]}"
-            
-            # Find subtitle file
-            import glob
-            vtt_files = glob.glob(f"{video_id}*.vtt")
-            
-            if not vtt_files:
-                return "N/A - No English subtitles found for this video."
-            
-            subtitle_file = vtt_files[0]
-            print(f"✓ Found subtitle file: {subtitle_file}")
-            
-            # Parse VTT
-            with open(subtitle_file, 'r', encoding='utf-8') as f:
-                content = f.read()
-            
-            lines = content.split('\n')
-            transcript_parts = []
-            
-            for line in lines:
-                line = line.strip()
-                if (line and 
-                    not line.startswith('WEBVTT') and 
-                    not '-->' in line and
-                    not line.isdigit() and
-                    not line.startswith('Kind:') and
-                    not line.startswith('Language:')):
-                    transcript_parts.append(line)
-            
-            full_transcript = " ".join(transcript_parts)
-            
-            # Cleanup
-            for vtt_file in vtt_files:
-                try:
-                    os.remove(vtt_file)
-                except:
-                    pass
-            
-            if not full_transcript:
-                return "Error: Transcript was empty."
-            
-            print(f"✓ Transcript extracted: {len(full_transcript)} chars")
-            return f"Transcript:\n{truncate_if_needed(full_transcript)}"
-            
-        except subprocess.TimeoutExpired:
-            if attempt < max_retries - 1:
-                continue
-            return "N/A - YouTube request timed out."
-        except FileNotFoundError:
-            return "Error: yt-dlp not installed."
-        except Exception as e:
-            if attempt < max_retries - 1:
-                time.sleep(2 ** attempt)
-                continue
-            print(f"❌ Error: {str(e)}")
-            return f"Error: {str(e)}"
+    # Get API key
+    YOUTUBE_API_KEY = os.getenv("YOUTUBE_API_KEY")
+    if not YOUTUBE_API_KEY:
+        return "Error: YOUTUBE_API_KEY not set in Space secrets."
     
-    return "N/A - YouTube transcript unavailable after multiple attempts."
+    try:
+        # Extract video ID
+        video_id = None
+        if "watch?v=" in video_url:
+            video_id = video_url.split("v=")[1].split("&")[0]
+        elif "youtu.be/" in video_url:
+            video_id = video_url.split("youtu.be/")[1].split("?")[0]
+        
+        if not video_id:
+            return "Error: Could not extract video ID from URL."
+        
+        print(f"   Video ID: {video_id}")
+        
+        # Initialize YouTube API
+        youtube = build('youtube', 'v3', developerKey=YOUTUBE_API_KEY)
+        
+        # Get caption tracks
+        captions_response = youtube.captions().list(
+            part='snippet',
+            videoId=video_id
+        ).execute()
+        
+        if not captions_response.get('items'):
+            return "N/A - No captions available for this video."
+        
+        # Find English caption track
+        caption_id = None
+        for caption in captions_response['items']:
+            lang = caption['snippet'].get('language', '')
+            if lang.startswith('en'):
+                caption_id = caption['id']
+                print(f"   Found English captions: {lang}")
+                break
+        
+        if not caption_id:
+            # Try first available caption
+            caption_id = captions_response['items'][0]['id']
+            print(f"   Using first available caption track")
+        
+        # Download caption content
+        caption_content = youtube.captions().download(
+            id=caption_id,
+            tfmt='srt'  # or 'vtt'
+        ).execute()
+        
+        # Parse SRT format (remove timestamps and numbers)
+        lines = caption_content.decode('utf-8').split('\n')
+        transcript_parts = []
+        
+        for line in lines:
+            line = line.strip()
+            # Skip line numbers, timestamps, and empty lines
+            if (line and 
+                not line.isdigit() and 
+                '-->' not in line):
+                transcript_parts.append(line)
+        
+        full_transcript = ' '.join(transcript_parts)
+        
+        if not full_transcript:
+            return "Error: Transcript was empty."
+        
+        print(f"✓ Transcript retrieved: {len(full_transcript)} chars")
+        return f"Transcript:\n{truncate_if_needed(full_transcript)}"
+        
+    except HttpError as e:
+        if e.resp.status == 403:
+            return "Error: YouTube API quota exceeded or captions are disabled for this video."
+        elif e.resp.status == 404:
+            return "Error: Video not found or captions not available."
+        else:
+            return f"YouTube API error: {str(e)}"
+    except Exception as e:
+        print(f"❌ Error: {str(e)}")
+        return f"Error: {str(e)}"
 
 
 class ScrapeInput(BaseModel):
@@ -1448,7 +1442,7 @@ REMEMBER: One tool per turn. No reasoning without tools. Exact answer format.
             self.llm_with_tools = ChatGroq(
                 temperature=0,
                 groq_api_key=GROQ_API_KEY,
-                model_name="qwen/qwen3-32b",
+                model_name="llama-3.3-70b-versatile",
                 max_tokens=4096,
                 timeout=60
                 ).bind_tools(self.tools, tool_choice="auto")