Update app.py

app.py

@@ -1,222 +1,170 @@
-import logging
 from fastapi import FastAPI, HTTPException
 from fastapi.middleware.cors import CORSMiddleware
 from pydantic import BaseModel
+from transformers import pipeline
 import yt_dlp
 import whisper
 import os
-import subprocess
-from transformers import pipeline
+import uuid
+import re
 from typing import Optional
-import traceback
 
-# Configure logging
-logging.basicConfig(level=logging.INFO)
-logger = logging.getLogger(__name__)
+app = FastAPI(title="TikTok Content Analyzer")
 
-app = FastAPI(title="TikTok Content Analyzer API")
-
-# CORS Configuration
+# Configure CORS
 app.add_middleware(
     CORSMiddleware,
     allow_origins=["*"],
+    allow_credentials=True,
     allow_methods=["*"],
     allow_headers=["*"],
 )
 
 class VideoRequest(BaseModel):
-    url: str
+    video_url: str
 
 class AnalysisResult(BaseModel):
     transcription: str
-    category: Optional[str] = None
-    confidence: Optional[float] = None
-    error: Optional[str] = None
+    category: Optional[str]
+    confidence: Optional[float]
+    error: Optional[str]
 
+# Delete temporary files
 def clean_temp_files():
-    """Remove temporary files if they exist"""
     temp_files = ["temp_video.mp4", "temp_audio.mp3"]
     for file in temp_files:
-        try:
-            if os.path.exists(file):
-                os.remove(file)
-                logger.info(f"Removed temporary file: {file}")
-        except Exception as e:
-            logger.warning(f"Could not remove {file}: {str(e)}")
+        if os.path.exists(file):
+            os.remove(file)
 
-def download_video(video_url: str) -> Optional[str]:
-    """Download TikTok video using yt-dlp"""
+# Download TikTok video
+def download_video(video_url):
     try:
         ydl_opts = {
-            'format': 'best[ext=mp4]',
+            'format': 'bestvideo[ext=mp4]+bestaudio[ext=m4a]/best[ext=mp4]',
             'outtmpl': 'temp_video.mp4',
             'quiet': True,
             'no_warnings': True,
-            'extract_flat': True,
-            'socket_timeout': 30,
-            'retries': 3
+            'extractor_args': {'tiktok': {'skip_watermark': True}}
         }
         
         with yt_dlp.YoutubeDL(ydl_opts) as ydl:
-            info = ydl.extract_info(video_url, download=True)
-            logger.info(f"Successfully downloaded video: {info.get('title', 'No title')}")
-        
-        return "temp_video.mp4" if os.path.exists("temp_video.mp4") else None
-    
-    except yt_dlp.utils.DownloadError as e:
-        logger.error(f"YT-DLP Download Error: {str(e)}")
+            ydl.download([video_url])
+        return "temp_video.mp4"
     except Exception as e:
-        logger.error(f"Unexpected download error: {str(e)}")
-        logger.error(traceback.format_exc())
-    
-    return None
+        print(f"Download error: {e}")
+        return None
 
-def extract_audio(video_path: str) -> Optional[str]:
-    """Extract audio from video using ffmpeg"""
-    try:
-        result = subprocess.run([
-            "ffmpeg",
-            "-i", video_path,
-            "-vn",
-            "-acodec", "libmp3lame",
-            "-q:a", "3",
-            "temp_audio.mp3",
-            "-y"
-        ], capture_output=True, text=True, check=True)
-        
-        logger.info("FFmpeg output: " + result.stdout)
-        if result.stderr:
-            logger.warning("FFmpeg warnings: " + result.stderr)
-        
-        return "temp_audio.mp3" if os.path.exists("temp_audio.mp3") else None
-    
-    except subprocess.CalledProcessError as e:
-        logger.error(f"FFmpeg error: {str(e)}")
-        logger.error(f"FFmpeg stderr: {e.stderr}")
-    except Exception as e:
-        logger.error(f"Unexpected audio extraction error: {str(e)}")
-    
-    return None
+# Extract audio
+def extract_audio(video_path):
+    os.system(f"ffmpeg -i \"{video_path}\" -vn -acodec libmp3lame -q:a 3 \"temp_audio.mp3\" -y")
+    return "temp_audio.mp3" if os.path.exists("temp_audio.mp3") else None
 
-def transcribe_audio(audio_path: str) -> Optional[str]:
-    """Transcribe audio using Whisper"""
+# Transcribe audio
+def transcribe_audio(audio_path):
     try:
-        logger.info("Loading Whisper model...")
-        model = whisper.load_model("tiny")  # Using tiny model for Hugging Face Spaces
-        
-        logger.info("Starting transcription...")
+        model = whisper.load_model("base")
         result = model.transcribe(audio_path)
-        
-        if 'text' in result:
-            logger.info("Transcription successful")
-            return result['text']
-        else:
-            logger.error("Whisper returned no text")
-    
+        return result['text']
     except Exception as e:
-        logger.error(f"Transcription error: {str(e)}")
-        logger.error(traceback.format_exc())
-    
-    return None
+        print(f"Transcription error: {e}")
+        return None
 
