Initial commit — MarineMaintenance v1.0
Marine maintenance management: work orders with photos, ISM/SWP procedures, MSDS, inventory, RFQ/purchases, vessel history, bilingual PDF reports. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,125 @@
|
||||
"""
|
||||
auth.py — Login y control de acceso sin flask-login
|
||||
Usa sesiones de Flask + werkzeug para hash de contraseñas
|
||||
"""
|
||||
from functools import wraps
|
||||
from flask import session, redirect, url_for, flash, request
|
||||
from werkzeug.security import generate_password_hash, check_password_hash
|
||||
import sqlite3, os
|
||||
|
||||
DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'marine_maintenance.db')
|
||||
|
||||
def get_db():
|
||||
conn = sqlite3.connect(DB_PATH)
|
||||
conn.row_factory = sqlite3.Row
|
||||
conn.execute("PRAGMA foreign_keys = ON")
|
||||
return conn
|
||||
|
||||
# ── Sesión ────────────────────────────────────────────────────────────────────
|
||||
def login_user(user):
|
||||
session['user_id'] = user['id']
|
||||
session['username'] = user['username']
|
||||
session['full_name'] = user['full_name'] or user['username']
|
||||
session['role'] = user['role']
|
||||
session['company_id'] = user['company_id']
|
||||
session['company_name'] = user['company_name'] if 'company_name' in user.keys() else None
|
||||
# update last_login
|
||||
conn = get_db()
|
||||
conn.execute("UPDATE users SET last_login=CURRENT_TIMESTAMP WHERE id=?", (user['id'],))
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
def logout_user():
|
||||
session.clear()
|
||||
|
||||
def current_user():
|
||||
if 'user_id' not in session:
|
||||
return None
|
||||
return {
|
||||
'id': session.get('user_id'),
|
||||
'username': session.get('username'),
|
||||
'full_name': session.get('full_name'),
|
||||
'role': session.get('role'),
|
||||
'company_id': session.get('company_id'),
|
||||
'company_name': session.get('company_name'),
|
||||
}
|
||||
|
||||
def is_logged_in():
|
||||
return 'user_id' in session
|
||||
|
||||
def is_superadmin():
|
||||
return session.get('role') == 'superadmin'
|
||||
|
||||
def is_admin():
|
||||
return session.get('role') in ('superadmin', 'admin')
|
||||
|
||||
# ── Decoradores ───────────────────────────────────────────────────────────────
|
||||
def login_required(f):
|
||||
@wraps(f)
|
||||
def decorated(*args, **kwargs):
|
||||
if not is_logged_in():
|
||||
return redirect(url_for('auth_login', next=request.url))
|
||||
return f(*args, **kwargs)
|
||||
return decorated
|
||||
|
||||
def admin_required(f):
|
||||
@wraps(f)
|
||||
def decorated(*args, **kwargs):
|
||||
if not is_logged_in():
|
||||
return redirect(url_for('auth_login'))
|
||||
if not is_admin():
|
||||
return redirect(url_for('dashboard'))
|
||||
return f(*args, **kwargs)
|
||||
return decorated
|
||||
|
||||
def superadmin_required(f):
|
||||
@wraps(f)
|
||||
def decorated(*args, **kwargs):
|
||||
if not is_logged_in():
|
||||
return redirect(url_for('auth_login'))
|
||||
if not is_superadmin():
|
||||
return redirect(url_for('dashboard'))
|
||||
return f(*args, **kwargs)
|
||||
return decorated
|
||||
|
||||
# ── Filtro de compañía ────────────────────────────────────────────────────────
|
||||
def company_filter():
|
||||
"""
|
||||
Retorna (sql_where, params) para filtrar por compañía del usuario.
|
||||
Si es superadmin, no filtra.
|
||||
"""
|
||||
if is_superadmin():
|
||||
return "", []
|
||||
cid = session.get('company_id')
|
||||
if cid:
|
||||
return "WHERE company_id = ?", [cid]
|
||||
return "WHERE 1=0", [] # sin compañía asignada no ve nada
|
||||
|
||||
def vessel_filter():
|
||||
"""WHERE clause para embarcaciones según compañía del usuario."""
|
||||
if is_superadmin():
|
||||
return "", []
|
||||
cid = session.get('company_id')
|
||||
if cid:
|
||||
return "WHERE v.company_id = ?", [cid]
|
||||
return "WHERE 1=0", []
|
||||
|
||||
# ── Hash de contraseñas ───────────────────────────────────────────────────────
|
||||
def hash_password(password):
|
||||
return generate_password_hash(password)
|
||||
|
||||
def verify_password(password, hashed):
|
||||
return check_password_hash(hashed, password)
|
||||
|
||||
# ── Crear superadmin inicial ──────────────────────────────────────────────────
|
||||
def create_initial_superadmin(username='admin', password='admin123', email='admin@marine.local'):
|
||||
conn = get_db()
|
||||
existing = conn.execute("SELECT id FROM users WHERE role='superadmin'").fetchone()
|
||||
if not existing:
|
||||
conn.execute("""
|
||||
INSERT INTO users (company_id, username, email, password_hash, full_name, role)
|
||||
VALUES (NULL, ?, ?, ?, 'Super Administrator', 'superadmin')
|
||||
""", (username, email, hash_password(password)))
|
||||
conn.commit()
|
||||
print(f"[AUTH] Superadmin creado: {username} / {password}")
|
||||
conn.close()
|
||||
Reference in New Issue
Block a user