| from fastapi import FastAPI, HTTPException |
| from fastapi.middleware.cors import CORSMiddleware |
| from youtube_transcript_api import YouTubeTranscriptApi, TranscriptsDisabled, NoTranscriptFound, VideoUnavailable |
| import re |
|
|
| app = FastAPI() |
|
|
| |
| app.add_middleware( |
| CORSMiddleware, |
| allow_origins=["*"], |
| allow_credentials=True, |
| allow_methods=["*"], |
| allow_headers=["*"], |
| ) |
|
|
| def extract_video_id(url: str) -> str: |
| pattern = r'(?:youtube\.com\/(?:[^\/]+\/.+\/|(?:v|e(?:mbed)?)\/|.*[?&]v=)|youtu\.be\/)([^"&?\/\s]{11})' |
| match = re.match(pattern, url) |
| if match: |
| return match.group(1) |
| return None |
|
|
| @app.post("/api/transcript") |
| async def get_transcript(request: dict): |
| url = request.get("url") |
| if not url: |
| raise HTTPException(status_code=400, detail="Missing URL") |
|
|
| video_id = extract_video_id(url) |
| if not video_id: |
| raise HTTPException(status_code=400, detail="Invalid YouTube URL") |
|
|
| try: |
| |
| transcript_list = YouTubeTranscriptApi.list_transcripts(video_id) |
| |
| |
| transcript = None |
| try: |
| transcript = transcript_list.find_manually_created_transcript(['en']) |
| except NoTranscriptFound: |
| try: |
| transcript = transcript_list.find_generated_transcript(['en']) |
| except NoTranscriptFound: |
| |
| for t in transcript_list: |
| transcript = t |
| break |
| |
| if not transcript: |
| raise HTTPException(status_code=404, detail="No captions found for this video.") |
|
|
| |
| transcript_data = transcript.fetch() |
| |
| |
| full_text = " ".join([line['text'] for line in transcript_data]) |
| |
| |
| full_text = full_text.replace("'", "'").replace(""", '"').replace("&", "&").replace("'", "'") |
| |
| return {"text": full_text} |
|
|
| except TranscriptsDisabled: |
| raise HTTPException(status_code=403, detail="Captions are disabled for this video.") |
| except VideoUnavailable: |
| raise HTTPException(status_code=404, detail="Video is unavailable.") |
| except Exception as e: |
| raise HTTPException(status_code=500, detail=f"Failed to fetch transcript: {str(e)}") |
|
|
| |
| |
| |
| @app.get("/api/health") |
| async def health_check(): |
| return {"status": "ok", "message": "Transcript API is running!"} |