"""SSH file browser UI.""" from __future__ import annotations import json from pathlib import PurePosixPath from typing import Callable import pandas as pd import streamlit as st from st_aggrid import AgGrid, DataReturnMode, GridOptionsBuilder, GridUpdateMode, JsCode from media_library_viewer.utils import human_size, timestamp_to_local FILE_BROWSER_FILTER_KEYS = [ "file_browser_kind_filter", "file_browser_search", "file_browser_extension_filter", "file_browser_sort", "file_browser_descending", "file_browser_page", ] def reset_file_browser_filters() -> None: """Clear directory-local filters before opening another folder.""" for key in FILE_BROWSER_FILTER_KEYS: st.session_state.pop(key, None) def set_file_browser_path(path: str, selected_path: str | None = None, reset_filters: bool = True) -> None: """Set current directory and selected path for the File browser.""" if reset_filters: st.session_state["file_browser_reset_filters_pending"] = True st.session_state["file_browser_current_dir"] = path st.session_state["file_browser_selected_path"] = selected_path or path st.session_state["file_browser_sync_path_input"] = True def aggrid_selected_rows(response: dict) -> list[dict]: """Return selected rows from a streamlit-aggrid response.""" selected_rows = response.get("selected_rows") if selected_rows is None: return [] if isinstance(selected_rows, pd.DataFrame): return selected_rows.to_dict("records") return list(selected_rows) def render_file_browser(cached_dir_listing: Callable[..., list[dict]], ssh_args: tuple, initial_path: str) -> str: """Render the File browser and return the selected file/folder path.""" host, username, port, key_filename, password = ssh_args st.subheader("Remote filesystem") if "file_browser_current_dir" not in st.session_state: st.session_state["file_browser_current_dir"] = initial_path or "/" if "file_browser_selected_path" not in st.session_state: st.session_state["file_browser_selected_path"] = st.session_state["file_browser_current_dir"] if st.session_state.pop("file_browser_reset_filters_pending", False): reset_file_browser_filters() if "file_browser_path_input" not in st.session_state or st.session_state.pop("file_browser_sync_path_input", False): st.session_state["file_browser_path_input"] = st.session_state["file_browser_current_dir"] current_dir = st.session_state["file_browser_current_dir"] or "/" selected = st.session_state.get("file_browser_selected_path", current_dir) status_col, selected_col = st.columns([1, 2]) status_col.caption(f"Current folder: `{current_dir}`") selected_col.caption(f"Selected path: `{selected}`") path_col, up_col, go_col, select_col, refresh_col = st.columns([5, 1, 1, 1.5, 1.5]) path_input = path_col.text_input("Remote path", key="file_browser_path_input", label_visibility="collapsed") requested_path = path_input or "/" if up_col.button("Up", key="file_browser_up", use_container_width=True): set_file_browser_path(str(PurePosixPath(current_dir).parent)) st.rerun() if go_col.button("Go", key="file_browser_go", use_container_width=True): set_file_browser_path(requested_path) st.rerun() if select_col.button("Select folder", key="file_browser_select_current_folder", use_container_width=True): if requested_path != current_dir: set_file_browser_path(requested_path, requested_path) else: st.session_state["file_browser_selected_path"] = current_dir st.rerun() if refresh_col.button("Refresh", key="file_browser_refresh", use_container_width=True): cached_dir_listing.clear() if requested_path != current_dir: set_file_browser_path(requested_path) st.rerun() try: rows = cached_dir_listing(host, username, port, key_filename, password, current_dir) except Exception as exc: st.error(f"Could not list directory `{current_dir}`: {exc}") return st.session_state.get("file_browser_selected_path") display_rows = [] for row in rows: kind = "dir" if row["type"] == "d" else "file" name = row["name"] extension = PurePosixPath(name).suffix.lower() if kind == "file" else "" full_path = str(PurePosixPath(current_dir) / name) display_rows.append( { "kind": kind, "label": "[DIR]" if kind == "dir" else "[FILE]", "name": name, "display_name": f"{'[DIR]' if kind == 'dir' else '[FILE]'} {name}", "extension": extension, "size_bytes": int(row["size"]), "size": "-" if kind == "dir" else human_size(row["size"]), "mtime": float(row["mtime"]), "modified": timestamp_to_local(row["mtime"]), "path": full_path, } ) total_count = len(display_rows) dir_count = sum(1 for r in display_rows if r["kind"] == "dir") file_count = total_count - dir_count total_file_size = sum(r["size_bytes"] for r in display_rows if r["kind"] == "file") st.caption( f"Entries: {total_count} | Directories: {dir_count} | Files: {file_count} | File size: {human_size(total_file_size)}" ) with st.container(border=True): filter_col, search_col, ext_col, sort_col, order_col, page_size_col = st.columns([1.1, 2.3, 1.1, 1.2, 1, 1]) kind_filter = filter_col.selectbox("Show", ["All", "Directories", "Files"], key="file_browser_kind_filter", label_visibility="collapsed") search_term = search_col.text_input("Search", placeholder="Search names", key="file_browser_search", label_visibility="collapsed") known_exts = sorted({r["extension"] for r in display_rows if r["extension"]}) extension_filter = ext_col.selectbox("Ext", ["All"] + known_exts, key="file_browser_extension_filter", label_visibility="collapsed") sort_by = sort_col.selectbox("Sort", ["Name", "Kind", "Size", "Modified"], key="file_browser_sort", label_visibility="collapsed") descending = order_col.toggle("Desc", value=False, key="file_browser_descending") page_size = page_size_col.selectbox("Rows", [10, 25, 50, 100, 200], index=1, key="file_browser_page_size", label_visibility="collapsed") filtered_rows = display_rows if kind_filter == "Directories": filtered_rows = [r for r in filtered_rows if r["kind"] == "dir"] elif kind_filter == "Files": filtered_rows = [r for r in filtered_rows if r["kind"] == "file"] if search_term: needle = search_term.lower() filtered_rows = [r for r in filtered_rows if needle in r["name"].lower()] if extension_filter != "All": filtered_rows = [r for r in filtered_rows if r["extension"] == extension_filter] sort_key_map = { "Name": lambda r: (r["kind"] != "dir", r["name"].lower()), "Kind": lambda r: (r["kind"], r["name"].lower()), "Size": lambda r: (r["kind"] != "dir", r["size_bytes"]), "Modified": lambda r: r["mtime"], } filtered_rows.sort(key=sort_key_map[sort_by], reverse=descending) filtered_count = len(filtered_rows) page_count = max(1, (filtered_count + page_size - 1) // page_size) page_col, summary_col = st.columns([1, 5]) if st.session_state.get("file_browser_page", 1) > page_count: st.session_state["file_browser_page"] = page_count page_number = page_col.number_input("Page", min_value=1, max_value=page_count, value=1, step=1, key="file_browser_page") start = (int(page_number) - 1) * page_size end = start + page_size page_rows = filtered_rows[start:end] summary_col.caption(f"Showing {start + 1 if filtered_count else 0}-{min(end, filtered_count)} of {filtered_count} matching entries") parent_path = str(PurePosixPath(current_dir).parent) visible_rows = [] if current_dir != "/": visible_rows.append( { "kind": "up", "label": "[UP]", "name": "..", "display_name": "[UP] ..", "extension": "", "size_bytes": 0, "size": "-", "mtime": 0.0, "modified": "", "path": parent_path, } ) visible_rows.extend(page_rows) if not display_rows: st.info("Directory is empty.") elif not page_rows: st.info("No entries match the current filters.") if not visible_rows: with st.expander("File browser diagnostics"): st.write(f"Current folder: `{current_dir}`") st.write(f"Path input: `{requested_path}`") st.write(f"Selected path: `{selected}`") st.write(f"Raw entries returned by remote listing: {len(rows)}") return selected table_rows = [ { "type": row["kind"], "name": row["name"], "ext": row["extension"], "size": row["size"], "modified": row["modified"], "path": row["path"], } for row in visible_rows ] table_df = pd.DataFrame(table_rows) grid_builder = GridOptionsBuilder.from_dataframe(table_df) grid_builder.configure_default_column(editable=False, resizable=True, sortable=False, filter=False) grid_builder.configure_column("type", width=90) grid_builder.configure_column("name", flex=2) grid_builder.configure_column("ext", width=90) grid_builder.configure_column("size", width=120) grid_builder.configure_column("modified", width=180) grid_builder.configure_column("path", hide=True) grid_builder.configure_selection(selection_mode="single", use_checkbox=False) grid_options = grid_builder.build() grid_options["rowSelection"] = { "mode": "singleRow", "checkboxes": False, "headerCheckbox": False, "enableClickSelection": True, } grid_options["suppressRowClickSelection"] = False grid_options["suppressCellFocus"] = True grid_options["onCellClicked"] = JsCode( """ function(event) { if (event && event.node) { event.node.setSelected(true, true); } } """ ) grid_response = AgGrid( table_df, gridOptions=grid_options, height=min(360, 33 * (len(table_rows) + 1)), fit_columns_on_grid_load=True, data_return_mode=DataReturnMode.AS_INPUT, update_mode=GridUpdateMode.SELECTION_CHANGED, key="file_browser_grid", theme="streamlit", allow_unsafe_jscode=True, ) selected_rows = aggrid_selected_rows(grid_response) if selected_rows: picked_row = selected_rows[0] picked_path = picked_row.get("path") picked_type = picked_row.get("type") action_token = f"{picked_type}:{picked_path}" # Open directories immediately on row select, including the [UP] row. if picked_type in {"dir", "up"} and picked_path and picked_path != current_dir: if st.session_state.get("file_browser_last_row_action") != action_token: st.session_state["file_browser_last_row_action"] = action_token set_file_browser_path(picked_path) st.rerun() elif picked_path: st.session_state["file_browser_selected_path"] = picked_path st.session_state["file_browser_last_row_action"] = action_token st.caption("Select a directory row to open it (including [UP] ..). Select a file row to target metadata/jobs.") return st.session_state.get("file_browser_selected_path", current_dir)