commit d7416d2bae65472b2e977a3b49d8cc1f8631ead9 Author: Russell Ballestrini Date: Sun Dec 29 15:21:59 2024 -0500 init new file: .gitignore new file: app.py new file: requirements.txt new file: templates/base.html.j2 new file: templates/edit_media.html.j2 new file: templates/home.html.j2 new file: templates/import_user_record.html.j2 new file: templates/list_media.html.j2 new file: templates/login.html.j2 new file: templates/profile.html.j2 new file: templates/upload_media.html.j2 new file: templates/user_media.html.j2 new file: templates/verify.html.j2 new file: templates/view_media.html.j2 new file: templates/view_media_details.html.j2 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..98e6ef6 --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +*.db diff --git a/app.py b/app.py new file mode 100644 index 0000000..dc66acc --- /dev/null +++ b/app.py @@ -0,0 +1,962 @@ +# app.py + +import os +import base64 +import datetime +import random +import string +import bcrypt +import re +import uuid +import hashlib +import smtplib +import mimetypes +import json +import logging # Import the logging module +from email.mime.text import MIMEText + +from pyramid.config import Configurator +from pyramid.view import view_config +from pyramid.response import Response +from pyramid.httpexceptions import HTTPFound, HTTPForbidden +from pyramid.session import SignedCookieSessionFactory +from sqlalchemy import ( + create_engine, + Column, + String, + DateTime, + Boolean, + Integer, + Index, + func, +) +from sqlalchemy.orm import declarative_base +from sqlalchemy.orm import sessionmaker +from waitress import serve + +# Import renderers for templates +from pyramid.renderers import render_to_response + +# Import for events +from pyramid.events import subscriber +from pyramid_jinja2 import IJinja2Environment + +################################################################################ +# Set up logging +################################################################################ + +log = logging.getLogger(__name__) + +################################################################################ +# Helper Functions +################################################################################ + +def slugify(text): + text = text.lower() + text = re.sub(r"\s+", "-", text) + text = re.sub(r"[^\w\-]", "", text) + return text + +def get_gravatar_url(email, size=100): + email = email.strip().lower() + hash_code = hashlib.md5(email.encode("utf-8")).hexdigest() + return f"https://www.gravatar.com/avatar/{hash_code}?s={size}&d=identicon" + +def send_email(to_email, subject, body): + # For testing purposes, print the email content to the console + log.debug("======= Email Sent =======") + log.debug(f"To: {to_email}") + log.debug(f"Subject: {subject}") + log.debug(f"Body:\n{body}") + log.debug("==========================") + + msg = MIMEText(body) + msg["Subject"] = subject + msg["From"] = "noreply@example.com" + msg["To"] = to_email + + try: + s = smtplib.SMTP("localhost", 25) + s.sendmail("noreply@example.com", [to_email], msg.as_string()) + s.quit() + except Exception as e: + log.error(f"Error sending email: {e}") + +def admin_required(view_func): + def wrapper(request): + user = request.user + if not user or not user.is_admin: + return HTTPForbidden("You must be an admin to access this page.") + return view_func(request) + return wrapper + +def get_current_user(request): + """Return the current user (authenticated or guest) from session.""" + user_id = request.session.get("user_id") + s = request.dbsession + + if user_id: + # Try to get the user from the database + user = s.query(User).filter_by(id=user_id).first() + if user: + return user + else: + # User ID in session does not exist in the database; remove it + del request.session["user_id"] + + # No valid user in session; create a guest user + # Generate a unique guest username + while True: + suffix = "".join(random.choices(string.ascii_uppercase + string.digits, k=8)) + guest_username = f"Guest-{suffix}" + existing_user = s.query(User).filter_by(username=guest_username).first() + if not existing_user: + break # Unique username found + + # Generate UUID and short ID + user_uuid = uuid.uuid4() + user_id = str(user_uuid) + short_id = uuid_to_short_id(user_uuid) + + # Create a new guest user + guest_user = User( + id=user_id, + short_id=short_id, + email=None, # Guests don't have an email + username=guest_username, + is_verified=False, + ) + s.add(guest_user) + s.commit() + + # Store the user ID in the session + request.session["user_id"] = guest_user.id + + return guest_user + +def get_mime_type(filename): + # Guess the MIME type based on the file extension + mime_type, _ = mimetypes.guess_type(filename) + if not mime_type: + mime_type = "application/octet-stream" + return mime_type + + +def uuid_to_short_id(u): + """Encode UUID to a URL-safe base64 string without padding.""" + return base64.urlsafe_b64encode(u.bytes).decode("ascii").rstrip("=") + +def short_id_to_uuid(sid): + """Decode the short ID back to UUID, trying different padding lengths.""" + # Try with 0 to 3 padding characters + for padding_length in range(5): + try: + padded = sid + ("=" * padding_length) + bytes_data = base64.urlsafe_b64decode(padded) + if len(bytes_data) == 16: # UUID is 16 bytes + return uuid.UUID(bytes=bytes_data) + except Exception as e: + continue + + # If we get here, none of the padding attempts worked + log.error(f"Failed to convert short_id {sid} to UUID after trying all padding lengths") + return None + +def get_user_db_url(user_id): + """Return the database URL for the user's SQLite database.""" + APP_DIR = os.path.dirname(os.path.abspath(__file__)) + db_file = os.path.join(APP_DIR, f"user_{user_id}.db") + return f"sqlite:///{db_file}" + +def filesizeformat(value): + """Returns the human-readable file size.""" + for unit in ["bytes", "KB", "MB", "GB", "TB"]: + if value < 1024.0: + return f"{value:.2f} {unit}" + value /= 1024.0 + return f"{value:.2f} PB" + +################################################################################ +# Database Setup +################################################################################ + +APP_DIR = os.path.dirname(os.path.abspath(__file__)) +DB_URL = f"sqlite:///{os.path.join(APP_DIR, 'main.db')}" +log.debug(f"Using database URL: {DB_URL}") # For debugging + +Base = declarative_base() + +class User(Base): + __tablename__ = "users" + id = Column(String, primary_key=True) # UUID + short_id = Column(String, unique=True, nullable=False) + email = Column(String, unique=True, nullable=True) # Guests don't have an email. + username = Column(String, unique=True, nullable=False) # user-chosen handle + code_hash = Column(String, nullable=True) # bcrypt hash of code + code_expires = Column(DateTime, nullable=True) # time limit for code + is_verified = Column(Boolean, default=False) + enable_gravatar = Column(Boolean, default=False) # Gravatar support + is_admin = Column(Boolean, default=False) # Admin flag + + # Indexes for faster lookup + __table_args__ = ( + Index("ix_users_id", "id"), + Index("ix_users_short_id", "short_id"), + ) + + def __repr__(self): + return f"" + +class Media(Base): + __tablename__ = "media" + id = Column(String, primary_key=True) # UUID + short_id = Column(String, unique=True, nullable=False) + user_id = Column(String, nullable=False) # Owner's UUID + filename = Column(String, nullable=False) + title = Column(String, nullable=True) # Optional title + media_type = Column(String, nullable=False) # 'image', 'audio', 'video' + media_b64 = Column(String, nullable=False) + upload_date = Column(DateTime, default=datetime.datetime.utcnow) + is_public = Column(Boolean, default=True) + size = Column(Integer, nullable=False) # Size in bytes + + # Indexes for faster lookup + __table_args__ = ( + Index("ix_media_id", "id"), + Index("ix_media_short_id", "short_id"), + ) + +################################################################################ +# Main Database (for Users) +################################################################################ + +engine = create_engine(DB_URL, echo=False) +Session = sessionmaker(bind=engine) +Base.metadata.create_all(engine) + +################################################################################ +# Jinja2 Environment and Custom Filters +################################################################################ + +@subscriber(IJinja2Environment) +def add_jinja2_filters(event): + env = event.environment + env.filters["filesizeformat"] = filesizeformat + +################################################################################ +# Routes +################################################################################ + +@view_config(route_name="home", renderer="home.html.j2") +def home_view(request): + return { + "request": request, + } + +################################################################################ +# Authentication Views +################################################################################ + +@view_config(route_name="login", request_method="GET", renderer="login.html.j2") +def login_get_view(request): + return {"request": request} + +@view_config(route_name="login", request_method="POST") +def login_post_view(request): + email = request.POST.get("email", "").strip().lower() + if not email: + return Response("Email required.", status=400) + + session = request.dbsession + user = session.query(User).filter_by(email=email).first() + + if not user: + # Generate UUID and short ID + user_uuid = uuid.uuid4() + user_id = str(user_uuid) + short_id = uuid_to_short_id(user_uuid) + # Create new user + user = User( + id=user_id, + short_id=short_id, + email=email, + username=email.split("@")[0], + is_verified=False, + ) + session.add(user) + session.commit() + + # Generate 6-digit code + code_str = f"{random.randint(0,999999):06d}" + code_hash = bcrypt.hashpw(code_str.encode("utf-8"), bcrypt.gensalt()).decode( + "utf-8" + ) + user.code_hash = code_hash + user.code_expires = datetime.datetime.now() + datetime.timedelta(minutes=15) + user.is_verified = False + session.commit() + + # Send code via email + email_body = f"Your verification code is: {code_str}" + send_email(user.email, "Your Verification Code", email_body) + + # Store the email in the session for verification + request.session["login_email"] = email + + return HTTPFound(location=request.route_url("verify")) + +@view_config(route_name="verify", request_method="GET", renderer="verify.html.j2") +def verify_get_view(request): + return {"request": request} + +@view_config(route_name="verify", request_method="POST") +def verify_post_view(request): + code_entered = request.POST.get("code", "").strip() + if not code_entered or len(code_entered) != 6: + return Response("Invalid code.", status=400) + + email = request.session.get("login_email") + if not email: + return Response( + "No email found in session. Please start the login process again.", + status=400, + ) + + s = request.dbsession + user = ( + s.query(User) + .filter( + User.email == email, + User.code_expires > datetime.datetime.now(), + User.code_hash != None, + ) + .first() + ) + + if not user: + return Response("Code not found or expired.", status=400) + + if not bcrypt.checkpw(code_entered.encode("utf-8"), user.code_hash.encode("utf-8")): + return Response("Invalid code.", status=400) + + user.is_verified = True + user.code_hash = None + user.code_expires = None + s.commit() + + # Remove the email from the session + del request.session["login_email"] + + request.session["user_id"] = user.id + return HTTPFound(location=request.route_url("home")) + +@view_config(route_name="logout") +def logout_view(request): + request.session.invalidate() + return HTTPFound(location=request.route_url("home")) + +################################################################################ +# Profile and Media Upload +################################################################################ + +@view_config(route_name="profile", request_method="GET", renderer="profile.html.j2") +def profile_get_view(request): + user = request.user + if not user: + return Response("You must be logged in to access your profile.", status=403) + gravatar_url = get_gravatar_url(user.email) if user.enable_gravatar else "" + + # Calculate stats + total_uploads = 0 + total_size = 0 # in bytes + + user_db_url = get_user_db_url(user.id) + db_file = os.path.join(APP_DIR, f"user_{user.id}.db") + if os.path.exists(db_file): + user_engine = create_engine(user_db_url) + UserSession = sessionmaker(bind=user_engine) + Base.metadata.create_all(user_engine) + user_dbsession = UserSession() + + media_items = user_dbsession.query(Media).all() + total_uploads = len(media_items) + total_size = sum(media.size for media in media_items) + + return { + "request": request, + "user": user, + "gravatar_url": gravatar_url, + "total_uploads": total_uploads, + "total_size": total_size, + } + +@view_config(route_name="profile", request_method="POST") +def profile_post_view(request): + if not request.user: + return Response("You must be logged in to update your profile.", status=403) + if not request.user.is_verified: + return Response( + "This account is in guest mode, log in to update your profile.", status=403 + ) + s = request.dbsession + enable_gravatar = request.POST.get("enable_gravatar") == "on" + request.user.enable_gravatar = enable_gravatar + + # Handle username update if provided + new_username = request.POST.get("new_username", "").strip() + if new_username: + # Check if the new username is already taken + existing = s.query(User).filter(User.username == new_username).first() + if existing and existing.id != request.user.id: + return Response("Username is already in use.", status=400) + request.user.username = new_username + + s.commit() + return HTTPFound(location=request.route_url("profile")) + +@view_config(route_name="download_database") +def download_database_view(request): + user = request.user + if not user or not user.is_verified: + return Response("You must be logged in to download your database.", status=403) + db_file = os.path.join(APP_DIR, f"user_{user.id}.db") + if not os.path.exists(db_file): + return Response("Database file not found.", status=404) + with open(db_file, "rb") as f: + data = f.read() + response = Response(body=data, content_type="application/octet-stream") + response.headers["Content-Disposition"] = f'attachment; filename="user_{user.id}.db"' + return response + +################################################################################ +# User Record Export (Users) and Import (Admins) +################################################################################ + +@view_config(route_name="export_user_record") +def export_user_record_view(request): + user = request.user + if not user or not user.is_verified: + return Response("You must be logged in to export your user record.", status=403) + + # Create a JSON representation of the user record + user_record = { + "id": user.id, + "short_id": user.short_id, + "email": user.email, + "username": user.username, + "enable_gravatar": user.enable_gravatar, + "is_admin": user.is_admin, + } + + # Convert to JSON string + user_json = json.dumps(user_record).encode("utf-8") + + # Create response + response = Response(body=user_json, content_type="application/json") + response.headers[ + "Content-Disposition" + ] = 'attachment; filename="user_record.json"' + return response + +@view_config(route_name="import_user_record", request_method="GET", renderer="import_user_record.html.j2") +@admin_required +def import_user_record_get_view(request): + return {"request": request} + +@view_config(route_name="import_user_record", request_method="POST") +@admin_required +def import_user_record_post_view(request): + # Admin-only import process + + # Get the uploaded user record file + user_record_file = request.POST.get("user_record_file") + if user_record_file is None or not getattr(user_record_file, "filename", "").strip(): + return Response("No user record file uploaded.", status=400) + + # Read and parse the JSON data + try: + user_record_data = user_record_file.file.read() + user_record = json.loads(user_record_data) + except Exception as e: + return Response(f"Invalid user record file: {e}", status=400) + + s = request.dbsession + + # Check if user already exists + existing_user = s.query(User).filter_by(id=user_record["id"]).first() + if existing_user: + return Response("User already exists in the database.", status=400) + + # Create new user with provided data + user = User( + id=user_record["id"], + short_id=user_record["short_id"], + email=user_record["email"], + username=user_record["username"], + enable_gravatar=user_record.get("enable_gravatar", False), + is_admin=user_record.get("is_admin", False), + is_verified=True, # Assume verified + ) + s.add(user) + s.commit() + + request.session.flash(f"User {user.username} imported successfully.") + return HTTPFound(location=request.route_url("home")) + +################################################################################ +# Media Upload, Listing, and Management +################################################################################ + +@view_config( + route_name="upload_media", request_method="GET", renderer="upload_media.html.j2" +) +def upload_media_get_view(request): + user = request.user + if not user or not user.is_verified: + return Response("You must be logged in to upload media.", status=403) + return {"request": request} + +@view_config(route_name="upload_media", request_method="POST") +def upload_media_post_view(request): + user = request.user + if not user or not user.is_verified: + return Response("You must be logged in to upload media.", status=403) + + media_file = request.POST.get("media_file") + if media_file is None or not getattr(media_file, "filename", "").strip(): + return Response("No file uploaded.", status=400) + + raw_bytes = media_file.file.read() + max_size = 30 * 1024 * 1024 # 30 MB + if len(raw_bytes) > max_size: + return Response("File size exceeds the 30MB limit.", status=400) + + file_size = len(raw_bytes) # Store the size in bytes + + # Determine media type based on MIME type + filename = media_file.filename + mime_type = get_mime_type(filename) + if not mime_type.startswith(("image/", "audio/", "video/")): + return Response("Unsupported media type.", status=400) + + media_type = mime_type.split("/")[0] + + # Get title from form + title = request.POST.get("title", "").strip() + + # Encode content to base64 + encoded_str = base64.b64encode(raw_bytes).decode("utf-8") + + # Handle user's database + user_db_url = get_user_db_url(user.id) + user_engine = create_engine(user_db_url) + UserSession = sessionmaker(bind=user_engine) + Base.metadata.create_all(user_engine) + user_dbsession = UserSession() + + # Generate UUID and short ID for the media + media_uuid = uuid.uuid4() + media_id = str(media_uuid) + media_short_id = uuid_to_short_id(media_uuid) + + is_public = request.POST.get("is_public") == "on" + + media = Media( + id=media_id, + short_id=media_short_id, + user_id=user.id, + filename=filename, + title=title, + media_type=media_type, + media_b64=encoded_str, + is_public=is_public, + size=file_size, + ) + user_dbsession.add(media) + user_dbsession.commit() + + return HTTPFound( + location=request.route_url( + "view_media_details", + user_short_id=user.short_id, + media_short_id=media.short_id, + ) + ) + +@view_config(route_name="list_media", renderer="list_media.html.j2") +def list_media_view(request): + # Aggregate public media from all users + s = request.dbsession + users = s.query(User).filter(User.is_verified == True).all() + media_list = [] + for user in users: + user_db_url = get_user_db_url(user.id) + db_file = os.path.join(APP_DIR, f"user_{user.id}.db") + if not os.path.exists(db_file): + continue + user_engine = create_engine(user_db_url) + UserSession = sessionmaker(bind=user_engine) + Base.metadata.create_all(user_engine) + user_dbsession = UserSession() + + media_items = user_dbsession.query(Media).filter(Media.is_public == True).all() + for media in media_items: + media_list.append( + { + "media": media, + "username": user.username, + "user_short_id": user.short_id, + } + ) + + # Sort media by upload date (recent first) + media_list.sort(key=lambda x: x["media"].upload_date, reverse=True) + + return { + "request": request, + "media_list": media_list, + } + +@view_config(route_name="user_media", renderer="user_media.html.j2") +def user_media_view(request): + # View uploads by a particular user + user_short_id = request.matchdict.get("user_short_id") + log.debug(f"Looking up user with short_id: {user_short_id}") # Debug + + try: + # First convert short_id to UUID + user_uuid = short_id_to_uuid(user_short_id) + if not user_uuid: + log.error(f"Could not convert short_id {user_short_id} to UUID") + return Response("User not found.", status=404) + + s = request.dbsession + user = s.query(User).filter_by(id=str(user_uuid)).first() + log.debug(f"User found: {user}") # Debug + + if not user: + return Response("User not found.", status=404) + + viewer = request.user + is_owner = viewer and viewer.id == user.id + + user_db_url = get_user_db_url(user.id) + db_file = os.path.join(APP_DIR, f"user_{user.id}.db") + if not os.path.exists(db_file): + return Response("User has no uploads.", status=404) + user_engine = create_engine(user_db_url) + UserSession = sessionmaker(bind=user_engine) + Base.metadata.create_all(user_engine) + user_dbsession = UserSession() + + if is_owner: + # Show all media (public and private) + media_items = user_dbsession.query(Media).all() + else: + # Show only public media + media_items = user_dbsession.query(Media).filter(Media.is_public == True).all() + + # Sort media by upload date (recent first) + media_items.sort(key=lambda media: media.upload_date, reverse=True) + + return { + "request": request, + "media_items": media_items, + "user": user, + "is_owner": is_owner, + } + + except Exception as e: + log.exception(f"Error processing user_short_id {user_short_id}") + return Response("Error processing request.", status=500) + +@view_config(route_name="view_media_details", renderer="view_media_details.html.j2") +def view_media_details_view(request): + media_short_id = request.matchdict.get("media_short_id") + user_short_id = request.matchdict.get("user_short_id") + if not user_short_id or not media_short_id: + return Response("Invalid request.", status=400) + + s = request.dbsession + user = s.query(User).filter_by(short_id=user_short_id).first() + if not user: + return Response("User not found.", status=404) + user_id = user.id + + user_db_url = get_user_db_url(user_id) + db_file = os.path.join(APP_DIR, f"user_{user_id}.db") + if not os.path.exists(db_file): + return Response("User database not found.", status=404) + user_engine = create_engine(user_db_url) + UserSession = sessionmaker(bind=user_engine) + user_dbsession = UserSession() + + # Lookup media by short_id + media = user_dbsession.query(Media).filter_by(short_id=media_short_id).first() + if not media: + return Response("Media not found.", status=404) + + viewer = request.user + is_owner = viewer and viewer.id == user_id + + if not media.is_public and not is_owner: + return Response("Media not available.", status=403) + + return { + "request": request, + "media": media, + "username": user.username, + "user_short_id": user.short_id, + "is_owner": is_owner, + } + +@view_config(route_name="delete_media", request_method="POST") +def delete_media_view(request): + media_short_id = request.matchdict.get("media_short_id") + user_short_id = request.matchdict.get("user_short_id") + + viewer = request.user + if not viewer or not viewer.is_verified: + return Response("You must be logged in.", status=403) + + if viewer.short_id != user_short_id: + return Response("You are not authorized to delete this media.", status=403) + + user_db_url = get_user_db_url(viewer.id) + user_engine = create_engine(user_db_url) + UserSession = sessionmaker(bind=user_engine) + user_dbsession = UserSession() + + media = user_dbsession.query(Media).filter_by(short_id=media_short_id).first() + if not media: + return Response("Media not found.", status=404) + + # Delete the media + user_dbsession.delete(media) + user_dbsession.commit() + + return HTTPFound( + location=request.route_url("user_media", user_short_id=viewer.short_id) + ) + +@view_config(route_name="edit_media", request_method="GET", renderer="edit_media.html.j2") +def edit_media_get_view(request): + media_short_id = request.matchdict.get("media_short_id") + user_short_id = request.matchdict.get("user_short_id") + + viewer = request.user + if not viewer or not viewer.is_verified: + return Response("You must be logged in.", status=403) + + if viewer.short_id != user_short_id: + return Response("You are not authorized to edit this media.", status=403) + + user_db_url = get_user_db_url(viewer.id) + user_engine = create_engine(user_db_url) + UserSession = sessionmaker(bind=user_engine) + user_dbsession = UserSession() + + media = user_dbsession.query(Media).filter_by(short_id=media_short_id).first() + if not media: + return Response("Media not found.", status=404) + + return { + "request": request, + "media": media, + } + +@view_config(route_name="edit_media", request_method="POST") +def edit_media_post_view(request): + media_short_id = request.matchdict.get("media_short_id") + user_short_id = request.matchdict.get("user_short_id") + + viewer = request.user + if not viewer or not viewer.is_verified: + return Response("You must be logged in.", status=403) + + if viewer.short_id != user_short_id: + return Response("You are not authorized to edit this media.", status=403) + + user_db_url = get_user_db_url(viewer.id) + user_engine = create_engine(user_db_url) + UserSession = sessionmaker(bind=user_engine) + user_dbsession = UserSession() + + media = user_dbsession.query(Media).filter_by(short_id=media_short_id).first() + if not media: + return Response("Media not found.", status=404) + + # Handle title update + new_title = request.POST.get("title", "").strip() + media.title = new_title + + # Handle media file update + new_media_file = request.POST.get("media_file") + if new_media_file and getattr(new_media_file, "filename", "").strip(): + raw_bytes = new_media_file.file.read() + max_size = 30 * 1024 * 1024 # 30 MB + if len(raw_bytes) > max_size: + return Response("File size exceeds the 30MB limit.", status=400) + file_size = len(raw_bytes) + filename = new_media_file.filename + mime_type = get_mime_type(filename) + if not mime_type.startswith(("image/", "audio/", "video/")): + return Response("Unsupported media type.", status=400) + media_type = mime_type.split("/")[0] + encoded_str = base64.b64encode(raw_bytes).decode("utf-8") + + # Update media fields + media.filename = filename + media.media_type = media_type + media.media_b64 = encoded_str + media.size = file_size + + # Handle public/private update + is_public = request.POST.get("is_public") == "on" + media.is_public = is_public + + user_dbsession.commit() + + return HTTPFound( + location=request.route_url( + "view_media_details", + user_short_id=viewer.short_id, + media_short_id=media.short_id, + ) + ) + +################################################################################ +# Media Viewing and Downloading +################################################################################ + +@view_config(route_name="view_media") +def view_media_view(request): + media_short_id = request.matchdict.get("media_short_id") + user_short_id = request.matchdict.get("user_short_id") + if not user_short_id or not media_short_id: + return Response("Invalid request.", status=400) + + s = request.dbsession + user = s.query(User).filter_by(short_id=user_short_id).first() + if not user: + return Response("User not found.", status=404) + user_id = user.id + + user_db_url = get_user_db_url(user_id) + db_file = os.path.join(APP_DIR, f"user_{user_id}.db") + if not os.path.exists(db_file): + return Response("User database not found.", status=404) + user_engine = create_engine(user_db_url) + UserSession = sessionmaker(bind=user_engine) + user_dbsession = UserSession() + + # Lookup media by short_id + media = user_dbsession.query(Media).filter_by(short_id=media_short_id).first() + if not media: + return Response("Media not found.", status=404) + + viewer = request.user + is_owner = viewer and viewer.id == user.id + + if not media.is_public and not is_owner: + return Response("Media not available.", status=403) + + # Decode base64 content + media_data = base64.b64decode(media.media_b64) + mime_type = get_mime_type(media.filename) + + # Prepare filename for download + if media.title: + # Use title as filename, slugify it + file_extension = os.path.splitext(media.filename)[1] + download_filename = f"{slugify(media.title)}{file_extension}" + else: + # Use original filename + download_filename = media.filename + + # Serve the media content with appropriate headers + response = Response(body=media_data, content_type=mime_type) + response.headers.update( + { + "Access-Control-Allow-Origin": "*", + "Content-Disposition": f'inline; filename="{download_filename}"', + } + ) + return response + +################################################################################ +# Main +################################################################################ + +def main(global_config=None, **settings): + from pyramid.decorator import reify + + # Configure logging + logging.basicConfig(level=logging.DEBUG) + + # Set up the session factory with the specified settings + session_factory = SignedCookieSessionFactory( + secret="it-is-a-secret-you-must-change", + hashalg="sha512", + timeout=31104000, # Approx. one year in seconds + max_age=31104000, # Set Max-Age attribute on cookie + reissue_time=15552000, # Approx. six months + samesite=None, # Allows cross-site requests if needed + httponly=True, # Helps mitigate XSS attacks + secure=False, # Set to True if using HTTPS + ) + + if not settings: + settings = {} + settings["sqlalchemy.url"] = DB_URL + + config = Configurator(settings=settings, session_factory=session_factory) + config.include("pyramid_jinja2") + + # Add .html.j2 extension for Jinja2 templates + # Set up Jinja2 template search path + config.add_jinja2_renderer(".j2") + config.add_jinja2_search_path("templates", name=".j2") + + # The Jinja2 filters are added via the event subscriber above + + # Add user to all requests. + config.add_request_method(callable=get_current_user, name="user", reify=True) + + def dbsession(request): + return Session() + + config.add_request_method(dbsession, "dbsession", reify=True) + + # Routes + config.add_route("home", "/") + + # Auth + config.add_route("login", "/auth/login") + config.add_route("verify", "/auth/verify") + config.add_route("logout", "/auth/logout") + config.add_route("profile", "/auth/profile") + + # Export and Import User Record (Admin only for import) + config.add_route("export_user_record", "/auth/export_user_record") + config.add_route("import_user_record", "/admin/import_user_record") + + # Media - Reordered routes with most specific first + config.add_route("upload_media", "/media/upload") + config.add_route("list_media", "/media/list") + config.add_route("user_media", "/media/user/{user_short_id}") # Moved earlier + config.add_route("view_media_details", "/media/{user_short_id}/{media_short_id}/details") + config.add_route("edit_media", "/media/{user_short_id}/{media_short_id}/edit") + config.add_route("delete_media", "/media/{user_short_id}/{media_short_id}/delete") + config.add_route("view_media", "/media/{user_short_id}/{media_short_id}") + + # Database Download + config.add_route("download_database", "/auth/download_db") + + config.scan() + return config.make_wsgi_app() + +if __name__ == "__main__": + app = main() + log.info("Serving on http://localhost:6544") + serve(app, host="0.0.0.0", port=6544) diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..43e487a --- /dev/null +++ b/requirements.txt @@ -0,0 +1,10 @@ +plaster_pastedeploy +pyramid +pyramid_jinja2 +pyramid_debugtoolbar +waitress +pyramid_retry + +bcrypt +sqlalchemy +werkzeug diff --git a/templates/base.html.j2 b/templates/base.html.j2 new file mode 100644 index 0000000..24ae59c --- /dev/null +++ b/templates/base.html.j2 @@ -0,0 +1,164 @@ + + + + + {% block title %}Media Hosting App{% endblock %} + + + + + + + + + + + + + + {% if request.session.peek_flash() %} + {% for message in request.session.pop_flash() %} +
{{ message }}
+ {% endfor %} + {% endif %} + +
+ {% block content %} + {% endblock %} +
+ + + + + diff --git a/templates/edit_media.html.j2 b/templates/edit_media.html.j2 new file mode 100644 index 0000000..8d9f056 --- /dev/null +++ b/templates/edit_media.html.j2 @@ -0,0 +1,35 @@ +{% extends 'base.html.j2' %} + +{% block title %}Edit Media{% endblock %} + +{% block content %} + +

