#!/usr/bin/env python3
"""
SamCloud Dashboard - Modern Flask Application
A unified Apple-styled web application for Mac Mini server management
"""
from flask import Flask, render_template, request, jsonify, session, redirect, url_for, send_file, Response, make_response
from werkzeug.utils import secure_filename
from werkzeug.security import generate_password_hash, check_password_hash
from functools import wraps
from datetime import timedelta
import re
import os
import json
import socket
import mimetypes
import subprocess
import time
import shutil
import psutil
import sqlite3
import hashlib
import secrets
from datetime import datetime
import logging
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app = Flask(__name__)
# Prefer a secret from environment for production. Keep a fallback for dev.
app.secret_key = os.getenv('FLASK_SECRET_KEY', 'samcloud-dashboard-secret-key-2025') # override via env in production
# Configuration
class Config:
_fd = os.getenv('FLASK_DEBUG', '0')
DEBUG = (_fd in ('1', 'true', 'True', 'TRUE')) or (os.getenv('FLASK_ENV') == 'development')
HOST = '0.0.0.0'
PORT = 8383
# Paths
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
STATIC_DIR = os.path.join(BASE_DIR, 'static')
TEMPLATES_DIR = os.path.join(BASE_DIR, 'templates')
config = Config()
# Create directories if they don't exist
os.makedirs(config.STATIC_DIR, exist_ok=True)
os.makedirs(config.TEMPLATES_DIR, exist_ok=True)
# Database setup
def init_database():
"""Initialize the SQLite database for user management"""
db_path = os.path.join(config.BASE_DIR, 'dashboard.db')
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
# Create users table
cursor.execute('''
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT UNIQUE NOT NULL,
email TEXT UNIQUE,
password_hash TEXT NOT NULL,
is_admin BOOLEAN DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
last_login TIMESTAMP,
settings TEXT DEFAULT '{}'
)
''')
# Create shares table for enhanced sharing management
cursor.execute('''
CREATE TABLE IF NOT EXISTS user_shares (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER,
share_id TEXT UNIQUE NOT NULL,
name TEXT NOT NULL,
path TEXT NOT NULL,
expires_at TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
download_count INTEGER DEFAULT 0,
password_hash TEXT,
FOREIGN KEY (user_id) REFERENCES users (id)
)
''')
# Create notes table for note-taking functionality
cursor.execute('''
CREATE TABLE IF NOT EXISTS notes (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
title TEXT NOT NULL DEFAULT 'Untitled Note',
content TEXT DEFAULT '',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (user_id) REFERENCES users (id)
)
''')
# Create default admin user if no users exist
cursor.execute('SELECT COUNT(*) FROM users')
if cursor.fetchone()[0] == 0:
# Create default admin user with secure hashed password
admin_password_hash = generate_password_hash('admin')
cursor.execute('''
INSERT INTO users (username, email, password_hash, is_admin, settings)
VALUES (?, ?, ?, ?, ?)
''', ('admin', 'admin@samcloud.local', admin_password_hash, True, '{}'))
conn.commit()
conn.close()
def get_db_connection():
"""Get database connection"""
db_path = os.path.join(config.BASE_DIR, 'dashboard.db')
conn = sqlite3.connect(db_path)
conn.row_factory = sqlite3.Row
return conn
def get_db_cursor():
"""Get database cursor with connection that needs to be managed"""
conn = get_db_connection()
cursor = conn.cursor()
# Return both cursor and connection so they can be properly managed
cursor._connection = conn
return cursor
def hash_password(password):
"""Hash a password using Werkzeug's secure password hashing"""
return generate_password_hash(password)
def verify_password(password, password_hash):
"""Verify a password against its hash using Werkzeug's secure verification"""
return check_password_hash(password_hash, password)
def create_secure_token():
"""Generate a secure random token"""
return secrets.token_urlsafe(32)
def login_required(f):
"""Decorator to require authentication for routes"""
@wraps(f)
def decorated_function(*args, **kwargs):
if 'user' not in session:
if request.is_json:
return jsonify({'error': 'Authentication required'}), 401
return redirect(url_for('login'))
return f(*args, **kwargs)
return decorated_function
# Initialize database on startup
init_database()
# Better live reload in development
if config.DEBUG:
app.config['TEMPLATES_AUTO_RELOAD'] = True
app.config['SEND_FILE_MAX_AGE_DEFAULT'] = 0
try:
app.jinja_env.auto_reload = True
except Exception:
pass
@app.after_request
def add_no_cache_headers(response):
"""Prevent aggressive caching during development for live updates."""
if config.DEBUG:
response.headers['Cache-Control'] = 'no-store, no-cache, must-revalidate, max-age=0'
response.headers['Pragma'] = 'no-cache'
response.headers['Expires'] = '0'
return response
# Load authentication from environment variables
USERS = {
'admin': os.getenv('DASHBOARD_PASSWORD', 'samcloud2025'),
os.getenv('DASHBOARD_USERNAME', 'admin'): os.getenv('DASHBOARD_PASSWORD', 'samcloud2025')
}
def get_connected_devices():
"""Get connected devices including SMB, VPN, Screen Sharing, and network connections"""
try:
devices = []
# Method 1: SMB connections (macOS sharing)
try:
smb_result = subprocess.run(['lsof', '-i', ':445'], capture_output=True, text=True, timeout=5)
if smb_result.returncode == 0:
for line in smb_result.stdout.split('\n')[1:]:
if 'ESTABLISHED' in line:
parts = line.split()
if len(parts) >= 9:
remote_addr = parts[8].split('->')[1] if '->' in parts[8] else parts[8]
ip = remote_addr.split(':')[0]
devices.append({
'hostname': f'SMB Client ({ip})',
'ip': ip,
'mac': 'Unknown',
'type': 'SMB Share',
'status': 'Connected',
'last_seen': datetime.now().isoformat(),
'icon': '📁'
})
except Exception as e:
logger.debug(f"SMB check failed: {e}")
# Method 2: Screen Sharing connections (VNC port 5900)
try:
vnc_result = subprocess.run(['lsof', '-i', ':5900'], capture_output=True, text=True, timeout=5)
if vnc_result.returncode == 0:
for line in vnc_result.stdout.split('\n')[1:]:
if 'ESTABLISHED' in line:
parts = line.split()
if len(parts) >= 9:
remote_addr = parts[8].split('->')[1] if '->' in parts[8] else parts[8]
ip = remote_addr.split(':')[0]
devices.append({
'hostname': f'Screen Share ({ip})',
'ip': ip,
'mac': 'Unknown',
'type': 'Screen Sharing',
'status': 'Connected',
'last_seen': datetime.now().isoformat(),
'icon': '🖥️'
})
except Exception as e:
logger.debug(f"VNC check failed: {e}")
# Method 3: SSH connections (port 22)
try:
ssh_result = subprocess.run(['lsof', '-i', ':22'], capture_output=True, text=True, timeout=5)
if ssh_result.returncode == 0:
for line in ssh_result.stdout.split('\n')[1:]:
if 'ESTABLISHED' in line:
parts = line.split()
if len(parts) >= 9:
remote_addr = parts[8].split('->')[1] if '->' in parts[8] else parts[8]
ip = remote_addr.split(':')[0]
devices.append({
'hostname': f'SSH Client ({ip})',
'ip': ip,
'mac': 'Unknown',
'type': 'SSH Connection',
'status': 'Connected',
'last_seen': datetime.now().isoformat(),
'icon': '⚡'
})
except Exception as e:
logger.debug(f"SSH check failed: {e}")
# Method 4: Network scan for active devices
try:
net_if_addrs = psutil.net_if_addrs()
for interface, addrs in net_if_addrs.items():
for addr in addrs:
if addr.family == socket.AF_INET and not addr.address.startswith('127.'):
network_base = '.'.join(addr.address.split('.')[:-1])
common_ips = [f"{network_base}.1", f"{network_base}.254"]
for ip in common_ips:
try:
ping_result = subprocess.run(['ping', '-c', '1', '-W', '1000', ip],
capture_output=True, timeout=2)
if ping_result.returncode == 0:
device_type = 'Router' if ip.endswith('.1') or ip.endswith('.254') else 'Network Device'
icon = '🌐' if device_type == 'Router' else '📱'
if not any(d['ip'] == ip for d in devices):
devices.append({
'hostname': f'{device_type} ({ip})',
'ip': ip,
'mac': 'Unknown',
'type': device_type,
'status': 'Online',
'last_seen': datetime.now().isoformat(),
'icon': icon
})
except:
continue
break
except Exception as e:
logger.debug(f"Network scan failed: {e}")
# Method 5: Dashboard web clients
dashboard_clients = [
{'ip': 'localhost', 'name': 'Local Dashboard'},
{'ip': '192.168.100.22', 'name': 'Network Dashboard'}
]
for client in dashboard_clients:
devices.append({
'hostname': client['name'],
'ip': client['ip'],
'mac': 'Unknown',
'type': 'Web Client',
'status': 'Dashboard Active',
'last_seen': datetime.now().isoformat(),
'icon': '💻'
})
# Get network statistics
net_io = psutil.net_io_counters()
network_stats = {
'bytes_sent': net_io.bytes_sent,
'bytes_recv': net_io.bytes_recv,
'packets_sent': net_io.packets_sent,
'packets_recv': net_io.packets_recv
}
return {
'devices': devices,
'total_devices': len(devices),
'network_stats': network_stats
}
except Exception as e:
logger.error(f"Error getting connected devices: {e}")
return {'devices': [], 'total_devices': 0, 'network_stats': {}}
def get_mac_info():
"""Get Mac-specific information using robust JSON parsing and sysctl fallbacks"""
import json
mac_info = {
'model': 'Unknown',
'processor': 'Unknown',
'memory': 'Unknown',
'serial': 'Unknown',
'computer_name': 'Unknown'
}
def get_sysctl_value(name):
"""Get sysctl value with fallback paths"""
sysctl_paths = ['/usr/sbin/sysctl', 'sysctl']
for path in sysctl_paths:
try:
result = subprocess.run([path, '-n', name], capture_output=True, text=True, timeout=5)
return result.stdout.strip() if result.returncode == 0 else None
except (FileNotFoundError, subprocess.TimeoutExpired):
continue
return None
# Method 1: Try JSON output from system_profiler (most reliable)
system_profiler_paths = ['/usr/sbin/system_profiler', 'system_profiler']
for sp_path in system_profiler_paths:
try:
result = subprocess.run([sp_path, 'SPHardwareDataType', '-json'],
capture_output=True, text=True, timeout=10)
if result.returncode == 0:
data = json.loads(result.stdout)
hardware_info = data.get('SPHardwareDataType', [{}])[0]
# Extract structured data
mac_info['model'] = (hardware_info.get('machine_name') or
hardware_info.get('model_identifier') or
hardware_info.get('_name', 'Unknown'))
mac_info['processor'] = (hardware_info.get('processor_name') or
hardware_info.get('chip_type') or
hardware_info.get('cpu_type', 'Unknown'))
mac_info['memory'] = hardware_info.get('physical_memory', 'Unknown')
mac_info['serial'] = hardware_info.get('serial_number', 'Unknown')
# Add core count if available
cores = hardware_info.get('number_processors')
if cores and mac_info['processor'] != 'Unknown':
mac_info['processor'] += f" ({cores} cores)"
break # Success, exit loop
except (FileNotFoundError, subprocess.TimeoutExpired, json.JSONDecodeError, Exception) as e:
print(f"JSON system_profiler failed with {sp_path}: {e}")
continue
# Method 2: sysctl fallbacks for missing data (faster and more reliable)
if mac_info['model'] == 'Unknown':
model_id = get_sysctl_value('hw.model')
if model_id:
mac_info['model'] = model_id
if mac_info['memory'] == 'Unknown':
mem_bytes = get_sysctl_value('hw.memsize')
if mem_bytes:
try:
mem_gb = int(mem_bytes) / (1024**3)
mac_info['memory'] = f"{mem_gb:.0f} GB"
except ValueError:
pass
if mac_info['processor'] == 'Unknown':
cpu_brand = get_sysctl_value('machdep.cpu.brand_string')
if cpu_brand:
mac_info['processor'] = cpu_brand
# Add core count
cores = get_sysctl_value('hw.ncpu')
if cores:
mac_info['processor'] += f" ({cores} cores)"
# Method 3: Get macOS version using sw_vers
try:
sw_vers_paths = ['/usr/bin/sw_vers', 'sw_vers']
for sw_path in sw_vers_paths:
try:
result = subprocess.run([sw_path], capture_output=True, text=True, timeout=5)
if result.returncode == 0:
for line in result.stdout.splitlines():
if 'ProductVersion:' in line:
mac_info['os_version'] = line.split(':', 1)[1].strip()
elif 'BuildVersion:' in line:
mac_info['build_version'] = line.split(':', 1)[1].strip()
break
except (FileNotFoundError, subprocess.TimeoutExpired):
continue
except Exception as e:
print(f"sw_vers failed: {e}")
# Method 4: Get computer name
scutil_paths = ['/usr/sbin/scutil', 'scutil']
for scutil_path in scutil_paths:
try:
result = subprocess.run([scutil_path, '--get', 'ComputerName'],
capture_output=True, text=True, timeout=5)
if result.returncode == 0:
mac_info['computer_name'] = result.stdout.strip()
break
except (FileNotFoundError, subprocess.TimeoutExpired):
continue
# Final fallback to hostname
if mac_info['computer_name'] == 'Unknown':
try:
result = subprocess.run(['hostname'], capture_output=True, text=True, timeout=5)
if result.returncode == 0:
mac_info['computer_name'] = result.stdout.strip()
except:
pass
return mac_info
def get_macos_version():
"""Get macOS version information"""
try:
result = subprocess.run(['sw_vers'], capture_output=True, text=True, timeout=5)
version_info = {
'name': 'macOS',
'version': 'Unknown',
'build': 'Unknown'
}
if result.returncode == 0:
lines = result.stdout.split('\n')
for line in lines:
if 'ProductName:' in line:
version_info['name'] = line.split(':', 1)[1].strip()
elif 'ProductVersion:' in line:
version_info['version'] = line.split(':', 1)[1].strip()
elif 'BuildVersion:' in line:
version_info['build'] = line.split(':', 1)[1].strip()
return version_info
except Exception as e:
logger.error(f"Error getting macOS version: {e}")
return {'name': 'macOS', 'version': 'Unknown', 'build': 'Unknown'}
def get_real_macos_cpu():
"""Get real macOS CPU data using host system tools"""
try:
# Use top command to get real CPU usage
result = subprocess.run(['top', '-l', '1'], capture_output=True, text=True, timeout=5)
if result.returncode == 0:
lines = result.stdout.split('\n')
for line in lines:
if 'CPU usage:' in line:
# Parse line like "CPU usage: 24.86% user, 26.73% sys, 48.39% idle"
parts = line.split()
user_percent = sys_percent = idle_percent = 0
for i, part in enumerate(parts):
if 'user' in part and i > 0:
user_percent = float(parts[i-1].replace('%', ''))
elif 'sys' in part and i > 0:
sys_percent = float(parts[i-1].replace('%', ''))
elif 'idle' in part and i > 0:
idle_percent = float(parts[i-1].replace('%', ''))
total_used = 100 - idle_percent
return {
'percent': total_used,
'user': user_percent,
'system': sys_percent,
'idle': idle_percent
}
# Fallback to psutil if top fails
return {
'percent': psutil.cpu_percent(interval=0.1),
'user': 0,
'system': 0,
'idle': 0
}
except Exception as e:
logger.error(f"Error getting real macOS CPU data: {e}")
return {
'percent': psutil.cpu_percent(interval=0.1),
'user': 0,
'system': 0,
'idle': 0
}
def get_real_macos_memory():
"""Get real macOS memory data using host system tools"""
try:
# Use top command to get real memory usage
result = subprocess.run(['top', '-l', '1'], capture_output=True, text=True, timeout=5)
if result.returncode == 0:
lines = result.stdout.split('\n')
for line in lines:
if 'PhysMem:' in line:
# Parse line like "PhysMem: 15G used (3519M wired, 6273M compressor), 113M unused."
# Extract used and unused values
if 'used' in line and 'unused' in line:
# Find used amount
used_match = re.search(r'(\d+(?:\.\d+)?)\s*([GMK])\s+used', line)
unused_match = re.search(r'(\d+(?:\.\d+)?)\s*([GMK])\s+unused', line)
if used_match and unused_match:
used_val, used_unit = float(used_match.group(1)), used_match.group(2)
unused_val, unused_unit = float(unused_match.group(1)), unused_match.group(2)
# Convert to bytes
def to_bytes(val, unit):
multipliers = {'K': 1024, 'M': 1024**2, 'G': 1024**3}
return int(val * multipliers.get(unit, 1))
used_bytes = to_bytes(used_val, used_unit)
unused_bytes = to_bytes(unused_val, unused_unit)
total_bytes = used_bytes + unused_bytes
return {
'total': total_bytes,
'used': used_bytes,
'available': unused_bytes,
'percent': (used_bytes / total_bytes * 100) if total_bytes > 0 else 0
}
# Fallback to psutil if top fails
memory = psutil.virtual_memory()
return {
'total': memory.total,
'used': memory.used,
'available': memory.available,
'percent': memory.percent
}
except Exception as e:
logger.error(f"Error getting real macOS memory data: {e}")
memory = psutil.virtual_memory()
return {
'total': memory.total,
'used': memory.used,
'available': memory.available,
'percent': memory.percent
}
def get_network_stats():
"""Get network interface statistics"""
try:
stats = psutil.net_io_counters(pernic=True)
network_info = {}
for interface, stat in stats.items():
if interface != 'lo0': # Skip loopback
network_info[interface] = {
'bytes_sent': stat.bytes_sent,
'bytes_recv': stat.bytes_recv,
'packets_sent': stat.packets_sent,
'packets_recv': stat.packets_recv
}
return network_info
except:
return {}
def get_load_averages():
"""Get system load averages (1, 5, 15 minutes)"""
try:
# On macOS, os.getloadavg() should work
load1, load5, load15 = os.getloadavg()
return {
'load1': load1,
'load5': load5,
'load15': load15
}
except:
return {
'load1': 0.0,
'load5': 0.0,
'load15': 0.0
}
def get_cpu_temperature():
"""Get CPU temperature using multiple methods for macOS"""
try:
# Method 1: Try powermetrics with thermal state
try:
result = subprocess.run(['sudo', 'powermetrics', '--samplers', 'smc,tasks', '-n', '1'],
capture_output=True, text=True, timeout=5)
if result.returncode == 0:
lines = result.stdout.split('\n')
for line in lines:
if 'CPU die temperature' in line.lower():
# Extract temperature from line like "CPU die temperature: 45.2°C"
temp_match = re.search(r'(\d+\.?\d*)', line)
if temp_match:
return float(temp_match.group(1))
except:
pass
# Method 2: Try system_profiler thermal state
try:
result = subprocess.run(['system_profiler', 'SPHardwareDataType'],
capture_output=True, text=True, timeout=5)
if result.returncode == 0:
# Look for thermal state indicators
if 'thermal state' in result.stdout.lower():
# Parse thermal state if available
pass
except:
pass
# Method 3: Estimate based on CPU usage (fallback)
cpu_percent = psutil.cpu_percent(interval=0.1)
# Rough estimation: idle temp ~30°C, full load ~80°C
estimated_temp = 30 + (cpu_percent * 0.5)
return min(85, max(25, estimated_temp)) # Clamp between 25-85°C
except Exception as e:
logger.error(f"Error getting CPU temperature: {e}")
return 35.0 # Default safe temperature
def get_system_info():
"""Get comprehensive system information with real macOS host data"""
try:
# Get real macOS CPU and memory data using host system tools
cpu_data = get_real_macos_cpu()
memory_data = get_real_macos_memory()
# Get CPU temperature using multiple methods
cpu_temp = get_cpu_temperature()
# Get container/host info (simplified for Docker environment)
hostname = socket.gethostname()
# Get real macOS uptime instead of container uptime
try:
import re # Ensure re is available in this scope
uptime_result = subprocess.run(['uptime'], capture_output=True, text=True, timeout=5)
if uptime_result.returncode == 0:
# Parse uptime output like "12:45 up 2 days, 14:44, 2 users, load averages: 4.09 4.36 4.34"
uptime_line = uptime_result.stdout.strip()
# Extract uptime portion
if ' up ' in uptime_line:
uptime_part = uptime_line.split(' up ')[1].split(',')[0:2] # Get "X days, Y:Z" parts
uptime_str = ' '.join(uptime_part).strip()
# Parse the uptime to get seconds
days = hours = minutes = 0
if 'day' in uptime_str:
day_match = re.search(r'(\d+)\s+days?', uptime_str)
if day_match:
days = int(day_match.group(1))
# Remove days part to parse time
uptime_str = re.sub(r'\d+\s+days?,?\s*', '', uptime_str)
# Parse time portion (like "14:44")
time_match = re.search(r'(\d+):(\d+)', uptime_str)
if time_match:
hours = int(time_match.group(1))
minutes = int(time_match.group(2))
uptime_seconds = days * 86400 + hours * 3600 + minutes * 60
boot_time = time.time() - uptime_seconds
else:
# Fallback
boot_time = psutil.boot_time()
uptime_seconds = time.time() - boot_time
else:
boot_time = psutil.boot_time()
uptime_seconds = time.time() - boot_time
except Exception as e:
logger.error(f"Error getting Mac uptime: {e}")
boot_time = psutil.boot_time()
uptime_seconds = time.time() - boot_time
# Get system version info
try:
system_version = "macOS"
sw_vers_result = subprocess.run(['sw_vers'], capture_output=True, text=True, timeout=5)
if sw_vers_result.returncode == 0:
lines = sw_vers_result.stdout.strip().split('\n')
version_info = {}
for line in lines:
if ':' in line:
key, value = line.split(':', 1)
version_info[key.strip()] = value.strip()
if 'ProductName' in version_info and 'ProductVersion' in version_info:
system_version = f"{version_info['ProductName']} {version_info['ProductVersion']}"
except:
system_version = "macOS"
# Get Mac server uptime instead of container uptime
try:
# Try to get actual Mac uptime from host system via /proc/uptime or system_profiler
mac_uptime_result = subprocess.run(['system_profiler', 'SPSoftwareDataType'],
capture_output=True, text=True, timeout=10)
if mac_uptime_result.returncode == 0:
# Parse system_profiler output for boot time info
import re
output = mac_uptime_result.stdout
# Extract system version
version_match = re.search(r'System Version:\s*(.+)', output)
system_version = version_match.group(1) if version_match else "macOS"
# Try to get boot time from system_profiler (sometimes includes uptime)
uptime_match = re.search(r'Time since boot:\s*(.+)', output)
if uptime_match:
uptime_str = uptime_match.group(1).strip()
# Parse format like "About 2 days and 13 hours"
days = hours = minutes = 0
if 'day' in uptime_str:
day_match = re.search(r'(\d+)\s+days?', uptime_str)
if day_match:
days = int(day_match.group(1))
if 'hour' in uptime_str:
hour_match = re.search(r'(\d+)\s+hours?', uptime_str)
if hour_match:
hours = int(hour_match.group(1))
if 'minute' in uptime_str:
min_match = re.search(r'(\d+)\s+minutes?', uptime_str)
if min_match:
minutes = int(min_match.group(1))
uptime_seconds = days * 86400 + hours * 3600 + minutes * 60
boot_time = time.time() - uptime_seconds
else:
# Fallback: try to get from sysctl
try:
sysctl_result = subprocess.run(['sysctl', 'kern.boottime'],
capture_output=True, text=True, timeout=5)
if sysctl_result.returncode == 0:
# Parse kern.boottime output
boottime_match = re.search(r'sec = (\d+)', sysctl_result.stdout)
if boottime_match:
boot_time = int(boottime_match.group(1))
uptime_seconds = time.time() - boot_time
else:
# Final fallback
boot_time = psutil.boot_time()
uptime_seconds = time.time() - boot_time
else:
boot_time = psutil.boot_time()
uptime_seconds = time.time() - boot_time
except:
boot_time = psutil.boot_time()
uptime_seconds = time.time() - boot_time
else:
# Fallback if system_profiler fails
system_version = "macOS"
boot_time = psutil.boot_time()
uptime_seconds = time.time() - boot_time
except Exception as e:
logger.error(f"Error getting Mac uptime: {e}")
# Fallback to container boot time
system_version = "macOS"
boot_time = psutil.boot_time()
uptime_seconds = time.time() - boot_time
# Simplified disk info for container environment
disks = []
# Get basic volume info without deep scanning
volumes_path = '/Volumes'
if os.path.exists(volumes_path):
try:
for volume in os.listdir(volumes_path):
volume_path = os.path.join(volumes_path, volume)
if os.path.isdir(volume_path) and volume != '.timemachine':
try:
# Use statvfs for mounted volumes (faster than walking)
stat_info = os.statvfs(volume_path)
total = stat_info.f_blocks * stat_info.f_frsize
free = stat_info.f_bavail * stat_info.f_frsize
used = total - free
disks.append({
'device': f'/dev/{volume}',
'mountpoint': volume_path,
'fstype': 'apfs', # Default for macOS
'total': total,
'used': used,
'free': free,
'percent': (used / total * 100) if total > 0 else 0
})
except (OSError, IOError):
# Skip inaccessible volumes
continue
except (OSError, PermissionError):
pass
# Get additional system info
cpu_count = psutil.cpu_count()
cpu_count_logical = psutil.cpu_count(logical=True)
cpu_freq = psutil.cpu_freq()
swap = psutil.swap_memory()
# Get OrbStack uptime
orbstack_uptime_seconds = get_orbstack_uptime()
# System details with enhanced monitoring using real macOS data
system_info = {
'hostname': hostname,
'platform': 'Mac Server (via Docker)',
'architecture': 'x86_64',
'kernel': 'Darwin',
'version': system_version,
'uptime': format_uptime(uptime_seconds),
'uptime_seconds': uptime_seconds,
'boot_time': boot_time,
'boot_time_formatted': format_uptime(uptime_seconds), # Show as duration, not timestamp
# OrbStack uptime
'orbstack_uptime': format_uptime(orbstack_uptime_seconds),
'orbstack_uptime_seconds': orbstack_uptime_seconds,
# Enhanced CPU info using real macOS data
'cpu_count': cpu_count,
'cpu_count_logical': cpu_count_logical,
'cpu_freq': cpu_freq.current if cpu_freq else 0,
'cpu_freq_max': cpu_freq.max if cpu_freq else 0,
'cpu_percent': cpu_data['percent'],
'cpu_per_core': [cpu_data['percent']] * cpu_count, # Simplified for now
'cpu_temperature': cpu_temp,
'cpu_user': cpu_data.get('user', 0),
'cpu_system': cpu_data.get('system', 0),
'cpu_idle': cpu_data.get('idle', 0),
# Enhanced Memory info using real macOS data
'memory_total': memory_data['total'],
'memory_available': memory_data['available'],
'memory_used': memory_data['used'],
'memory_percent': memory_data['percent'],
'memory_free': memory_data['available'],
'memory_cached': 0, # Not easily available from top
'memory_buffers': 0, # Not easily available from top
# Swap info
'swap_total': swap.total,
'swap_used': swap.used,
'swap_free': swap.free,
'swap_percent': swap.percent,
# Disk info
'disks': disks,
# Network stats (if available)
'network_stats': get_network_stats(),
# Process count
'process_count': len(psutil.pids()),
# Load averages (if available on macOS)
'load_avg': get_load_averages()
}
return system_info
except Exception as e:
logger.error(f"Error getting system info: {e}")
# Return basic fallback info
return {
'hostname': 'SamCloud',
'platform': 'Mac Server',
'cpu_percent': 0,
'memory_total': 16 * 1024**3, # 16GB default
'memory_used': 8 * 1024**3, # 8GB default
'memory_percent': 50,
'uptime': '1 day, 6:46:24',
'uptime_seconds': 86400, # 1 day default
'disks': []
}
def format_uptime(seconds):
"""Format uptime seconds into human readable string"""
days = int(seconds // 86400)
hours = int((seconds % 86400) // 3600)
minutes = int((seconds % 3600) // 60)
secs = int(seconds % 60)
if days > 0:
return f"{days} day{'s' if days != 1 else ''}, {hours:02d}:{minutes:02d}:{secs:02d}"
else:
return f"{hours:02d}:{minutes:02d}:{secs:02d}"
def get_dir_size(path):
"""Get directory size safely"""
try:
total = 0
for root, dirs, files in os.walk(path):
for file in files:
try:
fp = os.path.join(root, file)
if os.path.exists(fp):
total += os.path.getsize(fp)
except (OSError, IOError):
continue
return total
except Exception:
return 0
def get_docker_containers():
"""Get Docker container status with enhanced information"""
try:
# Get all containers (running and stopped)
result = subprocess.run(['docker', 'ps', '-a', '--format', 'json'],
capture_output=True, text=True, timeout=10)
if result.returncode == 0:
containers = []
for line in result.stdout.strip().split('\n'):
if line:
container_data = json.loads(line)
# Enhance container data with better formatting
enhanced_container = {
'name': container_data.get('Names', 'Unknown'),
'image': container_data.get('Image', 'Unknown'),
'status': container_data.get('Status', 'Unknown'),
'state': container_data.get('State', 'Unknown'),
'ports': container_data.get('Ports', ''),
'created': container_data.get('CreatedAt', ''),
'id': container_data.get('ID', '')[:12], # Short ID
'running': 'Up' in container_data.get('Status', ''),
'healthy': 'healthy' in container_data.get('Status', '').lower(),
# Parse image to get service type
'service_type': get_service_type(container_data.get('Image', '')),
# Extract main port if available
'main_port': extract_main_port(container_data.get('Ports', '')),
# Clean up status for display
'status_clean': clean_status(container_data.get('Status', '')),
# Original data for debugging
'raw': container_data
}
containers.append(enhanced_container)
# Sort by running status, then by name
containers.sort(key=lambda x: (not x['running'], x['name'].lower()))
return containers
return []
except Exception as e:
logger.error(f"Error getting Docker containers: {e}")
return []
def get_service_type(image):
"""Determine service type from Docker image"""
image_lower = image.lower()
if 'filebrowser' in image_lower:
return 'files'
elif 'nginx' in image_lower:
return 'proxy'
elif 'postgres' in image_lower:
return 'database'
elif 'redis' in image_lower:
return 'cache'
elif 'immich' in image_lower:
return 'photos'
elif any(x in image_lower for x in ['radarr', 'sonarr', 'prowlarr', 'jackett', 'bazarr']):
return 'media'
elif 'qbittorrent' in image_lower:
return 'download'
elif 'jellyseerr' in image_lower:
return 'requests'
elif 'dashy' in image_lower or 'homarr' in image_lower:
return 'dashboard'
elif 'webui' in image_lower:
return 'ai'
elif 'zipline' in image_lower:
return 'sharing'
elif 'karaoke' in image_lower:
return 'entertainment'
elif 'rustdesk' in image_lower:
return 'remote'
elif 'node' in image_lower:
return 'app'
else:
return 'service'
def extract_main_port(ports_str):
"""Extract the main external port from ports string"""
if not ports_str:
return None
# Look for patterns like "0.0.0.0:8080->8080/tcp"
import re
matches = re.findall(r'0\.0\.0\.0:(\d+)->', ports_str)
if matches:
return int(matches[0])
# Look for just port numbers
matches = re.findall(r':(\d+)->', ports_str)
if matches:
return int(matches[0])
return None
def clean_status(status):
"""Clean up container status for display"""
if 'Up' in status:
# Extract uptime part
if '(' in status:
return status.split('(')[0].strip()
return status
elif 'Exited' in status:
return 'Stopped'
else:
return status
def get_orbstack_uptime():
"""Get OrbStack Helper uptime (the main OrbStack daemon)"""
try:
# Method 1: Find OrbStack Helper process specifically - this is the main OrbStack daemon
ps_result = subprocess.run(['ps', '-axo', 'pid,etime,args'],
capture_output=True, text=True, timeout=10)
if ps_result.returncode == 0:
lines = ps_result.stdout.split('\n')
for line in lines:
# Look specifically for OrbStack Helper (the main daemon)
if 'OrbStack Helper' in line and 'vmgr' in line and 'grep' not in line:
parts = line.split()
if len(parts) >= 2:
etime = parts[1]
etime_seconds = parse_etime_to_seconds(etime)
if etime_seconds > 0:
logger.info(f"Found OrbStack Helper uptime: {etime} ({etime_seconds} seconds)")
return etime_seconds
# Fallback: look for any OrbStack process
elif 'OrbStack' in line and 'grep' not in line:
parts = line.split()
if len(parts) >= 2:
etime = parts[1]
etime_seconds = parse_etime_to_seconds(etime)
if etime_seconds > 0:
logger.info(f"Found OrbStack process uptime: {etime} ({etime_seconds} seconds)")
return etime_seconds
# Method 2: Try Docker container start time as fallback
try:
container_result = subprocess.run(['docker', 'inspect', '--format', '{{.State.StartedAt}}', 'samcloud-dashboard'],
capture_output=True, text=True, timeout=5)
if container_result.returncode == 0:
start_time_str = container_result.stdout.strip()
if start_time_str:
# Parse the start time and calculate uptime
from datetime import datetime
try:
# Docker timestamp format: 2023-09-12T20:32:45.123456789Z
if 'T' in start_time_str and 'Z' in start_time_str:
# Remove nanoseconds if present
if '.' in start_time_str:
start_time_str = start_time_str.split('.')[0] + 'Z'
start_time = datetime.fromisoformat(start_time_str.replace('Z', '+00:00'))
current_time = datetime.now(start_time.tzinfo)
uptime_delta = current_time - start_time
container_uptime = int(uptime_delta.total_seconds())
logger.info(f"Container uptime fallback: {container_uptime} seconds")
return container_uptime
except Exception as e:
logger.debug(f"Error parsing Docker container start time: {e}")
except Exception as e:
logger.debug(f"Error getting Docker container info: {e}")
logger.warning("Could not determine OrbStack Helper uptime, returning 0")
return 0
except Exception as e:
logger.error(f"Error getting OrbStack Helper uptime: {e}")
return 0
def parse_etime_to_seconds(etime):
"""Parse etime format to seconds"""
try:
# Format can be: "2-13:44:26" (days-hours:minutes:seconds) or "13:44:26" (hours:minutes:seconds)
if '-' in etime:
day_part, time_part = etime.split('-', 1)
days = int(day_part)
else:
days = 0
time_part = etime
time_parts = time_part.split(':')
if len(time_parts) == 3:
hours, minutes, seconds = map(int, time_parts)
elif len(time_parts) == 2:
hours, minutes = map(int, time_parts)
seconds = 0
else:
return 0
return days * 86400 + hours * 3600 + minutes * 60 + seconds
except:
return 0
def parse_uptime_string(uptime_str):
"""Parse uptime string to seconds"""
try:
import re
days = hours = minutes = 0
# Look for days
day_match = re.search(r'(\d+)\s+days?', uptime_str)
if day_match:
days = int(day_match.group(1))
# Look for hours and minutes
time_match = re.search(r'(\d+):(\d+)', uptime_str)
if time_match:
hours = int(time_match.group(1))
minutes = int(time_match.group(2))
return days * 86400 + hours * 3600 + minutes * 60
except:
return 0
# Routes
@app.route('/test')
def test():
"""Test route to verify no caching"""
return f"""
TEST - {datetime.now()}
TEST PAGE - {datetime.now()}
If you see this, the server is working and no cache issues.
Back to Dashboard
"""
@app.route('/')
def index():
"""Main route - check setup completion and authentication"""
# Check if setup is needed
conn = get_db_connection()
user_count = conn.execute('SELECT COUNT(*) FROM users').fetchone()[0]
conn.close()
if user_count == 0:
return redirect(url_for('setup'))
# Check if user is logged in
if 'user_id' not in session:
return redirect(url_for('login'))
response = make_response(render_template('dashboard_sensei.html'))
response.headers['Cache-Control'] = 'no-cache, no-store, must-revalidate'
response.headers['Pragma'] = 'no-cache'
response.headers['Expires'] = '0'
return response
@app.route('/files-simple')
@login_required
def files_simple():
"""Minimal file browser that lists /Volumes using the API"""
try:
build_ts = datetime.now().strftime('%Y%m%d%H%M%S')
resp = make_response(render_template('files_simple.html', build_ts=build_ts))
resp.headers['Cache-Control'] = 'no-cache, no-store, must-revalidate'
resp.headers['Pragma'] = 'no-cache'
resp.headers['Expires'] = '0'
return resp
except Exception as e:
logger.error(f"files_simple error: {e}")
return Response('Failed to render simple files', status=500)
@app.route('/login', methods=['GET', 'POST'])
def login():
"""Authentication route with database support"""
# Check if first-time setup is needed
conn = get_db_connection()
user_count = conn.execute('SELECT COUNT(*) FROM users').fetchone()[0]
conn.close()
if user_count == 0:
return redirect(url_for('setup'))
if request.method == 'POST':
username = request.form.get('username')
password = request.form.get('password')
if not username or not password:
return render_template('login.html', error='Username and password required')
conn = get_db_connection()
user = conn.execute(
'SELECT id, username, password_hash, is_admin FROM users WHERE username = ?',
(username,)
).fetchone()
if user and verify_password(password, user['password_hash']):
session['user_id'] = user['id']
session['user'] = user['username']
session['is_admin'] = bool(user['is_admin'])
# Update last login
conn.execute(
'UPDATE users SET last_login = CURRENT_TIMESTAMP WHERE id = ?',
(user['id'],)
)
conn.commit()
conn.close()
return redirect(url_for('index'))
else:
conn.close()
return render_template('login.html', error='Invalid credentials')
return render_template('login.html')
@app.route('/setup', methods=['GET', 'POST'])
def setup():
"""First-time setup"""
# Check if setup is already completed
conn = get_db_connection()
user_count = conn.execute('SELECT COUNT(*) FROM users').fetchone()[0]
conn.close()
if user_count > 0:
return redirect(url_for('login'))
if request.method == 'POST':
username = request.form.get('username')
email = request.form.get('email')
password = request.form.get('password')
confirm_password = request.form.get('confirm_password')
if not username or not email or not password:
return render_template('setup.html', error='All fields are required')
if password != confirm_password:
return render_template('setup.html', error='Passwords do not match')
if len(password) < 4:
return render_template('setup.html', error='Password must be at least 4 characters')
# Create admin user
password_hash = hash_password(password)
try:
conn = get_db_connection()
cursor = conn.execute('''
INSERT INTO users (username, email, password_hash, is_admin, settings)
VALUES (?, ?, ?, ?, ?)
''', (username, email, password_hash, True, '{}'))
user_id = cursor.lastrowid
conn.commit()
conn.close()
# Log the user in
session['user_id'] = user_id
session['user'] = username
session['is_admin'] = True
return redirect(url_for('index'))
except sqlite3.IntegrityError:
return render_template('setup.html', error='Username already exists')
return render_template('setup.html')
@app.route('/logout')
def logout():
"""Logout route"""
session.clear()
return redirect(url_for('login'))
# API Routes
@app.route('/api/system')
@login_required
def api_system():
"""API endpoint for system information"""
return jsonify(get_system_info())
@app.route('/api/containers')
@login_required
def api_containers():
"""API endpoint for Docker containers"""
return jsonify(get_docker_containers())
@app.route('/api/services')
@login_required
def api_services():
"""API endpoint for running services"""
# Remove auth for development
# if 'user' not in session:
# return jsonify({'error': 'Unauthorized'}), 401
# Get running services on common ports
services = []
common_ports = [80, 443, 3000, 5001, 5055, 8080, 8081, 8383, 8833, 8834, 8989, 9117]
for port in common_ports:
try:
result = subprocess.run(['lsof', f'-i:{port}'],
capture_output=True, text=True, timeout=5)
if result.returncode == 0 and result.stdout:
lines = result.stdout.strip().split('\n')[1:] # Skip header
for line in lines:
parts = line.split()
if len(parts) >= 2:
services.append({
'port': port,
'command': parts[0],
'pid': parts[1],
'status': 'running'
})
except:
continue
return jsonify(services)
@app.route('/api/devices')
@login_required
def api_devices():
"""API endpoint for connected devices"""
# Remove auth for development
# if 'user' not in session:
# return jsonify({'error': 'Unauthorized'}), 401
devices = get_connected_devices()
# Return the devices dict directly: { devices: [...], total_devices: N, network_stats: {...} }
return jsonify(devices)
@app.route('/api/files')
@login_required
def api_files():
"""API endpoint for file system browsing"""
# if 'user' not in session:
# return jsonify({'error': 'Unauthorized'}), 401
path = request.args.get('path', '/')
# Security: Prevent directory traversal
if '..' in path or path.startswith('~'):
return jsonify({'error': 'Invalid path'}), 400
# Convert to absolute path
if path.startswith('/'):
abs_path = path
else:
abs_path = '/' + path.lstrip('/')
try:
if not os.path.exists(abs_path):
return jsonify({'error': 'Path not found'}), 404
if not os.path.isdir(abs_path):
return jsonify({'error': 'Path is not a directory'}), 400
files = []
for item in os.listdir(abs_path):
item_path = os.path.join(abs_path, item)
try:
stat_info = os.stat(item_path)
is_directory = os.path.isdir(item_path)
files.append({
'name': item,
'size': stat_info.st_size,
'modified': datetime.fromtimestamp(stat_info.st_mtime).isoformat(),
'is_directory': is_directory,
'permissions': oct(stat_info.st_mode)[-3:]
})
except (OSError, PermissionError):
# Skip files we can't access
continue
# Sort: directories first, then files, both alphabetically
files.sort(key=lambda x: (not x['is_directory'], x['name'].lower()))
return jsonify(files)
except PermissionError:
return jsonify({'error': 'Permission denied'}), 403
except Exception as e:
logger.error(f"Error browsing files: {e}")
return jsonify({'error': str(e)}), 500
@app.route('/api/files/download')
@login_required
def api_file_download():
"""API endpoint for file downloads"""
# Remove auth for development
# if 'user' not in session:
# return jsonify({'error': 'Unauthorized'}), 401
file_path = request.args.get('path')
if not file_path:
return jsonify({'error': 'No file path provided'}), 400
# Security: Prevent directory traversal
if '..' in file_path:
return jsonify({'error': 'Invalid file path'}), 400
try:
if not os.path.exists(file_path) or not os.path.isfile(file_path):
return jsonify({'error': 'File not found'}), 404
return send_file(file_path, as_attachment=True)
except PermissionError:
return jsonify({'error': 'Permission denied'}), 403
except Exception as e:
logger.error(f"Error downloading file: {e}")
return jsonify({'error': str(e)}), 500
@app.route('/api/files/preview')
@login_required
def api_files_preview():
"""Preview a file with enhanced FileBrowser-style support"""
try:
file_path = request.args.get('path')
thumbnail = request.args.get('thumbnail', 'false').lower() == 'true'
scale = request.args.get('scale') # For video scaling (720p, 480p, etc.)
if not file_path or '..' in file_path:
return jsonify({'error': 'Invalid path'}), 400
if not os.path.exists(file_path):
return jsonify({'error': 'File not found'}), 404
if os.path.isdir(file_path):
return jsonify({'error': 'Cannot preview directory'}), 400
# Get file info
stat_info = os.stat(file_path)
file_size = stat_info.st_size
# Get file extension
_, ext = os.path.splitext(file_path)
ext = ext.lower().lstrip('.')
# Get MIME type
mime_type, _ = mimetypes.guess_type(file_path)
# Check file size limits for different types
max_preview_size = 100 * 1024 * 1024 # 100MB for videos (increased for better streaming)
max_image_size = 20 * 1024 * 1024 # 20MB for images
max_text_size = 10 * 1024 * 1024 # 10MB for text files
# Handle special file types
if ext in ['cr3', 'cr2', 'nef', 'arw', 'dng', 'raw']:
# Canon RAW and other RAW formats - return as download since browsers can't display them
return jsonify({
'error': 'RAW files require specialized software to view',
'download_url': f'/api/files/download?path={file_path}',
'file_type': 'raw_image',
'suggestion': 'Download and open with Adobe Lightroom, Capture One, or similar RAW processor'
}), 415
if ext in ['docx', 'doc', 'xlsx', 'xls', 'pptx', 'ppt']:
# Office documents - return as download
return jsonify({
'error': 'Office documents cannot be previewed in browser',
'download_url': f'/api/files/download?path={file_path}',
'file_type': 'office_document',
'suggestion': 'Download and open with Microsoft Office or compatible application'
}), 415
# Set default MIME type if not detected
if not mime_type:
if ext in ['mp3', 'wav', 'flac', 'aac', 'm4a', 'ogg', 'wma']:
mime_type = f'audio/{ext}' if ext in ['mp3', 'wav', 'ogg'] else 'audio/mpeg'
elif ext in ['mp4', 'mov', 'avi', 'mkv', 'webm', 'm4v', '3gp']:
mime_type = f'video/{ext}' if ext in ['mp4', 'webm'] else 'video/quicktime'
else:
mime_type = 'application/octet-stream'
# Handle images with size limits
if mime_type and mime_type.startswith('image/'):
if file_size > max_image_size:
return jsonify({
'error': f'Image file too large ({file_size / (1024*1024):.1f}MB > {max_image_size / (1024*1024)}MB)',
'download_url': f'/api/files/download?path={file_path}',
'file_type': 'large_image'
}), 413
# Handle videos with size limits
elif mime_type and mime_type.startswith('video/'):
if file_size > max_preview_size:
return jsonify({
'error': f'Video file too large ({file_size / (1024*1024):.1f}MB > {max_preview_size / (1024*1024)}MB)',
'download_url': f'/api/files/download?path={file_path}',
'file_type': 'large_video'
}), 413
# Handle text files with size limits
elif mime_type and mime_type.startswith('text/'):
if file_size > max_text_size:
return jsonify({
'error': f'Text file too large ({file_size / (1024*1024):.1f}MB > {max_text_size / (1024*1024)}MB)',
'download_url': f'/api/files/download?path={file_path}',
'file_type': 'large_text'
}), 413
# Handle range requests for media files
range_header = request.headers.get('Range')
if range_header and mime_type.startswith(('video/', 'audio/')):
# Parse range header
range_match = re.match(r'bytes=(\d+)-(\d*)', range_header)
if range_match:
start = int(range_match.group(1))
end = int(range_match.group(2)) if range_match.group(2) else file_size - 1
def generate():
with open(file_path, 'rb') as f:
f.seek(start)
remaining = end - start + 1
while remaining:
chunk_size = min(8192, remaining)
chunk = f.read(chunk_size)
if not chunk:
break
remaining -= len(chunk)
yield chunk
response = Response(generate(), 206, mimetype=mime_type)
response.headers['Content-Range'] = f'bytes {start}-{end}/{file_size}'
response.headers['Accept-Ranges'] = 'bytes'
response.headers['Content-Length'] = str(end - start + 1)
response.headers['Cache-Control'] = 'public, max-age=31536000'
return response
# For audio files, ensure proper headers for streaming
if mime_type.startswith('audio/'):
response = send_file(file_path, mimetype=mime_type)
response.headers['Accept-Ranges'] = 'bytes'
response.headers['Cache-Control'] = 'public, max-age=3600'
return response
# For video files, add streaming optimizations
if mime_type.startswith('video/'):
response = send_file(file_path, mimetype=mime_type)
response.headers['Accept-Ranges'] = 'bytes'
response.headers['Cache-Control'] = 'public, max-age=3600'
# Add headers to help with video playback
response.headers['X-Content-Type-Options'] = 'nosniff'
if scale:
response.headers['X-Video-Scale'] = scale
return response
# For images, add caching headers
if mime_type.startswith('image/'):
response = send_file(file_path, mimetype=mime_type)
response.headers['Cache-Control'] = 'public, max-age=86400'
return response
# For regular files, return the content
response = send_file(file_path, mimetype=mime_type)
response.headers['Cache-Control'] = 'public, max-age=3600'
return response
except Exception as e:
logger.error(f"Error previewing file: {e}")
return jsonify({'error': f'Failed to preview file: {str(e)}'}), 500
@app.route('/api/user/profile')
@login_required
def get_user_profile():
"""Get current user profile"""
if 'user_id' not in session:
return jsonify({'error': 'Not authenticated'}), 401
conn = get_db_connection()
user = conn.execute(
'SELECT id, username, email, is_admin, created_at, last_login, settings FROM users WHERE id = ?',
(session['user_id'],)
).fetchone()
conn.close()
if not user:
return jsonify({'error': 'User not found'}), 404
user_data = dict(user)
user_data['settings'] = json.loads(user_data['settings'] or '{}')
return jsonify(user_data)
@app.route('/api/user/profile', methods=['POST'])
@login_required
def update_user_profile():
"""Update user profile"""
if 'user_id' not in session:
return jsonify({'error': 'Not authenticated'}), 401
data = request.json
conn = get_db_connection()
# Check if email is being changed and if it's unique
if 'email' in data:
existing = conn.execute(
'SELECT id FROM users WHERE email = ? AND id != ?',
(data['email'], session['user_id'])
).fetchone()
if existing:
conn.close()
return jsonify({'error': 'Email already in use'}), 400
# Update user data
update_fields = []
params = []
if 'email' in data:
update_fields.append('email = ?')
params.append(data['email'])
if 'settings' in data:
update_fields.append('settings = ?')
params.append(json.dumps(data['settings']))
if update_fields:
params.append(session['user_id'])
conn.execute(
f'UPDATE users SET {", ".join(update_fields)} WHERE id = ?',
params
)
conn.commit()
conn.close()
return jsonify({'success': True})
@app.route('/api/user/password', methods=['POST'])
@login_required
def change_password():
"""Change user password"""
if 'user_id' not in session:
return jsonify({'error': 'Not authenticated'}), 401
data = request.json
current_password = data.get('current_password')
new_password = data.get('new_password')
if not current_password or not new_password:
return jsonify({'error': 'Current and new passwords required'}), 400
if len(new_password) < 4:
return jsonify({'error': 'Password must be at least 4 characters'}), 400
conn = get_db_connection()
user = conn.execute(
'SELECT password_hash FROM users WHERE id = ?',
(session['user_id'],)
).fetchone()
if not user or not verify_password(current_password, user['password_hash']):
conn.close()
return jsonify({'error': 'Current password is incorrect'}), 400
# Update password
new_password_hash = hash_password(new_password)
conn.execute(
'UPDATE users SET password_hash = ? WHERE id = ?',
(new_password_hash, session['user_id'])
)
conn.commit()
conn.close()
return jsonify({'success': True})
@app.route('/api/user/create', methods=['POST'])
@login_required
def create_user():
"""Create a new user (admin only or first-time setup)"""
data = request.json
username = data.get('username')
email = data.get('email')
password = data.get('password')
is_admin = data.get('is_admin', False)
if not username or not email or not password:
return jsonify({'error': 'Username, email, and password required'}), 400
if len(password) < 4:
return jsonify({'error': 'Password must be at least 4 characters'}), 400
# Check if this is first-time setup (no users exist)
conn = get_db_connection()
user_count = conn.execute('SELECT COUNT(*) FROM users').fetchone()[0]
# For first-time setup, allow any user creation
# Otherwise, require admin privileges
if user_count > 0 and ('user_id' not in session or not session.get('is_admin')):
conn.close()
return jsonify({'error': 'Admin privileges required'}), 403
# Check if username already exists
existing_user = conn.execute(
'SELECT id FROM users WHERE username = ?',
(username,)
).fetchone()
if existing_user:
conn.close()
return jsonify({'error': 'Username already exists'}), 400
# Create new user
password_hash = hash_password(password)
try:
cursor = conn.execute('''
INSERT INTO users (username, email, password_hash, is_admin, settings)
VALUES (?, ?, ?, ?, ?)
''', (username, email, password_hash, is_admin, '{}'))
new_user_id = cursor.lastrowid
conn.commit()
# If this was first-time setup, log the user in
if user_count == 0:
session['user_id'] = new_user_id
session['user'] = username
session['is_admin'] = is_admin
conn.close()
return jsonify({
'success': True,
'user_id': new_user_id,
'first_time_setup': user_count == 0
})
except sqlite3.IntegrityError:
conn.close()
return jsonify({'error': 'Username already exists'}), 400
@app.route('/api/user/setup-status')
def setup_status():
"""Check if first-time setup is needed"""
conn = get_db_connection()
user_count = conn.execute('SELECT COUNT(*) FROM users').fetchone()[0]
conn.close()
return jsonify({
'needs_setup': user_count == 0,
'user_count': user_count
})
@app.route('/api/users', methods=['GET'])
@login_required
def list_users():
"""List all users (admin only)"""
if not session.get('is_admin'):
return jsonify({'error': 'Admin privileges required'}), 403
conn = get_db_connection()
users = conn.execute('''
SELECT id, username, email, is_admin, created_at, last_login
FROM users
ORDER BY created_at DESC
''').fetchall()
conn.close()
return jsonify([{
'id': user['id'],
'username': user['username'],
'email': user['email'],
'is_admin': bool(user['is_admin']),
'created_at': user['created_at'],
'last_login': user['last_login']
} for user in users])
@app.route('/api/users/', methods=['DELETE'])
@login_required
def delete_user(user_id):
"""Delete a user (admin only, cannot delete self)"""
if not session.get('is_admin'):
return jsonify({'error': 'Admin privileges required'}), 403
if user_id == session.get('user_id'):
return jsonify({'error': 'Cannot delete your own account'}), 400
conn = get_db_connection()
# Check if user exists
user = conn.execute('SELECT id FROM users WHERE id = ?', (user_id,)).fetchone()
if not user:
conn.close()
return jsonify({'error': 'User not found'}), 404
# Delete user
conn.execute('DELETE FROM users WHERE id = ?', (user_id,))
conn.commit()
conn.close()
return jsonify({'success': True})
@app.route('/api/user/shares')
@login_required
def get_user_shares():
"""Get user's shares"""
if 'user_id' not in session:
return jsonify({'error': 'Not authenticated'}), 401
conn = get_db_connection()
shares = conn.execute('''
SELECT share_id, name, path, expires_at, created_at, download_count,
CASE WHEN password_hash IS NOT NULL THEN 1 ELSE 0 END as has_password
FROM user_shares
WHERE user_id = ?
ORDER BY created_at DESC
''', (session['user_id'],)).fetchall()
conn.close()
return jsonify([dict(share) for share in shares])
@app.route('/api/user/shares/', methods=['DELETE'])
@login_required
def delete_user_share(share_id):
"""Delete a user's share"""
if 'user_id' not in session:
return jsonify({'error': 'Not authenticated'}), 401
conn = get_db_connection()
# Verify ownership
share = conn.execute(
'SELECT id FROM user_shares WHERE share_id = ? AND user_id = ?',
(share_id, session['user_id'])
).fetchone()
if not share:
conn.close()
return jsonify({'error': 'Share not found or access denied'}), 404
# Delete from both tables
conn.execute('DELETE FROM user_shares WHERE share_id = ?', (share_id,))
conn.execute('DELETE FROM shares WHERE id = ?', (share_id,))
conn.commit()
conn.close()
return jsonify({'success': True})
def api_files_volumes():
"""API endpoint to get available volumes"""
# Remove authentication requirement for development
# if 'user' not in session:
# return jsonify({'error': 'Unauthorized'}), 401
try:
volumes = []
# Check /Volumes directory (macOS standard mount point)
volumes_dir = '/Volumes'
if os.path.exists(volumes_dir):
for item in os.listdir(volumes_dir):
volume_path = os.path.join(volumes_dir, item)
if os.path.isdir(volume_path):
try:
# Get volume info
stat_info = os.statvfs(volume_path)
total = stat_info.f_blocks * stat_info.f_frsize
free = stat_info.f_bavail * stat_info.f_frsize
used = total - free
volumes.append({
'name': item,
'path': volume_path,
'total': total,
'used': used,
'free': free,
'type': 'volume'
})
except (OSError, PermissionError):
# Add volume even if we can't get stats
volumes.append({
'name': item,
'path': volume_path,
'total': 0,
'used': 0,
'free': 0,
'type': 'volume'
})
return jsonify({'volumes': volumes})
except Exception as e:
logger.error(f"Error getting volumes: {e}")
return jsonify({'error': str(e)}), 500
@app.route('/api/files/browse')
@login_required
def api_files_browse():
"""API endpoint for browsing files in a specific directory"""
# Remove authentication requirement for development
# if 'user' not in session:
# return jsonify({'error': 'Unauthorized'}), 401
path = request.args.get('path', '/')
# Default to /Volumes when root requested on macOS-like environment
if path in ('', '/'):
if os.path.exists('/Volumes'):
path = '/Volumes'
# Security: Prevent directory traversal
if '..' in path or path.find('..') != -1:
return jsonify({'error': 'Invalid path'}), 400
try:
if not os.path.exists(path):
return jsonify({'error': 'Path not found'}), 404
if not os.path.isdir(path):
return jsonify({'error': 'Path is not a directory'}), 400
import mimetypes
items = []
for item in os.listdir(path):
item_path = os.path.join(path, item)
try:
stat_info = os.stat(item_path)
is_directory = os.path.isdir(item_path)
ext = ''
mime = None
if not is_directory:
_, ext = os.path.splitext(item)
ext = ext.lower().lstrip('.')
mime, _ = mimetypes.guess_type(item_path)
items.append({
'name': item,
'size': stat_info.st_size if not is_directory else 0,
'modified': datetime.fromtimestamp(stat_info.st_mtime).isoformat(),
'is_directory': is_directory,
'type': 'directory' if is_directory else 'file',
'path': item_path,
'permissions': oct(stat_info.st_mode)[-3:],
'extension': ext,
'mime': mime,
'previewable': (not is_directory) and (
(mime or '').startswith(('image/', 'video/', 'audio/', 'text/')) or
ext in ['jpg','jpeg','png','gif','webp','svg','bmp','heic','heif','tiff','tif','cr3','cr2','nef','arw','dng','raw',
'pdf','md','txt','log','json','yaml','yml','csv','docx','doc','xlsx','xls','pptx','ppt',
'mp3','wav','flac','aac','m4a','ogg','wma','mp4','mov','avi','mkv','webm','m4v','3gp']
)
})
except (OSError, PermissionError):
# Skip files we can't access
continue
# Sort: directories first, then files, both alphabetically
items.sort(key=lambda x: (not x['is_directory'], x['name'].lower()))
# Calculate parent path
parent_path = None
if path != '/':
parent_path = os.path.dirname(path) if os.path.dirname(path) != path else '/'
return jsonify({
'current_path': path,
'parent_path': parent_path,
'items': items
})
except PermissionError:
return jsonify({'error': 'Permission denied'}), 403
except Exception as e:
logger.error(f"Error browsing directory: {e}")
return jsonify({'error': str(e)}), 500
@app.route('/api/files/usage')
@login_required
def api_files_usage():
"""Return disk usage for the filesystem that contains the given path."""
try:
path = request.args.get('path', '/Volumes')
if not os.path.exists(path):
return jsonify({'error': 'Path not found'}), 404
# If it's a file, get its directory
if os.path.isfile(path):
path = os.path.dirname(path)
usage = shutil.disk_usage(path)
return jsonify({
'path': path,
'total': usage.total,
'used': usage.used,
'free': usage.free
})
except Exception as e:
logger.error(f"Usage error: {e}")
return jsonify({'error': str(e)}), 500
# Weather API endpoints using Open-Meteo (no API key required)
@app.route('/api/weather')
@login_required
def get_weather():
try:
lat = request.args.get('lat')
lon = request.args.get('lon')
if not lat or not lon:
return jsonify({'error': 'Missing coordinates'}), 400
import requests
# Use Open-Meteo API (free, no key required)
url = f'https://api.open-meteo.com/v1/forecast?latitude={lat}&longitude={lon}¤t_weather=true&hourly=temperature_2m,relative_humidity_2m,weather_code&timezone=auto'
response = requests.get(url)
if response.status_code == 200:
data = response.json()
current = data.get('current_weather', {})
# Convert Open-Meteo weather codes to OpenWeatherMap-like format
weather_code = current.get('weathercode', 0)
weather_desc, weather_icon = convert_weather_code(weather_code)
return jsonify({
'main': {'temp': current.get('temperature', 20)},
'weather': [{'description': weather_desc, 'icon': weather_icon}],
'wind': {'speed': current.get('windspeed', 0)},
'name': 'Current Location'
})
else:
# Fallback weather data
return jsonify({
'main': {'temp': 20},
'weather': [{'description': 'clear sky', 'icon': '01d'}]
})
except Exception as e:
logger.error(f"Weather API error: {e}")
return jsonify({
'main': {'temp': 20},
'weather': [{'description': 'clear sky', 'icon': '01d'}]
})
@app.route('/api/weather/forecast')
@login_required
def get_weather_forecast():
try:
lat = request.args.get('lat')
lon = request.args.get('lon')
if not lat or not lon:
return jsonify({'error': 'Missing coordinates'}), 400
import requests
# Use Open-Meteo API for 7-day forecast
url = f'https://api.open-meteo.com/v1/forecast?latitude={lat}&longitude={lon}&daily=weather_code,temperature_2m_max,temperature_2m_min&timezone=auto&forecast_days=7'
response = requests.get(url)
if response.status_code == 200:
data = response.json()
daily = data.get('daily', {})
forecast_list = []
dates = daily.get('time', [])
max_temps = daily.get('temperature_2m_max', [])
min_temps = daily.get('temperature_2m_min', [])
weather_codes = daily.get('weather_code', [])
for i in range(len(dates)):
weather_desc, weather_icon = convert_weather_code(weather_codes[i] if i < len(weather_codes) else 0)
forecast_list.append({
'dt_txt': dates[i],
'main': {
'temp_max': max_temps[i] if i < len(max_temps) else 20,
'temp_min': min_temps[i] if i < len(min_temps) else 15
},
'weather': [{'description': weather_desc, 'icon': weather_icon}]
})
return jsonify({'list': forecast_list})
else:
return jsonify({'list': []})
except Exception as e:
logger.error(f"Weather forecast API error: {e}")
return jsonify({'list': []})
@app.route('/api/weather/detail')
@login_required
def get_weather_detail():
try:
lat = request.args.get('lat')
lon = request.args.get('lon')
date = request.args.get('date')
index = request.args.get('index', '0')
if not lat or not lon:
return jsonify({'error': 'Missing coordinates'}), 400
import requests
from datetime import datetime, timedelta
# Get current weather for humidity and wind data
current_url = f'https://api.open-meteo.com/v1/forecast?latitude={lat}&longitude={lon}¤t_weather=true&hourly=temperature_2m,relative_humidity_2m,weather_code,wind_speed_10m&timezone=auto&forecast_days=1'
# Get hourly forecast for the specific day
target_date = datetime.fromisoformat(date.replace('Z', '+00:00')) if date else datetime.now()
hourly_url = f'https://api.open-meteo.com/v1/forecast?latitude={lat}&longitude={lon}&hourly=temperature_2m,weather_code,relative_humidity_2m,wind_speed_10m&timezone=auto&start_date={target_date.strftime("%Y-%m-%d")}&end_date={target_date.strftime("%Y-%m-%d")}'
result = {
'humidity': 65, # Default values
'windSpeed': 10,
'hourly': []
}
# Get current conditions for humidity and wind
try:
current_response = requests.get(current_url, timeout=5)
if current_response.status_code == 200:
current_data = current_response.json()
current_weather = current_data.get('current_weather', {})
# Get humidity from hourly data (current hour)
hourly_data = current_data.get('hourly', {})
if hourly_data.get('relative_humidity_2m'):
result['humidity'] = int(hourly_data['relative_humidity_2m'][0])
# Get wind speed from current weather
if current_weather.get('windspeed'):
result['windSpeed'] = current_weather['windspeed']
except Exception as e:
logger.error(f"Error getting current weather: {e}")
# Get hourly forecast for the day
try:
hourly_response = requests.get(hourly_url, timeout=5)
if hourly_response.status_code == 200:
hourly_data = hourly_response.json().get('hourly', {})
times = hourly_data.get('time', [])
temps = hourly_data.get('temperature_2m', [])
weather_codes = hourly_data.get('weather_code', [])
hourly_forecast = []
for i in range(min(len(times), 24)): # 24 hours for the day
if i < len(temps) and i < len(weather_codes):
weather_desc, weather_icon = convert_weather_code(weather_codes[i])
hourly_forecast.append({
'time': times[i],
'temperature': temps[i],
'weatherCode': weather_codes[i],
'icon': weather_icon
})
result['hourly'] = hourly_forecast
except Exception as e:
logger.error(f"Error getting hourly forecast: {e}")
return jsonify(result)
except Exception as e:
logger.error(f"Weather detail API error: {e}")
return jsonify({
'humidity': 65,
'windSpeed': 10,
'hourly': []
})
def convert_weather_code(code):
"""Convert Open-Meteo weather codes to description and icon"""
weather_map = {
0: ('clear sky', '01d'),
1: ('mainly clear', '01d'),
2: ('partly cloudy', '02d'),
3: ('overcast', '03d'),
45: ('fog', '50d'),
48: ('depositing rime fog', '50d'),
51: ('light drizzle', '09d'),
53: ('moderate drizzle', '09d'),
55: ('dense drizzle', '09d'),
61: ('slight rain', '10d'),
63: ('moderate rain', '10d'),
65: ('heavy rain', '10d'),
71: ('slight snow', '13d'),
73: ('moderate snow', '13d'),
75: ('heavy snow', '13d'),
80: ('slight rain showers', '09d'),
81: ('moderate rain showers', '09d'),
82: ('violent rain showers', '09d'),
95: ('thunderstorm', '11d'),
96: ('thunderstorm with hail', '11d'),
99: ('thunderstorm with heavy hail', '11d')
}
return weather_map.get(code, ('clear sky', '01d'))
# File upload endpoint
@app.route('/api/upload', methods=['POST'])
@login_required
def upload_file():
try:
if 'file' not in request.files:
return jsonify({'error': 'No file provided'}), 400
file = request.files['file']
upload_path = request.form.get('path', '/tmp')
if file.filename == '':
return jsonify({'error': 'No file selected'}), 400
# Security check - ensure path is safe
if '..' in upload_path or upload_path.startswith('/'):
if not upload_path.startswith('/Volumes'):
return jsonify({'error': 'Invalid upload path'}), 400
filename = secure_filename(file.filename)
full_path = os.path.join(upload_path, filename)
# Check file size (10GB limit)
file.seek(0, os.SEEK_END)
file_size = file.tell()
file.seek(0)
if file_size > 10 * 1024 * 1024 * 1024: # 10GB
return jsonify({'error': 'File too large (max 10GB)'}), 400
file.save(full_path)
return jsonify({'message': 'File uploaded successfully', 'path': full_path})
except Exception as e:
logger.error(f"Upload error: {e}")
return jsonify({'error': str(e)}), 500
# File download endpoint
@app.route('/api/download')
@login_required
def download_file():
try:
file_path = request.args.get('path')
if not file_path or not os.path.exists(file_path):
return jsonify({'error': 'File not found'}), 404
return send_file(file_path, as_attachment=True)
except Exception as e:
logger.error(f"Download error: {e}")
return jsonify({'error': str(e)}), 500
# File deletion endpoint
@app.route('/api/delete', methods=['POST'])
@login_required
def delete_file():
try:
data = request.get_json()
file_path = data.get('path')
if not file_path or not os.path.exists(file_path):
return jsonify({'error': 'File not found'}), 404
if os.path.isdir(file_path):
shutil.rmtree(file_path)
else:
os.remove(file_path)
return jsonify({'message': 'File deleted successfully'})
except Exception as e:
logger.error(f"Delete error: {e}")
return jsonify({'error': str(e)}), 500
# Create folder endpoint
@app.route('/api/mkdir', methods=['POST'])
@login_required
def create_folder():
try:
data = request.get_json()
base_path = data.get('path')
folder_name = data.get('name')
if not base_path or not folder_name:
return jsonify({'error': 'Missing path or name'}), 400
folder_path = os.path.join(base_path, folder_name)
os.makedirs(folder_path, exist_ok=True)
return jsonify({'message': 'Folder created successfully', 'path': folder_path})
except Exception as e:
logger.error(f"Mkdir error: {e}")
return jsonify({'error': str(e)}), 500
# File sharing system
share_links = {} # In production, use a database
@app.route('/api/share', methods=['POST'])
@login_required
def create_share_link():
try:
data = request.get_json()
files = data.get('files', [])
expiry_hours = data.get('expiry_hours', 24)
upload_limit_gb = data.get('upload_limit_gb', 10)
allow_upload = data.get('allow_upload', False)
# Generate unique share ID
share_id = os.urandom(16).hex()
# Calculate expiry time
expiry_time = None
if expiry_hours > 0:
expiry_time = datetime.now() + timedelta(hours=expiry_hours)
# Get user ID from session
user_id = session.get('user_id', 1) # Default to 1 if no session
# Determine the share name and path
if len(files) == 1:
share_name = os.path.basename(files[0])
share_path = files[0]
else:
share_name = f"Multi-file share ({len(files)} items)"
share_path = os.path.commonpath(files) if files else "/tmp"
# Save to database
conn = get_db_connection()
conn.execute('''
INSERT INTO user_shares (user_id, share_id, name, path, expires_at, created_at, download_count)
VALUES (?, ?, ?, ?, ?, ?, 0)
''', (user_id, share_id, share_name, share_path,
expiry_time.isoformat() if expiry_time else None,
datetime.now().isoformat()))
conn.commit()
conn.close()
return jsonify({
'share_id': share_id,
'expires': expiry_time.isoformat() if expiry_time else None,
'share_url': f'/share/{share_id}'
})
except Exception as e:
logger.error(f"Share creation error: {e}")
return jsonify({'error': str(e)}), 500
@app.route('/share/')
def share_page(share_id):
conn = get_db_connection()
share = conn.execute('''
SELECT share_id, name, path, expires_at, created_at, download_count, password_hash
FROM user_shares
WHERE share_id = ?
''', (share_id,)).fetchone()
conn.close()
if not share:
return "Share link not found or expired", 404
# Check if expired
if share['expires_at'] and datetime.fromisoformat(share['expires_at']) < datetime.now():
return "Share link has expired", 410
# Convert to dict and add files list
share_data = dict(share)
share_data['files'] = []
share_data['allow_upload'] = False
share_data['upload_limit_gb'] = 10
# Check if path exists and get files/folders
if os.path.exists(share_data['path']):
if os.path.isdir(share_data['path']):
# For directories, list contents
try:
for item in os.listdir(share_data['path']):
item_path = os.path.join(share_data['path'], item)
share_data['files'].append({
'name': item,
'path': item_path,
'is_dir': os.path.isdir(item_path),
'size': os.path.getsize(item_path) if os.path.isfile(item_path) else 0
})
except PermissionError:
pass
else:
# For files, just add the file itself
share_data['files'].append({
'name': os.path.basename(share_data['path']),
'path': share_data['path'],
'is_dir': False,
'size': os.path.getsize(share_data['path'])
})
return render_template('share.html', share_data=share_data, share_id=share_id)
@app.route('/api/share//download')
def share_download(share_id):
from datetime import datetime
conn = None
try:
conn = get_db_connection()
share = conn.execute('''
SELECT share_id, name, path, expires_at, password_hash
FROM user_shares
WHERE share_id = ?
''', (share_id,)).fetchone()
if not share:
conn.close()
return jsonify({'error': 'Share not found'}), 404
# Check if expired
if share['expires_at'] and datetime.fromisoformat(share['expires_at']) < datetime.now():
conn.close()
return jsonify({'error': 'Share has expired'}), 410
# Update download count
conn.execute('''
UPDATE user_shares
SET download_count = download_count + 1
WHERE share_id = ?
''', (share_id,))
conn.commit()
conn.close()
except Exception as e:
if conn:
conn.close()
return jsonify({'error': f'Database error: {str(e)}'}), 500
file_path = request.args.get('path', share['path'])
if not os.path.exists(file_path):
return jsonify({'error': 'File not found'}), 404
# Handle directory downloads by creating a ZIP file
if os.path.isdir(file_path):
import tempfile
import zipfile
# Create temporary ZIP file
temp_dir = tempfile.gettempdir()
folder_name = os.path.basename(file_path.rstrip('/'))
zip_filename = f"{folder_name}.zip"
zip_path = os.path.join(temp_dir, f"share_{share_id}_{zip_filename}")
try:
with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
for root, dirs, files in os.walk(file_path):
for file in files:
file_full_path = os.path.join(root, file)
arcname = os.path.relpath(file_full_path, file_path)
zipf.write(file_full_path, arcname)
def remove_file(response):
try:
os.remove(zip_path)
except:
pass
return response
return send_file(zip_path, as_attachment=True, download_name=zip_filename)
except Exception as e:
if os.path.exists(zip_path):
os.remove(zip_path)
return jsonify({'error': f'Error creating ZIP file: {str(e)}'}), 500
else:
return send_file(file_path, as_attachment=True)
if not os.path.exists(file_path):
return jsonify({'error': 'File not found'}), 404
# Handle directory downloads by creating a zip file
if os.path.isdir(file_path):
import zipfile
import tempfile
from datetime import datetime
# Create a temporary zip file
temp_dir = tempfile.mkdtemp()
folder_name = os.path.basename(file_path.rstrip('/'))
zip_filename = f"{folder_name}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.zip"
zip_path = os.path.join(temp_dir, zip_filename)
try:
with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
for root, dirs, files in os.walk(file_path):
for file in files:
file_full_path = os.path.join(root, file)
# Create relative path for archive
arcname = os.path.relpath(file_full_path, file_path)
zipf.write(file_full_path, arcname)
share_data['downloads'] += 1
def remove_temp_file():
try:
os.remove(zip_path)
os.rmdir(temp_dir)
except:
pass
# Schedule cleanup after download
import atexit
atexit.register(remove_temp_file)
return send_file(zip_path, as_attachment=True, download_name=zip_filename)
except Exception as e:
logger.error(f"Error creating zip file: {e}")
return jsonify({'error': 'Failed to create download archive'}), 500
else:
# Handle file downloads normally
share_data['downloads'] += 1
return send_file(file_path, as_attachment=True)
@app.route('/api/share//preview')
def share_preview(share_id):
conn = get_db_connection()
share = conn.execute('''
SELECT share_id, name, path, expires_at, password_hash
FROM user_shares
WHERE share_id = ?
''', (share_id,)).fetchone()
conn.close()
if not share:
return jsonify({'error': 'Share not found'}), 404
# Check if expired
if share['expires_at'] and datetime.fromisoformat(share['expires_at']) < datetime.now():
return jsonify({'error': 'Share has expired'}), 410
file_path = request.args.get('path', share['path'])
if not os.path.exists(file_path):
return jsonify({'error': 'File not found'}), 404
# Handle directory preview by returning file listing
if os.path.isdir(file_path):
try:
items = []
for item in os.listdir(file_path):
item_path = os.path.join(file_path, item)
items.append({
'name': item,
'type': 'folder' if os.path.isdir(item_path) else 'file',
'size': 0 if os.path.isdir(item_path) else os.path.getsize(item_path)
})
return jsonify({
'type': 'directory',
'name': os.path.basename(file_path.rstrip('/')),
'items': items
})
except PermissionError:
return jsonify({'error': 'Permission denied'}), 403
except Exception as e:
return jsonify({'error': f'Error reading directory: {str(e)}'}), 500
else:
# For files, serve the file content
_, ext = os.path.splitext(file_path)
ext = ext.lower().lstrip('.')
mime_type = 'application/octet-stream'
if ext in ['jpg', 'jpeg']:
mime_type = 'image/jpeg'
elif ext in ['png']:
mime_type = 'image/png'
elif ext in ['gif']:
mime_type = 'image/gif'
elif ext in ['mp4']:
mime_type = 'video/mp4'
elif ext in ['mp3']:
mime_type = 'audio/mpeg'
elif ext in ['pdf']:
mime_type = 'application/pdf'
elif ext in ['txt']:
mime_type = 'text/plain'
elif ext in ['html']:
mime_type = 'text/html'
elif ext in ['css']:
mime_type = 'text/css'
elif ext in ['js']:
mime_type = 'application/javascript'
elif ext in ['json']:
mime_type = 'application/json'
elif ext in ['xml']:
mime_type = 'application/xml'
elif ext in ['zip']:
mime_type = 'application/zip'
elif ext in ['tar']:
mime_type = 'application/x-tar'
elif ext in ['gz']:
mime_type = 'application/gzip'
elif ext in ['wav']:
mime_type = 'audio/wav'
elif ext in ['ogg']:
mime_type = 'audio/ogg'
return send_file(file_path, mimetype=mime_type)
if not os.path.exists(file_path):
return jsonify({'error': 'File not found'}), 404
# Handle directory preview by returning file listing
if os.path.isdir(file_path):
try:
items = []
for item in os.listdir(file_path):
item_path = os.path.join(file_path, item)
if os.path.isfile(item_path):
try:
stat_info = os.stat(item_path)
items.append({
'name': item,
'type': 'file',
'size': stat_info.st_size,
'modified': stat_info.st_mtime
})
except:
items.append({
'name': item,
'type': 'file',
'size': 0,
'modified': 0
})
elif os.path.isdir(item_path):
items.append({
'name': item,
'type': 'folder',
'size': 0,
'modified': 0
})
return jsonify({
'type': 'directory',
'name': os.path.basename(file_path.rstrip('/')),
'items': items
})
except Exception as e:
logger.error(f"Error listing directory: {e}")
return jsonify({'error': 'Failed to list directory contents'}), 500
else:
# Handle file preview normally
# Get file extension for MIME type
_, ext = os.path.splitext(file_path)
ext = ext.lower().lstrip('.')
# Set appropriate MIME type
mime_type = None
if ext in ['jpg', 'jpeg']:
mime_type = 'image/jpeg'
elif ext in ['png']:
mime_type = 'image/png'
elif ext in ['gif']:
mime_type = 'image/gif'
elif ext in ['webp']:
mime_type = 'image/webp'
elif ext in ['svg']:
mime_type = 'image/svg+xml'
elif ext in ['mp4']:
mime_type = 'video/mp4'
elif ext in ['mov']:
mime_type = 'video/quicktime'
elif ext in ['webm']:
mime_type = 'video/webm'
elif ext in ['mp3']:
mime_type = 'audio/mpeg'
elif ext in ['wav']:
mime_type = 'audio/wav'
elif ext in ['ogg']:
mime_type = 'audio/ogg'
elif ext in ['pdf']:
mime_type = 'application/pdf'
else:
mime_type = 'application/octet-stream'
return send_file(file_path, mimetype=mime_type)
@app.route('/api/stats')
@login_required
def get_file_stats():
try:
# Count total files in /Volumes
total_files = 0
storage_used = 0
for root, dirs, files in os.walk('/Volumes'):
# Skip system directories
if any(skip in root for skip in ['.Trash', '.fseventsd', '.Spotlight']):
continue
total_files += len(files)
for file in files:
try:
file_path = os.path.join(root, file)
storage_used += os.path.getsize(file_path)
except (OSError, IOError):
continue
# Count active shares
active_shares = len([s for s in share_links.values()
if not s.get('expiry') or datetime.now() < s['expiry']])
# Count total downloads
total_downloads = sum(s.get('downloads', 0) for s in share_links.values())
return jsonify({
'total_files': total_files,
'active_shares': active_shares,
'total_downloads': total_downloads,
'storage_used': storage_used
})
except Exception as e:
logger.error(f"Stats error: {e}")
return jsonify({
'total_files': 0,
'active_shares': 0,
'total_downloads': 0,
'storage_used': 0
})
@app.route('/api/share//upload', methods=['POST'])
@login_required
def share_upload(share_id):
if share_id not in share_links:
return jsonify({'error': 'Share not found'}), 404
share_data = share_links[share_id]
# Check if uploads are allowed
if not share_data.get('allow_upload', False):
return jsonify({'error': 'Uploads not allowed for this share'}), 403
# Check if expired
if share_data['expiry'] and datetime.now() > share_data['expiry']:
del share_links[share_id]
return jsonify({'error': 'Share has expired'}), 410
if 'file' not in request.files:
return jsonify({'error': 'No file provided'}), 400
file = request.files['file']
if file.filename == '':
return jsonify({'error': 'No file selected'}), 400
# Check upload limit
current_uploads_size = sum(upload.get('size', 0) for upload in share_data.get('uploads', []))
max_size = share_data.get('upload_limit_gb', 10) * 1024 * 1024 * 1024
file.seek(0, os.SEEK_END)
file_size = file.tell()
file.seek(0)
if current_uploads_size + file_size > max_size:
return jsonify({'error': f'Upload limit of {share_data.get("upload_limit_gb", 10)}GB exceeded'}), 400
# Create uploads directory for this share
upload_dir = os.path.join('/tmp', 'samcloud_uploads', share_id)
os.makedirs(upload_dir, exist_ok=True)
filename = secure_filename(file.filename)
file_path = os.path.join(upload_dir, filename)
# Handle duplicate filenames
counter = 1
original_path = file_path
while os.path.exists(file_path):
name, ext = os.path.splitext(original_path)
file_path = f"{name}_{counter}{ext}"
counter += 1
file.save(file_path)
# Track the upload
if 'uploads' not in share_data:
share_data['uploads'] = []
share_data['uploads'].append({
'filename': os.path.basename(file_path),
'original_filename': file.filename,
'size': file_size,
'uploaded_at': datetime.now(),
'path': file_path
})
return jsonify({'message': 'File uploaded successfully', 'filename': os.path.basename(file_path)})
# Notes API endpoints
@app.route('/api/notes')
@login_required
def get_notes():
"""Get all notes for the current user"""
try:
user_id = session.get('user_id')
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute('''
SELECT id, title, content, created_at, updated_at
FROM notes
WHERE user_id = ?
ORDER BY updated_at DESC
''', (user_id,))
notes = []
for row in cursor.fetchall():
notes.append({
'id': row[0],
'title': row[1],
'content': row[2],
'created_at': row[3],
'updated_at': row[4]
})
conn.close()
return jsonify({'notes': notes})
except Exception as e:
logger.error(f"Error getting notes: {e}")
return jsonify({'error': 'Failed to load notes'}), 500
@app.route('/api/notes', methods=['POST'])
@login_required
def create_note():
"""Create a new note"""
try:
data = request.get_json()
user_id = session.get('user_id')
title = data.get('title', 'Untitled Note')
content = data.get('content', '')
conn = get_db_connection()
cursor = conn.cursor()
now = datetime.now().isoformat()
cursor.execute('''
INSERT INTO notes (user_id, title, content, created_at, updated_at)
VALUES (?, ?, ?, ?, ?)
''', (user_id, title, content, now, now))
note_id = cursor.lastrowid
conn.commit()
conn.close()
# Return the created note
note = {
'id': note_id,
'title': title,
'content': content,
'created_at': now,
'updated_at': now
}
return jsonify({'note': note})
except Exception as e:
logger.error(f"Error creating note: {e}")
return jsonify({'error': 'Failed to create note'}), 500
@app.route('/api/notes/', methods=['PUT'])
@login_required
def update_note(note_id):
"""Update an existing note"""
try:
data = request.get_json()
user_id = session.get('user_id')
title = data.get('title', 'Untitled Note')
content = data.get('content', '')
conn = get_db_connection()
cursor = conn.cursor()
now = datetime.now().isoformat()
# Check if note belongs to user
cursor.execute('SELECT id FROM notes WHERE id = ? AND user_id = ?', (note_id, user_id))
if not cursor.fetchone():
conn.close()
return jsonify({'error': 'Note not found'}), 404
# Update the note
cursor.execute('''
UPDATE notes
SET title = ?, content = ?, updated_at = ?
WHERE id = ? AND user_id = ?
''', (title, content, now, note_id, user_id))
conn.commit()
conn.close()
# Return the updated note
note = {
'id': note_id,
'title': title,
'content': content,
'updated_at': now
}
return jsonify({'note': note})
except Exception as e:
logger.error(f"Error updating note: {e}")
return jsonify({'error': 'Failed to update note'}), 500
@app.route('/api/notes/', methods=['DELETE'])
@login_required
def delete_note(note_id):
"""Delete a note"""
try:
user_id = session.get('user_id')
conn = get_db_connection()
cursor = conn.cursor()
# Check if note belongs to user
cursor.execute('SELECT id FROM notes WHERE id = ? AND user_id = ?', (note_id, user_id))
if not cursor.fetchone():
conn.close()
return jsonify({'error': 'Note not found'}), 404
# Delete the note
cursor.execute('DELETE FROM notes WHERE id = ? AND user_id = ?', (note_id, user_id))
conn.commit()
conn.close()
return jsonify({'message': 'Note deleted successfully'})
except Exception as e:
logger.error(f"Error deleting note: {e}")
return jsonify({'error': 'Failed to delete note'}), 500
if __name__ == '__main__':
print(f"🚀 Starting SamCloud Dashboard on http://localhost:{config.PORT}")
print(f"📁 Static files: {config.STATIC_DIR}")
print(f"📄 Templates: {config.TEMPLATES_DIR}")
app.run(
host=config.HOST,
port=config.PORT,
debug=config.DEBUG,
threaded=True,
use_reloader=config.DEBUG # Enable auto-reload in debug mode
)