Upload auth.py

auth.py

@@ -1,292 +1,318 @@
-"""
-Módulo de autenticación para la aplicación Veureu.
-Gestiona usuarios, verificación de contraseñas y sincronización de usuarios por defecto.
-"""
-import sys
-from datetime import datetime
-from pathlib import Path
-
-import streamlit as st
-
-from databases import get_user, create_user, update_user_password, get_all_users, log_action
-from mobile_verification import (
-    initialize_sms_state,
-    render_mobile_verification_screen,
-    get_user_permissions,
-    show_verification_status_in_sidebar
-)
-from persistent_data_gate import confirm_changes_and_logout
-from compliance_client import compliance_client
-import yaml
-
-
-def log(msg: str):
-    """Helper per logging amb timestamp"""
-    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
-    sys.stderr.write(f"[{timestamp}] {msg}\n")
-    sys.stderr.flush()
-
-
-def verify_password(password: str, stored_password: str) -> bool:
-    """Verifica la contraseña como texto plano."""
-    return password == stored_password
-
-
-def create_default_users_if_needed():
-    """Asegura que existan los usuarios por defecto y sus contraseñas esperadas (texto plano)."""
-    log("Sincronizando usuarios per defecte (sense detalls sensibles)...")
-    users_to_create = [
-        ("verd", "verd123", "verd"),
-        ("groc", "groc123", "groc"),
-        ("taronja", "taronja123", "taronja"),
-        ("blau", "blau123", "blau"),
-        ("vermell", "vermell123", "vermell"),
-    ]
-    for username, password, role in users_to_create:
-        try:
-            row = get_user(username)
-            if row:
-                update_user_password(username, password)
-            else:
-                create_user(username, password, role)
-        except Exception as e:
-            log(f"Error sincronizando usuario {username}: {e}")
-    log("Sincronització d'usuaris per defecte completada.")
-
-
-def initialize_auth_system(db_path: str):
-    """Inicializa el sistema de autenticación y sincroniza usuarios."""
-    if 'users_synced' not in st.session_state:
-        create_default_users_if_needed()
-        st.session_state['users_synced'] = True
-
-    # Inicializar estado de verificación SMS
-    initialize_sms_state()
-
-    # Diagnòstic de base de dades simplificat (sense dades sensibles)
-    if 'diag_logged' not in st.session_state:
-        log("Base de dades d'usuaris inicialitzada correctament.")
-        st.session_state['diag_logged'] = True
-
-
-def require_login(login_form_func):
-    """Requiere que el usuario esté autenticado."""
-    if not st.session_state.user:
-        st.info("Por favor, inicia sesión para continuar.")
-        login_form_func()
-        st.stop()
-
-
-def render_login_form():
-    """Renderiza el formulario de login con logs de depuración."""
-    st.subheader("Inici de sessió")
-    username = st.text_input("Usuari")
-    password = st.text_input("Contrasenya", type="password")
-    
-    if st.button("Entrar", type="primary"):
-        row = get_user(username)
-
-        # Logs de depuración
-        log("\n--- INTENTO DE LOGIN ---")
-        log(f"Usuario introducido: '{username}'")
-        log(f"Contraseña introducida: {'Sí' if password else 'No'}")
-
-        if row:
-            log(f"Usuario encontrado en BD: '{row['username']}'")
-            stored_pw = (row["password_hash"] or "")
-            log(f"Password almacenado (longitud): {len(stored_pw)}")
-            is_valid = verify_password(password, stored_pw)
-            log(f"Resultado de verify_password: {is_valid}")
-        else:
-            log("Usuario no encontrado en la BD.")
-            is_valid = False
-        
-        log("--- FIN INTENTO DE LOGIN ---\n")
-
-        if is_valid:
-            st.session_state.user = {
-                "id": row["id"],
-                "username": row["username"],
-                "role": row["role"]
-            }
-            # Desa la darrera contrasenya per poder-la registrar als esdeveniments
-            st.session_state.last_password = password
-
-            # Registre d'esdeveniment de login a events.db
-            try:
-                session_id = st.session_state.get("session_id", "")
-                phone = (
-                    st.session_state.get("sms_phone_verified")
-                    or st.session_state.get("sms_phone")
-                    or ""
-                )
-                log_action(
-                    session=session_id,
-                    user=username or "",
-                    phone=phone,
-                    action="login",
-                    sha1sum="",
-                )
-            except Exception as e:
-                log(f"Error registrant esdeveniment de login: {e}")
-
-            st.success(f"Benvingut/da, {row['username']}")
-            st.rerun()
-        else:
-            st.error("Credencials invàlides")
-
-
-def render_sidebar():
-    """Renderiza la barra lateral con información de usuario y navegación."""
-    role = st.session_state.user["role"] if st.session_state.user else None
-    
-    with st.sidebar:
-        logo_path = Path(__file__).parent / "images" / "veureu.png"
-        if logo_path.exists():
-            st.image(str(logo_path), width=140)
-        else:
-            st.title("Veureu")
-        if st.session_state.user:
-            st.write(f"Usuari: **{st.session_state.user['username']}** (rol: {st.session_state.user['role']})")
-            
-            # Mostrar estado de verificación SMS
-            show_verification_status_in_sidebar()
-        
-        if st.session_state.user:
-            # Obtener permisos del usuario
-            permissions = get_user_permissions(
-                role,
-                st.session_state.get('sms_verified')
-            )
-            
-            # Construir opciones de navegación según permisos
-            page_options = []
-            
-            if permissions["analizar"]:
-                page_options.append("Analitzar audiodescripcions")
-            
-            if permissions["procesar_videos"]:
-                page_options.append("Processar vídeo nou")
-            
-            if permissions["estadisticas"]:
-                page_options.append("Estadístiques")
-
-            if permissions["validar"]:
-                page_options.append("Validació")
-
-            # Opció de configuració avançada (sala de màquines), només per a l'usuari 'verd'
-            if st.session_state.user.get("username") == "verd":
-                page_options.append("Sala de màquines")
-            
-            # Si no hay opciones disponibles, mostrar solo análisis
-            if not page_options:
-                page_options = ["Analitzar audiodescripcions"]
-            
-            page = st.radio(
-                "Navegació",
-                page_options,
-                index=0
-            )
-            st.markdown("---")
-
-            if st.button(
-                "Confirmar canvis i tancar sessió",
-                key="confirmar_canvis_tancar",
-                use_container_width=True,
-                type="primary",
-            ):
-                # Persistir canvis de la sessió actual abans de tancar
-                try:
-                    base_dir = Path(__file__).parent
-                    session_id = st.session_state.get("session_id", "")
-                    api_client = st.session_state.get("api_client")
-                    digest_info = confirm_changes_and_logout(base_dir, api_client, session_id)
-                except Exception:
-                    digest_info = None
-
-                # Llegir flag public_blockchain_enabled de config.yaml
-                try:
-                    cfg_path = Path(__file__).parent / "config.yaml"
-                    with cfg_path.open("r", encoding="utf-8") as f:
-                        cfg = yaml.safe_load(f) or {}
-                    comp_cfg = cfg.get("compliance", {}) or {}
-                    public_blockchain_enabled = bool(
-                        comp_cfg.get(
-                            "public_blockchain_enabled",
-                            comp_cfg.get("public_blockchain_enable", False),
-                        )
-                    )
-                except Exception:
-                    public_blockchain_enabled = False
-
-                blockchain_published = False
-                polygonscan_url = None
-
-                if public_blockchain_enabled and digest_info and digest_info.get("events_digest"):
-                    try:
-                        session_id = st.session_state.get("session_id", "")
-                        resp = compliance_client.publish_events_digest(
-                            session_id=session_id,
-                            digest_hash=digest_info["events_digest"],
-                        )
-                        if resp:
-                            polygonscan_url = resp.get("transaction_url")
-                            blockchain_published = bool(resp.get("transaction_hash"))
-                    except Exception:
-                        blockchain_published = False
-
-                # Registrar desconnexió a events.db
-                try:
-                    current_user = st.session_state.user or {}
-                    session_id = st.session_state.get("session_id", "")
-                    phone = (
-                        st.session_state.get("sms_phone_verified")
-                        or st.session_state.get("sms_phone")
-                        or ""
-                    )
-                    last_password = st.session_state.get("last_password", "")
-                    log_action(
-                        session=session_id,
-                        user=current_user.get("username", ""),
-                        phone=phone,
-                        action="logout",
-                        sha1sum="",
-                    )
-                except Exception as e:
-                    log(f"Error registrant esdeveniment de logout: {e}")
-
-                # Anotar al log el resultat de la publicació a Polygon (si aplica)
-                digest_hash = digest_info.get("events_digest") if digest_info else None
-                events_count = digest_info.get("events_count") if digest_info else None
-                log(
-                    "Logout completat: "
-                    f"session={session_id or '-'} "
-                    f"events_digest={digest_hash or '-'} "
-                    f"events_count={events_count if events_count is not None else '-'} "
-                    f"polygon_published={'sí' if blockchain_published else 'no'} "
-                    f"polygon_url={polygonscan_url or '-'}"
-                )
-
-                # Netejar sessió d'usuari
-                st.session_state.user = None
-                st.session_state.sms_verified = None
-
-                # Mostrar missatge segons estat de publicació
-                if public_blockchain_enabled and blockchain_published and polygonscan_url:
-                    st.success(
-                        "✅ Els canvis s'han desat i s'han publicat a la cadena de blocs de Polygon. "
-                        f"Pots consultar la transacció a: {polygonscan_url}"
-                    )
-                elif public_blockchain_enabled:
-                    st.info(
-                        "✅ Els canvis s'han desat, però no s'ha pogut completar la publicació "
-                        "a la cadena de blocs de Polygon en aquest moment."
-                    )
-                else:
-                    st.info("✅ Canvis desats (sense publicació a blockchain).")
-
-                st.rerun()
-        else:
-            page = None
-    
-    return page, role
+"""
+Módulo de autenticación para la aplicación Veureu.
+Gestiona usuarios, verificación de contraseñas y sincronización de usuarios por defecto.
+"""
+import sys
+from datetime import datetime
+from pathlib import Path
+
+import streamlit as st
+
+from databases import get_user, create_user, update_user_password, get_all_users, log_action
+from mobile_verification import (
+    initialize_sms_state,
+    render_mobile_verification_screen,
+    get_user_permissions,
+    show_verification_status_in_sidebar
+)
+from persistent_data_gate import confirm_changes_and_logout
+from compliance_client import compliance_client
+import yaml
+
+
+def log(msg: str):
+    """Helper per logging amb timestamp"""
+    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+    sys.stderr.write(f"[{timestamp}] {msg}\n")
+    sys.stderr.flush()
+
+
+def verify_password(password: str, stored_password: str) -> bool:
+    """Verifica la contraseña como texto plano."""
+    return password == stored_password
+
+
+def create_default_users_if_needed():
+    """Asegura que existan los usuarios por defecto y sus contraseñas esperadas (texto plano)."""
+    log("Sincronizando usuarios per defecte (sense detalls sensibles)...")
+    users_to_create = [
+        ("verd", "verd123", "verd"),
+        ("groc", "groc123", "groc"),
+        ("taronja", "taronja123", "taronja"),
+        ("blau", "blau123", "blau"),
+        ("vermell", "vermell123", "vermell"),
+    ]
+    for username, password, role in users_to_create:
+        try:
+            row = get_user(username)
+            if row:
+                update_user_password(username, password)
+            else:
+                create_user(username, password, role)
+        except Exception as e:
+            log(f"Error sincronizando usuario {username}: {e}")
+    log("Sincronització d'usuaris per defecte completada.")
+
+
+def initialize_auth_system(db_path: str):
+    """Inicializa el sistema de autenticación y sincroniza usuarios."""
+    if 'users_synced' not in st.session_state:
+        create_default_users_if_needed()
+        st.session_state['users_synced'] = True
+
+    # Inicializar estado de verificación SMS
+    initialize_sms_state()
+
+    # Diagnòstic de base de dades simplificat (sense dades sensibles)
+    if 'diag_logged' not in st.session_state:
+        log("Base de dades d'usuaris inicialitzada correctament.")
+        st.session_state['diag_logged'] = True
+
+
+def require_login(login_form_func):
+    """Requiere que el usuario esté autenticado."""
+    if not st.session_state.user:
+        st.info("Por favor, inicia sesión para continuar.")
+        login_form_func()
+        st.stop()
+
+
+def render_login_form():
+    """Renderiza el formulario de login con logs de depuración."""
+    st.subheader("Inici de sessió")
+    username = st.text_input("Usuari")
+    password = st.text_input("Contrasenya", type="password")
+    
+    if st.button("Entrar", type="primary"):
+        row = get_user(username) if username else None
+
+        # Logs de depuración
+        log("\n--- INTENTO DE LOGIN ---")
+        log(f"Usuario introducido: '{username}'")
+        log(f"Contraseña introducida: {'Sí' if password else 'No'}")
+
+        if row:
+            log(f"Usuario encontrado en BD: '{row['username']}'")
+            stored_pw = (row["password_hash"] or "")
+            log(f"Password almacenado (longitud): {len(stored_pw)}")
+            is_valid = verify_password(password, stored_pw)
+            log(f"Resultado de verify_password: {is_valid}")
+        else:
+            log("Usuario no encontrado en la BD o nombre de usuario vacío. Se asignará rol 'vermell'.")
+            is_valid = False
+        
+        log("--- FIN INTENTO DE LOGIN ---\n")
+
+        if is_valid:
+            st.session_state.user = {
+                "id": row["id"],
+                "username": row["username"],
+                "role": row["role"]
+            }
+            # Desa la darrera contrasenya per poder-la registrar als esdeveniments
+            st.session_state.last_password = password
+
+            # Registre d'esdeveniment de login a events.db
+            try:
+                session_id = st.session_state.get("session_id", "")
+                phone = (
+                    st.session_state.get("sms_phone_verified")
+                    or st.session_state.get("sms_phone")
+                    or ""
+                )
+                log_action(
+                    session=session_id,
+                    user=username or "",
+                    phone=phone,
+                    action="login",
+                    sha1sum="",
+                )
+            except Exception as e:
+                log(f"Error registrant esdeveniment de login: {e}")
+
+            st.success(f"Benvingut/da, {row['username']}")
+            st.rerun()
+        else:
+            # Si l'usuari està buit o no existeix, entrar com a "vermell" (visitant)
+            st.session_state.user = {
+                "id": None,
+                "username": "vermell",
+                "role": "vermell",
+            }
+            st.session_state.last_password = password
+
+            try:
+                session_id = st.session_state.get("session_id", "")
+                phone = (
+                    st.session_state.get("sms_phone_verified")
+                    or st.session_state.get("sms_phone")
+                    or ""
+                )
+                log_action(
+                    session=session_id,
+                    user="vermell",
+                    phone=phone,
+                    action="login",
+                    sha1sum="",
+                )
+            except Exception as e:
+                log(f"Error registrant esdeveniment de login per usuari 'vermell': {e}")
+
+            st.success("Benvingut/da, vermell")
+            st.rerun()
+
+
+def render_sidebar():
+    """Renderiza la barra lateral con información de usuario y navegación."""
+    role = st.session_state.user["role"] if st.session_state.user else None
+    
+    with st.sidebar:
+        logo_path = Path(__file__).parent / "images" / "veureu.png"
+        if logo_path.exists():
+            st.image(str(logo_path), width=140)
+        else:
+            st.title("Veureu")
+        if st.session_state.user:
+            st.write(f"Usuari: **{st.session_state.user['username']}** (rol: {st.session_state.user['role']})")
+            
+            # Mostrar estado de verificación SMS
+            show_verification_status_in_sidebar()
+        
+        if st.session_state.user:
+            # Obtener permisos del usuario
+            permissions = get_user_permissions(
+                role,
+                st.session_state.get('sms_verified')
+            )
+            
+            # Construir opciones de navegación según permisos
+            page_options = []
+            
+            if permissions["analizar"]:
+                page_options.append("Analitzar audiodescripcions")
+            
+            if permissions["procesar_videos"]:
+                page_options.append("Processar vídeo nou")
+            
+            if permissions["estadisticas"]:
+                page_options.append("Estadístiques")
+
+            if permissions["validar"]:
+                page_options.append("Validació")
+
+            # Opció de configuració avançada (sala de màquines), només per a l'usuari 'verd'
+            if st.session_state.user.get("username") == "verd":
+                page_options.append("Sala de màquines")
+            
+            # Si no hay opciones disponibles, mostrar solo análisis
+            if not page_options:
+                page_options = ["Analitzar audiodescripcions"]
+            
+            page = st.radio(
+                "Navegació",
+                page_options,
+                index=0
+            )
+            st.markdown("---")
+
+            if st.button(
+                "Confirmar canvis i tancar sessió",
+                key="confirmar_canvis_tancar",
+                use_container_width=True,
+                type="primary",
+            ):
+                # Persistir canvis de la sessió actual abans de tancar
+                try:
+                    base_dir = Path(__file__).parent
+                    session_id = st.session_state.get("session_id", "")
+                    api_client = st.session_state.get("api_client")
+                    digest_info = confirm_changes_and_logout(base_dir, api_client, session_id)
+                except Exception:
+                    digest_info = None
+
+                # Llegir flag public_blockchain_enabled de config.yaml
+                try:
+                    cfg_path = Path(__file__).parent / "config.yaml"
+                    with cfg_path.open("r", encoding="utf-8") as f:
+                        cfg = yaml.safe_load(f) or {}
+                    comp_cfg = cfg.get("compliance", {}) or {}
+                    public_blockchain_enabled = bool(
+                        comp_cfg.get(
+                            "public_blockchain_enabled",
+                            comp_cfg.get("public_blockchain_enable", False),
+                        )
+                    )
+                except Exception:
+                    public_blockchain_enabled = False
+
+                blockchain_published = False
+                polygonscan_url = None
+
+                if public_blockchain_enabled and digest_info and digest_info.get("events_digest"):
+                    try:
+                        session_id = st.session_state.get("session_id", "")
+                        resp = compliance_client.publish_events_digest(
+                            session_id=session_id,
+                            digest_hash=digest_info["events_digest"],
+                        )
+                        if resp:
+                            polygonscan_url = resp.get("transaction_url")
+                            blockchain_published = bool(resp.get("transaction_hash"))
+                    except Exception:
+                        blockchain_published = False
+
+                # Registrar desconnexió a events.db
+                try:
+                    current_user = st.session_state.user or {}
+                    session_id = st.session_state.get("session_id", "")
+                    phone = (
+                        st.session_state.get("sms_phone_verified")
+                        or st.session_state.get("sms_phone")
+                        or ""
+                    )
+                    last_password = st.session_state.get("last_password", "")
+                    log_action(
+                        session=session_id,
+                        user=current_user.get("username", ""),
+                        phone=phone,
+                        action="logout",
+                        sha1sum="",
+                    )
+                except Exception as e:
+                    log(f"Error registrant esdeveniment de logout: {e}")
+
+                # Anotar al log el resultat de la publicació a Polygon (si aplica)
+                digest_hash = digest_info.get("events_digest") if digest_info else None
+                events_count = digest_info.get("events_count") if digest_info else None
+                log(
+                    "Logout completat: "
+                    f"session={session_id or '-'} "
+                    f"events_digest={digest_hash or '-'} "
+                    f"events_count={events_count if events_count is not None else '-'} "
+                    f"polygon_published={'sí' if blockchain_published else 'no'} "
+                    f"polygon_url={polygonscan_url or '-'}"
+                )
+
+                # Netejar sessió d'usuari
+                st.session_state.user = None
+                st.session_state.sms_verified = None
+
+                # Mostrar missatge segons estat de publicació
+                if public_blockchain_enabled and blockchain_published and polygonscan_url:
+                    st.success(
+                        "✅ Els canvis s'han desat i s'han publicat a la cadena de blocs de Polygon. "
+                        f"Pots consultar la transacció a: {polygonscan_url}"
+                    )
+                elif public_blockchain_enabled:
+                    st.info(
+                        "✅ Els canvis s'han desat, però no s'ha pogut completar la publicació "
+                        "a la cadena de blocs de Polygon en aquest moment."
+                    )
+                else:
+                    st.info("✅ Canvis desats (sense publicació a blockchain).")
+
+                st.rerun()
+        else:
+            page = None
+    
+    return page, role