import os import uuid import logging from datetime import datetime from fastapi import APIRouter, Depends, HTTPException, Query, Header, Request from fastapi.responses import FileResponse, StreamingResponse, Response from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select from ..dependencies import get_db, get_current_user from ..models.user import User from ..models.media_item import LibraryItem, BookFile, Chapter from ..models.session import PlaybackSession from ..models.progress import MediaProgress from ..services.auth import decode_token logger = logging.getLogger(__name__) router = APIRouter(tags=["stream"]) _MIME = { ".mp3": "audio/mpeg", ".m4a": "audio/mp4", ".m4b": "audio/mp4", ".aac": "audio/aac", ".ogg": "audio/ogg", ".opus": "audio/ogg", ".flac": "audio/flac", ".wav": "audio/wav", } @router.post("/api/items/{item_id}/play") async def start_playback( item_id: str, body: dict | None = None, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): body = body or {} result = await db.execute(select(LibraryItem).where(LibraryItem.id == item_id)) item = result.scalar_one_or_none() if not item: raise HTTPException(status_code=404, detail="Item not found") files_result = await db.execute( select(BookFile).where(BookFile.library_item_id == item_id).order_by(BookFile.track_index) ) files = files_result.scalars().all() if not files: raise HTTPException(status_code=400, detail="Keine Audio-Dateien vorhanden") chapters_result = await db.execute( select(Chapter).where(Chapter.library_item_id == item_id).order_by(Chapter.chapter_index) ) chapters = chapters_result.scalars().all() progress_result = await db.execute( select(MediaProgress).where( MediaProgress.user_id == current_user.id, MediaProgress.library_item_id == item_id, ) ) progress = progress_result.scalar_one_or_none() current_time = progress.current_time if progress else 0.0 if body.get("startTime") is not None: current_time = float(body["startTime"]) session_id = str(uuid.uuid4()) session = PlaybackSession( id=session_id, user_id=current_user.id, library_item_id=item_id, media_type=item.media_type, current_time=current_time, duration=item.duration_seconds or 0.0, device_id=body.get("deviceId", ""), device_info=body.get("deviceInfo", {}), media_player=body.get("mediaPlayer", ""), hls_session_path=None, is_active=True, ) db.add(session) await db.commit() chapters_out = [ {"id": c.chapter_index, "start": c.start_seconds, "end": c.end_seconds, "title": c.title} for c in chapters ] audio_tracks = [] offset = 0.0 for i, f in enumerate(files): dur = f.duration_seconds or 0.0 ext = os.path.splitext(f.filename or "")[1].lower() audio_tracks.append({ "index": i, "startOffset": offset, "duration": dur, "title": f.filename or f"Part {i + 1}", "contentUrl": f"/api/stream/{session_id}/{i}", "mimeType": _MIME.get(ext, "audio/mpeg"), }) offset += dur total_duration = item.duration_seconds or offset logger.info( f"Playback gestartet: item={item.title!r} session={session_id} " f"files={len(files)} duration={total_duration:.1f}s startAt={current_time:.1f}s" ) return { "id": session_id, "userId": current_user.id, "libraryId": item.library_id, "libraryItemId": item_id, "episodeId": None, "mediaType": item.media_type, "chapters": chapters_out, "displayTitle": item.title, "displayAuthor": item.author, "coverPath": f"/api/items/{item_id}/cover" if item.cover_path else None, "duration": total_duration, "playMethod": 2, "mediaPlayer": body.get("mediaPlayer", ""), "deviceInfo": body.get("deviceInfo", {}), "serverVersion": "2.4.0", "date": datetime.utcnow().strftime("%Y-%m-%d"), "dayOfWeek": datetime.utcnow().strftime("%A"), "timeListening": 0, "startTime": current_time, "currentTime": current_time, "startedAt": int(datetime.utcnow().timestamp() * 1000), "updatedAt": int(datetime.utcnow().timestamp() * 1000), "audioTracks": audio_tracks, "videoTrack": None, } def _parse_range(range_header: str, file_size: int) -> tuple[int, int] | None: """Parsed 'bytes=START-END'. Gibt (start, end) zurück oder None bei Fehler.""" if not range_header or not range_header.startswith("bytes="): return None try: spec = range_header[6:].strip() # nur erste Range spec = spec.split(",")[0] start_s, _, end_s = spec.partition("-") if start_s == "": # Suffix-Range: bytes=-500 → letzte 500 Bytes length = int(end_s) start = max(0, file_size - length) end = file_size - 1 else: start = int(start_s) end = int(end_s) if end_s else file_size - 1 if start < 0 or end < start or start >= file_size: return None end = min(end, file_size - 1) return start, end except (ValueError, IndexError): return None async def _stream_file_range(path: str, start: int, end: int, chunk_size: int = 64 * 1024): """Streamt einen Byte-Bereich aus einer Datei.""" with open(path, "rb") as f: f.seek(start) remaining = end - start + 1 while remaining > 0: chunk = f.read(min(chunk_size, remaining)) if not chunk: break remaining -= len(chunk) yield chunk @router.get("/api/stream/{session_id}/{track}") async def stream_file( session_id: str, track: int, request: Request, token: str | None = Query(None), authorization: str | None = Header(None), db: AsyncSession = Depends(get_db), ): """Audio-Streaming mit nativen HTTP Range Requests (206 Partial Content).""" raw = token if not raw and authorization: parts = authorization.split(" ", 1) if len(parts) == 2 and parts[0].lower() == "bearer": raw = parts[1] if not raw or not decode_token(raw): logger.warning(f"Stream 401: session={session_id} track={track}") raise HTTPException(status_code=401, detail="Nicht autorisiert") result = await db.execute( select(PlaybackSession).where(PlaybackSession.id == session_id) ) session = result.scalar_one_or_none() if not session: raise HTTPException(status_code=404, detail="Session nicht gefunden") files_result = await db.execute( select(BookFile) .where(BookFile.library_item_id == session.library_item_id) .order_by(BookFile.track_index) ) files = files_result.scalars().all() if track < 0 or track >= len(files): raise HTTPException(status_code=404, detail="Track nicht gefunden") path = files[track].path if not os.path.exists(path): logger.warning(f"Audio-Datei nicht gefunden: {path}") raise HTTPException(status_code=404, detail="Audio-Datei nicht gefunden") file_size = os.path.getsize(path) ext = os.path.splitext(path)[1].lower() media_type = _MIME.get(ext, "audio/mpeg") range_header = request.headers.get("range") or request.headers.get("Range") parsed = _parse_range(range_header, file_size) if range_header else None common_headers = { "Accept-Ranges": "bytes", "Cache-Control": "no-cache", } if parsed: start, end = parsed chunk_size = end - start + 1 logger.info( f"Stream Range: session={session_id} track={track} " f"bytes={start}-{end}/{file_size}" ) return StreamingResponse( _stream_file_range(path, start, end), status_code=206, media_type=media_type, headers={ **common_headers, "Content-Range": f"bytes {start}-{end}/{file_size}", "Content-Length": str(chunk_size), }, ) logger.info(f"Stream full: session={session_id} track={track} size={file_size}") return FileResponse( path, media_type=media_type, headers={**common_headers, "Content-Length": str(file_size)}, ) @router.head("/api/stream/{session_id}/{track}") async def stream_head( session_id: str, track: int, token: str | None = Query(None), authorization: str | None = Header(None), db: AsyncSession = Depends(get_db), ): """HEAD-Request für Audio-Datei (Metadaten ohne Body).""" raw = token if not raw and authorization: parts = authorization.split(" ", 1) if len(parts) == 2 and parts[0].lower() == "bearer": raw = parts[1] if not raw or not decode_token(raw): raise HTTPException(status_code=401, detail="Nicht autorisiert") result = await db.execute( select(PlaybackSession).where(PlaybackSession.id == session_id) ) session = result.scalar_one_or_none() if not session: raise HTTPException(status_code=404, detail="Session nicht gefunden") files_result = await db.execute( select(BookFile) .where(BookFile.library_item_id == session.library_item_id) .order_by(BookFile.track_index) ) files = files_result.scalars().all() if track < 0 or track >= len(files): raise HTTPException(status_code=404, detail="Track nicht gefunden") path = files[track].path if not os.path.exists(path): raise HTTPException(status_code=404, detail="Audio-Datei nicht gefunden") file_size = os.path.getsize(path) ext = os.path.splitext(path)[1].lower() media_type = _MIME.get(ext, "audio/mpeg") return Response( headers={ "Accept-Ranges": "bytes", "Content-Length": str(file_size), "Content-Type": media_type, } ) @router.post("/api/playback-session/{session_id}/sync") async def sync_session( session_id: str, body: dict | None = None, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): body = body or {} result = await db.execute( select(PlaybackSession).where( PlaybackSession.id == session_id, PlaybackSession.user_id == current_user.id, ) ) session = result.scalar_one_or_none() if not session: raise HTTPException(status_code=404, detail="Session not found") current_time = float(body.get("currentTime", session.current_time)) duration = float(body.get("duration", session.duration)) session.current_time = current_time session.duration = duration session.updated_at = datetime.utcnow() progress_result = await db.execute( select(MediaProgress).where( MediaProgress.user_id == current_user.id, MediaProgress.library_item_id == session.library_item_id, MediaProgress.episode_id == session.episode_id, ) ) progress = progress_result.scalar_one_or_none() if not progress: progress = MediaProgress( user_id=current_user.id, library_item_id=session.library_item_id, episode_id=session.episode_id, duration=duration, started_at=datetime.utcnow(), ) db.add(progress) progress.current_time = current_time progress.duration = duration progress.last_update = datetime.utcnow() is_finished = duration > 0 and (current_time / duration) >= 0.99 if is_finished and not progress.is_finished: progress.is_finished = True progress.finished_at = datetime.utcnow() elif not is_finished: progress.is_finished = False await db.commit() return {"id": session_id, "currentTime": current_time} @router.delete("/api/playback-session/{session_id}") async def close_session( session_id: str, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): result = await db.execute( select(PlaybackSession).where( PlaybackSession.id == session_id, PlaybackSession.user_id == current_user.id, ) ) session = result.scalar_one_or_none() if not session: raise HTTPException(status_code=404, detail="Session not found") session.is_active = False await db.commit() return {"success": True}