feat(stream): implement stream processing service with routes for managing streaming URLs; add support for concurrent processing and metadata extraction
This commit is contained in:
JSC
210
app/routes/stream.py
Normal file
210
app/routes/stream.py
Normal file
@@ -0,0 +1,210 @@
|
||||
"""Stream routes for managing streaming service links."""
|
||||
|
||||
from flask import Blueprint, jsonify, request
|
||||
|
||||
from app.database import db
|
||||
from app.models.stream import Stream
|
||||
from app.services.decorators import require_auth
|
||||
|
||||
bp = Blueprint("stream", __name__)
|
||||
|
||||
|
||||
@bp.route("/add-url", methods=["POST"])
|
||||
@require_auth
|
||||
def add_url():
|
||||
"""Add a URL to the stream processing queue."""
|
||||
try:
|
||||
data = request.get_json()
|
||||
|
||||
if not data or "url" not in data:
|
||||
return jsonify({"error": "URL is required"}), 400
|
||||
|
||||
url = data["url"].strip()
|
||||
if not url:
|
||||
return jsonify({"error": "URL cannot be empty"}), 400
|
||||
|
||||
# Check if URL already exists
|
||||
existing_stream = Stream.query.filter_by(url=url).first()
|
||||
if existing_stream:
|
||||
return (
|
||||
jsonify(
|
||||
{
|
||||
"error": "URL already exists in stream",
|
||||
"stream": existing_stream.to_dict(),
|
||||
}
|
||||
),
|
||||
409,
|
||||
)
|
||||
|
||||
# Try to extract basic metadata to check for service/service_id duplicates
|
||||
from app.services.stream_processing_service import (
|
||||
StreamProcessingService,
|
||||
)
|
||||
|
||||
try:
|
||||
metadata, _ = StreamProcessingService._extract_metadata(url)
|
||||
if metadata:
|
||||
service = metadata.get("service")
|
||||
service_id = metadata.get("service_id")
|
||||
|
||||
if service and service_id:
|
||||
existing_service_stream = Stream.query.filter_by(
|
||||
service=service, service_id=service_id
|
||||
).first()
|
||||
|
||||
if existing_service_stream:
|
||||
return (
|
||||
jsonify(
|
||||
{
|
||||
"error": f"Stream already exists with {service} ID: {service_id}",
|
||||
"existing_stream": existing_service_stream.to_dict(),
|
||||
}
|
||||
),
|
||||
409,
|
||||
)
|
||||
except Exception as e:
|
||||
# If metadata extraction fails here, we'll let the background process handle it
|
||||
# This is just an early check to prevent obvious duplicates
|
||||
pass
|
||||
|
||||
# Create stream entry with pending status
|
||||
stream = Stream.create_stream(url=url, status="pending", commit=True)
|
||||
|
||||
# Add to processing queue (will be implemented next)
|
||||
from app.services.stream_processing_service import (
|
||||
StreamProcessingService,
|
||||
)
|
||||
|
||||
StreamProcessingService.add_to_queue(stream.id)
|
||||
|
||||
return (
|
||||
jsonify(
|
||||
{
|
||||
"message": "URL added to processing queue",
|
||||
"stream": stream.to_dict(),
|
||||
}
|
||||
),
|
||||
201,
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
db.session.rollback()
|
||||
return jsonify({"error": str(e)}), 500
|
||||
|
||||
|
||||
@bp.route("/", methods=["GET"])
|
||||
@require_auth
|
||||
def list_streams():
|
||||
"""List all streams with optional filtering."""
|
||||
try:
|
||||
status = request.args.get("status")
|
||||
service = request.args.get("service")
|
||||
|
||||
query = Stream.query
|
||||
|
||||
if status:
|
||||
query = query.filter_by(status=status)
|
||||
if service:
|
||||
query = query.filter_by(service=service)
|
||||
|
||||
streams = query.order_by(Stream.created_at.desc()).all()
|
||||
|
||||
return (
|
||||
jsonify({"streams": [stream.to_dict() for stream in streams]}),
|
||||
200,
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
return jsonify({"error": str(e)}), 500
|
||||
|
||||
|
||||
@bp.route("/<int:stream_id>", methods=["GET"])
|
||||
@require_auth
|
||||
def get_stream(stream_id):
|
||||
"""Get a specific stream by ID."""
|
||||
try:
|
||||
stream = Stream.query.get_or_404(stream_id)
|
||||
return jsonify({"stream": stream.to_dict()}), 200
|
||||
|
||||
except Exception as e:
|
||||
return jsonify({"error": str(e)}), 500
|
||||
|
||||
|
||||
@bp.route("/<int:stream_id>", methods=["PUT"])
|
||||
@require_auth
|
||||
def update_stream(stream_id):
|
||||
"""Update stream metadata."""
|
||||
try:
|
||||
stream = Stream.query.get_or_404(stream_id)
|
||||
data = request.get_json()
|
||||
|
||||
if not data:
|
||||
return jsonify({"error": "No data provided"}), 400
|
||||
|
||||
# Update allowed fields
|
||||
updatable_fields = [
|
||||
"title",
|
||||
"track",
|
||||
"artist",
|
||||
"album",
|
||||
"genre",
|
||||
"status",
|
||||
]
|
||||
for field in updatable_fields:
|
||||
if field in data:
|
||||
setattr(stream, field, data[field])
|
||||
|
||||
db.session.commit()
|
||||
|
||||
return (
|
||||
jsonify(
|
||||
{
|
||||
"message": "Stream updated successfully",
|
||||
"stream": stream.to_dict(),
|
||||
}
|
||||
),
|
||||
200,
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
db.session.rollback()
|
||||
return jsonify({"error": str(e)}), 500
|
||||
|
||||
|
||||
@bp.route("/<int:stream_id>", methods=["DELETE"])
|
||||
@require_auth
|
||||
def delete_stream(stream_id):
|
||||
"""Delete a stream."""
|
||||
try:
|
||||
stream = Stream.query.get_or_404(stream_id)
|
||||
|
||||
# If stream is being processed, mark for deletion instead
|
||||
if stream.status == "processing":
|
||||
stream.status = "cancelled"
|
||||
db.session.commit()
|
||||
return jsonify({"message": "Stream marked for cancellation"}), 200
|
||||
|
||||
db.session.delete(stream)
|
||||
db.session.commit()
|
||||
|
||||
return jsonify({"message": "Stream deleted successfully"}), 200
|
||||
|
||||
except Exception as e:
|
||||
db.session.rollback()
|
||||
return jsonify({"error": str(e)}), 500
|
||||
|
||||
|
||||
@bp.route("/queue/status", methods=["GET"])
|
||||
@require_auth
|
||||
def queue_status():
|
||||
"""Get the current processing queue status."""
|
||||
try:
|
||||
from app.services.stream_processing_service import (
|
||||
StreamProcessingService,
|
||||
)
|
||||
|
||||
status = StreamProcessingService.get_queue_status()
|
||||
return jsonify(status), 200
|
||||
|
||||
except Exception as e:
|
||||
return jsonify({"error": str(e)}), 500
|
||||
Reference in New Issue
Block a user