import os import json import uuid from datetime import datetime, timedelta from typing import List, Dict, Optional, Any import secrets from fastapi import FastAPI, Depends, HTTPException, status, UploadFile, File, Form from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm from fastapi.staticfiles import StaticFiles from fastapi.responses import FileResponse, RedirectResponse from fastapi.middleware.cors import CORSMiddleware from jose import JWTError, jwt from passlib.context import CryptContext from pydantic import BaseModel, Field # --- Configuration --- # In a real production app, use Hugging Face Space Secrets to set this! JWT_SECRET_KEY = os.environ.get("JWT_SECRET_KEY", secrets.token_hex(32)) ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24 * 7 # 7 days # --- Persistent Data Paths (Hugging Face Spaces use /data for persistent storage) --- DATA_DIR = "data" USERS_DB_FILE = os.path.join(DATA_DIR, "users.json") UPLOAD_DIR = os.path.join(DATA_DIR, "uploads") # Create persistent directories if they don't exist os.makedirs(DATA_DIR, exist_ok=True) os.makedirs(UPLOAD_DIR, exist_ok=True) # --- Security & Hashing --- pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") # --- Pydantic Models (Data Schemas) --- class Token(BaseModel): access_token: str token_type: str class TokenData(BaseModel): username: Optional[str] = None class EpisodeUpdate(BaseModel): """Data required to add an episode to the watch history.""" show_id: str season_number: int episode_number: int class WatchedEpisode(EpisodeUpdate): """Represents a single episode in the user's watch history, including the timestamp.""" watch_timestamp: datetime class UserBase(BaseModel): username: str class UserCreate(UserBase): password: str class UserInDB(UserBase): hashed_password: str profile_picture_url: Optional[str] = None watch_history: List[Dict[str, Any]] = Field(default_factory=list) class UserPublic(UserBase): profile_picture_url: Optional[str] = None watch_history_detailed: Dict[str, Any] = Field(default_factory=dict) # --- Database Helper Functions (using JSON file) --- def load_users() -> Dict[str, Dict]: if not os.path.exists(USERS_DB_FILE): return {} with open(USERS_DB_FILE, "r") as f: try: return json.load(f) except json.JSONDecodeError: return {} def save_users(users_db: Dict[str, Dict]): with open(USERS_DB_FILE, "w") as f: json.dump(users_db, f, indent=4) # --- Password & Token Functions --- def verify_password(plain_password, hashed_password): return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password): return pwd_context.hash(password) def create_access_token(data: dict, expires_delta: Optional[timedelta] = None): to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=15) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, JWT_SECRET_KEY, algorithm=ALGORITHM) return encoded_jwt # --- Dependency to get current user --- async def get_current_user(token: str = Depends(oauth2_scheme)) -> UserInDB: credentials_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) try: payload = jwt.decode(token, JWT_SECRET_KEY, algorithms=[ALGORITHM]) username: str = payload.get("sub") if username is None: raise credentials_exception token_data = TokenData(username=username) except JWTError: raise credentials_exception users_db = load_users() user_data = users_db.get(token_data.username) if user_data is None: raise credentials_exception return UserInDB(**user_data) # --- FastAPI App Initialization --- app = FastAPI(title="Media Auth API") # CORS middleware to allow the frontend to communicate with the API # This is crucial for web apps. app.add_middleware( CORSMiddleware, allow_origins=["*"], # Allows all origins allow_credentials=True, allow_methods=["*"], # Allows all methods allow_headers=["*"], # Allows all headers ) # --- Helper function to structure watch history for the frontend --- def structure_watch_history(history_list: List[Dict]) -> Dict: """Transforms a flat list of watched episodes into a nested dictionary.""" structured = {} for item in history_list: show_id = item.get("show_id") show_title = item.get("show_title", "Unknown Show") season_num = item.get("season_number") episode_num = item.get("episode_number") if show_id not in structured: structured[show_id] = { "show_id": show_id, "title": show_title, "seasons": {} } if season_num not in structured[show_id]["seasons"]: structured[show_id]["seasons"][season_num] = { "season_number": season_num, "episodes": {} } structured[show_id]["seasons"][season_num]["episodes"][episode_num] = True return structured # --- API Endpoints --- @app.post("/token", response_model=Token, tags=["Authentication"]) async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()): """ Standard OAuth2 login. Takes username and password from a form. """ users_db = load_users() user_data = users_db.get(form_data.username) if not user_data or not verify_password(form_data.password, user_data["hashed_password"]): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect username or password", headers={"WWW-Authenticate": "Bearer"}, ) access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) access_token = create_access_token( data={"sub": user_data["username"]}, expires_delta=access_token_expires ) return {"access_token": access_token, "token_type": "bearer"} @app.post("/signup", status_code=status.HTTP_201_CREATED, tags=["Authentication"]) async def signup_user(user: UserCreate): """ Creates a new user account. """ users_db = load_users() if user.username in users_db: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Username already registered", ) hashed_password = get_password_hash(user.password) new_user = UserInDB(username=user.username, hashed_password=hashed_password) users_db[user.username] = new_user.dict() save_users(users_db) return {"message": "User created successfully. Please login."} @app.get("/users/me", response_model=UserPublic, tags=["User"]) async def read_users_me(current_user: UserInDB = Depends(get_current_user)): """ Fetch the profile of the currently authenticated user. """ # The frontend expects a detailed, nested history object. We transform it here. detailed_history = structure_watch_history(current_user.watch_history) # We create a UserPublic object to avoid sending the hashed_password user_public_data = UserPublic( username=current_user.username, profile_picture_url=current_user.profile_picture_url, watch_history_detailed=detailed_history ) return user_public_data @app.post("/users/me/profile-picture", response_model=UserPublic, tags=["User"]) async def upload_profile_picture( file: UploadFile = File(...), current_user: UserInDB = Depends(get_current_user) ): """ Upload or update the user's profile picture. """ # Generate a unique filename to prevent overwrites and add extension file_extension = os.path.splitext(file.filename)[1] unique_filename = f"{uuid.uuid4()}{file_extension}" file_path = os.path.join(UPLOAD_DIR, unique_filename) # Save the file with open(file_path, "wb") as buffer: buffer.write(await file.read()) # Update user data with the URL to the new picture # The URL must match the static file mount path profile_picture_url = f"/uploads/{unique_filename}" users_db = load_users() users_db[current_user.username]["profile_picture_url"] = profile_picture_url save_users(users_db) # Return the updated user profile current_user.profile_picture_url = profile_picture_url detailed_history = structure_watch_history(current_user.watch_history) return UserPublic( username=current_user.username, profile_picture_url=current_user.profile_picture_url, watch_history_detailed=detailed_history ) import status from fastapi import Depends, APIRouter from pydantic import BaseModel from datetime import datetime, timezone from typing import List # Assume these are defined elsewhere in your application # from .dependencies import get_current_user, load_users, save_users # from .models import UserInDB # --- Pydantic Models --- # It's good practice to define your data models clearly. @app.get("/users/me/watch-history", response_model=List[WatchedEpisode]) async def get_watch_history( current_user: UserInDB = Depends(get_current_user) ): """ Retrieves the current user's watch history, sorted by most recently watched. """ users_db = load_users() user_data = users_db.get(current_user.username, {}) # Get the watch history, defaulting to an empty list if it doesn't exist watch_history = user_data.get("watch_history", []) # Optional: Sort the history so the most recently watched items are first # The `key` lambda function handles cases where an old item might not have a timestamp. watch_history.sort( key=lambda item: item.get("watch_timestamp", "1970-01-01T00:00:00Z"), reverse=True ) return watch_history @app.post("/users/me/watch-history", status_code=status.HTTP_200_OK) async def update_watch_history( episode_data: EpisodeUpdate, current_user: UserInDB = Depends(get_current_user) ): """ Adds a new episode with a timestamp to the user's watch history. If the episode already exists, its timestamp is not updated. """ users_db = load_users() user_data = users_db[current_user.username] # Ensure watch_history list exists if "watch_history" not in user_data: user_data["watch_history"] = [] # Create a unique identifier for the episode to prevent duplicates episode_id = f"{episode_data.show_id}_{episode_data.season_number}_{episode_data.episode_number}" # Check if this exact episode is already in the history is_already_watched = any( (f"{item.get('show_id')}_{item.get('season_number')}_{item.get('episode_number')}" == episode_id) for item in user_data["watch_history"] ) if not is_already_watched: # Convert the Pydantic model to a dictionary to add the new field new_episode_record = episode_data.dict() # Add the UTC timestamp in standard ISO format new_episode_record["watch_timestamp"] = datetime.now(timezone.utc).isoformat() user_data["watch_history"].append(new_episode_record) save_users(users_db) return {"message": "Watch history updated."} return {"message": "Episode already in watch history."} # --- Static File Serving --- # This serves the uploaded profile pictures from the persistent /data/uploads directory app.mount("/uploads", StaticFiles(directory=UPLOAD_DIR), name="uploads") # This serves the main login/settings pages from the /static directory # 'html=True' makes it so that visiting "/" serves "index.html" (we'll name login.html as index.html) app.mount("/", StaticFiles(directory="static", html=True), name="static") # Redirect root to login page for convenience @app.get("/", include_in_schema=False) def root(): return RedirectResponse(url="/login.html")