Edit Media

+
+ +

+ + +

+ +

+ + +
+ +

+ + View | + Download | + Back to Media Details +

+ +
+ +
+ +
+{% endblock %} diff --git a/templates/home.html.j2 b/templates/home.html.j2 new file mode 100644 index 0000000..21457e6 --- /dev/null +++ b/templates/home.html.j2 @@ -0,0 +1,15 @@ +{% extends 'base.html.j2' %} + +{% block title %}Home{% endblock %} + +{% block content %} +

Welcome to the Media Hosting App

+

+ {% if request.user and request.user.is_verified %} + Hello, {{ request.user.username }}! + {% else %} + Welcome, Guest! + {% endif %} +

+

This application allows authenticated users to upload images, audio, and short videos. Guests can browse and download publicly available media.

+{% endblock %} diff --git a/templates/import_user_record.html.j2 b/templates/import_user_record.html.j2 new file mode 100644 index 0000000..93e96ca --- /dev/null +++ b/templates/import_user_record.html.j2 @@ -0,0 +1,12 @@ +{% extends 'base.html.j2' %} + +{% block title %}Import User Record{% endblock %} + +{% block content %} +

Import User Record

+
+ +

+ +
+{% endblock %} diff --git a/templates/list_media.html.j2 b/templates/list_media.html.j2 new file mode 100644 index 0000000..1c63f01 --- /dev/null +++ b/templates/list_media.html.j2 @@ -0,0 +1,30 @@ +{% extends 'base.html.j2' %} + +{% block title %}Media List{% endblock %} + +{% block content %} +

