"""UI logic for the "Analitzar audiodescripcions" page.""" from __future__ import annotations import csv import io from pathlib import Path from typing import Dict, Optional import hashlib import runpy import streamlit as st import yaml from utils import save_bytes from persistent_data_gate import ensure_media_for_video from databases import ( get_videos_from_audiodescriptions, get_audiodescription, get_audiodescription_history, update_audiodescription_text, update_audiodescription_info_ad, insert_demo_feedback_row, get_feedback_score_labels, log_action, insert_action, get_latest_user_phone_for_session, get_video_owner_by_sha1, ) def _load_labels_from_config() -> Dict[str, str]: """Carrega les etiquetes dels sliders des de config.yaml. Retorna un dict amb claus score_1..score_6. """ # config.yaml es troba a la carpeta arrel de "demo" cfg_path = Path(__file__).resolve().parent.parent / "config.yaml" try: with cfg_path.open("r", encoding="utf-8") as f: cfg = yaml.safe_load(f) or {} except FileNotFoundError: cfg = {} labels_cfg = cfg.get("labels", {}) or {} return { "score_1": labels_cfg.get("score_1", "Precisió Descriptiva"), "score_2": labels_cfg.get("score_2", "Sincronització Temporal"), "score_3": labels_cfg.get("score_3", "Claredat i Concisió"), "score_4": labels_cfg.get("score_4", "Inclusió de Diàleg/So"), "score_5": labels_cfg.get("score_5", "Contextualització"), "score_6": labels_cfg.get("score_6", "Flux i Ritme de la Narració"), } def _find_best_file_for_version(vid_dir: Path, version: str, filename: str) -> Optional[Path]: """Busca un fitxer dins de temp/media/// amb prioritat. Ordre de cerca de subtipus: "HITL OK" -> "HITL Test" -> "Original" -> arrel de . """ preferred_subtypes = ["HITL OK", "HITL Test", "Original"] for subtype in preferred_subtypes: candidate = vid_dir / version / subtype / filename if candidate.exists(): return candidate legacy = vid_dir / version / filename if legacy.exists(): return legacy return None def _file_for_hist_choice(vid_dir: Path, version: str, filename: str, hist_choice: str) -> Optional[Path]: """Retorna el fitxer per al subtype seleccionat (Original/HITL OK/HITL Test). Si no existeix al subtype triat, fa servir el comportament per defecte de _find_best_file_for_version. """ # Map hist_choice -> subcarpeta física subtype_map = { "Original": "Original", "HITL OK": "HITL OK", "HITL Test": "HITL Test", } subtype = subtype_map.get(hist_choice) if subtype: candidate = vid_dir / version / subtype / filename if candidate.exists(): return candidate return _find_best_file_for_version(vid_dir, version, filename) def load_eval_values(vid_dir: Path, version: str, eval_content: Optional[str] = None) -> Optional[Dict[str, int]]: """Carga los valores de evaluación desde eval (DB o CSV) si existe. Args: vid_dir: Directorio del vídeo version: Versión seleccionada (MoE/Salamandra) Returns: Diccionario con los valores de evaluación o None si no existe el CSV """ csv_path = vid_dir / version / "eval.csv" try: if eval_content is not None: f_obj = io.StringIO(eval_content) elif csv_path.exists(): f_obj = open(csv_path, 'r', encoding='utf-8') else: return None with f_obj as f: reader = csv.DictReader(f) row = next(reader, None) if not row: return None # Mapeo de nombres de columnas CSV a claves internas # Usamos las etiquetas actuales de config.yaml, pero # mantenemos compatibilidad con nombres antiguos. labels = _load_labels_from_config() mappings = { 'transcripcio': [ labels["score_1"], 'Precisió Descriptiva', ], 'identificacio': [ labels["score_2"], 'Sincronització Temporal', ], 'localitzacions': [ labels["score_3"], 'Claredat i Concisió', ], 'activitats': [ labels["score_4"], 'Inclusió de Diàleg/So', 'Inclusió de Diàleg', ], 'narracions': [ labels["score_5"], 'Contextualització', ], 'expressivitat': [ labels["score_6"], 'Flux i Ritme de la Narració', ], } values = {} for key, possible_names in mappings.items(): for name in possible_names: if name in row and row[name]: try: # Convertir a int y asegurar que está en rango 0-7 val = int(float(row[name])) values[key] = max(0, min(7, val)) break except (ValueError, TypeError): continue # Si no se encontró valor, usar 7 por defecto if key not in values: values[key] = 7 return values except Exception: # Si hay cualquier error, simplemente ignorar y devolver None return None def render_analyze_audiodescriptions_page(api, permissions: Dict[str, bool]) -> None: st.header("Analitzar audiodescripcions") # Llista de vídeos disponibles segons demo/temp/db/audiodescriptions.db accessible_rows = get_videos_from_audiodescriptions() # Base de media: demo/temp/media/ base_dir = Path(__file__).resolve().parent.parent base_media_dir = base_dir / "temp" / "media" # Ja no filtrem per media: mostrem tots els vídeos presents a audiodescriptions.db filtered_rows = accessible_rows if not filtered_rows: st.info("No hi ha cap vídeo disponible per analitzar a audiodescriptions.db.") st.stop() if "current_video" not in st.session_state: st.session_state.current_video = None seleccio_row = st.selectbox( "Selecciona un vídeo:", filtered_rows, index=0, format_func=lambda r: r["video_name"], ) # sha1sum per accedir a la carpeta física; video_name per a etiquetes/BD selected_sha1 = seleccio_row["sha1sum"] selected_video_name = seleccio_row["video_name"] or seleccio_row["sha1sum"] if selected_sha1 != st.session_state.current_video: st.session_state.current_video = selected_sha1 if "version_selector" in st.session_state: del st.session_state["version_selector"] st.session_state.add_ad_checkbox = False # Limpiar valores de evaluación al cambiar de vídeo if "eval_values" in st.session_state: del st.session_state["eval_values"] st.rerun() ensure_media_for_video(base_dir, api, selected_sha1) vid_dir = base_media_dir / selected_sha1 mp4s = sorted(vid_dir.glob("*.mp4")) col_video, col_txt = st.columns([2, 1], gap="large") with col_video: # Llistar subcarpetes de versions d'AD, excloent explícitament HITL subcarpetas_ad = ["Salamandra", "MoE"] # Per defecte, prioritzar Salamandra si existeix default_index_sub = 0 subcarpeta_seleccio = st.selectbox( "Selecciona una versió d'audiodescripció:", subcarpetas_ad, index=default_index_sub, key="version_selector", ) # Cargar valores de evaluación cuando cambia la versión if subcarpeta_seleccio: ensure_media_for_video(base_dir, api, selected_sha1, subcarpeta_seleccio) # Clau única per vídeo (sha1) + versió seleccionada current_version_key = f"{selected_sha1}_{subcarpeta_seleccio}" if "last_version_key" not in st.session_state or st.session_state.last_version_key != current_version_key: st.session_state.last_version_key = current_version_key # Intentar llegir eval des de audiodescriptions.db ad_row = get_audiodescription(selected_sha1, subcarpeta_seleccio) eval_text = ad_row["eval"] if ad_row is not None and "eval" in ad_row.keys() else None eval_values = load_eval_values(vid_dir, subcarpeta_seleccio, eval_text) if eval_values: st.session_state.eval_values = eval_values elif "eval_values" in st.session_state: del st.session_state["eval_values"] st.markdown("---") st.markdown("#### Accions") c1, c2, c3 = st.columns(3) with c1: if st.button("Reconstruir àudio amb narració lliure", use_container_width=True, key="rebuild_free_ad"): # Genera nuevo MP3 desde el texto editado en el editor if subcarpeta_seleccio: # Fer servir sempre el subtype triat a l'optionbox d'historial hist_choice = st.session_state.get("ad_hist_choice_" + hist_key_suffix, "HITL OK") free_ad_txt_path = _file_for_hist_choice(vid_dir, subcarpeta_seleccio, "free_ad.txt", hist_choice) if free_ad_txt_path is not None and free_ad_txt_path.exists(): with st.spinner("Generant àudio de la narració lliure..."): # Leer el texto actual del archivo (puede haber sido editado) text_content = free_ad_txt_path.read_text(encoding="utf-8") voice = "central/grau" print(f"🎙️ Reconstruyendo audio desde texto") print(f"📝 Longitud: {len(text_content)} caracteres") response = api.tts_matxa(text=text_content, voice=voice) if "mp3_bytes" in response: subtype_for_files = "HITL Test" output_path = vid_dir / subcarpeta_seleccio / subtype_for_files / "free_ad.mp3" output_path.parent.mkdir(parents=True, exist_ok=True) mp3_bytes = response["mp3_bytes"] save_bytes(output_path, mp3_bytes) # Actualitzar test_free_ad a audiodescriptions.db amb el text utilitzat try: update_audiodescription_text( selected_sha1, subcarpeta_seleccio, test_free_ad=text_content, ) except Exception: pass # Registrar acció "New correction" a actions.db try: session_id = st.session_state.get("session_id", "") user_obj = st.session_state.get("user") or {} username = user_obj.get("username") if isinstance(user_obj, dict) else str(user_obj or "") phone = ( st.session_state.get("sms_phone_verified") or st.session_state.get("sms_phone") or "" ) insert_action( session=session_id or "", user=username or "", phone=phone, action="New correction", sha1sum=selected_sha1, ) except Exception: pass # Posar l'optionbox d'historial automàticament a "HITL Test" st.session_state["ad_hist_choice_" + hist_key_suffix] = "HITL Test" st.success(f"✅ Àudio generat i desat a: {output_path}") st.rerun() else: st.error(f"❌ Error en la generació de l'àudio: {response.get('error', 'Desconegut')}") else: st.warning("⚠️ No s'ha trobat el fitxer 'free_ad.txt' en aquesta versió.") with c2: if st.button("Reconstruir vídeo amb audiodescripció", use_container_width=True, key="rebuild_video_ad"): # Genera video con AD usando el SRT y el video original if subcarpeta_seleccio and mp4s: # Fer servir sempre el subtype triat a l'optionbox d'historial hist_choice = st.session_state.get("ad_hist_choice_" + hist_key_suffix, "HITL OK") une_srt_path = _file_for_hist_choice(vid_dir, subcarpeta_seleccio, "une_ad.srt", hist_choice) video_original_path = mp4s[0] # El único MP4 en videos/ if une_srt_path is not None and une_srt_path.exists(): with st.spinner( "Reconstruint el vídeo amb l'audiodescripció... Aquesta operació pot trigar una estona." ): print(f"🎬 Reconstruyendo video con AD") print(f"📹 Video: {video_original_path}") print(f"📝 SRT: {une_srt_path}") response = api.rebuild_video_with_ad( video_path=str(video_original_path), srt_path=str(une_srt_path), ) if "video_bytes" in response: subtype_for_files = "HITL Test" output_path = vid_dir / subcarpeta_seleccio / subtype_for_files / "une_ad.mp4" output_path.parent.mkdir(parents=True, exist_ok=True) video_bytes = response["video_bytes"] save_bytes(output_path, video_bytes) # Actualitzar test_une_ad a audiodescriptions.db amb el contingut del SRT try: une_text_for_db = "" try: une_text_for_db = une_srt_path.read_text(encoding="utf-8") if une_srt_path is not None else "" except Exception: if une_srt_path is not None: une_text_for_db = une_srt_path.read_text(errors="ignore") if une_text_for_db: update_audiodescription_text( selected_sha1, subcarpeta_seleccio, test_une_ad=une_text_for_db, ) except Exception: pass # Registrar acció "New correction" a actions.db try: session_id = st.session_state.get("session_id", "") user_obj = st.session_state.get("user") or {} username = user_obj.get("username") if isinstance(user_obj, dict) else str(user_obj or "") phone = ( st.session_state.get("sms_phone_verified") or st.session_state.get("sms_phone") or "" ) insert_action( session=session_id or "", user=username or "", phone=phone, action="New correction", sha1sum=selected_sha1, ) except Exception: pass # Posar l'optionbox d'historial automàticament a "HITL Test" st.session_state["ad_hist_choice_" + hist_key_suffix] = "HITL Test" st.success(f"✅ Vídeo amb AD generat i desat a: {output_path}") st.info( "Pots visualitzar-lo activant la casella 'Afegir audiodescripció' i seleccionant el nou fitxer si cal." ) st.rerun() else: st.error(f"❌ Error en la reconstrucció del vídeo: {response.get('error', 'Desconegut')}") else: st.warning("⚠️ No s'ha trobat el fitxer 'une_ad.srt' en aquesta versió.") # Botó per revocar permisos d'ús del vídeo actual with c3: if st.button("Revocar permisos d'ús del vídeo", use_container_width=True, key="revoke_video_permits"): session_id = st.session_state.get("session_id", "") if not session_id: st.error("No s'ha pogut determinar la sessió actual.") else: try: # Telèfon associat al login actual (actions.db) user_for_session, phone_for_session = get_latest_user_phone_for_session(session_id) except Exception: user_for_session, phone_for_session = "", "" try: owner_phone = get_video_owner_by_sha1(selected_sha1) except Exception: owner_phone = "" if phone_for_session and owner_phone and str(phone_for_session) == str(owner_phone): # Registrar acció de revocació i informar a l'usuari try: username = ( st.session_state.get("user", {}).get("username") if isinstance(st.session_state.get("user"), dict) else str(st.session_state.get("user", "")) ) insert_action( session=session_id, user=username or "", phone=str(phone_for_session), action="Revocation of permits", sha1sum=selected_sha1, ) except Exception: pass st.success( "Els permisos per utilitzar el vídeo han estat revocats. " "Has de desar els canvis i en breu rebràs un SMS de confirmació." ) else: st.warning( "No es poden revocar els permisos: el teu telèfon no coincideix amb el propietari del vídeo." ) with col_txt: # Selector de versió temporal de l'audiodescripció (històric) hist_options = ["Original", "HITL OK", "HITL Test"] # Per defecte, mostrar la versió validada per HITL (OK) default_hist_index = 1 # "HITL OK" hist_key_suffix = f"{selected_sha1}_{subcarpeta_seleccio or 'none'}" hist_choice = st.radio( "Edició a mostrar", hist_options, index=default_hist_index, key=f"ad_hist_choice_{hist_key_suffix}", horizontal=True, ) # Seleccionar el vídeo amb AD segons el subtype triat (si existeix) video_ad_path = None if subcarpeta_seleccio: video_ad_path = _file_for_hist_choice(vid_dir, subcarpeta_seleccio, "une_ad.mp4", hist_choice) is_ad_video_available = video_ad_path is not None and video_ad_path.exists() add_ad_video = st.checkbox( "Afegir audiodescripció", disabled=not is_ad_video_available, key="add_ad_checkbox", ) video_to_show = None if add_ad_video and is_ad_video_available: video_to_show = video_ad_path elif mp4s: video_to_show = mp4s[0] if video_to_show: st.video(str(video_to_show)) else: st.warning("No s'ha trobat cap fitxer **.mp4** a la carpeta seleccionada.") tipus_ad_options = ["narració lliure", "UNE-153010"] tipus_ad_seleccio = st.selectbox("Fitxer d'audiodescripció a editar:", tipus_ad_options) text_content = "" if subcarpeta_seleccio: # Llegir des de audiodescriptions.db (totes les versions per sha1+versió) rows = get_audiodescription_history(selected_sha1, subcarpeta_seleccio) selected_row = None if rows: # Mantindrem la mateixa fila segons l'ordre temporal/edicions, # però les diferents opcions del radi mapegen a columnes diferents. if hist_choice == "Original": # Versió original: Llegir dels camps base une_ad/free_ad selected_row = rows[0] elif hist_choice == "HITL OK": # Versió validada per HITL: utilitza els camps ok_* selected_row = rows[-1] else: # "HITL Test": versió de prova selected_row = rows[1] if len(rows) > 1 else rows[0] if selected_row is not None: if hist_choice == "Original": src_une = "une_ad" src_free = "free_ad" elif hist_choice == "HITL OK": src_une = "ok_une_ad" src_free = "ok_free_ad" else: # "HITL Test" src_une = "test_une_ad" src_free = "test_free_ad" if tipus_ad_seleccio == "narració lliure": text_content = selected_row[src_free] if src_free in selected_row.keys() else "" else: text_content = selected_row[src_une] if src_une in selected_row.keys() else "" else: st.info("No s'ha trobat cap registre d'audiodescripció a la base de dades per a aquest vídeo i versió.") else: st.warning("Selecciona una versió per veure els fitxers.") new_text = st.text_area( f"Contingut de {tipus_ad_seleccio}", value=text_content, height=500, key=f"editor_{selected_sha1}_{subcarpeta_seleccio}_{tipus_ad_seleccio}", ) # Solo mostrar botón de reproducción para narración libre if tipus_ad_seleccio == "narració lliure": if st.button( "▶️ Reproduir narració", use_container_width=True, disabled=not subcarpeta_seleccio, key="play_button_editor", ): # Reproduir el MP3 existent segons el subtype d'historial triat if subcarpeta_seleccio: mp3_path = _file_for_hist_choice(vid_dir, subcarpeta_seleccio, "free_ad.mp3", hist_choice) if mp3_path.exists(): try: print(f"🎵 Reproduciendo MP3: {mp3_path}") audio_bytes = mp3_path.read_bytes() st.audio(audio_bytes, format="audio/mp3") except Exception as e: print(f"❌ Error leyendo MP3: {e}") st.error(f"Error llegint l'àudio: {e}") else: st.warning("⚠️ No s'ha trobat el fitxer 'free_ad.mp3'. Genera'l primer amb 'Reconstruir àudio amb narració lliure'.") else: st.info("📝 L'audiodescripció UNE-153010 està en format SRT. Per reproduir-la, genera el vídeo amb el botó de reconstrucció.") if st.button("Desar canvis", use_container_width=True, type="primary"): if not subcarpeta_seleccio: st.error("No s'ha seleccionat una versió vàlida per desar.") else: try: # Actualitzar a la base de dades, mapejant l'opció d'historial # a les columnes corresponents (Original / HITL OK / HITL Test). kwargs = {} if hist_choice == "Original": if tipus_ad_seleccio == "narració lliure": kwargs["free_ad"] = new_text else: kwargs["une_ad"] = new_text elif hist_choice == "HITL OK": if tipus_ad_seleccio == "narració lliure": kwargs["ok_free_ad"] = new_text else: kwargs["ok_une_ad"] = new_text else: # "HITL Test" if tipus_ad_seleccio == "narració lliure": kwargs["test_free_ad"] = new_text else: kwargs["test_une_ad"] = new_text if kwargs: update_audiodescription_text( selected_sha1, subcarpeta_seleccio, **kwargs, ) st.success("Text d'audiodescripció desat correctament a la base de dades.") st.rerun() except Exception as e: st.error(f"No s'ha pogut desar el text a la base de dades: {e}") st.markdown("---") st.subheader("Avaluació de la qualitat de l'audiodescripció") can_rate = permissions.get("valorar", False) controls_disabled = not can_rate # Obtener valores por defecto desde el CSV si existen eval_values = st.session_state.get("eval_values", {}) # Captions governats des de config.yaml labels = _load_labels_from_config() c1, c2, c3 = st.columns(3) with c1: transcripcio = st.slider( labels["score_1"], 0, 7, eval_values.get('transcripcio', 7), disabled=controls_disabled, ) identificacio = st.slider( labels["score_2"], 0, 7, eval_values.get('identificacio', 7), disabled=controls_disabled, ) with c2: localitzacions = st.slider( labels["score_3"], 0, 7, eval_values.get('localitzacions', 7), disabled=controls_disabled, ) activitats = st.slider( labels["score_4"], 0, 7, eval_values.get('activitats', 7), disabled=controls_disabled, ) with c3: narracions = st.slider( labels["score_5"], 0, 7, eval_values.get('narracions', 7), disabled=controls_disabled, ) expressivitat = st.slider( labels["score_6"], 0, 7, eval_values.get('expressivitat', 7), disabled=controls_disabled, ) comments = st.text_area( "Comentaris (opcional)", placeholder="Escriu els teus comentaris lliures…", height=120, disabled=controls_disabled, ) if not can_rate: st.info("El teu rol no permet enviar valoracions.") else: if st.button("Enviar valoració", type="primary", use_container_width=True): try: from databases import add_feedback_ad # Guardar en la base de datos agregada d'AD add_feedback_ad( video_name=selected_video_name, user_id=st.session_state.user["id"], transcripcio=transcripcio, identificacio=identificacio, localitzacions=localitzacions, activitats=activitats, narracions=narracions, expressivitat=expressivitat, comments=comments or None, ) # Determinar versió i llegir UNE/free per a la inserció detallada version = subcarpeta_seleccio or "MoE" video_dir = base_media_dir / selected_sha1 une_path = _find_best_file_for_version(video_dir, version, "une_ad.srt") free_path = _find_best_file_for_version(video_dir, version, "free_ad.txt") try: une_ad_text = ( une_path.read_text(encoding="utf-8") if une_path is not None and une_path.exists() else "" ) except Exception: une_ad_text = ( une_path.read_text(errors="ignore") if une_path is not None and une_path.exists() else "" ) try: free_ad_text = ( free_path.read_text(encoding="utf-8") if free_path is not None and free_path.exists() else "" ) except Exception: free_ad_text = ( free_path.read_text(errors="ignore") if free_path is not None and free_path.exists() else "" ) user_name = ( st.session_state.user.get("username") if isinstance(st.session_state.get("user"), dict) else str(st.session_state.get("user", "")) ) session_id = st.session_state.get("session_id", "") # Inserció detallada a demo/data/feedback.db insert_demo_feedback_row( user=user_name or "", session=session_id or "", video_name=selected_video_name, version=version, une_ad=une_ad_text, free_ad=free_ad_text, comments=comments or None, transcripcio=transcripcio, identificacio=identificacio, localitzacions=localitzacions, activitats=activitats, narracions=narracions, expressivitat=expressivitat, ) # Registrar també l'esdeveniment de feedback a events.db perquè # el digest de sessió el pugui tenir en compte. try: feedback_payload = { "user": user_name or "", "session": session_id or "", "video_name": selected_video_name, "version": version, "transcripcio": transcripcio, "identificacio": identificacio, "localitzacions": localitzacions, "activitats": activitats, "narracions": narracions, "expressivitat": expressivitat, "comments": comments or "", } feedback_str = repr(sorted(feedback_payload.items())) feedback_hash = hashlib.sha1(feedback_str.encode("utf-8")).hexdigest() user_obj = st.session_state.get("user") or {} ip = st.session_state.get("client_ip", "") password = st.session_state.get("last_password", "") phone = ( st.session_state.get("sms_phone_verified") or st.session_state.get("sms_phone") or "" ) log_action( session=session_id or "", user=(user_obj.get("username") if isinstance(user_obj, dict) else str(user_obj or "")), phone=phone, action="Feedback for AD", sha1sum=feedback_hash, ) except Exception: # No interrompre la UX si falla el logging de feedback pass # También guardar en CSV (reubicado en demo/temp/media) csv_path = video_dir / version / "eval.csv" csv_data = [ {"Caracteristica": labels["score_1"], "Valoracio (0-7)": str(transcripcio), "Justificacio": comments or ""}, {"Caracteristica": labels["score_2"], "Valoracio (0-7)": str(identificacio), "Justificacio": comments or ""}, {"Caracteristica": labels["score_3"], "Valoracio (0-7)": str(localitzacions), "Justificacio": comments or ""}, {"Caracteristica": labels["score_4"], "Valoracio (0-7)": str(activitats), "Justificacio": comments or ""}, {"Caracteristica": labels["score_5"], "Valoracio (0-7)": str(narracions), "Justificacio": comments or ""}, {"Caracteristica": labels["score_6"], "Valoracio (0-7)": str(expressivitat), "Justificacio": comments or ""}, ] csv_path.parent.mkdir(parents=True, exist_ok=True) with open(csv_path, 'w', newline='', encoding='utf-8') as f: writer = csv.DictWriter(f, fieldnames=["Caracteristica", "Valoracio (0-7)", "Justificacio"]) writer.writeheader() writer.writerows(csv_data) st.success("Gràcies! La teva valoració s'ha desat correctament.") except Exception as e: st.error(f"S'ha produït un error en desar la valoració: {e}")