Russell Ballestrini
d7416d2bae
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
962 lines
33 KiB
Python
962 lines
33 KiB
Python
# 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"<User(username='{self.username}', email='{self.email}')>"
|
|
|
|
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)
|