Browse the data

1import os
2import uuid
3import json
4from fastapi import FastAPI, HTTPException, Header
5from fastapi.responses import FileResponse
6from pydantic import BaseModel
7from typing import Optional
8 
9app = FastAPI()
10 
11users = {}
12tokens = {}
13students = {}
14recordings = {}
15access_log = {}
16 
17AUDIO_DIR = "audio_lessons"
18os.makedirs(AUDIO_DIR, exist_ok=True)
19 
20class SignupRequest(BaseModel):
21    username: str
22    password: str
23 
24class LoginRequest(BaseModel):
25    username: str
26    password: str
27 
28class StudentCreate(BaseModel):
29    name: str
30 
31class RecordingCreate(BaseModel):
32    student_id: int
33    file_name: str
34 
35def get_current_user(authorization: Optional[str] = Header(None)):
36    if not authorization:
37        raise HTTPException(status_code=401, detail="Missing auth header")
38    token = authorization.replace("Bearer ", "")
39    if token not in tokens:
40        raise HTTPException(status_code=401, detail="Invalid token")
41    return tokens[token]
42 
43@app.post("/signup")
44def signup(req: SignupRequest):
45    if req.username in users:
46        raise HTTPException(status_code=400, detail="User exists")
47    users[req.username] = {"username": req.username, "password": req.password}
48    return {"message": "User created"}
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 = str(uuid.uuid4())
56    tokens[token] = req.username
57    return {"token": token}
58 
59@app.post("/students")
60def create_student(req: StudentCreate, authorization: Optional[str] = Header(None)):
61    get_current_user(authorization)
62    student_id = len(students) + 1
63    students[student_id] = {"id": student_id, "name": req.name}
64    access_log[student_id] = []
65    return students[student_id]
66 
67@app.get("/students/{student_id}")
68def get_student(student_id: int, authorization: Optional[str] = Header(None)):
69    get_current_user(authorization)
70    if student_id not in students:
71        raise HTTPException(status_code=404, detail="Student not found")
72    return students[student_id]
73 
74@app.post("/recordings")
75def create_recording(req: RecordingCreate, authorization: Optional[str] = Header(None)):
76    get_current_user(authorization)
77    if req.student_id not in students:
78        raise HTTPException(status_code=404, detail="Student not found")
79    recording_id = len(recordings) + 1
80    recordings[recording_id] = {"id": recording_id, "student_id": req.student_id, "file_name": req.file_name}
81    return recordings[recording_id]
82 
83@app.get("/recordings/{file_id}")
84def get_recording(file_id: int, authorization: Optional[str] = Header(None)):
85    get_current_user(authorization)
86    if file_id not in recordings:
87        raise HTTPException(status_code=404, detail="Recording not found")
88    recording = recordings[file_id]
89    file_path = os.path.join(AUDIO_DIR, recording["file_name"])
90    if not os.path.exists(file_path):
91        raise HTTPException(status_code=404, detail="File not found on disk")
92    student_id = recording["student_id"]
93    if student_id not in access_log:
94        access_log[student_id] = []
95    access_log[student_id].append(file_id)
96    return FileResponse(file_path, media_type="audio/mpeg")
97 
98@app.get("/students/{student_id}/history")
99def get_student_history(student_id: int, authorization: Optional[str] = Header(None)):
100    get_current_user(authorization)
101    if student_id not in students:
102        raise HTTPException(status_code=404, detail="Student not found")
103    recent_files = access_log.get(student_id, [])
104    result = [recordings[fid] for fid in recent_files if fid in recordings]
105    return result

1fastapi
2uvicorn