Browse the data

1import os
2import random
3import string
4from fastapi import FastAPI, HTTPException, Header
5from pydantic import BaseModel
6 
7app = FastAPI()
8 
9users = {}
10tokens = {}
11videos = {}
12chunks = {}
13 
14def generate_token():
15    return ''.join(random.choices(string.ascii_letters + string.digits, k=32))
16 
17def get_current_user(authorization: str = Header(None)):
18    if not authorization or not authorization.startswith("Bearer "):
19        raise HTTPException(status_code=401, detail="Invalid token")
20    token = authorization[7:]
21    user_id = tokens.get(token)
22    if user_id is None:
23        raise HTTPException(status_code=401, detail="Invalid token")
24    return user_id
25 
26class SignupRequest(BaseModel):
27    username: str
28    password: str
29 
30class LoginRequest(BaseModel):
31    username: str
32    password: str
33 
34class VideoCreate(BaseModel):
35    title: str
36    description: str = ""
37 
38class ChunkCreate(BaseModel):
39    filename: str
40    filepath: str
41 
42@app.post("/signup")
43def signup(req: SignupRequest):
44    if req.username in users:
45        raise HTTPException(status_code=400, detail="User already exists")
46    user_id = len(users) + 1
47    users[req.username] = {"id": user_id, "password": req.password}
48    return {"id": user_id, "username": req.username}
49 
50@app.post("/login")
51def login(req: LoginRequest):
52    user = users.get(req.username)
53    if not user or user["password"] != req.password:
54        raise HTTPException(status_code=401, detail="Invalid credentials")
55    token = generate_token()
56    tokens[token] = user["id"]
57    return {"token": token}
58 
59@app.get("/video/{video_id}")
60def get_video(video_id: int, authorization: str = Header(None)):
61    user_id = get_current_user(authorization)
62    video = videos.get(video_id)
63    if not video:
64        raise HTTPException(status_code=404, detail="Video not found")
65    return video
66 
67@app.post("/video")
68def create_video(req: VideoCreate, authorization: str = Header(None)):
69    user_id = get_current_user(authorization)
70    video_id = len(videos) + 1
71    videos[video_id] = {"id": video_id, "title": req.title, "description": req.description, "user_id": user_id}
72    chunks[video_id] = []
73    return videos[video_id]
74 
75@app.get("/video/{video_id}/manifest")
76def get_manifest(video_id: int, authorization: str = Header(None)):
77    user_id = get_current_user(authorization)
78    video = videos.get(video_id)
79    if not video:
80        raise HTTPException(status_code=404, detail="Video not found")
81    video_chunks = chunks.get(video_id, [])
82    return {"video_id": video_id, "chunks": [c["filename"] for c in video_chunks]}
83 
84@app.get("/video/{video_id}/chunks/{filename}")
85def get_chunk(video_id: int, filename: str, authorization: str = Header(None)):
86    user_id = get_current_user(authorization)
87    video = videos.get(video_id)
88    if not video:
89        raise HTTPException(status_code=404, detail="Video not found")
90    video_chunks = chunks.get(video_id, [])
91    chunk = next((c for c in video_chunks if c["filename"] == filename), None)
92    if not chunk:
93        raise HTTPException(status_code=404, detail="Chunk not found")
94    file_path = os.path.join("/var/videos", str(video_id), filename)
95    if not os.path.exists(file_path):
96        raise HTTPException(status_code=404, detail="File not found on disk")
97    with open(file_path, "rb") as f:
98        content = f.read()
99    return Response(content=content, media_type="video/mp2t")
100 
101@app.post("/video/{video_id}/chunks")
102def upload_chunk(video_id: int, req: ChunkCreate, authorization: str = Header(None)):
103    user_id = get_current_user(authorization)
104    video = videos.get(video_id)
105    if not video:
106        raise HTTPException(status_code=404, detail="Video not found")
107    chunk_id = len(chunks.get(video_id, [])) + 1
108    chunk_data = {"id": chunk_id, "filename": req.filename, "filepath": req.filepath}
109    if video_id not in chunks:
110        chunks[video_id] = []
111    chunks[video_id].append(chunk_data)
112    return chunk_data

1fastapi
2uvicorn