Media List

+ +{% if media_list %} + +{% else %} +

No media found.

+{% endif %} + +{% endblock %} diff --git a/templates/login.html.j2 b/templates/login.html.j2 new file mode 100644 index 0000000..085aa1a --- /dev/null +++ b/templates/login.html.j2 @@ -0,0 +1,13 @@ +{% extends 'base.html.j2' %} + +{% block title %}Login{% endblock %} + +{% block content %} +

Login

+

Enter your email to receive a 6-digit verification code.

+
+ + + +
+{% endblock %} diff --git a/templates/profile.html.j2 b/templates/profile.html.j2 new file mode 100644 index 0000000..e356e8b --- /dev/null +++ b/templates/profile.html.j2 @@ -0,0 +1,37 @@ +{% extends 'base.html.j2' %} + +{% block title %}Profile{% endblock %} + +{% block content %} +

Your Profile

+{% if gravatar_url %} + Gravatar Image +{% endif %} +

Username: {{ user.username or "Not set" }}

+ +
+

+ +

+ + +
+ +

Your Stats

+ + +

Your Media

+

View Your Uploads

+

Upload New Media

+

Download Your Database

+ +{% endblock %} diff --git a/templates/upload_media.html.j2 b/templates/upload_media.html.j2 new file mode 100644 index 0000000..ad8731c --- /dev/null +++ b/templates/upload_media.html.j2 @@ -0,0 +1,22 @@ +{% extends 'base.html.j2' %} + +{% block title %}Upload Media{% endblock %} + +{% block content %} +

