############################################################################### # app.py - Enhanced PyraFiles Application with Scheduled Actions and More ############################################################################### 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 unicodedata import threading import time import signal import sys 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, HTTPNotFound from pyramid.session import SignedCookieSessionFactory from sqlalchemy import ( create_engine, Column, String, DateTime, Boolean, Integer, Index, ForeignKey, Text, or_, inspect, ) from sqlalchemy.orm import ( declarative_base, sessionmaker, scoped_session, relationship, load_only, ) from sqlalchemy.pool import StaticPool from waitress import serve from pyramid.renderers import render_to_response from pyramid.events import subscriber from pyramid_jinja2 import IJinja2Environment import transaction # Import transaction management from zope.sqlalchemy import register # Import zope.sqlalchemy import jwt # Import PyJWT library from jwt import PyJWTError ################################################################################ # Set up logging ################################################################################ logging.basicConfig(level=logging.INFO) log = logging.getLogger(__name__) ################################################################################ # Environment Variables and Defaults ################################################################################ # Application directory APP_DIR = os.path.dirname(os.path.abspath(__file__)) # Data directory DATA_DIR = os.path.join(APP_DIR, "data") # Ensure DATA_DIR exists if not os.path.exists(DATA_DIR): os.makedirs(DATA_DIR) # Paths to the secret files APP_SECRET_FILE = os.path.join(DATA_DIR, "pyrafiles_secret.txt") JWT_SECRET_FILE = os.path.join(DATA_DIR, "jwt_secret.txt") ALLOWED_MIME_PREFIXES = ("image/", "audio/", "video/", "text/", "application/") def get_or_create_secret(env_var_name, secret_file_path): """ Retrieves the secret from an environment variable, or loads it from the specified file. If neither is available, generates a new secret, saves it to the file, and returns it. """ # Check environment variable secret = os.environ.get(env_var_name, "") if secret: log.info(f"Using {env_var_name} from environment variable.") return secret # Check if the secret file exists if os.path.exists(secret_file_path): with open(secret_file_path, "r") as f: secret = f.read().strip() if secret: log.info(f"Loaded {env_var_name} from {secret_file_path}") return secret else: log.warning(f"{secret_file_path} is empty. Generating new secret.") else: log.info(f"{secret_file_path} does not exist. Generating new secret.") # Generate a new secret secret = "".join(random.choices(string.ascii_letters + string.digits, k=64)) # Save the secret to the file with open(secret_file_path, "w") as f: f.write(secret) log.info(f"Generated and saved new {env_var_name} to {secret_file_path}") return secret # Retrieve or generate the secrets app_secret = get_or_create_secret("PYRAFILES_SECRET", APP_SECRET_FILE) JWT_SECRET = get_or_create_secret("PYRAFILES_JWT_SECRET", JWT_SECRET_FILE) JWT_ALGORITHM = "HS256" # Database URL can be overridden by environment variable default_main_db_url = f"sqlite:///{os.path.join(DATA_DIR, 'main.db')}" DB_URL = os.environ.get("PYRAFILES_DB_URL", default_main_db_url) # Host and port for the application HOST = os.environ.get("PYRAFILES_HOST", "0.0.0.0") PORT = int(os.environ.get("PYRAFILES_PORT", "6544")) # SMTP host/port smtp_host = os.environ.get("PYRAFILES_SMTP_HOST", "localhost") smtp_port = int(os.environ.get("PYRAFILES_SMTP_PORT", "25")) ################################################################################ # 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, from_email=None): if from_email is None: from_email = os.environ.get("PYRAFILES_FROM_EMAIL", "master@master.unturf.com") msg = MIMEText(body) msg["Subject"] = subject msg["From"] = from_email msg["To"] = to_email try: s = smtplib.SMTP(smtp_host, smtp_port) s.sendmail(from_email, [to_email], msg.as_string()) s.quit() except Exception as e: log = logging.getLogger(__name__) log.info("======= Email Sent =======") log.info(f"To: {to_email}") log.info(f"Subject: {subject}") log.info(f"Body:\n{body}") log.info("==========================") log.error(f"Error sending email: {e}") 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 is_ascii_file(mime_type): """Check if the file is ASCII-based (text files).""" ascii_types = [ "text/", "application/json", "application/xml", "application/javascript", "application/x-yaml", "application/yaml", ] return any(mime_type.startswith(t) for t in ascii_types) 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.""" for padding_length in range(6): 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: continue log.error(f"Failed to convert short_id {sid} to UUID after trying all paddings") return None def get_namespace_db_url(namespace_id): """Return the database URL for the namespace's SQLite database.""" db_file = os.path.join(DATA_DIR, f"namespace_{namespace_id}.db") return f"sqlite:///{db_file}" def sanitize_filename_for_http_header(filename): """ Ensure that the filename is safe for Waitress (Latin-1 headers). Converts to ASCII, replacing or removing characters that won't encode. """ normalized = unicodedata.normalize("NFKD", filename) ascii_bytes = normalized.encode("ascii", "ignore") # drop non-ASCII safe = ascii_bytes.decode("ascii") # Replace any remaining bad chars with underscores # e.g. keep alphanumerics, dots, underscores, hyphens, etc. safe = re.sub(r"[^A-Za-z0-9._-]+", "_", safe) return safe or "download" 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" def generate_jwt_token(agent): """Generate a JWT for the given agent without an expiration time.""" payload = { "agent_id": agent.id, "agent_name": agent.name, "namespace_id": agent.namespace_id, "namespace_short_id": agent.namespace.short_id, "role": agent.role, "token_version": agent.token_version, "iat": datetime.datetime.utcnow(), } token = jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM) return token def verify_jwt_token(token): """Verify the JWT and return the payload if valid.""" try: payload = jwt.decode( token, JWT_SECRET, algorithms=[JWT_ALGORITHM], options={"verify_exp": False}, # Disable expiration verification ) return payload except PyJWTError: return None ################################################################################ # Database Setup ################################################################################ log.debug(f"Using database URL: {DB_URL}") # For debugging Base = declarative_base() # Association class for Namespace <-> User (with roles) class NamespaceUserAssociation(Base): __tablename__ = "namespace_user_association" namespace_id = Column(String, ForeignKey("namespaces.id"), primary_key=True) user_id = Column(String, ForeignKey("users.id"), primary_key=True) role = Column(String, nullable=False) # 'owner', 'editor', 'reader' namespace = relationship("Namespace", back_populates="user_associations") user = relationship("User", back_populates="namespace_associations") 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) username = Column(String, unique=True, nullable=False) 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 # Namespaces the user is associated with namespace_associations = relationship( "NamespaceUserAssociation", back_populates="user" ) namespaces = relationship( "Namespace", secondary="namespace_user_association", back_populates="users", ) # Attribute to hold dbsession in permission checks dbsession = None def __repr__(self): return f"" class Namespace(Base): __tablename__ = "namespaces" id = Column(String, primary_key=True) # UUID short_id = Column(String, unique=True, nullable=False) name = Column(String, unique=True, nullable=False) is_public = Column(Boolean, default=False) # Users associated with the namespace user_associations = relationship( "NamespaceUserAssociation", back_populates="namespace" ) users = relationship( "User", secondary="namespace_user_association", back_populates="namespaces", ) # Agents associated with the namespace agents = relationship("Agent", back_populates="namespace") def __repr__(self): return f"" class Agent(Base): __tablename__ = "agents" id = Column(String, primary_key=True) # UUID name = Column(String, nullable=False) namespace_id = Column(String, ForeignKey("namespaces.id")) role = Column(String, nullable=False) # 'owner', 'editor', 'reader' token_version = Column(Integer, default=0) created_at = Column(DateTime, default=datetime.datetime.utcnow) status = Column(String, default="active") # 'active' or 'revoked' namespace = relationship("Namespace", back_populates="agents") def __repr__(self): return ( f"" ) # New table for scheduler lock class SchedulerLock(Base): __tablename__ = "scheduler_lock" id = Column(Integer, primary_key=True) locked = Column(Boolean, default=False) locked_at = Column(DateTime, nullable=True) process_id = Column(String, nullable=True) ################################################################################ # Namespace Database Models ################################################################################ NamespaceBase = declarative_base() class Media(NamespaceBase): __tablename__ = "media" id = Column(String, primary_key=True) # UUID short_id = Column(String, unique=True, nullable=False) filename = Column(String, nullable=False) title = Column(String, nullable=True) # Optional title media_type = Column(String, nullable=False) # 'image', 'audio', 'video' mime_type = Column(String, nullable=True) # MIME type for compatibility media_b64 = Column(Text, nullable=False) upload_date = Column(DateTime, default=datetime.datetime.utcnow) visibility = Column(String, default="public") # 'public', 'private', 'unlisted' size = Column(Integer, nullable=False) # Size in bytes def __repr__(self): return f"" class ScheduledAction(NamespaceBase): __tablename__ = "scheduled_actions" id = Column(String, primary_key=True) # UUID media_id = Column(String, ForeignKey("media.id"), nullable=False) action_type = Column( String, nullable=False ) # 'set_public', 'set_private', 'set_unlisted', 'delete' scheduled_date = Column(DateTime, nullable=False) created_at = Column(DateTime, default=datetime.datetime.utcnow) status = Column(String, default="pending") # 'pending', 'completed', 'failed' media = relationship("Media") def __repr__(self): return f"" ################################################################################ # Jinja2 Environment and Custom Filters ################################################################################ @subscriber(IJinja2Environment) def add_jinja2_filters(event): env = event.environment env.filters["filesizeformat"] = filesizeformat ################################################################################ # Request Methods ################################################################################ def get_current_user(request): """ Return the current user from session (for users). """ s = request.dbsession user_id = request.session.get("user_id") if user_id: 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; return None return None def get_current_agent(request): """ Return the current agent based on the JWT token. """ s = request.dbsession auth_header = request.headers.get("Authorization") if auth_header and auth_header.startswith("Bearer "): token = auth_header[len("Bearer ") :].strip() payload = verify_jwt_token(token) if payload: agent_id = payload.get("agent_id") if not agent_id: return None agent = s.query(Agent).filter_by(id=agent_id).first() if agent and agent.status == "active": # Check token version if payload.get("token_version") != agent.token_version: return None # Token has been revoked request.jwt_payload = payload return agent return None def get_namespace(request): """Get the namespace from the route parameter 'namespace_short_id'.""" namespace_short_id = request.matchdict.get("namespace_short_id") if not namespace_short_id: return None s = request.dbsession namespace = s.query(Namespace).filter_by(short_id=namespace_short_id).first() return namespace def get_user_or_agent_namespace_role(request): """ Retrieve the role of the current user or agent in the namespace. """ namespace = request.namespace if not namespace: return None # First, check if an agent is authenticated agent = request.agent if agent and agent.namespace_id == namespace.id: return agent.role # Next, check if a user is authenticated via session user = request.user if user: s = request.dbsession association = ( s.query(NamespaceUserAssociation) .filter( NamespaceUserAssociation.user_id == user.id, NamespaceUserAssociation.namespace_id == namespace.id, ) .first() ) if association: return association.role # No role found return None def get_namespace_dbsession(request): """Adds namespace_dbsession to request if namespace is set.""" namespace = request.namespace if namespace: namespace_dbsession = get_namespace_dbsession_by_namespace_id( namespace.id, request ) return namespace_dbsession else: return None # No namespace selected def get_namespace_dbsession_by_namespace_id(namespace_id, request): """Helper function to get a namespace_dbsession for a given namespace_id.""" namespace_db_url = get_namespace_db_url(namespace_id) db_file = os.path.join(DATA_DIR, f"namespace_{namespace_id}.db") if not os.path.exists(db_file): # Create the namespace database if it doesn't exist engine = create_engine( namespace_db_url, connect_args={"check_same_thread": False}, poolclass=StaticPool, ) NamespaceBase.metadata.create_all(engine) engine.dispose() namespace_engine = create_engine( namespace_db_url, connect_args={"check_same_thread": False}, poolclass=StaticPool, ) NamespaceSessionFactory = sessionmaker(bind=namespace_engine) namespace_dbsession = scoped_session(NamespaceSessionFactory) register(namespace_dbsession) # Register with zope.sqlalchemy # Attach cleanup callbacks def cleanup(_request): namespace_dbsession.remove() namespace_engine.dispose() request.add_finished_callback(cleanup) return namespace_dbsession def check_namespace_permission(request, required_role): """ Check if the user or agent has the required role in the namespace. Supports both session-based user authentication and JWT-based agent authentication. """ namespace = request.namespace if not namespace: return False # Get role from user or agent role = get_user_or_agent_namespace_role(request) if role: roles_hierarchy = {"owner": 3, "editor": 2, "reader": 1} return roles_hierarchy.get(role, 0) >= roles_hierarchy.get(required_role, 0) # If no role, check if the namespace is public and required_role is 'reader' if namespace.is_public and required_role == "reader": return True return False def owner_required(view_func): def wrapper(request): if check_namespace_permission(request, "owner"): return view_func(request) else: return HTTPForbidden("You must be an owner to access this page.") return wrapper def editor_required(view_func): def wrapper(request): if check_namespace_permission(request, "editor"): return view_func(request) else: return HTTPForbidden("You must be an editor to access this page.") return wrapper def reader_required(view_func): def wrapper(request): if check_namespace_permission(request, "reader"): return view_func(request) else: return HTTPForbidden("You do not have access to this namespace.") return wrapper ################################################################################ # Scheduler Functions ################################################################################ def acquire_scheduler_lock(dbsession): """Acquire the scheduler lock.""" process_id = f"{os.getpid()}_{int(time.time())}" # Try to get existing lock lock = dbsession.query(SchedulerLock).first() if not lock: # Create new lock lock = SchedulerLock( locked=True, locked_at=datetime.datetime.utcnow(), process_id=process_id ) dbsession.add(lock) dbsession.commit() return True # Check if lock is stale (older than 5 minutes) if lock.locked and lock.locked_at: if datetime.datetime.utcnow() - lock.locked_at > datetime.timedelta(minutes=5): log.warning("Detected stale scheduler lock, releasing it") lock.locked = False lock.locked_at = None lock.process_id = None dbsession.commit() if not lock.locked: lock.locked = True lock.locked_at = datetime.datetime.utcnow() lock.process_id = process_id dbsession.commit() return True return False def release_scheduler_lock(dbsession): """Release the scheduler lock.""" lock = dbsession.query(SchedulerLock).first() if lock and lock.locked: lock.locked = False lock.locked_at = None lock.process_id = None dbsession.commit() def process_scheduled_actions(): """Process all scheduled actions across all namespaces.""" main_engine = create_engine( DB_URL, connect_args={"check_same_thread": False}, poolclass=StaticPool, ) MainSessionFactory = sessionmaker(bind=main_engine) main_dbsession = MainSessionFactory() try: if not acquire_scheduler_lock(main_dbsession): return # Another process is already running log.info("Scheduler acquired lock, processing scheduled actions...") # Get all namespaces namespaces = main_dbsession.query(Namespace).all() for namespace in namespaces: process_namespace_scheduled_actions(namespace.id) log.info("Finished processing scheduled actions") except Exception as e: log.error(f"Error in scheduler: {e}") finally: release_scheduler_lock(main_dbsession) main_dbsession.close() main_engine.dispose() def process_namespace_scheduled_actions(namespace_id): """Process scheduled actions for a specific namespace.""" namespace_db_url = get_namespace_db_url(namespace_id) try: namespace_engine = create_engine( namespace_db_url, connect_args={"check_same_thread": False}, poolclass=StaticPool, ) NamespaceSessionFactory = sessionmaker(bind=namespace_engine) namespace_dbsession = NamespaceSessionFactory() # Check if scheduled_actions table exists inspector = inspect(namespace_engine) if "scheduled_actions" not in inspector.get_table_names(): log.debug( f"No scheduled_actions table in namespace {namespace_id}, skipping" ) return # Get pending actions that are due now = datetime.datetime.utcnow() pending_actions = ( namespace_dbsession.query(ScheduledAction) .filter( ScheduledAction.status == "pending", ScheduledAction.scheduled_date <= now, ) .all() ) for action in pending_actions: try: media = ( namespace_dbsession.query(Media) .filter_by(id=action.media_id) .first() ) if not media: action.status = "failed" continue if action.action_type == "delete": namespace_dbsession.delete(media) log.info(f"Scheduled delete executed for media {media.filename}") elif action.action_type == "set_public": media.visibility = "public" log.info( f"Scheduled set_public executed for media {media.filename}" ) elif action.action_type == "set_private": media.visibility = "private" log.info( f"Scheduled set_private executed for media {media.filename}" ) elif action.action_type == "set_unlisted": media.visibility = "unlisted" log.info( f"Scheduled set_unlisted executed for media {media.filename}" ) action.status = "completed" except Exception as e: log.error(f"Error processing scheduled action {action.id}: {e}") action.status = "failed" namespace_dbsession.commit() except Exception as e: log.error(f"Error processing namespace {namespace_id}: {e}") finally: if "namespace_dbsession" in locals(): namespace_dbsession.close() if "namespace_engine" in locals(): namespace_engine.dispose() # Global variables for scheduler thread scheduler_thread = None scheduler_running = False def scheduler_worker(): """Background scheduler worker that runs every minute.""" global scheduler_running while scheduler_running: try: process_scheduled_actions() except Exception as e: log.error(f"Scheduler worker error: {e}") # Sleep for 60 seconds for i in range(60): if not scheduler_running: break time.sleep(1) def start_scheduler(): """Start the background scheduler.""" global scheduler_thread, scheduler_running if not scheduler_running: scheduler_running = True scheduler_thread = threading.Thread(target=scheduler_worker, daemon=True) scheduler_thread.start() log.info("Scheduler started") def stop_scheduler(): """Stop the background scheduler.""" global scheduler_thread, scheduler_running if scheduler_running: scheduler_running = False if scheduler_thread: scheduler_thread.join(timeout=5) log.info("Scheduler stopped") ################################################################################ # Routes and Views ################################################################################ @view_config(route_name="home", renderer="home.html.j2") def home_view(request): s = request.dbsession # Get public namespaces public_namespaces = s.query(Namespace).filter(Namespace.is_public == True).all() user_namespaces = [] if request.user and request.user.is_verified: # Get namespaces the user has access to along with their roles user_namespaces = [] for association in request.user.namespace_associations: ns = association.namespace user_namespaces.append( { "name": ns.name, "short_id": ns.short_id, "role": association.role, } ) return { "request": request, "public_namespaces": public_namespaces, "user_namespaces": user_namespaces, } ################################################################################ # 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.flush() # 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.flush() # 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.flush() # 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", request_method="POST", require_csrf=True) def logout_view(request): request.session.invalidate() return HTTPFound(location=request.route_url("home")) ################################################################################ # Profile and Namespace Management ################################################################################ @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 "" # Get namespaces where the user is an owner owner_namespaces = [] for association in user.namespace_associations: if association.role == "owner": owner_namespaces.append(association.namespace) return { "request": request, "user": user, "gravatar_url": gravatar_url, "owner_namespaces": owner_namespaces, } @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.flush() return HTTPFound(location=request.route_url("profile")) @view_config( route_name="create_namespace", request_method="GET", renderer="create_namespace.html.j2", ) def create_namespace_get_view(request): if not request.user or not request.user.is_verified: return Response("You must be logged in to create a namespace.", status=403) return {"request": request} @view_config(route_name="create_namespace", request_method="POST") def create_namespace_post_view(request): user = request.user if not user or not user.is_verified: return Response("You must be logged in to create a namespace.", status=403) name = request.POST.get("name", "").strip() if not name: return Response("Namespace name is required.", status=400) s = request.dbsession existing_namespace = s.query(Namespace).filter(Namespace.name == name).first() if existing_namespace: return Response("Namespace name already exists.", status=400) # Generate UUID and short ID for the namespace namespace_uuid = uuid.uuid4() namespace_id = str(namespace_uuid) namespace_short_id = uuid_to_short_id(namespace_uuid) is_public = request.POST.get("is_public") == "on" namespace = Namespace( id=namespace_id, short_id=namespace_short_id, name=name, is_public=is_public, ) s.add(namespace) # Add the user as an owner association = NamespaceUserAssociation( namespace=namespace, user=user, role="owner", ) s.add(association) s.flush() # Create namespace database namespace_db_url = get_namespace_db_url(namespace.id) engine = create_engine( namespace_db_url, connect_args={"check_same_thread": False}, poolclass=StaticPool, ) NamespaceBase.metadata.create_all(engine) engine.dispose() return HTTPFound( location=request.route_url( "manage_namespace", namespace_short_id=namespace.short_id ) ) @view_config(route_name="manage_namespace", renderer="manage_namespace.html.j2") @editor_required def manage_namespace_view(request): namespace = request.namespace s = request.dbsession # Get users and their roles in the namespace associations = ( s.query(NamespaceUserAssociation) .filter(NamespaceUserAssociation.namespace_id == namespace.id) .all() ) users = [] for association in associations: user = association.user users.append({"user": user, "role": association.role}) # Get agents associated with the namespace (only active agents) agents = ( s.query(Agent) .filter(Agent.namespace_id == namespace.id, Agent.status == "active") .all() ) return { "request": request, "namespace": namespace, "users": users, "agents": agents, } @view_config(route_name="change_member_role", request_method="POST") @owner_required def change_member_role_view(request): s = request.dbsession namespace = request.namespace current_user = request.user user_id = request.POST.get("user_id") new_role = request.POST.get("role") if not user_id or not new_role: return Response("User ID and new role are required.", status=400) if new_role not in ["owner", "editor", "reader"]: return Response("Invalid role.", status=400) # Prevent owners from changing their own role if user_id == current_user.id: return Response("Owners cannot change their own role.", status=400) # Ensure the user is a member of the namespace association = ( s.query(NamespaceUserAssociation) .filter( NamespaceUserAssociation.namespace_id == namespace.id, NamespaceUserAssociation.user_id == user_id, ) .first() ) if not association: return Response("User is not a member of this namespace.", status=400) # Update the user's role association.role = new_role s.flush() request.session.flash(f"User's role has been updated to {new_role}.") return HTTPFound( location=request.route_url( "manage_namespace", namespace_short_id=namespace.short_id ) ) @view_config(route_name="update_namespace", request_method="POST") @owner_required def update_namespace_view(request): namespace = request.namespace s = request.dbsession # Update namespace properties is_public = request.POST.get("is_public") == "on" namespace.is_public = is_public s.flush() return HTTPFound( location=request.route_url( "manage_namespace", namespace_short_id=namespace.short_id ) ) @view_config(route_name="invite_user", request_method="POST") @owner_required def invite_user_view(request): namespace = request.namespace s = request.dbsession email = request.POST.get("email", "").strip().lower() role = request.POST.get("role", "").strip().lower() if role not in ["owner", "editor", "reader"]: return Response("Invalid role.", status=400) # Find or create the user user = s.query(User).filter(User.email == email).first() if not user: user_uuid = uuid.uuid4() user_id = str(user_uuid) short_id = uuid_to_short_id(user_uuid) user = User( id=user_id, short_id=short_id, email=email, username=email.split("@")[0], is_verified=False, ) s.add(user) s.flush() # Check if the user already has an association with the namespace existing_association = ( s.query(NamespaceUserAssociation) .filter( NamespaceUserAssociation.namespace_id == namespace.id, NamespaceUserAssociation.user_id == user.id, ) .first() ) if existing_association: # Update the role if the user is already associated existing_association.role = role else: # Create a new association association = NamespaceUserAssociation( namespace=namespace, user=user, role=role, ) s.add(association) s.flush() request.session.flash(f"{user.email} was invited as {role} role.") # Send invitation email email_body = ( f"You have been invited as a {role} to namespace " f"'{namespace.name}'. Please log in to access it." ) send_email(user.email, "Namespace Invitation", email_body) return HTTPFound( location=request.route_url( "manage_namespace", namespace_short_id=namespace.short_id ) ) @view_config(route_name="remove_user", request_method="POST") @owner_required def remove_user_view(request): s = request.dbsession namespace = request.namespace current_user = request.user # The owner initiating the removal # Get the user ID to remove from the POST data user_id_to_remove = request.POST.get("user_id") if not user_id_to_remove: return Response("User ID is required.", status=400) # Ensure that the user exists and is a member of the namespace user_to_remove = s.query(User).filter(User.id == user_id_to_remove).first() if not user_to_remove: return Response("User not found.", status=404) # Prevent owners from removing themselves if user_to_remove.id == current_user.id: return Response("Owners cannot remove themselves.", status=400) # Check if the user to remove is a member of the namespace association = ( s.query(NamespaceUserAssociation) .filter( NamespaceUserAssociation.namespace_id == namespace.id, NamespaceUserAssociation.user_id == user_id_to_remove, ) .first() ) if not association: return Response("User is not a member of this namespace.", status=400) # Remove the association s.delete(association) s.flush() # Provide a success message request.session.flash( f"User '{user_to_remove.username}' has been removed from the namespace." ) return HTTPFound( location=request.route_url( "manage_namespace", namespace_short_id=namespace.short_id ) ) @view_config( route_name="generate_agent_jwt", request_method="POST", renderer="display_agent_jwt.html.j2", ) @owner_required def generate_agent_jwt_view(request): namespace = request.namespace s = request.dbsession agent_name = request.POST.get("agent_name", "").strip() agent_role = request.POST.get("agent_role", "").strip().lower() if not agent_name: return Response("Agent name is required.", status=400) if agent_role not in ["owner", "editor", "reader"]: return Response("Invalid agent role.", status=400) # Check if an agent with the same name exists in the namespace agent = ( s.query(Agent) .filter( Agent.name == agent_name, Agent.namespace_id == namespace.id, ) .first() ) if agent: # Agent exists, increment token_version and regenerate JWT agent.token_version += 1 agent.status = "active" # Ensure the agent is active agent.role = agent_role # Update the role s.flush() jwt_token = generate_jwt_token(agent) message = ( f"A new JWT has been generated for existing agent '{agent.name}'. " "Any previous tokens have been revoked." ) else: # Generate UUID for the new agent agent_uuid = uuid.uuid4() agent_id = str(agent_uuid) # Create a new agent entry agent = Agent( id=agent_id, name=agent_name, namespace_id=namespace.id, role=agent_role, token_version=0, status="active", ) s.add(agent) s.flush() # Generate JWT for the agent including the namespace_id jwt_token = generate_jwt_token(agent) message = f"A new agent '{agent.name}' has been created." return { "request": request, "namespace": namespace, "agent_name": agent_name, "jwt_token": jwt_token, "message": message, } @view_config(route_name="revoke_agent", request_method="POST") @owner_required def revoke_agent_view(request): namespace = request.namespace s = request.dbsession agent_id = request.POST.get("agent_id") if not agent_id: return Response("Agent ID is required.", status=400) # Get the agent agent = ( s.query(Agent) .filter( Agent.id == agent_id, Agent.namespace_id == namespace.id, ) .first() ) if not agent: return Response("Agent not found.", status=404) # Set the agent's status to 'revoked' to hide it from the dashboard agent.status = "revoked" s.flush() request.session.flash(f"Agent '{agent.name}' has been revoked.") return HTTPFound( location=request.route_url( "manage_namespace", namespace_short_id=namespace.short_id ) ) ################################################################################ # Media Upload, Listing, and Management ################################################################################ @view_config( route_name="upload_media", request_method="GET", renderer="upload_media.html.j2", ) @editor_required def upload_media_get_view(request): return {"request": request} @view_config(route_name="upload_media", request_method="POST") @editor_required def upload_media_post_view(request): namespace = request.namespace namespace_dbsession = request.namespace_dbsession 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 = 310 * 1024 * 1024 # 310 MB if len(raw_bytes) > max_size: return Response("File size exceeds the 310MB limit.", status=400) file_size = len(raw_bytes) # Determine media type based on MIME type filename = media_file.filename mime_type = get_mime_type(filename) if not mime_type.startswith(ALLOWED_MIME_PREFIXES): return Response("Unsupported media type.", status=400) media_type = mime_type.split("/")[0] # Get title from form title = request.POST.get("title", "").strip() # If user did not enter a title, default to the original filename if not title: title = filename # Encode content to base64 encoded_str = base64.b64encode(raw_bytes).decode("utf-8") # 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) visibility = request.POST.get("visibility", "public") if visibility not in ["public", "private", "unlisted"]: visibility = "public" media = Media( id=media_id, short_id=media_short_id, filename=filename, title=title, media_type=media_type, mime_type=mime_type, media_b64=encoded_str, visibility=visibility, size=file_size, ) namespace_dbsession.add(media) # Handle scheduled actions scheduled_public = request.POST.get("scheduled_public") scheduled_private = request.POST.get("scheduled_private") scheduled_unlisted = request.POST.get("scheduled_unlisted") scheduled_delete = request.POST.get("scheduled_delete") # Check if scheduled_actions table exists before adding scheduled actions try: inspector = inspect(namespace_dbsession.bind) has_scheduled_actions = "scheduled_actions" in inspector.get_table_names() except: has_scheduled_actions = False if has_scheduled_actions: if scheduled_public: try: scheduled_date = datetime.datetime.fromisoformat(scheduled_public) action_uuid = uuid.uuid4() action = ScheduledAction( id=str(action_uuid), media_id=media_id, action_type="set_public", scheduled_date=scheduled_date, ) namespace_dbsession.add(action) except ValueError: pass if scheduled_private: try: scheduled_date = datetime.datetime.fromisoformat(scheduled_private) action_uuid = uuid.uuid4() action = ScheduledAction( id=str(action_uuid), media_id=media_id, action_type="set_private", scheduled_date=scheduled_date, ) namespace_dbsession.add(action) except ValueError: pass if scheduled_unlisted: try: scheduled_date = datetime.datetime.fromisoformat(scheduled_unlisted) action_uuid = uuid.uuid4() action = ScheduledAction( id=str(action_uuid), media_id=media_id, action_type="set_unlisted", scheduled_date=scheduled_date, ) namespace_dbsession.add(action) except ValueError: pass if scheduled_delete: try: scheduled_date = datetime.datetime.fromisoformat(scheduled_delete) action_uuid = uuid.uuid4() action = ScheduledAction( id=str(action_uuid), media_id=media_id, action_type="delete", scheduled_date=scheduled_date, ) namespace_dbsession.add(action) except ValueError: pass namespace_dbsession.flush() return HTTPFound( location=request.route_url( "view_media_details", namespace_short_id=namespace.short_id, media_short_id=media.short_id, ) ) @view_config(route_name="list_media", renderer="list_media.html.j2") def list_media_view(request): namespace = request.namespace namespace_dbsession = request.namespace_dbsession # Check access if not check_namespace_permission(request, "reader"): return HTTPForbidden("You do not have access to this namespace.") # Get the media items, excluding the media_b64 column # Only show public media unless user has editor+ access query = namespace_dbsession.query(Media).options( load_only( Media.id, Media.short_id, Media.filename, Media.title, Media.media_type, Media.upload_date, Media.visibility, Media.size, ) ) # Filter based on user permissions user_role = get_user_or_agent_namespace_role(request) if user_role in ["owner", "editor"]: # Owners and editors can see all media except unlisted media_items = ( query.filter(Media.visibility != "unlisted") .order_by(Media.upload_date.desc()) .all() ) else: # Readers and public users can only see public media media_items = ( query.filter(Media.visibility == "public") .order_by(Media.upload_date.desc()) .all() ) return { "request": request, "namespace": namespace, "media_items": media_items, } @view_config(route_name="view_media_details", renderer="view_media_details.html.j2") def view_media_details_view(request): namespace = request.namespace namespace_dbsession = request.namespace_dbsession media_short_id = request.matchdict.get("media_short_id") media = namespace_dbsession.query(Media).filter_by(short_id=media_short_id).first() if not media: return Response("Media not found.", status=404) role = get_user_or_agent_namespace_role(request) is_owner_or_editor = role in ["owner", "editor"] # Check visibility permissions if media.visibility == "private" and not check_namespace_permission( request, "reader" ): return Response("Media not available.", status=403) elif media.visibility == "unlisted" and not is_owner_or_editor: return Response("Media not available.", status=403) elif media.visibility == "public": # Public media is always accessible pass # Check if this is an ASCII file and get preview content ascii_preview = None mime_type = get_mime_type(media.filename) if is_ascii_file(mime_type): try: raw_content = base64.b64decode(media.media_b64).decode("utf-8") lines = raw_content.split("\n") if len(lines) > 100: ascii_preview = "\n".join(lines[:100]) + "\n... (truncated)" else: ascii_preview = raw_content except (UnicodeDecodeError, Exception): ascii_preview = None # Get scheduled actions for this media scheduled_actions = [] if is_owner_or_editor: try: # Check if scheduled_actions table exists inspector = inspect(namespace_dbsession.bind) if "scheduled_actions" in inspector.get_table_names(): scheduled_actions = ( namespace_dbsession.query(ScheduledAction) .filter( ScheduledAction.media_id == media.id, ScheduledAction.status == "pending", ) .order_by(ScheduledAction.scheduled_date) .all() ) except Exception as e: log.debug(f"Could not query scheduled actions: {e}") scheduled_actions = [] return { "request": request, "media": media, "namespace": namespace, "is_owner_or_editor": is_owner_or_editor, "ascii_preview": ascii_preview, "scheduled_actions": scheduled_actions, } @view_config(route_name="delete_media", request_method="POST") @editor_required def delete_media_view(request): namespace = request.namespace namespace_dbsession = request.namespace_dbsession media_short_id = request.matchdict.get("media_short_id") media = namespace_dbsession.query(Media).filter_by(short_id=media_short_id).first() if not media: return Response("Media not found.", status=404) # Delete associated scheduled actions if table exists try: inspector = inspect(namespace_dbsession.bind) if "scheduled_actions" in inspector.get_table_names(): scheduled_actions = ( namespace_dbsession.query(ScheduledAction) .filter_by(media_id=media.id) .all() ) for action in scheduled_actions: namespace_dbsession.delete(action) except Exception as e: log.debug(f"Could not delete scheduled actions: {e}") namespace_dbsession.delete(media) namespace_dbsession.flush() return HTTPFound( location=request.route_url("list_media", namespace_short_id=namespace.short_id) ) @view_config( route_name="edit_media", request_method="GET", renderer="edit_media.html.j2" ) @editor_required def edit_media_get_view(request): namespace = request.namespace namespace_dbsession = request.namespace_dbsession media_short_id = request.matchdict.get("media_short_id") media = namespace_dbsession.query(Media).filter_by(short_id=media_short_id).first() if not media: return Response("Media not found.", status=404) # Check if this is an ASCII file and get editable content ascii_content = None mime_type = get_mime_type(media.filename) if is_ascii_file(mime_type): try: ascii_content = base64.b64decode(media.media_b64).decode("utf-8") except (UnicodeDecodeError, Exception): ascii_content = None return { "request": request, "media": media, "namespace": namespace, "ascii_content": ascii_content, } @view_config(route_name="edit_media", request_method="POST") @editor_required def edit_media_post_view(request): namespace = request.namespace namespace_dbsession = request.namespace_dbsession media_short_id = request.matchdict.get("media_short_id") media = namespace_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 ASCII content update ascii_content = request.POST.get("ascii_content", "").strip() mime_type = get_mime_type(media.filename) if ascii_content and is_ascii_file(mime_type): try: # Update the media content with the new ASCII content encoded_str = base64.b64encode(ascii_content.encode("utf-8")).decode( "utf-8" ) media.media_b64 = encoded_str media.size = len(ascii_content.encode("utf-8")) except Exception as e: return Response(f"Error updating content: {e}", status=400) # Handle media file update new_media_file = request.POST.get("media_file") if new_media_file is not None and getattr(new_media_file, "filename", "").strip(): raw_bytes = new_media_file.file.read() # 310 MB max_size = 310 * 1024 * 1024 if len(raw_bytes) > max_size: return Response("File size exceeds the 310MB limit.", status=400) file_size = len(raw_bytes) filename = new_media_file.filename mime_type = get_mime_type(filename) if not mime_type.startswith(ALLOWED_MIME_PREFIXES): return Response("Unsupported media type.", status=400) media_type = mime_type.split("/")[0] encoded_str = base64.b64encode(raw_bytes).decode("utf-8") media.filename = filename media.media_type = media_type media.mime_type = mime_type # Update mime_type too media.media_b64 = encoded_str media.size = file_size # Handle visibility update visibility = request.POST.get("visibility", "public") if visibility in ["public", "private", "unlisted"]: media.visibility = visibility namespace_dbsession.flush() return HTTPFound( location=request.route_url( "view_media_details", namespace_short_id=namespace.short_id, media_short_id=media.short_id, ) ) ################################################################################ # Media Viewing and Downloading ################################################################################ @view_config(route_name="view_media") def view_media_view(request): namespace = request.namespace namespace_dbsession = request.namespace_dbsession media_short_id = request.matchdict.get("media_short_id") media = namespace_dbsession.query(Media).filter_by(short_id=media_short_id).first() if not media: return Response("Media not found.", status=404) role = get_user_or_agent_namespace_role(request) is_owner_or_editor = role in ["owner", "editor"] # Check visibility permissions if media.visibility == "private" and not check_namespace_permission( request, "reader" ): return Response("Media not available.", status=403) elif media.visibility == "unlisted": # Unlisted media is accessible via direct link but not listed if not is_owner_or_editor and not check_namespace_permission(request, "reader"): return Response("Media not available.", status=403) elif media.visibility == "public": # Public media is always accessible via direct link pass media_data = base64.b64decode(media.media_b64) mime_type = get_mime_type(media.filename) # Build a safe filename (avoid Unicode issues in the header) if media.title: file_extension = os.path.splitext(media.filename)[1] raw_title = media.title download_filename = sanitize_filename_for_http_header( f"{raw_title}{file_extension}" ) else: download_filename = sanitize_filename_for_http_header(media.filename) # Check if user wants attachment or inline download = request.GET.get("download", "false").lower() == "true" content_disposition = "attachment" if download else "inline" response = Response(body=media_data, content_type=mime_type) response.headers.update( { "Access-Control-Allow-Origin": "*", "Content-Disposition": f'{content_disposition}; filename="{download_filename}"', } ) return response ################################################################################ # Scheduled Action Management ################################################################################ @view_config(route_name="delete_scheduled_action", request_method="POST") @editor_required def delete_scheduled_action_view(request): namespace = request.namespace namespace_dbsession = request.namespace_dbsession media_short_id = request.matchdict.get("media_short_id") action_id = request.matchdict.get("action_id") # Verify media exists media = namespace_dbsession.query(Media).filter_by(short_id=media_short_id).first() if not media: return Response("Media not found.", status=404) # Check if scheduled_actions table exists try: inspector = inspect(namespace_dbsession.bind) if "scheduled_actions" not in inspector.get_table_names(): return Response("Scheduled actions not supported.", status=404) except: return Response("Scheduled actions not supported.", status=404) # Find and delete the scheduled action scheduled_action = ( namespace_dbsession.query(ScheduledAction) .filter(ScheduledAction.id == action_id, ScheduledAction.media_id == media.id) .first() ) if not scheduled_action: return Response("Scheduled action not found.", status=404) namespace_dbsession.delete(scheduled_action) namespace_dbsession.flush() request.session.flash( f"Scheduled {scheduled_action.action_type.replace('set_', '').replace('_', ' ')} action deleted." ) return HTTPFound( location=request.route_url( "view_media_details", namespace_short_id=namespace.short_id, media_short_id=media.short_id, ) ) @view_config( route_name="edit_scheduled_action", request_method="GET", renderer="edit_scheduled_action.html.j2", ) @editor_required def edit_scheduled_action_get_view(request): namespace = request.namespace namespace_dbsession = request.namespace_dbsession media_short_id = request.matchdict.get("media_short_id") action_id = request.matchdict.get("action_id") # Verify media exists media = namespace_dbsession.query(Media).filter_by(short_id=media_short_id).first() if not media: return Response("Media not found.", status=404) # Check if scheduled_actions table exists try: inspector = inspect(namespace_dbsession.bind) if "scheduled_actions" not in inspector.get_table_names(): return Response("Scheduled actions not supported.", status=404) except: return Response("Scheduled actions not supported.", status=404) # Find the scheduled action scheduled_action = ( namespace_dbsession.query(ScheduledAction) .filter(ScheduledAction.id == action_id, ScheduledAction.media_id == media.id) .first() ) if not scheduled_action: return Response("Scheduled action not found.", status=404) return { "request": request, "namespace": namespace, "media": media, "scheduled_action": scheduled_action, } @view_config(route_name="edit_scheduled_action", request_method="POST") @editor_required def edit_scheduled_action_post_view(request): namespace = request.namespace namespace_dbsession = request.namespace_dbsession media_short_id = request.matchdict.get("media_short_id") action_id = request.matchdict.get("action_id") # Verify media exists media = namespace_dbsession.query(Media).filter_by(short_id=media_short_id).first() if not media: return Response("Media not found.", status=404) # Check if scheduled_actions table exists try: inspector = inspect(namespace_dbsession.bind) if "scheduled_actions" not in inspector.get_table_names(): return Response("Scheduled actions not supported.", status=404) except: return Response("Scheduled actions not supported.", status=404) # Find the scheduled action scheduled_action = ( namespace_dbsession.query(ScheduledAction) .filter(ScheduledAction.id == action_id, ScheduledAction.media_id == media.id) .first() ) if not scheduled_action: return Response("Scheduled action not found.", status=404) # Update the scheduled action new_action_type = request.POST.get("action_type") new_scheduled_date = request.POST.get("scheduled_date") if new_action_type not in ["set_public", "set_private", "set_unlisted", "delete"]: return Response("Invalid action type.", status=400) try: scheduled_date = datetime.datetime.fromisoformat(new_scheduled_date) except ValueError: return Response("Invalid date format.", status=400) scheduled_action.action_type = new_action_type scheduled_action.scheduled_date = scheduled_date namespace_dbsession.flush() request.session.flash( f"Scheduled action updated to {new_action_type.replace('set_', '').replace('_', ' ')} on {scheduled_date.strftime('%Y-%m-%d %H:%M')}." ) return HTTPFound( location=request.route_url( "view_media_details", namespace_short_id=namespace.short_id, media_short_id=media.short_id, ) ) @view_config( route_name="add_scheduled_action", request_method="GET", renderer="add_scheduled_action.html.j2", ) @editor_required def add_scheduled_action_get_view(request): namespace = request.namespace namespace_dbsession = request.namespace_dbsession media_short_id = request.matchdict.get("media_short_id") # Verify media exists media = namespace_dbsession.query(Media).filter_by(short_id=media_short_id).first() if not media: return Response("Media not found.", status=404) # Check if scheduled_actions table exists try: inspector = inspect(namespace_dbsession.bind) if "scheduled_actions" not in inspector.get_table_names(): return Response("Scheduled actions not supported.", status=404) except: return Response("Scheduled actions not supported.", status=404) return { "request": request, "namespace": namespace, "media": media, } @view_config(route_name="add_scheduled_action", request_method="POST") @editor_required def add_scheduled_action_post_view(request): namespace = request.namespace namespace_dbsession = request.namespace_dbsession media_short_id = request.matchdict.get("media_short_id") # Verify media exists media = namespace_dbsession.query(Media).filter_by(short_id=media_short_id).first() if not media: return Response("Media not found.", status=404) # Check if scheduled_actions table exists try: inspector = inspect(namespace_dbsession.bind) if "scheduled_actions" not in inspector.get_table_names(): return Response("Scheduled actions not supported.", status=404) except: return Response("Scheduled actions not supported.", status=404) # Get form data action_type = request.POST.get("action_type") scheduled_date_str = request.POST.get("scheduled_date") if action_type not in ["set_public", "set_private", "set_unlisted", "delete"]: return Response("Invalid action type.", status=400) try: scheduled_date = datetime.datetime.fromisoformat(scheduled_date_str) except ValueError: return Response("Invalid date format.", status=400) # Create new scheduled action action_uuid = uuid.uuid4() scheduled_action = ScheduledAction( id=str(action_uuid), media_id=media.id, action_type=action_type, scheduled_date=scheduled_date, ) namespace_dbsession.add(scheduled_action) namespace_dbsession.flush() request.session.flash( f"Scheduled {action_type.replace('set_', '').replace('_', ' ')} action added for {scheduled_date.strftime('%Y-%m-%d %H:%M')}." ) return HTTPFound( location=request.route_url( "view_media_details", namespace_short_id=namespace.short_id, media_short_id=media.short_id, ) ) ################################################################################ # Main ################################################################################ def main(*config, **settings): # Configure logging logging.basicConfig(level=logging.INFO) # Set up the session factory session_factory = SignedCookieSessionFactory( secret=app_secret, 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_tm") config.include("pyramid_openapi3") config.pyramid_openapi3_spec("openapi.yaml", route="/openapi.yaml") config.pyramid_openapi3_add_explorer(route="/docs") # Add .html.j2 extension for Jinja2 templates config.include("pyramid_jinja2") config.add_jinja2_renderer(".j2") config.add_jinja2_search_path("templates", name=".j2") # The Jinja2 filters are added via the event subscriber above # Set up SQLAlchemy engine = create_engine( DB_URL, connect_args={"check_same_thread": False}, poolclass=StaticPool, ) session_factory_ = sessionmaker(bind=engine) Base.metadata.bind = engine # Create tables if they don't exist Base.metadata.create_all(engine) DBSession = scoped_session(session_factory_) # Register with zope.sqlalchemy register(DBSession) # Provide dbsession to requests def dbsession(request): return DBSession config.add_request_method(dbsession, "dbsession", reify=True) # Add user, agent, namespace, user_namespace_role to requests config.add_request_method(get_current_user, "user", reify=True) config.add_request_method(get_current_agent, "agent", reify=True) config.add_request_method(get_namespace, "namespace", reify=True) config.add_request_method( get_user_or_agent_namespace_role, "user_namespace_role", reify=True ) # Add namespace_dbsession to requests if namespace is set config.add_request_method( get_namespace_dbsession, "namespace_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") # Namespace Management config.add_route("create_namespace", "/namespace/create") config.add_route("manage_namespace", "/namespace/{namespace_short_id}/manage") config.add_route("update_namespace", "/namespace/{namespace_short_id}/update") config.add_route("invite_user", "/namespace/{namespace_short_id}/invite") config.add_route("remove_user", "/namespace/{namespace_short_id}/remove_user") config.add_route( "change_member_role", "/namespace/{namespace_short_id}/change_member_role" ) # Agent Management config.add_route("revoke_agent", "/namespace/{namespace_short_id}/revoke_agent") config.add_route( "generate_agent_jwt", "/namespace/{namespace_short_id}/generate_agent_jwt" ) # Media config.add_route("upload_media", "/namespace/{namespace_short_id}/media/upload") config.add_route("list_media", "/namespace/{namespace_short_id}/media/list") config.add_route( "view_media_details", "/namespace/{namespace_short_id}/media/{media_short_id}/details", ) config.add_route( "edit_media", "/namespace/{namespace_short_id}/media/{media_short_id}/edit", ) config.add_route( "delete_media", "/namespace/{namespace_short_id}/media/{media_short_id}/delete", ) config.add_route( "view_media", "/namespace/{namespace_short_id}/media/{media_short_id}", ) # Scheduled Actions Management config.add_route( "delete_scheduled_action", "/namespace/{namespace_short_id}/media/{media_short_id}/scheduled_action/{action_id}/delete", ) config.add_route( "edit_scheduled_action", "/namespace/{namespace_short_id}/media/{media_short_id}/scheduled_action/{action_id}/edit", ) config.add_route( "add_scheduled_action", "/namespace/{namespace_short_id}/media/{media_short_id}/add_scheduled_action", ) config.scan() # Start the scheduler start_scheduler() return config.make_wsgi_app() # Signal handler for graceful shutdown def signal_handler(signum, frame): log.info("Received shutdown signal, stopping scheduler...") stop_scheduler() sys.exit(0) # Register signal handlers signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGTERM, signal_handler) # Expose the WSGI application callable for uwsgi uwsgi_app = main({}) if __name__ == "__main__": app = main() log.info(f"Serving on http://{HOST}:{PORT}") serve(app, host=HOST, port=PORT)