yt-dlp-proxy/app.py

import logging
import os
from flask import Flask, render_template, request, Response, abort, jsonify
from werkzeug.exceptions import HTTPException

import dlp
from utils import is_valid_url, get_error_message, get_video_id, resolve_video_id

app = Flask(__name__)

LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO")
logging.basicConfig(
    level=getattr(logging, LOG_LEVEL),
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

PORT = int(os.getenv("PORT", 5000))


@app.route("/")
def index():
    return render_template("index.html")


@app.route("/player")
def player():
    video_url = request.args.get("url")
    if not video_url:
        abort(400, description="Missing url parameter")
    
    if not is_valid_url(video_url):
        abort(400, description="Invalid or disallowed URL")
    
    try:
        stream_info = dlp.get_stream_info(video_url)
        video_id = get_video_id(video_url)
        hls_url = stream_info.get("hls_url")
        proxy_hls_url = f"/hls/{video_id}/index.m3u8" if hls_url else None
        
        return render_template(
            "player.html",
            video_url=video_url,
            proxy_hls_url=proxy_hls_url,
            direct_url=stream_info.get("direct_url"),
            title=stream_info.get("title", "Video"),
            thumbnail=stream_info.get("thumbnail"),
            # Pass all metadata to template
            description=stream_info.get("description"),
            uploader=stream_info.get("uploader"),
            uploader_url=stream_info.get("uploader_url"),
            duration=stream_info.get("duration"),
            duration_string=stream_info.get("duration_string"),
            upload_date=stream_info.get("upload_date"),
            view_count=stream_info.get("view_count"),
            like_count=stream_info.get("like_count"),
            dislike_count=stream_info.get("dislike_count"),
            comment_count=stream_info.get("comment_count"),
            age_limit=stream_info.get("age_limit"),
            categories=stream_info.get("categories"),
            tags=stream_info.get("tags"),
            language=stream_info.get("language"),
            license=stream_info.get("license"),
            channel=stream_info.get("channel"),
            channel_url=stream_info.get("channel_url"),
            channel_id=stream_info.get("channel_id"),
            extractor=stream_info.get("extractor"),
            extractor_key=stream_info.get("extractor_key"),
            display_id=stream_info.get("display_id"),
            url=stream_info.get("url"),
            fulltitle=stream_info.get("fulltitle"),
            resolution=stream_info.get("resolution"),
            format=stream_info.get("format"),
            format_note=stream_info.get("format_note"),
            filesize=stream_info.get("filesize"),
            filesize_approx=stream_info.get("filesize_approx"),
            hls_url=hls_url
        )
    except Exception as e:
        logger.error(f"Error getting stream info: {e}")
        abort(500, description=str(e))


@app.route("/hls/<video_id>/index.m3u8")
def hls_index(video_id):
    try:
        video_url = resolve_video_id(video_id)
        if not video_url:
            abort(400, description="Unknown video id")
        
        if not is_valid_url(video_url):
            abort(400, description="Invalid URL")
        playlist = dlp.get_hls_playlist(video_url)
        return Response(playlist, mimetype="application/vnd.apple.mpegurl", headers={"Cache-Control": "public, max-age=31536000"})
    
    except HTTPException:
        raise
    except ValueError as e:
        logger.warning(f"Validation error: {e}")
        abort(400, description=str(e))
    except Exception as e:
        logger.error(f"HLS proxy error: {e}")
        return Response(str(e), status=500, mimetype="text/plain")


@app.route("/hls/<video_id>/seg/<seg_id>")
def hls_segment(video_id, seg_id):
    try:
        video_url = resolve_video_id(video_id)
        if not video_url:
            abort(400, description="Unknown video id")

        if not is_valid_url(video_url):
            abort(400, description="Invalid URL")

        data = dlp.get_hls_segment_with_retry(video_url, str(seg_id))
        if data is None:
            abort(500, description="Failed to fetch segment")

        return Response(data, mimetype="video/mp2t", headers={"Cache-Control": "public, max-age=31536000"})

    except HTTPException:
        raise
    except ValueError as e:
        logger.warning(f"Validation error: {e}")
        abort(400, description=str(e))
    except Exception as e:
        logger.error(f"HLS segment error: {e}")
        return Response(str(e), status=500, mimetype="text/plain")


@app.errorhandler(Exception)
def handle_error(e):
    if isinstance(e, HTTPException):
        return jsonify({"error": get_error_message(e.code), "message": str(e.description)}), e.code
    logger.error(f"Unexpected error: {e}")
    return jsonify({"error": "Internal Server Error", "message": str(e)}), 500


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=PORT)