Update app.py

app.py

@@ -2,7 +2,8 @@ import re
 import json
 import numpy as np
 import faiss
-from flask import Flask, request, jsonify
+from fastapi import FastAPI, HTTPException
+from pydantic import BaseModel
 from transformers import (
     pipeline,
     AutoModelForSequenceClassification,
@@ -14,8 +15,18 @@ from transformers import (
 )
 from sentence_transformers import SentenceTransformer
 from bertopic import BERTopic
-from datasets import load_dataset, Features, Value
-import pyarrow.parquet as pq
+from datasets import Features, Value
+from googleapiclient.discovery import build
+from youtube_transcript_api import YouTubeTranscriptApi
+
+# Initialize FastAPI app
+app = FastAPI()
+
+# YouTube Data API setup
+API_KEY = "your_youtube_api_key"
+YOUTUBE_API_SERVICE_NAME = "youtube"
+YOUTUBE_API_VERSION = "v3"
+youtube = build(YOUTUBE_API_SERVICE_NAME, YOUTUBE_API_VERSION, developerKey=API_KEY)
 
 # Preprocessing function
 def preprocess_text(text):
@@ -128,9 +139,6 @@ class Chatbot:
         return response
 
 
-# Flask API for Chatbot Integration
-app = Flask(__name__)
-
 # Initialize models
 classifier = ContentClassifier()
 relevance_detector = RelevanceDetector()
@@ -139,93 +147,135 @@ search_engine = SearchEngine()
 topic_extractor = TopicExtractor()
 chatbot = Chatbot()
 
-# Load the yt-commons dataset
-
-
-# Define the schema
-features = Features({
-    "video_id": Value("string"),
-    "video_link": Value("string"),
-    "title": Value("string"),
-    "text": Value("string"),
-    "channel": Value("string"),
-    "channel_id": Value("string"),
-    "date": Value("string"),
-    "license": Value("string"),
-    "original_language": Value("string"),
-    "source_language": Value("string"),
-    "transcription_language": Value("string"),
-    "word_count": Value("int64"),
-    "character_count": Value("int64"),
-})
-
-# Load the dataset from Hugging Face Hub
-try:
-    dataset = load_dataset(
-        "PleIAs/YouTube-Commons",
-        features=features,
-        streaming=True,
+# Fetch video metadata using YouTube Data API
+def fetch_video_metadata(video_id):
+    request = youtube.videos().list(
+        part="snippet,statistics",
+        id=video_id
     )
+    response = request.execute()
+    return response["items"][0] if response["items"] else None
+
+
+# Fetch video transcript using youtube-transcript-api
+def fetch_video_transcript(video_id):
+    try:
+        transcript = YouTubeTranscriptApi.get_transcript(video_id)
+        return " ".join([entry["text"] for entry in transcript])
+    except Exception as e:
+        print(f"Error fetching transcript: {e}")
+        return None
+
+
+# Fetch and preprocess video data
+def fetch_and_preprocess_video_data(video_id):
+    metadata = fetch_video_metadata(video_id)
+    if not metadata:
+        return None
+
+    transcript = fetch_video_transcript(video_id)
+
+    # Preprocess the data
+    video_data = {
+        "video_id": video_id,
+        "video_link": f"https://www.youtube.com/watch?v={video_id}",
+        "title": metadata["snippet"]["title"],
+        "text": transcript if transcript else metadata["snippet"]["description"],
+        "channel": metadata["snippet"]["channelTitle"],
+        "channel_id": metadata["snippet"]["channelId"],
+        "date": metadata["snippet"]["publishedAt"],
+        "license": "Unknown",
+        "original_language": "Unknown",
+        "source_language": "Unknown",
+        "transcription_language": "Unknown",
+        "word_count": len(metadata["snippet"]["description"].split()),
+        "character_count": len(metadata["snippet"]["description"]),
+    }
+    return video_data
+
+
+# Pydantic models for request validation
+class VideoRequest(BaseModel):
+    video_id: str
+
+
+class TextRequest(BaseModel):
+    text: str
+
+
+class QueryRequest(BaseModel):
+    query: str
+
+
+class PromptRequest(BaseModel):
+    prompt: str
 
-    # Process the dataset
-    for example in dataset["train"]:
-        print(example)  # Process each example
-        break  # Stop after the first example for demonstration
-except Exception as e:
-    print(f"Error loading dataset: {e}")
 
 # API Endpoints
-@app.route("/classify", methods=["POST"])
-def classify():
-    text = request.json.get("text", "")
-    if not text:
-        return jsonify({"error": "No text provided"}), 400
-    result = classifier.classify(text)
-    return jsonify(result)
+@app.post("/classify")
+async def classify(request: VideoRequest):
+    video_id = request.video_id
+    video_data = fetch_and_preprocess_video_data(video_id)
+    if not video_data:
+        raise HTTPException(status_code=400, detail="Failed to fetch video data")
 
+    result = classifier.classify(video_data["text"])
+    return {"result": result}
 
-@app.route("/relevance", methods=["POST"])
-def relevance():
-    text = request.json.get("text", "")
-    if not text:
-        return jsonify({"error": "No text provided"}), 400
-    relevant = relevance_detector.detect_relevance(text)
-    return jsonify({"relevant": relevant})
 
+@app.post("/relevance")
+async def relevance(request: VideoRequest):
+    video_id = request.video_id
+    video_data = fetch_and_preprocess_video_data(video_id)
+    if not video_data:
+        raise HTTPException(status_code=400, detail="Failed to fetch video data")
+
+    relevant = relevance_detector.detect_relevance(video_data["text"])
+    return {"relevant": relevant}
 
-@app.route("/summarize", methods=["POST"])
-def summarize():
-    text = request.json.get("text", "")
-    if not text:
-        return jsonify({"error": "No text provided"}), 400
-    summary = summarizer.summarize(text)
-    return jsonify({"summary": summary})
 
+@app.post("/summarize")
+async def summarize(request: VideoRequest):
+    video_id = request.video_id
+    video_data = fetch_and_preprocess_video_data(video_id)
+    if not video_data:
+        raise HTTPException(status_code=400, detail="Failed to fetch video data")
 
-@app.route("/search", methods=["POST"])
-def search():
-    query = request.json.get("query", "")
+    summary = summarizer.summarize(video_data["text"])
+    return {"summary": summary}
+
+
+@app.post("/search")
+async def search(request: QueryRequest):
+    query = request.query
     if not query:
-        return jsonify({"error": "No query provided"}), 400
+        raise HTTPException(status_code=400, detail="No query provided")
+
     results = search_engine.search(query)
-    return jsonify({"results": results})
+    return {"results": results}
 
 
-@app.route("/topics", methods=["POST"])
-def topics():
-    result = topic_extractor.extract_topics(youtube_data)
-    return jsonify({"topics": result.to_dict()})
+@app.post("/topics")
+async def topics(request: TextRequest):
+    text = request.text
+    if not text:
+        raise HTTPException(status_code=400, detail="No text provided")
+
+    result = topic_extractor.extract_topics([text])
+    return {"topics": result.to_dict()}
 
 
-@app.route("/chat", methods=["POST"])
-def chat():
-    prompt = request.json.get("prompt", "")
+@app.post("/chat")
+async def chat(request: PromptRequest):
+    prompt = request.prompt
     if not prompt:
-        return jsonify({"error": "No prompt provided"}), 400
+        raise HTTPException(status_code=400, detail="No prompt provided")
+
     response = chatbot.generate_response(prompt)
-    return jsonify({"response": response})
+    return {"response": response}
 
 
-# Start the Flask API
+# Start the FastAPI app
 if __name__ == "__main__":
-    app.run(debug=True)
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8000)