Browse the data

1import os
2import zipfile
3import io
4import shutil
5from fastapi import FastAPI, HTTPException, Header, UploadFile, File, Form
6from fastapi.responses import StreamingResponse, FileResponse
7from pydantic import BaseModel
8import uuid
9import hashlib
10import time
11 
12app = FastAPI()
13 
14users = {}
15tokens = {}
16lessons = {}
17lesson_id_counter = 1
18user_id_counter = 1
19 
20AUDIO_DIR = "/var/lessons"
21 
22class SignupRequest(BaseModel):
23    username: str
24    password: str
25 
26class LoginRequest(BaseModel):
27    username: str
28    password: str
29 
30def generate_token():
31    return str(uuid.uuid4())
32 
33def hash_password(password: str):
34    return hashlib.sha256(password.encode()).hexdigest()
35 
36def get_user_from_token(authorization: str = Header(None)):
37    if not authorization:
38        raise HTTPException(status_code=401, detail="Missing authorization header")
39    token = authorization.replace("Bearer ", "")
40    if token not in tokens:
41        raise HTTPException(status_code=401, detail="Invalid token")
42    return tokens[token]
43 
44@app.post("/signup")
45def signup(req: SignupRequest):
46    global user_id_counter
47    if req.username in users:
48        raise HTTPException(status_code=400, detail="User already exists")
49    user_id = user_id_counter
50    user_id_counter += 1
51    users[req.username] = {
52        "id": user_id,
53        "username": req.username,
54        "password_hash": hash_password(req.password)
55    }
56    return {"id": user_id, "username": req.username}
57 
58@app.post("/login")
59def login(req: LoginRequest):
60    if req.username not in users:
61        raise HTTPException(status_code=401, detail="Invalid credentials")
62    user = users[req.username]
63    if user["password_hash"] != hash_password(req.password):
64        raise HTTPException(status_code=401, detail="Invalid credentials")
65    token = generate_token()
66    tokens[token] = req.username
67    return {"token": token}
68 
69@app.post("/lesson")
70def create_lesson(authorization: str = Header(None)):
71    user = get_user_from_token(authorization)
72    global lesson_id_counter
73    lesson_id = lesson_id_counter
74    lesson_id_counter += 1
75    lesson_dir = os.path.join(AUDIO_DIR, str(lesson_id))
76    os.makedirs(lesson_dir, exist_ok=True)
77    lessons[lesson_id] = {"id": lesson_id, "files": []}
78    return {"id": lesson_id}
79 
80@app.get("/lesson/{lesson_id}")
81def get_lesson(lesson_id: int, authorization: str = Header(None)):
82    user = get_user_from_token(authorization)
83    if lesson_id not in lessons:
84        raise HTTPException(status_code=404, detail="Lesson not found")
85    lesson = lessons[lesson_id]
86    lesson_dir = os.path.join(AUDIO_DIR, str(lesson_id))
87    files = []
88    if os.path.exists(lesson_dir):
89        files = os.listdir(lesson_dir)
90    return {"id": lesson_id, "files": files}
91 
92@app.post("/lesson/{lesson_id}/audio")
93async def upload_audio(lesson_id: int, file: UploadFile = File(...), authorization: str = Header(None)):
94    user = get_user_from_token(authorization)
95    if lesson_id not in lessons:
96        raise HTTPException(status_code=404, detail="Lesson not found")
97    lesson_dir = os.path.join(AUDIO_DIR, str(lesson_id))
98    os.makedirs(lesson_dir, exist_ok=True)
99    file_path = os.path.join(lesson_dir, file.filename)
100    with open(file_path, "wb") as f:
101        content = await file.read()
102        f.write(content)
103    return {"filename": file.filename}
104 
105@app.get("/lesson/{lesson_id}/audio/{file_name}")
106def stream_audio(lesson_id: int, file_name: str, authorization: str = Header(None)):
107    user = get_user_from_token(authorization)
108    if lesson_id not in lessons:
109        raise HTTPException(status_code=404, detail="Lesson not found")
110    file_path = os.path.join(AUDIO_DIR, str(lesson_id), file_name)
111    if not os.path.exists(file_path):
112        raise HTTPException(status_code=404, detail="File not found")
113    return FileResponse(file_path, media_type="audio/mpeg")
114 
115@app.get("/lesson/{lesson_id}/download")
116def download_lesson_zip(lesson_id: int, authorization: str = Header(None)):
117    user = get_user_from_token(authorization)
118    if lesson_id not in lessons:
119        raise HTTPException(status_code=404, detail="Lesson not found")
120    lesson_dir = os.path.join(AUDIO_DIR, str(lesson_id))
121    if not os.path.exists(lesson_dir):
122        raise HTTPException(status_code=404, detail="Lesson directory not found")
123    
124    zip_buffer = io.BytesIO()
125    with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf:
126        for root, dirs, files in os.walk(lesson_dir):
127            for file in files:
128                file_path = os.path.join(root, file)
129                arcname = os.path.relpath(file_path, lesson_dir)
130                zf.write(file_path, arcname)
131    zip_buffer.seek(0)
132    return StreamingResponse(
133        zip_buffer,
134        media_type="application/zip",
135        headers={"Content-Disposition": f"attachment; filename=lesson_{lesson_id}.zip"}
136    )

1fastapi
2uvicorn
3python-multipart