Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from typing import Optional
3import time
4import hashlib
5import random
6 
7app = FastAPI()
8 
9users = {}
10tokens = {}
11videos = {
12    1: {"id": 1, "title": "How to Code in Python", "duration": 3600, "thumbnail_url": "https://example.com/thumb1.jpg", "view_count": 0},
13    2: {"id": 2, "title": "FastAPI Crash Course", "duration": 1800, "thumbnail_url": "https://example.com/thumb2.jpg", "view_count": 0},
14    3: {"id": 3, "title": "Machine Learning Basics", "duration": 5400, "thumbnail_url": "https://example.com/thumb3.jpg", "view_count": 0},
15}
16next_video_id = 4
17watch_logs = []
18recommendations_map = {
19    1: [2, 3],
20    2: [1, 3],
21    3: [1, 2],
22}
23 
24def get_current_user(authorization: Optional[str] = Header(None)):
25    if not authorization:
26        raise HTTPException(status_code=401, detail="Missing auth header")
27    token = authorization.replace("Bearer ", "")
28    for uid, t in tokens.items():
29        if t == token:
30            return uid
31    raise HTTPException(status_code=401, detail="Invalid token")
32 
33@app.post("/signup")
34def signup(username: str, password: str):
35    if username in users:
36        raise HTTPException(status_code=400, detail="User exists")
37    users[username] = {"password": password, "id": len(users) + 1}
38    return {"message": "User created"}
39 
40@app.post("/login")
41def login(username: str, password: str):
42    if username not in users or users[username]["password"] != password:
43        raise HTTPException(status_code=401, detail="Invalid credentials")
44    token = hashlib.sha256(f"{username}{time.time()}".encode()).hexdigest()
45    tokens[users[username]["id"]] = token
46    return {"token": token}
47 
48@app.get("/videos/{video_id}")
49def get_video(video_id: int, authorization: Optional[str] = Header(None)):
50    get_current_user(authorization)
51    if video_id not in videos:
52        raise HTTPException(status_code=404, detail="Video not found")
53    return videos[video_id]
54 
55@app.post("/videos")
56def create_video(title: str, duration: int, thumbnail_url: str, authorization: Optional[str] = Header(None)):
57    get_current_user(authorization)
58    global next_video_id
59    video = {
60        "id": next_video_id,
61        "title": title,
62        "duration": duration,
63        "thumbnail_url": thumbnail_url,
64        "view_count": 0
65    }
66    videos[next_video_id] = video
67    recommendations_map[next_video_id] = []
68    next_video_id += 1
69    return video
70 
71@app.get("/videos")
72def list_videos(authorization: Optional[str] = Header(None)):
73    get_current_user(authorization)
74    return [{"id": v["id"], "title": v["title"], "duration": v["duration"], "thumbnail_url": v["thumbnail_url"], "view_count": v["view_count"]} for v in videos.values()]
75 
76@app.post("/videos/{video_id}/watch")
77def watch_video(video_id: int, watch_time: int, authorization: Optional[str] = Header(None)):
78    get_current_user(authorization)
79    if video_id not in videos:
80        raise HTTPException(status_code=404, detail="Video not found")
81    videos[video_id]["view_count"] += 1
82    watch_logs.append({"video_id": video_id, "watch_time": watch_time, "timestamp": time.time()})
83    return {"message": "Watch recorded"}
84 
85@app.get("/videos/{video_id}/recommendations")
86def get_recommendations(video_id: int, authorization: Optional[str] = Header(None)):
87    get_current_user(authorization)
88    if video_id not in videos:
89        raise HTTPException(status_code=404, detail="Video not found")
90    rec_ids = recommendations_map.get(video_id, [])
91    return [videos[rid] for rid in rec_ids if rid in videos]

1fastapi
2uvicorn