Upload Media

+
+ +

+ + +

+ + +

+ + +
+{% endblock %} diff --git a/templates/user_media.html.j2 b/templates/user_media.html.j2 new file mode 100644 index 0000000..6e51146 --- /dev/null +++ b/templates/user_media.html.j2 @@ -0,0 +1,33 @@ +{% extends 'base.html.j2' %} + +{% block title %}{{ user.username }}'s Media{% endblock %} + +{% block content %} +

{{ user.username }}'s Media

+ +{% if media_items %} + +{% else %} +

No media found.

+{% endif %} + +{% endblock %} diff --git a/templates/verify.html.j2 b/templates/verify.html.j2 new file mode 100644 index 0000000..90c143c --- /dev/null +++ b/templates/verify.html.j2 @@ -0,0 +1,13 @@ +{% extends 'base.html.j2' %} + +{% block title %}Verify Account{% endblock %} + +{% block content %} +

Verify Your Account

+

A 6-digit code has been sent to your email. Please enter it below to verify your account.

+
+ + + +
+{% endblock %} diff --git a/templates/view_media.html.j2 b/templates/view_media.html.j2 new file mode 100644 index 0000000..80be403 --- /dev/null +++ b/templates/view_media.html.j2 @@ -0,0 +1,38 @@ +{% extends 'base.html.j2' %} + +{% block title %}View Media{% endblock %} + +{% block content %} +

