import gradio as gr from pathlib import Path from datetime import datetime import os import json import uuid from collections import OrderedDict def _get_selected_idx(audio_infos, selected_idx): selected_idx = int(selected_idx) if selected_idx is not None else 0 if selected_idx >= len(audio_infos): selected_idx = len(audio_infos) -1 elif selected_idx < 0: selected_idx = 0 if len(audio_infos) == 0: selected_idx = -1 return selected_idx class AudioGallery: """ A custom Gradio component that displays an audio gallery with thumbnails. Args: audio_paths: List of audio file paths selected_index: Initially selected index (default: 0) max_thumbnails: Maximum number of thumbnails to display (default: 10) height: Height of the gallery in pixels (default: 400) label: Label for the component (default: "Audio Gallery") update_only: If True, only render the inner HTML/Audio (internal use) """ _timestamp_cache = OrderedDict() _timestamp_cache_max_entries = 50 _gradio_upload_mtime_patch_installed = False def __init__(self, audio_paths=None, selected_index=-1, max_thumbnails=10, height=400, label="Audio Gallery", update_only=False): self.audio_paths = audio_paths or [] self.selected_index = selected_index self.max_thumbnails = max_thumbnails self.height = height self.label = label self._render(update_only) # ------------------------------- # Public API # ------------------------------- @staticmethod def get_javascript(): """ Returns JavaScript code to append into Blocks(js=...). This code assumes it will be concatenated with other JS (no IIFE wrapper). """ return r""" // ===== AudioGallery: global-safe setup (no IIFE) ===== window.__agNS = window.__agNS || {}; const AG = window.__agNS; // Singleton-ish state & guards AG.state = AG.state || { manual: false, prevScroll: 0 }; AG.init = AG.init || false; AG.moMap = AG.moMap || {}; // by element id, if needed later AG.uploadPatchInit = AG.uploadPatchInit || false; // Helpers scoped on the namespace AG.root = function () { return document.querySelector('#audio_gallery_html'); }; AG.container = function () { const r = AG.root(); return r ? r.querySelector('.thumbnails-container') : null; }; // Install global listeners once if (!AG.init) { // Track scroll position of the thumbnails container document.addEventListener('scroll', (e) => { const c = AG.container(); if (c && e.target === c) AG.state.prevScroll = c.scrollLeft; }, true); // Delegate click handling to select thumbnails (manual interaction) document.addEventListener('click', function (e) { const thumb = e.target.closest('.audio-thumbnail'); if (!thumb) return; const c = AG.container(); if (c) AG.state.prevScroll = c.scrollLeft; // snapshot BEFORE backend updates const idx = thumb.getAttribute('data-index'); window.selectAudioThumbnail(idx); }); AG.init = true; } // Patch upload requests once to forward browser lastModified timestamps. if (!AG.uploadPatchInit) { AG._origFetch = AG._origFetch || window.fetch.bind(window); window.fetch = function (input, init) { let nextInit = init; try { const reqUrl = (typeof input === 'string') ? input : (input && input.url ? input.url : ''); const reqMethod = ((init && init.method) || (input && input.method) || 'GET').toUpperCase(); const reqBody = init && init.body; const isUploadRequest = reqMethod === 'POST' && typeof reqUrl === 'string' && (reqUrl.includes('/upload') || reqUrl.includes('/gradio_api/upload')); if (isUploadRequest && reqBody instanceof FormData && reqBody.has('files')) { const files = reqBody.getAll('files'); const lastModifiedValues = []; for (const oneFile of files) { if (typeof File !== 'undefined' && oneFile instanceof File) { lastModifiedValues.push(Number(oneFile.lastModified || 0)); } else { lastModifiedValues.push(0); } } if (lastModifiedValues.length > 0) { const headers = new Headers((init && init.headers) || {}); headers.set('x-wangp-file-last-modified', JSON.stringify(lastModifiedValues)); nextInit = Object.assign({}, init || {}, { headers }); } } } catch (_) {} return AG._origFetch(input, nextInit); }; AG.uploadPatchInit = true; } // Manual selection trigger (used by click handler and callable from elsewhere) window.selectAudioThumbnail = function (index) { const c = AG.container(); if (c) AG.state.prevScroll = c.scrollLeft; // snapshot BEFORE Gradio re-renders AG.state.manual = true; const hiddenTextbox = document.querySelector('#audio_gallery_click_data textarea'); const hiddenButton = document.querySelector('#audio_gallery_click_trigger'); if (hiddenTextbox && hiddenButton) { hiddenTextbox.value = String(index); hiddenTextbox.dispatchEvent(new Event('input', { bubbles: true })); hiddenButton.click(); } // Brief window marking this as manual to suppress auto-scroll setTimeout(() => { AG.state.manual = false; }, 500); }; // Ensure selected thumbnail is fully visible (programmatic only) AG.ensureVisibleIfNeeded = function () { const c = AG.container(); const r = AG.root(); if (!c || !r) return; const sel = r.querySelector('.audio-thumbnail.selected'); if (!sel) return; const left = sel.offsetLeft; const right = left + sel.clientWidth; const viewLeft = c.scrollLeft; const viewRight = viewLeft + c.clientWidth; // Already fully visible if (left >= viewLeft && right <= viewRight) return; // Animate from CURRENT position (which we restore first) const target = left - (c.clientWidth / 2) + (sel.clientWidth / 2); const start = c.scrollLeft; const dist = target - start; const duration = 300; let t0 = null; function step(ts) { if (t0 === null) t0 = ts; const p = Math.min((ts - t0) / duration, 1); const ease = p < 0.5 ? 2 * p * p : 1 - Math.pow(-2 * p + 2, 2) / 2; c.scrollLeft = start + dist * ease; if (p < 1) requestAnimationFrame(step); } requestAnimationFrame(step); }; // Observe Gradio's DOM replacement of the HTML component and restore scroll AG.installObserver = function () { const rootEl = AG.root(); if (!rootEl) return; // Reuse/detach previous observer tied to this element id if needed const key = 'audio_gallery_html'; if (AG.moMap[key]) { try { AG.moMap[key].disconnect(); } catch (_) {} } const mo = new MutationObserver(() => { const c = AG.container(); if (!c) return; // 1) Always restore the last known scroll immediately (prevents jump-to-0) if (typeof AG.state.prevScroll === 'number') c.scrollLeft = AG.state.prevScroll; // 2) Only auto-scroll for programmatic changes if (!AG.state.manual) requestAnimationFrame(AG.ensureVisibleIfNeeded); }); mo.observe(rootEl, { childList: true, subtree: true }); AG.moMap[key] = mo; }; // Try to install immediately; if not present yet, retry shortly AG.tryInstall = function () { if (AG.root()) { AG.installObserver(); } else { setTimeout(AG.tryInstall, 50); } }; // Kick things off AG.tryInstall(); // ===== end AudioGallery JS ===== """ @classmethod def install_gradio_upload_mtime_patch(cls): """Monkey-patch Gradio multipart parser to keep browser file lastModified as temp-file mtime.""" if cls._gradio_upload_mtime_patch_installed: return True try: from gradio.route_utils import GradioMultiPartParser, GradioUploadFile except Exception: return False original_parse = getattr(GradioMultiPartParser, "parse", None) if original_parse is None: return False if getattr(original_parse, "__wangp_upload_mtime_patch__", False): cls._gradio_upload_mtime_patch_installed = True return True header_name = "x-wangp-file-last-modified" async def patched_parse(parser_self, *args, **kwargs): form = await original_parse(parser_self, *args, **kwargs) try: header_payload = None if hasattr(parser_self, "headers") and parser_self.headers is not None: header_payload = parser_self.headers.get(header_name) if not header_payload: return form mtime_values = json.loads(header_payload) if not isinstance(mtime_values, list) or len(mtime_values) == 0: return form uploaded_files = form.getlist("files") for idx, file_obj in enumerate(uploaded_files): if idx >= len(mtime_values): break ts_ms = mtime_values[idx] if isinstance(ts_ms, dict): ts_ms = ts_ms.get("lastModified", ts_ms.get("last_modified", 0)) try: ts_ms = float(ts_ms) except Exception: continue if ts_ms <= 0: continue ts = ts_ms / 1000.0 if ts_ms > 100_000_000_000 else ts_ms tmp_path = None if isinstance(file_obj, GradioUploadFile): tmp_path = getattr(getattr(file_obj, "file", None), "name", None) elif hasattr(file_obj, "file"): tmp_path = getattr(file_obj.file, "name", None) if tmp_path and os.path.exists(tmp_path): os.utime(tmp_path, (ts, ts)) except Exception: pass return form patched_parse.__wangp_upload_mtime_patch__ = True GradioMultiPartParser.parse = patched_parse cls._gradio_upload_mtime_patch_installed = True return True def get_state(self): """Get the state components for use in other Gradio events. Returns: (state_paths, state_selected, refresh_trigger)""" return self.state_paths, self.state_selected, self.refresh_trigger def update( self, new_audio_paths=None, new_selected_index=None, current_paths_json=None, current_selected=None, ): """ Programmatically update the gallery with new audio paths and/or selected index. Returns: (state_paths_json, state_selected_index, refresh_trigger_id) """ # Decide which paths to use if new_audio_paths is not None: paths = new_audio_paths paths_json = json.dumps(paths) elif current_paths_json: paths_json = current_paths_json paths = json.loads(current_paths_json) else: paths = [] paths_json = json.dumps([]) audio_infos = self._process_audio_paths(paths) # Decide which selected index to use if new_selected_index is not None: try: selected_idx = int(new_selected_index) except Exception: selected_idx = 0 elif current_selected is not None: try: selected_idx = int(current_selected) except Exception: selected_idx = 0 else: selected_idx = 0 selected_idx = _get_selected_idx(audio_infos, selected_idx) # Trigger id to notify the frontend to refresh (observed via MutationObserver on HTML rerender) refresh_id = str(uuid.uuid4()) return paths_json, selected_idx, refresh_id # ------------------------------- # Internal plumbing # ------------------------------- def _render(self, update_only): """Internal render method called during initialization.""" with gr.Column() as self.component: # Persistent state components self.state_paths = gr.Textbox( value=json.dumps(self.audio_paths), visible=False, elem_id="audio_gallery_state_paths", ) selected_index = self.selected_index self.state_selected = gr.Number( value=selected_index, visible=False, elem_id="audio_gallery_state_selected", ) # Trigger for refreshing the gallery (programmatic) self.refresh_trigger = gr.Textbox( value="", visible=False, elem_id="audio_gallery_refresh_trigger", ) # Process audio paths (keep provided order) audio_infos = self._process_audio_paths(self.audio_paths) # Default selection default_audio = ( audio_infos[selected_index]["path"] if audio_infos and selected_index < len(audio_infos) else (audio_infos[0]["path"] if audio_infos else None) ) # Store for later use self.current_audio_infos = audio_infos # Wrapper for audio player with fixed height with gr.Column(elem_classes="audio-player-wrapper"): self.audio_player = gr.Audio( value=default_audio, label=self.label, type="filepath" ) # Create the gallery HTML (filename + thumbnails) self.gallery_html = gr.HTML( value=self._create_gallery_html(audio_infos, selected_index), elem_id="audio_gallery_html", # stable anchor for MutationObserver ) if update_only: return # Hidden textbox to capture clicks self.click_data = gr.Textbox( value="", visible=False, elem_id="audio_gallery_click_data" ) # Hidden button to trigger the update self.click_trigger = gr.Button( visible=False, elem_id="audio_gallery_click_trigger" ) # Set up the click handler self.click_trigger.click( fn=self._select_audio, inputs=[self.click_data, self.state_paths, self.state_selected], outputs=[ self.audio_player, self.gallery_html, self.state_paths, self.state_selected, self.click_data, ], show_progress="hidden", ) # Set up the refresh handler (programmatic updates) self.refresh_trigger.change( fn=self._refresh_gallery, inputs=[self.refresh_trigger, self.state_paths, self.state_selected], outputs=[self.audio_player, self.gallery_html, self.state_selected], show_progress="hidden", ) def _select_audio(self, click_value, paths_json, current_selected): """Handle thumbnail selection (manual).""" if not click_value: return self._render_from_state(paths_json, current_selected) try: paths = json.loads(paths_json) if paths_json else [] audio_infos = self._process_audio_paths(paths) if not audio_infos: return None, self._create_gallery_html([], 0), paths_json, -1, "" new_index = int(click_value) if 0 <= new_index < len(audio_infos): selected_path = audio_infos[new_index]["path"] if not os.path.exists(selected_path): selected_path = None return ( selected_path, self._create_gallery_html(audio_infos, new_index), paths_json, new_index, "", ) except Exception: pass return self._render_from_state(paths_json, current_selected) def _refresh_gallery(self, refresh_id, paths_json, selected_idx): """Refresh gallery based on state (programmatic).""" if not refresh_id: return self._render_from_state(paths_json, selected_idx)[:2], selected_idx try: paths = json.loads(paths_json) if paths_json else [] audio_infos = self._process_audio_paths(paths) if not audio_infos: return None, self._create_gallery_html([], 0), -1 selected_idx = _get_selected_idx(audio_infos, selected_idx) selected_path = audio_infos[selected_idx]["path"] gallery_html_content = self._create_gallery_html(audio_infos, selected_idx) return selected_path, gallery_html_content, selected_idx except Exception: return None, self._create_gallery_html([], 0), -1 def _get_audio_duration(self, audio_path): """Get audio duration in seconds. Returns formatted string.""" try: import wave import contextlib # Try WAV format first try: with contextlib.closing(wave.open(audio_path, "r")) as f: frames = f.getnframes() rate = f.getframerate() duration = frames / float(rate) return self._format_duration(duration) except Exception: pass # For other formats, try using mutagen if available try: from mutagen import File audio = File(audio_path) if audio and getattr(audio, "info", None): return self._format_duration(audio.info.length) except Exception: pass # Fallback to file size estimation (very rough) file_size = os.path.getsize(audio_path) estimated_duration = file_size / 32000 # bytes per second guess return self._format_duration(estimated_duration) except Exception: return "0:00" @classmethod def _get_cached_creation_datetime(cls, audio_path): try: stat = os.stat(audio_path) cache_key = (audio_path, int(stat.st_size), int(stat.st_mtime)) except Exception: return None cached_dt = cls._timestamp_cache.get(cache_key) if cached_dt is not None: cls._timestamp_cache.move_to_end(cache_key) return cached_dt dt = None try: from shared.utils.audio_metadata import read_audio_metadata, resolve_audio_creation_datetime wangp_metadata = read_audio_metadata(audio_path) dt = resolve_audio_creation_datetime(audio_path, wangp_metadata=wangp_metadata) except Exception: dt = None if dt is None: try: if os.name == "nt": ts = os.path.getctime(audio_path) else: stat = os.stat(audio_path) ts = stat.st_birthtime if hasattr(stat, "st_birthtime") else stat.st_mtime dt = datetime.fromtimestamp(ts) except Exception: return None cls._timestamp_cache[cache_key] = dt while len(cls._timestamp_cache) > cls._timestamp_cache_max_entries: cls._timestamp_cache.popitem(last=False) return dt def _format_duration(self, seconds): """Format duration in seconds to MM:SS format.""" mins = int(seconds // 60) secs = int(seconds % 60) return f"{mins}:{secs:02d}" def _get_file_info(self, audio_path, not_found=False): """Get file information: basename, date/time, duration.""" p = Path(audio_path) basename = p.name if not_found: timestamp = "" date_str = "Deleted" time_str = "00:00:00" duration = "0:00" else: dt = self._get_cached_creation_datetime(audio_path) if dt is None: dt = datetime.fromtimestamp(os.path.getmtime(audio_path)) date_str = dt.strftime("%Y-%m-%d") time_str = dt.strftime("%H:%M:%S") timestamp = dt.timestamp() # Get duration duration = self._get_audio_duration(audio_path) return { "basename": basename, "date": date_str, "time": time_str, "duration": duration, "path": audio_path, "timestamp": timestamp, } def _create_thumbnail_html(self, info, index, is_selected): """Create HTML for a thumbnail.""" selected_class = "selected" if is_selected else "" return f"""
{info['date']}
{info['time']}
{info['duration']}
""" def _create_gallery_html(self, audio_infos, selected_index): """Create the complete gallery HTML.""" thumbnails_html = "" num_thumbnails = len(audio_infos) # Calculate thumbnail width based on number of thumbnails if num_thumbnails > 0: thumbnail_width = max(80, min(150, 100 - (num_thumbnails - 1) * 2)) else: thumbnail_width = 100 for i, info in enumerate(audio_infos): is_selected = i == selected_index thumbnails_html += self._create_thumbnail_html(info, i, is_selected) selected_basename = ( audio_infos[selected_index]["basename"] if (audio_infos and 0 <= selected_index < len(audio_infos)) else "" ) gallery_html = f""" """ return gallery_html def _process_audio_paths(self, paths): """Process audio paths and return audio infos in the same order.""" audio_infos = [] if paths: for path in paths: try: if os.path.exists(path): audio_infos.append(self._get_file_info(path)) else: audio_infos.append(self._get_file_info(path, True)) except Exception: continue audio_infos = audio_infos[: self.max_thumbnails] return audio_infos def _render_from_state(self, paths_json, selected_idx): """Render gallery from state.""" try: paths = json.loads(paths_json) if paths_json else [] audio_infos = self._process_audio_paths(paths) if not audio_infos: return None, self._create_gallery_html([], 0), paths_json, -1, "" selected_idx = _get_selected_idx(audio_infos, selected_idx) selected_path = audio_infos[selected_idx]["path"] if not os.path.exists(selected_path): selected_path = None gallery_html_content = self._create_gallery_html(audio_infos, selected_idx) return selected_path, gallery_html_content, paths_json, selected_idx, "" except Exception: return None, self._create_gallery_html([], 0), paths_json, -1, ""