"""UI logic for the "Validació" page.""" from __future__ import annotations from datetime import datetime from pathlib import Path from typing import Dict import sys import shutil import streamlit as st from databases import ( get_accessible_videos_with_sha1, log_event, get_audiodescription_history, update_audiodescription_text, ) from persistent_data_gate import _load_data_origin def _log(msg: str) -> None: """Helper de logging a stderr amb timestamp (coherent amb auth.py).""" ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S") sys.stderr.write(f"[{ts}] {msg}\n") sys.stderr.flush() def render_validation_page( compliance_client, runtime_videos: Path, permissions: Dict[str, bool], username: str, ) -> None: if not permissions.get("validar", False): st.warning("⚠️ No tens permisos per accedir a aquesta secció de validació.") st.stop() st.header("🔍 Validació de Vídeos") tab_videos, tab_ads = st.tabs(["📹 Validar Vídeos", "🎬 Validar Audiodescripcions"]) base_dir = Path(__file__).resolve().parent.parent data_origin = _load_data_origin(base_dir) # Llista de vídeos accessibles (mode internal) o pendents al backend (mode external) session_id = st.session_state.get("session_id") accessible_rows = get_accessible_videos_with_sha1(session_id) if data_origin == "internal" else [] # Rutes base per a media i vídeos pendents base_media_dir = base_dir / "temp" / "media" pending_root = base_dir / "temp" / "pending_videos" with tab_videos: st.subheader("📹 Validar Vídeos Pujats") video_folders = [] # Botó per actualitzar manualment la llista de vídeos pendents des de l'engine col_refresh_list, _ = st.columns([1, 3]) with col_refresh_list: if st.button("🔄 Actualitzar llista de vídeos pendents", key="refresh_pending_videos_list"): st.rerun() if data_origin == "internal": # Mode intern: llistar carpetes de vídeos pendents des de temp/pending_videos if pending_root.exists() and pending_root.is_dir(): for folder in sorted(pending_root.iterdir()): if not folder.is_dir(): continue sha1 = folder.name video_files = list(folder.glob("*.mp4")) + list(folder.glob("*.avi")) + list(folder.glob("*.mov")) if not video_files: continue mod_time = folder.stat().st_mtime fecha = datetime.fromtimestamp(mod_time).strftime("%Y-%m-%d %H:%M") video_folders.append( { "sha1sum": sha1, "video_name": sha1, "path": str(folder), "created_at": fecha, "video_files": video_files, } ) else: # Mode external: llistar vídeos pendents des de l'engine api_client = st.session_state.get("api_client") if api_client is not None: try: resp = api_client.list_pending_videos() _log(f"[pending_videos] list_pending_videos raw resp type= {type(resp)}") _log(f"[pending_videos] list_pending_videos raw resp content= {repr(resp)}") except Exception as e_list: _log(f"[pending_videos] Error cridant list_pending_videos: {e_list}") resp = {"error": "exception"} pending_list = [] if isinstance(resp, dict) and not resp.get("error"): # Pot ser un dict amb clau "videos" o directament una llista if isinstance(resp.get("videos"), list): pending_list = resp["videos"] elif isinstance(resp.get("items"), list): pending_list = resp["items"] elif isinstance(resp.get("results"), list): pending_list = resp["results"] elif isinstance(resp, list): pending_list = resp elif isinstance(resp, list): pending_list = resp _log(f"[pending_videos] parsed pending_list length= {len(pending_list) if isinstance(pending_list, list) else 'N/A'}") if isinstance(pending_list, list) and pending_list: _log(f"[pending_videos] first items: {pending_list[:3]}") for item in pending_list: sha1 = item.get("sha1") or item.get("video_hash") or item.get("id") if not sha1: continue video_name = item.get("latest_video") or sha1 # Carpeta local on descarregarem el vídeo pendent si cal folder = pending_root / sha1 if folder.exists(): video_files = list(folder.glob("*.mp4")) else: video_files = [] created_at = item.get("created_at") or datetime.utcnow().strftime("%Y-%m-%d %H:%M") video_folders.append( { "sha1sum": sha1, "video_name": video_name, "path": str(folder), "created_at": created_at, "video_files": video_files, } ) if not video_folders: st.info("📝 No hi ha vídeos pujats pendents de validació.") else: opciones_video = [f"{video['video_name']} - {video['created_at']}" for video in video_folders] seleccion = st.selectbox( "Selecciona un vídeo per validar:", opciones_video, index=0 if opciones_video else None, ) if seleccion: indice = opciones_video.index(seleccion) video_seleccionat = video_folders[indice] col1, col2 = st.columns([2, 1]) with col1: st.markdown("### 📹 Informació del Vídeo") st.markdown(f"**Nom:** {video_seleccionat['video_name']}") st.markdown(f"**Data:** {video_seleccionat['created_at']}") st.markdown(f"**Arxius:** {len(video_seleccionat['video_files'])} vídeos trobats") # Assegurar que disposem del fitxer local en mode external if data_origin == "external" and not video_seleccionat["video_files"]: api_client = st.session_state.get("api_client") if api_client is not None: try: resp = api_client.download_pending_video(video_seleccionat["sha1sum"]) except Exception: resp = {"error": "exception"} video_bytes = ( resp.get("video_bytes") if isinstance(resp, dict) else None ) if video_bytes: local_folder = pending_root / video_seleccionat["sha1sum"] local_folder.mkdir(parents=True, exist_ok=True) local_path = local_folder / "video.mp4" with local_path.open("wb") as f: f.write(video_bytes) video_seleccionat["video_files"] = [local_path] if video_seleccionat["video_files"]: video_path = str(video_seleccionat["video_files"][0]) st.markdown("**Vídeo principal:**") st.video(video_path) else: st.warning("⚠️ No s'han trobat arxius de vídeo.") with col2: st.markdown("### 🔍 Accions de Validació") col_btn1, col_btn2 = st.columns(2) with col_btn1: if st.button("✅ Acceptar", type="primary", key=f"accept_video_{video_seleccionat['sha1sum']}"): # 1) Registrar decisió al servei de compliance success = compliance_client.record_validator_decision( document_id=f"video_{video_seleccionat['video_name']}", validator_email=f"{username}@veureu.local", decision="acceptat", comments=f"Vídeo validat per {username}", ) # 2) Registrar esdeveniment "video approval" a events.db session_id = st.session_state.get("session_id") or "" client_ip = st.session_state.get("client_ip") or "" phone = st.session_state.get("phone_number") or "" password = st.session_state.get("password") or "" try: log_event( session=session_id, ip=client_ip, user=username or "", password=password, phone=phone, action="video approval", sha1sum=video_seleccionat["sha1sum"], visibility=None, ) except Exception as e: st.warning(f"⚠️ No s'ha pogut registrar l'esdeveniment d'aprovació: {e}") if success: st.success("✅ Vídeo acceptat, registrat al servei de compliance i marcat com aprovat a events.db") else: st.error("❌ Error registrant el veredicte al servei de compliance") # 3) En mode external, moure el vídeo de temp/pending_videos a temp/media if data_origin == "external": sha1 = video_seleccionat["sha1sum"] local_pending_dir = pending_root / sha1 local_media_dir = base_media_dir / sha1 try: local_media_dir.mkdir(parents=True, exist_ok=True) src = local_pending_dir / "video.mp4" if src.exists(): dst = local_media_dir / "video.mp4" shutil.copy2(src, dst) if local_pending_dir.exists(): shutil.rmtree(local_pending_dir) except Exception: pass with col_btn2: if st.button("❌ Rebutjar", type="secondary", key=f"reject_video_{video_seleccionat['video_name']}"): success = compliance_client.record_validator_decision( document_id=f"video_{video_seleccionat['video_name']}", validator_email=f"{username}@veureu.local", decision="rebutjat", comments=f"Vídeo rebutjat per {username}", ) if success: st.success("✅ Vídeo rebutjat i registrat al servei de compliance") else: st.error("❌ Error registrant el veredicte") with tab_ads: st.subheader("🎬 Validar Audiodescripcions") # Llistar vídeos la darrera acció dels quals a events.db sigui "Waiting for UNE validation" # i permetre validar l'audiodescripció (UNE + narració lliure). # Construir mapa sha1 -> video_name a partir de vídeos accessibles sha1_to_name = {row["sha1sum"]: (row["video_name"] or row["sha1sum"]) for row in accessible_rows} # Llegir events.db directament per obtenir l'última acció per cada sha1 from databases import _connect_events_db # tipus intern però útil aquí pending_videos = [] try: with _connect_events_db() as conn: cur = conn.execute( """ SELECT sha1sum, MAX(timestamp) AS latest_ts FROM events GROUP BY sha1sum """ ) latest_by_sha1 = {row["sha1sum"]: row["latest_ts"] for row in cur.fetchall()} # Filtrar aquells on l'última acció és "Waiting for UNE validation" for sha1, latest_ts in latest_by_sha1.items(): if not sha1: continue row = conn.execute( "SELECT action FROM events WHERE sha1sum=? AND timestamp=?", (sha1, latest_ts), ).fetchone() if row and row["action"] == "Waiting for UNE validation": pending_videos.append({ "sha1sum": sha1, "video_name": sha1_to_name.get(sha1, sha1), }) except Exception as e_ev: _log(f"[UNE validation] Error llegint events.db: {e_ev}") if not pending_videos: st.info("📝 No hi ha audiodescripcions pendents de validació UNE.") else: options = [f"{v['video_name']} ({v['sha1sum']})" for v in pending_videos] seleccion_ad = st.selectbox( "Selecciona una audiodescripció per validar:", options, index=0 if options else None, ) if seleccion_ad: idx = options.index(seleccion_ad) sel = pending_videos[idx] sha1 = sel["sha1sum"] video_name = sel["video_name"] # Preferim la versió MoE si existeix; si no, Salamandra selected_version = None for v in ("MoE", "Salamandra"): rows = get_audiodescription_history(sha1, v) if rows: selected_version = v ad_rows = rows break if not selected_version: st.error("No s'ha trobat cap audiodescripció per a aquest vídeo a la base de dades.") else: # Agafem la darrera fila per a aquesta versió row_ad = ad_rows[-1] current_une = row_ad.get("une_ad") or "" current_free = row_ad.get("free_ad") or "" col1, col2 = st.columns([2, 1]) with col1: st.markdown("### 🎬 Informació de l'Audiodescripció") st.markdown(f"**Vídeo:** {video_name}") st.markdown(f"**SHA1:** {sha1}") st.markdown(f"**Versió:** {selected_version}") video_dir = base_media_dir / sha1 video_file = None if video_dir.exists(): for cand in [video_dir / "video.mp4", video_dir / "video.avi", video_dir / "video.mov"]: if cand.exists(): video_file = cand break if video_file is not None: st.video(str(video_file)) else: st.warning("⚠️ No s'ha trobat el fitxer de vídeo original.") st.markdown("#### 📝 Audiodescripció UNE-153010 (SRT)") new_une = st.text_area( "Text SRT UNE-153010", value=current_une, height=220, key=f"une_editor_{sha1}_{selected_version}", ) st.markdown("#### 🗒️ Narració lliure") new_free = st.text_area( "Text narració lliure", value=current_free, height=220, key=f"free_editor_{sha1}_{selected_version}", ) with col2: st.markdown("### 🔍 Accions de Validació") if st.button("✅ Acceptar", type="primary", key=f"accept_ad_{sha1}_{selected_version}"): try: # 1) Registrar decisió al servei de compliance (opcional, com abans) try: success = compliance_client.record_validator_decision( document_id=f"ad_{video_name}", validator_email=f"{username}@veureu.local", decision="acceptat", comments=f"Audiodescripció UNE validada per {username}", ) if not success: st.warning("⚠️ No s'ha pogut registrar el veredicte al servei de compliance.") except Exception as e_comp: _log(f"[UNE validation] Error amb compliance: {e_comp}") # 2) Actualitzar camps OK/TEST per a aquest vídeo/versió update_audiodescription_text( sha1sum=sha1, version=selected_version, ok_une_ad=new_une, test_une_ad=new_une, ok_free_ad=new_free, test_free_ad=new_free, ) # 3) Registrar event 'Validated AD UNE-153020' try: log_event( session=session_id or "", ip=st.session_state.get("client_ip", "") or "", user=username or "", password=st.session_state.get("last_password", "") or "", phone=st.session_state.get("sms_phone_verified") or st.session_state.get("sms_phone") or "", action="Validated AD UNE-153020", sha1sum=sha1, visibility=None, ) except Exception as e_evt: _log(f"[UNE validation] Error registrant event Validated AD UNE-153020: {e_evt}") st.success("✅ Audiodescripció UNE-153010 validada i desada (OK/TEST).") st.rerun() except Exception as e_val: st.error(f"❌ Error durant la validació de l'audiodescripció: {e_val}") st.markdown("---") st.markdown("### ℹ️ Informació del Procés de Validació") st.markdown( """ - **Tots els veredictes** es registren al servei de compliance per garantir la traçabilitat - **Cada validació** inclou veredicte, nom del vídeo i validador responsable - **Els registres** compleixen amb la normativa AI Act i GDPR """ )