{{ media.title if media.title else media.filename }}

+

Uploaded by {{ username }} on {{ media.upload_date.strftime('%Y-%m-%d %H:%M:%S') }}

+

Size: {{ media.size | filesizeformat }}

+ + +{% if media.media_type == 'image' %} + {{ media.filename }} +{% elif media.media_type == 'audio' %} + +{% elif media.media_type == 'video' %} + +{% endif %} + + +

+ View | + Download +

+ +{% if is_owner %} + +

+ Edit +

+{% endif %} + +{% endblock %} diff --git a/templates/view_media_details.html.j2 b/templates/view_media_details.html.j2 new file mode 100644 index 0000000..224653e --- /dev/null +++ b/templates/view_media_details.html.j2 @@ -0,0 +1,28 @@ +{% extends 'base.html.j2' %} + +{% block title %}Media Details{% endblock %} + +{% block content %} +

Media Details

+ +

Title: {{ media.title if media.title else 'Untitled' }}

+

Filename: {{ media.filename }}

+

Uploaded by: {{ username }}

+

Upload Date: {{ media.upload_date.strftime('%Y-%m-%d %H:%M:%S') }}

+

Status: {{ 'Public' if media.is_public else 'Private' }}

+

Size: {{ media.size | filesizeformat }}

+ + +

+ View | + Download +

+ +{% if is_owner %} + +

+ Edit +

+{% endif %} + +{% endblock %}