-def classify_content(text: str) -> tuple[Optional[str], Optional[float]]:
-    """Classify text content using transformers"""
-    if not text or len(text.strip()) == 0:
-        return None, None
-        
+# Classify content
+def classify_content(text):
     try:
-        logger.info("Loading classifier...")
-        classifier = pipeline(
-            "zero-shot-classification",
-            model="facebook/bart-large-mnli",
-            device="cpu"
-        )
+        if not text or len(text.strip()) == 0:
+            return None, None
+            
+        classifier = pipeline("zero-shot-classification", 
+                           model="facebook/bart-large-mnli")
         
-        labels = [
-            "educational", "entertainment", "news",
-            "political", "religious", "technical",
-            "advertisement", "social"
-        ]
+        labels = ["educational", "entertainment", "news", "political", 
+                "religious", "technical", "advertisement", "social"]
         
-        logger.info("Starting classification...")
-        result = classifier(
-            text,
-            candidate_labels=labels,
-            hypothesis_template="This text is about {}."
-        )
+        result = classifier(text, 
+                          candidate_labels=labels, 
+                          hypothesis_template="This text is about {}.")
         
-        if result and 'labels' in result and 'scores' in result:
-            return result['labels'][0], result['scores'][0]
-    
+        return result['labels'][0], result['scores'][0]
     except Exception as e:
-        logger.error(f"Classification error: {str(e)}")
-        logger.error(traceback.format_exc())
-    
-    return None, None
+        print(f"Classification error: {e}")
+        return None, None
 
 @app.post("/analyze", response_model=AnalysisResult)
 async def analyze_video(request: VideoRequest):
-    """Main endpoint for video analysis"""
     clean_temp_files()
     
-    # Validate URL
-    if not request.url or not any(domain in request.url for domain in ["tiktok.com", "vm.tiktok.com"]):
-        error_msg = "Invalid TikTok URL. Please provide a valid TikTok video URL."
-        logger.error(error_msg)
-        return AnalysisResult(error=error_msg)
+    video_url = request.video_url
     
-    try:
-        # Step 1: Download video
-        logger.info(f"Processing URL: {request.url}")
-        video_path = download_video(request.url)
-        if not video_path:
-            error_msg = "Failed to download video. The URL may be invalid or private."
-            logger.error(error_msg)
-            return AnalysisResult(error=error_msg)
-        
-        # Step 2: Extract audio
-        audio_path = extract_audio(video_path)
-        if not audio_path:
-            error_msg = "Failed to extract audio from video."
-            logger.error(error_msg)
-            clean_temp_files()
-            return AnalysisResult(error=error_msg)
-        
-        # Step 3: Transcribe audio
-        transcription = transcribe_audio(audio_path)
-        if not transcription:
-            error_msg = "Failed to transcribe audio content."
-            logger.error(error_msg)
-            clean_temp_files()
-            return AnalysisResult(error=error_msg)
-        
-        # Step 4: Classify content
-        category, confidence = classify_content(transcription)
-        
-        # Clean up
+    if not video_url or len(video_url.strip()) == 0:
+        return AnalysisResult(
+            transcription="",
+            category=None,
+            confidence=None,
+            error="Please enter a valid TikTok URL"
+        )
+    
+    if "tiktok.com" not in video_url and "vm.tiktok.com" not in video_url:
+        return AnalysisResult(
+            transcription="",
+            category=None,
+            confidence=None,
+            error="This app is for TikTok links only"
+        )
+    
+    # Download video
+    video_path = download_video(video_url)
+    if not video_path:
+        return AnalysisResult(
+            transcription="",
+            category=None,
+            confidence=None,
+            error="Failed to download video"
+        )
+    
+    # Extract audio
+    audio_path = extract_audio(video_path)
+    if not audio_path:
         clean_temp_files()
-        
         return AnalysisResult(
-            transcription=transcription,
-            category=category,
-            confidence=confidence
+            transcription="",
+            category=None,
+            confidence=None,
+            error="Failed to extract audio"
         )
     
-    except Exception as e:
-        error_msg = f"Internal server error: {str(e)}"
-        logger.error(error_msg)
-        logger.error(traceback.format_exc())
+    # Transcribe
+    transcription = transcribe_audio(audio_path)
+    if not transcription:
+        clean_temp_files()
+        return AnalysisResult(
+            transcription="",
+            category=None,
+            confidence=None,
+            error="Failed to transcribe audio"
+        )
+    
+    # Classify
+    category, confidence = classify_content(transcription)
+    if not category:
         clean_temp_files()
-        return AnalysisResult(error=error_msg)
+        return AnalysisResult(
+            transcription=transcription,
+            category=None,
+            confidence=None,
+            error="Failed to classify content"
+        )
+    
+    # Clean up
+    clean_temp_files()
+    
+    return AnalysisResult(
+        transcription=transcription,
+        category=category,
+        confidence=confidence,
+        error=None
+    )
 
 @app.get("/")
-async def health_check():
-    """Health check endpoint"""
-    return {
-        "status": "OK",
-        "message": "TikTok Analyzer API is running",
-        "endpoints": {
-            "analyze": "POST /analyze",
-            "health": "GET /"
-        }
-    }
+async def read_root():
+    return {"message": "TikTok Content Analyzer API is running"}