Browse the data

1import os
2import uuid
3import hashlib
4from fastapi import FastAPI, HTTPException, Header
5from fastapi.responses import StreamingResponse
6from pydantic import BaseModel
7import uvicorn
8 
9app = FastAPI()
10 
11users = {}
12tokens = {}
13tracks = {}
14playlists = {}
15next_user_id = 1
16next_track_id = 1
17next_playlist_id = 1
18 
19MUSIC_DIR = "/var/music"
20 
21def get_current_user(authorization: str = Header(None)):
22    if not authorization:
23        raise HTTPException(status_code=401, detail="Missing auth header")
24    token = authorization.replace("Bearer ", "")
25    user_id = tokens.get(token)
26    if user_id is None:
27        raise HTTPException(status_code=401, detail="Invalid token")
28    return user_id
29 
30class SignupRequest(BaseModel):
31    username: str
32    password: str
33 
34class LoginRequest(BaseModel):
35    username: str
36    password: str
37 
38class TrackCreate(BaseModel):
39    title: str
40    artist: str = ""
41 
42class PlaylistCreate(BaseModel):
43    name: str
44    track_ids: list[int] = []
45 
46@app.post("/signup")
47def signup(req: SignupRequest):
48    global next_user_id
49    for u in users.values():
50        if u["username"] == req.username:
51            raise HTTPException(status_code=400, detail="Username exists")
52    user_id = next_user_id
53    next_user_id += 1
54    users[user_id] = {
55        "id": user_id,
56        "username": req.username,
57        "password": hashlib.sha256(req.password.encode()).hexdigest()
58    }
59    return {"id": user_id, "username": req.username}
60 
61@app.post("/login")
62def login(req: LoginRequest):
63    for uid, u in users.items():
64        if u["username"] == req.username and u["password"] == hashlib.sha256(req.password.encode()).hexdigest():
65            token = str(uuid.uuid4())
66            tokens[token] = uid
67            return {"token": token}
68    raise HTTPException(status_code=401, detail="Invalid credentials")
69 
70@app.get("/track/{track_id}")
71def get_track(track_id: int, authorization: str = Header(None)):
72    user_id = get_current_user(authorization)
73    track = tracks.get(track_id)
74    if not track:
75        raise HTTPException(status_code=404, detail="Track not found")
76    return track
77 
78@app.post("/track")
79def create_track(track: TrackCreate, authorization: str = Header(None)):
80    user_id = get_current_user(authorization)
81    global next_track_id
82    tid = next_track_id
83    next_track_id += 1
84    tracks[tid] = {
85        "id": tid,
86        "title": track.title,
87        "artist": track.artist,
88        "owner_id": user_id
89    }
90    return tracks[tid]
91 
92@app.get("/track/{track_id}/audio/{file}")
93def stream_audio(track_id: int, file: str, authorization: str = Header(None)):
94    user_id = get_current_user(authorization)
95    track = tracks.get(track_id)
96    if not track:
97        raise HTTPException(status_code=404, detail="Track not found")
98    file_path = os.path.join(MUSIC_DIR, str(track_id), file)
99    if not os.path.exists(file_path):
100        raise HTTPException(status_code=404, detail="Audio file not found")
101    def iterfile():
102        with open(file_path, "rb") as f:
103            yield from f
104    return StreamingResponse(iterfile(), media_type="audio/mpeg")
105 
106@app.get("/playlist/{playlist_id}")
107def get_playlist(playlist_id: int, authorization: str = Header(None)):
108    user_id = get_current_user(authorization)
109    playlist = playlists.get(playlist_id)
110    if not playlist:
111        raise HTTPException(status_code=404, detail="Playlist not found")
112    return playlist
113 
114@app.post("/playlist")
115def create_playlist(playlist: PlaylistCreate, authorization: str = Header(None)):
116    user_id = get_current_user(authorization)
117    global next_playlist_id
118    pid = next_playlist_id
119    next_playlist_id += 1
120    playlists[pid] = {
121        "id": pid,
122        "name": playlist.name,
123        "track_ids": playlist.track_ids,
124        "owner_id": user_id
125    }
126    return playlists[pid]
127 
128@app.get("/playlist/{playlist_id}/files")
129def get_playlist_files(playlist_id: int, authorization: str = Header(None)):
130    user_id = get_current_user(authorization)
131    playlist = playlists.get(playlist_id)
132    if not playlist:
133        raise HTTPException(status_code=404, detail="Playlist not found")
134    files = []
135    for tid in playlist["track_ids"]:
136        track = tracks.get(tid)
137        if track:
138            track_dir = os.path.join(MUSIC_DIR, str(tid))
139            if os.path.isdir(track_dir):
140                for f in sorted(os.listdir(track_dir)):
141                    if f.endswith(".mp3"):
142                        files.append(f"/track/{tid}/audio/{f}")
143    return {"playlist_id": playlist_id, "name": playlist["name"], "files": files}

1fastapi
2uvicorn