Browse the data

1import secrets
2import time
3from datetime import datetime, timedelta
4from fastapi import FastAPI, HTTPException, Header
5from pydantic import BaseModel
6from typing import Optional
7 
8app = FastAPI()
9 
10users = {}
11tokens = {}
12playlists = {}
13playlist_videos = {}
14playlist_shares = {}
15video_id_counter = 1
16playlist_id_counter = 1
17 
18class SignupRequest(BaseModel):
19    username: str
20    password: str
21 
22class LoginRequest(BaseModel):
23    username: str
24    password: str
25 
26class CreatePlaylistRequest(BaseModel):
27    name: str
28 
29class AddVideoRequest(BaseModel):
30    url: str
31    title: str = ""
32 
33class SharePlaylistRequest(BaseModel):
34    playlist_id: int
35    expiry_hours: int = 24
36 
37@app.post("/signup")
38def signup(req: SignupRequest):
39    if req.username in users:
40        raise HTTPException(status_code=400, detail="User exists")
41    users[req.username] = {"password": req.password, "id": len(users) + 1}
42    return {"message": "User created"}
43 
44@app.post("/login")
45def login(req: LoginRequest):
46    user = users.get(req.username)
47    if not user or user["password"] != req.password:
48        raise HTTPException(status_code=401, detail="Invalid credentials")
49    token = secrets.token_hex(16)
50    tokens[token] = user["id"]
51    return {"token": token}
52 
53def get_user_id(authorization: str = Header(...)):
54    if not authorization.startswith("Bearer "):
55        raise HTTPException(status_code=401, detail="Invalid auth header")
56    token = authorization[7:]
57    user_id = tokens.get(token)
58    if not user_id:
59        raise HTTPException(status_code=401, detail="Invalid token")
60    return user_id
61 
62@app.post("/playlists")
63def create_playlist(req: CreatePlaylistRequest, authorization: str = Header(...)):
64    user_id = get_user_id(authorization)
65    global playlist_id_counter
66    pid = playlist_id_counter
67    playlist_id_counter += 1
68    playlists[pid] = {"id": pid, "name": req.name, "user_id": user_id, "created_at": datetime.now().isoformat()}
69    playlist_videos[pid] = []
70    return playlists[pid]
71 
72@app.get("/playlists/{playlist_id}")
73def get_playlist(playlist_id: int, authorization: str = Header(...)):
74    user_id = get_user_id(authorization)
75    pl = playlists.get(playlist_id)
76    if not pl:
77        raise HTTPException(status_code=404, detail="Playlist not found")
78    if pl["user_id"] != user_id:
79        raise HTTPException(status_code=403, detail="Not your playlist")
80    return {"playlist": pl, "videos": playlist_videos.get(playlist_id, [])}
81 
82@app.post("/playlists/{playlist_id}/videos")
83def add_video(playlist_id: int, req: AddVideoRequest, authorization: str = Header(...)):
84    user_id = get_user_id(authorization)
85    pl = playlists.get(playlist_id)
86    if not pl:
87        raise HTTPException(status_code=404, detail="Playlist not found")
88    if pl["user_id"] != user_id:
89        raise HTTPException(status_code=403, detail="Not your playlist")
90    global video_id_counter
91    vid = video_id_counter
92    video_id_counter += 1
93    video = {"id": vid, "url": req.url, "title": req.title, "resume_timestamp": 0}
94    playlist_videos[playlist_id].append(video)
95    return video
96 
97@app.get("/playlists/{playlist_id}/watch")
98def watch_playlist(playlist_id: int, authorization: str = Header(...)):
99    user_id = get_user_id(authorization)
100    pl = playlists.get(playlist_id)
101    if not pl:
102        raise HTTPException(status_code=404, detail="Playlist not found")
103    videos = playlist_videos.get(playlist_id, [])
104    recommendations = []
105    for pid, p in playlists.items():
106        if pid != playlist_id and p["user_id"] == user_id:
107            recommendations.append({"id": pid, "name": p["name"]})
108    return {
109        "playlist_name": pl["name"],
110        "videos": [{"url": v["url"], "resume_timestamp": v["resume_timestamp"]} for v in videos],
111        "recommendations": recommendations[:5]
112    }
113 
114@app.post("/playlists/share")
115def share_playlist(req: SharePlaylistRequest, authorization: str = Header(...)):
116    user_id = get_user_id(authorization)
117    pl = playlists.get(req.playlist_id)
118    if not pl:
119        raise HTTPException(status_code=404, detail="Playlist not found")
120    if pl["user_id"] != user_id:
121        raise HTTPException(status_code=403, detail="Not your playlist")
122    share_id = secrets.token_hex(8)
123    expires_at = datetime.now() + timedelta(hours=req.expiry_hours)
124    playlist_shares[share_id] = {
125        "playlist_id": req.playlist_id,
126        "expires_at": expires_at.isoformat(),
127        "created_by": user_id
128    }
129    return {"share_link": f"/shared/{share_id}", "expires_at": expires_at.isoformat()}
130 
131@app.get("/shared/{share_id}")
132def access_shared_playlist(share_id: str):
133    share = playlist_shares.get(share_id)
134    if not share:
135        raise HTTPException(status_code=404, detail="Share link not found")
136    if datetime.fromisoformat(share["expires_at"]) < datetime.now():
137        raise HTTPException(status_code=410, detail="Share link expired")
138    pl = playlists.get(share["playlist_id"])
139    if not pl:
140        raise HTTPException(status_code=404, detail="Playlist not found")
141    videos = playlist_videos.get(share["playlist_id"], [])
142    return {
143        "playlist_name": pl["name"],
144        "videos": [{"url": v["url"], "resume_timestamp": v["resume_timestamp"]} for v in videos]
145    }

1fastapi
2uvicorn