diff --git a/API/data/alldebrid.json b/API/data/alldebrid.json index 7194bbf..f88c33c 100644 --- a/API/data/alldebrid.json +++ b/API/data/alldebrid.json @@ -645,7 +645,7 @@ "(upload42\\.com/[0-9a-zA-Z]{12})" ], "regexp": "(upload42\\.com/[0-9a-zA-Z]{12})", - "status": true + "status": false }, "uploadbank": { "name": "uploadbank", @@ -679,7 +679,7 @@ "(uploadboy\\.com/[0-9a-zA-Z]{12})" ], "regexp": "(uploadboy\\.com/[0-9a-zA-Z]{12})", - "status": true + "status": false }, "uploader": { "name": "uploader", diff --git a/CLI.py b/CLI.py index b298c92..fb59e6b 100644 --- a/CLI.py +++ b/CLI.py @@ -798,10 +798,9 @@ class CmdletIntrospection: @staticmethod def store_choices(config: Dict[str, Any]) -> List[str]: try: - # Use config-only helper to avoid instantiating backends during completion - from Store.registry import list_configured_backend_names - - return list(list_configured_backend_names(config) or []) + # Use the cached startup check from SharedArgs + from cmdlet._shared import SharedArgs + return SharedArgs.get_store_choices(config) except Exception: return [] @@ -4206,6 +4205,14 @@ class MedeiaCLI: except Exception: pass + # Initialize the store choices cache at startup (filters disabled stores) + try: + from cmdlet._shared import SharedArgs + config = self._config_loader.load() + SharedArgs._refresh_store_choices_cache(config) + except Exception: + pass + self._cmdlet_executor = CmdletExecutor(config_loader=self._config_loader) self._pipeline_executor = PipelineExecutor(config_loader=self._config_loader) diff --git a/Provider/alldebrid.py b/Provider/alldebrid.py index 3882ea0..2c9eea9 100644 --- a/Provider/alldebrid.py +++ b/Provider/alldebrid.py @@ -11,6 +11,7 @@ from urllib.parse import urlparse from API.HTTP import HTTPClient, _download_direct_file from API.alldebrid import AllDebridClient, parse_magnet_or_hash, is_torrent_file from ProviderCore.base import Provider, SearchResult +from SYS.provider_helpers import TableProviderMixin from SYS.utils import sanitize_filename from SYS.logger import log, debug from SYS.models import DownloadError @@ -541,7 +542,32 @@ def adjust_output_dir_for_alldebrid( return output_dir -class AllDebrid(Provider): +class AllDebrid(TableProviderMixin, Provider): + """AllDebrid account provider with magnet folder/file browsing and downloads. + + This provider uses the new table system (strict ResultTable adapter pattern) for + consistent selection and auto-stage integration across all providers. It exposes + magnets as folder rows and files as file rows, with metadata enrichment for: + - magnet_id: For routing to _download_magnet_by_id + - status/ready: For showing sync state + - _selection_args/_selection_action: For @N expansion control + - relpath: For proper file hierarchy in downloads + + KEY FEATURES: + - Table system: Using ResultTable adapter for strict column/metadata handling + - Selection override: Full metadata control via _selection_args/_selection_action + - Auto-stages: download-file is auto-inserted when @N is used on magnet folders + - File unlocking: URLs with /f/ paths are automatically unlocked via API before download + - Drill-down: Selecting a folder row (@N) fetches and displays all files + + SELECTION FLOW: + 1. User runs: search-file -provider alldebrid "ubuntu" + 2. Results show magnet folders and (optionally) files + 3. User selects a row: @1 + 4. Selection metadata routes to download-file with -magnet-id + 5. download-file calls provider.download_items() with magnet_id + 6. Provider fetches files, unlocks locked URLs, and downloads + """ # Magnet URIs should be routed through this provider. TABLE_AUTO_STAGES = {"alldebrid": ["download-file"]} AUTO_STAGE_USE_SELECTION_ARGS = True @@ -1147,6 +1173,7 @@ class AllDebrid(Provider): "file": file_node, "provider": "alldebrid", "provider_view": "files", + # Selection metadata for table system "_selection_args": ["-magnet-id", str(magnet_id)], "_selection_action": ["download-file", "-provider", "alldebrid", "-magnet-id", str(magnet_id)], } @@ -1521,6 +1548,12 @@ try: def _columns_factory(rows: List[ResultModel]) -> List[ColumnSpec]: + """Build column specifications from available metadata in rows. + + This factory inspects all rows and creates ColumnSpec entries only + for metadata that is actually present in the result set. This avoids + empty columns in the display. + """ cols = [title_column()] if _has_metadata(rows, "magnet_name"): cols.append(metadata_column("magnet_name", "Magnet")) @@ -1531,7 +1564,7 @@ try: if _has_metadata(rows, "ready"): cols.append(metadata_column("ready", "Ready")) if _has_metadata(rows, "relpath"): - cols.append(metadata_column("relpath", "Relpath")) + cols.append(metadata_column("relpath", "File Path")) if _has_metadata(rows, "provider_view"): cols.append(metadata_column("provider_view", "View")) if _has_metadata(rows, "size"): @@ -1540,22 +1573,45 @@ try: def _selection_fn(row: ResultModel) -> List[str]: + """Return selection args for @N expansion and auto-download integration. + + Selection precedence: + 1. Explicit _selection_action (full command args) + 2. Explicit _selection_args (URL-specific args) + 3. Magic routing based on provider_view (files vs folders) + 4. Magnet ID routing for folder-type rows + 5. Direct URL for file rows + + This ensures that selector overrides all pre-codes and gives users full power. + """ metadata = row.metadata or {} + + # First try explicit action (full command) action = metadata.get("_selection_action") or metadata.get("selection_action") if isinstance(action, (list, tuple)) and action: return [str(x) for x in action if x is not None] + + # Next try explicit args (typically URL-based) args = metadata.get("_selection_args") or metadata.get("selection_args") if isinstance(args, (list, tuple)) and args: return [str(x) for x in args if x is not None] + + # Magic routing by view type view = metadata.get("provider_view") or metadata.get("view") or "" if view == "files": + # File rows: pass direct URL for immediate download if row.path: return ["-url", row.path] + + # Folder rows: use magnet_id to fetch and download all files magnet_id = metadata.get("magnet_id") if magnet_id is not None: return ["-magnet-id", str(magnet_id)] + + # Fallback: try direct URL if row.path: return ["-url", row.path] + return ["-title", row.title or ""] diff --git a/Provider/matrix.py b/Provider/matrix.py index 26bf6f7..3190828 100644 --- a/Provider/matrix.py +++ b/Provider/matrix.py @@ -1,15 +1,18 @@ from __future__ import annotations import mimetypes +import sys import time import uuid from pathlib import Path -from typing import Any, Dict, List, Optional, Tuple +from typing import Any, Dict, Iterable, List, Optional, Tuple from urllib.parse import quote import requests -from ProviderCore.base import Provider +from ProviderCore.base import Provider, SearchResult +from SYS.provider_helpers import TableProviderMixin +from SYS.logger import log _MATRIX_INIT_CHECK_CACHE: Dict[str, Tuple[bool, @@ -207,8 +210,29 @@ def _matrix_health_check(*, return False, str(exc) -class Matrix(Provider): - """File provider for Matrix (Element) chat rooms.""" +class Matrix(TableProviderMixin, Provider): + """Matrix (Element) room provider with file uploads and selection. + + This provider uses the new table system (strict ResultTable adapter pattern) for + consistent room listing and selection. It exposes Matrix joined rooms as selectable + rows with metadata enrichment for: + - room_id: Unique Matrix room identifier + - room_name: Human-readable room display name + - _selection_args: For @N expansion control and upload routing + + KEY FEATURES: + - Table system: Using ResultTable adapter for strict column/metadata handling + - Room discovery: search() or list_rooms() to enumerate joined rooms + - Selection integration: @N selection triggers upload_to_room() via selector() + - Deferred uploads: Files can be queued for upload to multiple rooms + - MIME detection: Automatic content type classification for Matrix msgtype + + SELECTION FLOW: + 1. User runs: search-file -provider matrix "room" (or .matrix -list-rooms) + 2. Results show available joined rooms + 3. User selects rooms: @1 @2 (or @1,2) + 4. Selection triggers upload of pending files to selected rooms + """ def __init__(self, config: Optional[Dict[str, Any]] = None): super().__init__(config) @@ -261,6 +285,70 @@ class Matrix(Provider): and (matrix_conf.get("access_token") or matrix_conf.get("password")) ) + def search( + self, + query: str, + limit: int = 50, + filters: Optional[Dict[str, Any]] = None, + **kwargs: Any, + ) -> List[SearchResult]: + """Search/list joined Matrix rooms. + + If query is empty or "*", returns all joined rooms. + Otherwise, filters rooms by name/ID matching the query. + """ + try: + rooms = self.list_rooms() + except Exception as exc: + log(f"[matrix] Failed to list rooms: {exc}", file=sys.stderr) + return [] + + q = (query or "").strip().lower() + needle = "" if q in {"*", "all", "list"} else q + + results: List[SearchResult] = [] + for room in rooms: + if len(results) >= limit: + break + + room_id = room.get("room_id") or "" + room_name = room.get("name") or "" + + # Filter by query if provided + if needle: + match_text = f"{room_name} {room_id}".lower() + if needle not in match_text: + continue + + if not room_id: + continue + + display_name = room_name or room_id + results.append( + SearchResult( + table="matrix", + title=display_name, + path=f"matrix:room:{room_id}", + detail=room_id if room_name else "", + annotations=["room"], + media_kind="folder", + columns=[ + ("Room", display_name), + ("ID", room_id), + ], + full_metadata={ + "room_id": room_id, + "room_name": room_name, + "provider": "matrix", + # Selection metadata for table system and @N expansion + "_selection_args": ["-room-id", room_id], + }, + ) + ) + + return results + + def _get_homeserver_and_token(self) -> Tuple[str, str]: matrix_conf = self.config.get("provider", {}).get("matrix", @@ -595,3 +683,100 @@ class Matrix(Provider): if any_failed: print("\nOne or more Matrix uploads failed\n") return True + + +# Minimal provider registration for the new table system +try: + from SYS.result_table_adapters import register_provider + from SYS.result_table_api import ResultModel, ColumnSpec, metadata_column, title_column + + def _convert_search_result_to_model(sr: Any) -> ResultModel: + """Convert Matrix SearchResult to ResultModel for strict table display.""" + d = sr.to_dict() if hasattr(sr, "to_dict") else (sr if isinstance(sr, dict) else {"title": getattr(sr, "title", str(sr))}) + title = d.get("title") or "" + path = d.get("path") or None + columns = d.get("columns") or getattr(sr, "columns", None) or [] + + # Extract metadata from columns and full_metadata + metadata: Dict[str, Any] = {} + for name, value in columns: + key = str(name or "").strip().lower() + if key in ("room_id", "room_name", "id", "name"): + metadata[key] = value + + try: + fm = d.get("full_metadata") or {} + if isinstance(fm, dict): + for k, v in fm.items(): + metadata[str(k).strip().lower()] = v + except Exception: + pass + + return ResultModel( + title=str(title), + path=str(path) if path else None, + ext=None, + size_bytes=None, + metadata=metadata, + source="matrix" + ) + + def _adapter(items: Iterable[Any]) -> Iterable[ResultModel]: + """Adapter to convert SearchResults to ResultModels.""" + for it in items: + try: + yield _convert_search_result_to_model(it) + except Exception: + continue + + def _has_metadata(rows: List[ResultModel], key: str) -> bool: + """Check if any row has a given metadata key with a non-empty value.""" + for row in rows: + md = row.metadata or {} + if key in md: + val = md[key] + if val is None: + continue + if isinstance(val, str) and not val.strip(): + continue + return True + return False + + def _columns_factory(rows: List[ResultModel]) -> List[ColumnSpec]: + """Build column specifications from available metadata in rows.""" + cols = [title_column()] + if _has_metadata(rows, "room_id"): + cols.append(metadata_column("room_id", "Room ID")) + if _has_metadata(rows, "room_name"): + cols.append(metadata_column("room_name", "Name")) + return cols + + def _selection_fn(row: ResultModel) -> List[str]: + """Return selection args for @N expansion and room selection. + + Uses explicit -room-id flag to identify the selected room for file uploads. + """ + metadata = row.metadata or {} + + # Check for explicit selection args first + args = metadata.get("_selection_args") or metadata.get("selection_args") + if isinstance(args, (list, tuple)) and args: + return [str(x) for x in args if x is not None] + + # Fallback to room_id + room_id = metadata.get("room_id") + if room_id: + return ["-room-id", str(room_id)] + + return ["-title", row.title or ""] + + register_provider( + "matrix", + _adapter, + columns=_columns_factory, + selection_fn=_selection_fn, + metadata={"description": "Matrix room provider with file uploads"}, + ) +except Exception: + # best-effort registration + pass diff --git a/Provider/youtube.py b/Provider/youtube.py index 8fa722b..68ee1a1 100644 --- a/Provider/youtube.py +++ b/Provider/youtube.py @@ -1,14 +1,32 @@ from __future__ import annotations import sys -from typing import Any, Dict, List, Optional +from typing import Any, Dict, Iterable, List, Optional from ProviderCore.base import Provider, SearchResult +from SYS.provider_helpers import TableProviderMixin from SYS.logger import log -class YouTube(Provider): - """Search provider for YouTube using the yt_dlp Python package.""" +class YouTube(TableProviderMixin, Provider): + """YouTube video search provider using yt_dlp. + + This provider uses the new table system (strict ResultTable adapter pattern) for + consistent selection and auto-stage integration. It exposes videos as SearchResult + rows with metadata enrichment for: + - video_id: Unique YouTube video identifier + - uploader: Channel/creator name + - duration: Video length in seconds + - view_count: Number of views + - _selection_args: For @N expansion control and download-file routing + + SELECTION FLOW: + 1. User runs: search-file -provider youtube "linux tutorial" + 2. Results show video rows with uploader, duration, views + 3. User selects a video: @1 + 4. Selection metadata routes to download-file with the YouTube URL + 5. download-file uses yt_dlp to download the video + """ TABLE_AUTO_STAGES = { "youtube": ["download-file"], @@ -79,6 +97,8 @@ class YouTube(Provider): "uploader": uploader, "duration": duration, "view_count": view_count, + # Selection metadata for table system and @N expansion + "_selection_args": ["-url", url], }, ) ) @@ -94,3 +114,102 @@ class YouTube(Provider): return True except Exception: return False + + +# Minimal provider registration for the new table system +try: + from SYS.result_table_adapters import register_provider + from SYS.result_table_api import ResultModel, ColumnSpec, metadata_column, title_column + + def _convert_search_result_to_model(sr: Any) -> ResultModel: + """Convert YouTube SearchResult to ResultModel for strict table display.""" + d = sr.to_dict() if hasattr(sr, "to_dict") else (sr if isinstance(sr, dict) else {"title": getattr(sr, "title", str(sr))}) + title = d.get("title") or "" + path = d.get("path") or None + columns = d.get("columns") or getattr(sr, "columns", None) or [] + + # Extract metadata from columns and full_metadata + metadata: Dict[str, Any] = {} + for name, value in columns: + key = str(name or "").strip().lower() + if key in ("uploader", "duration", "views", "video_id"): + metadata[key] = value + + try: + fm = d.get("full_metadata") or {} + if isinstance(fm, dict): + for k, v in fm.items(): + metadata[str(k).strip().lower()] = v + except Exception: + pass + + return ResultModel( + title=str(title), + path=str(path) if path else None, + ext=None, + size_bytes=None, + metadata=metadata, + source="youtube" + ) + + def _adapter(items: Iterable[Any]) -> Iterable[ResultModel]: + """Adapter to convert SearchResults to ResultModels.""" + for it in items: + try: + yield _convert_search_result_to_model(it) + except Exception: + continue + + def _has_metadata(rows: List[ResultModel], key: str) -> bool: + """Check if any row has a given metadata key with a non-empty value.""" + for row in rows: + md = row.metadata or {} + if key in md: + val = md[key] + if val is None: + continue + if isinstance(val, str) and not val.strip(): + continue + return True + return False + + def _columns_factory(rows: List[ResultModel]) -> List[ColumnSpec]: + """Build column specifications from available metadata in rows.""" + cols = [title_column()] + if _has_metadata(rows, "uploader"): + cols.append(metadata_column("uploader", "Uploader")) + if _has_metadata(rows, "duration"): + cols.append(metadata_column("duration", "Duration")) + if _has_metadata(rows, "views"): + cols.append(metadata_column("views", "Views")) + return cols + + def _selection_fn(row: ResultModel) -> List[str]: + """Return selection args for @N expansion and auto-download integration. + + Uses explicit -url flag to ensure the YouTube URL is properly routed + to download-file for yt_dlp download processing. + """ + metadata = row.metadata or {} + + # Check for explicit selection args first + args = metadata.get("_selection_args") or metadata.get("selection_args") + if isinstance(args, (list, tuple)) and args: + return [str(x) for x in args if x is not None] + + # Fallback to direct URL + if row.path: + return ["-url", row.path] + + return ["-title", row.title or ""] + + register_provider( + "youtube", + _adapter, + columns=_columns_factory, + selection_fn=_selection_fn, + metadata={"description": "YouTube video search using yt_dlp"}, + ) +except Exception: + # best-effort registration + pass diff --git a/TUI.py b/TUI.py index 51b2d8d..9b5b5bc 100644 --- a/TUI.py +++ b/TUI.py @@ -498,6 +498,15 @@ class PipelineHubApp(App): if self.worker_table: self.worker_table.add_columns("ID", "Type", "Status", "Details") + # Initialize the store choices cache at startup (filters disabled stores) + try: + from cmdlet._shared import SharedArgs + from SYS.config import load_config + config = load_config() + SharedArgs._refresh_store_choices_cache(config) + except Exception: + pass + self._populate_store_options() self._load_cmdlet_names() if self.executor.worker_manager: diff --git a/cmdlet/_shared.py b/cmdlet/_shared.py index 3bdf683..6d9f52d 100644 --- a/cmdlet/_shared.py +++ b/cmdlet/_shared.py @@ -205,36 +205,71 @@ class SharedArgs: def get_store_choices(config: Optional[Dict[str, Any]] = None) -> List[str]: """Get list of available store backend names. - This method dynamically discovers all configured storage backends - instead of using a static list. Should be called when building - autocomplete choices or validating store names. + This method returns the cached list of available backends from the most + recent startup check. Stores that failed to initialize are filtered out. + Users must restart to refresh the list if stores are enabled/disabled. Args: - config: Optional config dict. If not provided, will try to load from config module. + config: Ignored (kept for compatibility); uses cached startup result. Returns: List of backend names (e.g., ['default', 'test', 'home', 'work']) + Only includes backends that successfully initialized at startup. Example: SharedArgs.STORE.choices = SharedArgs.get_store_choices(config) """ + # Use the cached startup check result if available + if hasattr(SharedArgs, "_cached_available_stores"): + return SharedArgs._cached_available_stores or [] + + # Fallback to configured names if cache doesn't exist yet + # (This shouldn't happen in normal operation, but provides a safe fallback) try: - # Use the non-instantiating helper so autocomplete doesn't trigger backend init. from Store.registry import list_configured_backend_names - - # If no config provided, try to load it if config is None: try: from SYS.config import load_config config = load_config() except Exception: return [] - - return list_configured_backend_names(config) + return list_configured_backend_names(config) or [] except Exception: - # Fallback to empty list if FileStorage isn't available return [] + @staticmethod + def _refresh_store_choices_cache(config: Optional[Dict[str, Any]] = None) -> None: + """Refresh the cached store choices list. Should be called once at startup. + + This performs the actual StoreRegistry initialization check and caches the result. + Subsequent calls to get_store_choices() will use this cache. + + Args: + config: Config dict. If not provided, will try to load from config module. + """ + try: + if config is None: + try: + from SYS.config import load_config + config = load_config() + except Exception: + SharedArgs._cached_available_stores = [] + return + + # Initialize registry once to filter disabled stores + from Store.registry import Store as StoreRegistry + + try: + registry = StoreRegistry(config=config, suppress_debug=True) + available = registry.list_backends() + SharedArgs._cached_available_stores = available or [] + except Exception: + # If registry creation fails, fallback to configured names + from Store.registry import list_configured_backend_names + SharedArgs._cached_available_stores = list_configured_backend_names(config) or [] + except Exception: + SharedArgs._cached_available_stores = [] + LOCATION = CmdletArg( "location", type="enum", diff --git a/cmdlet/add_file.py b/cmdlet/add_file.py index d975855..8d7d403 100644 --- a/cmdlet/add_file.py +++ b/cmdlet/add_file.py @@ -321,12 +321,10 @@ class Add_File(Cmdlet): is_storage_backend_location = False if location: try: - # Use a config-only check to avoid instantiating backends (which may perform network checks). - from Store.registry import list_configured_backend_names - - is_storage_backend_location = location in ( - list_configured_backend_names(config) or [] - ) + # Check against the cached startup list of available backends + from cmdlet._shared import SharedArgs + available_backends = SharedArgs.get_store_choices(config) + is_storage_backend_location = location in available_backends except Exception: is_storage_backend_location = False @@ -546,7 +544,10 @@ class Add_File(Cmdlet): # Update pipe_obj with resolved path pipe_obj.path = str(media_path) - if not self._validate_source(media_path): + # When using -path (filesystem export), allow all file types. + # When using -store (backend), restrict to SUPPORTED_MEDIA_EXTENSIONS. + allow_all_files = not (location and is_storage_backend_location) + if not self._validate_source(media_path, allow_all_extensions=allow_all_files): failures += 1 continue @@ -1193,8 +1194,15 @@ class Add_File(Cmdlet): return files_info @staticmethod - def _validate_source(media_path: Optional[Path]) -> bool: - """Validate that the source file exists and is supported.""" + @staticmethod + def _validate_source(media_path: Optional[Path], allow_all_extensions: bool = False) -> bool: + """Validate that the source file exists and is supported. + + Args: + media_path: Path to the file to validate + allow_all_extensions: If True, skip file type filtering (used for -path exports). + If False, only allow SUPPORTED_MEDIA_EXTENSIONS (used for -store). + """ if media_path is None: return False @@ -1214,11 +1222,12 @@ class Add_File(Cmdlet): log(f"File not found: {media_path}") return False - # Validate file type - file_extension = media_path.suffix.lower() - if file_extension not in SUPPORTED_MEDIA_EXTENSIONS: - log(f"❌ Unsupported file type: {file_extension}", file=sys.stderr) - return False + # Validate file type: only when adding to -store backend, not for -path exports + if not allow_all_extensions: + file_extension = media_path.suffix.lower() + if file_extension not in SUPPORTED_MEDIA_EXTENSIONS: + log(f"❌ Unsupported file type: {file_extension}", file=sys.stderr) + return False return True diff --git a/cmdnat/config.py b/cmdnat/config.py index 380e7b8..be5a2e6 100644 --- a/cmdnat/config.py +++ b/cmdnat/config.py @@ -1,7 +1,9 @@ -from typing import List, Dict, Any +from typing import List, Dict, Any, Optional, Sequence from cmdlet._shared import Cmdlet, CmdletArg from SYS.config import load_config, save_config +from SYS import pipeline as ctx +from SYS.result_table import ResultTable CMDLET = Cmdlet( name=".config", @@ -22,28 +24,21 @@ CMDLET = Cmdlet( ) -def flatten_config(config: Dict[str, - Any], - parent_key: str = "", - sep: str = ".") -> List[Dict[str, - Any]]: - items = [] +def flatten_config(config: Dict[str, Any], parent_key: str = "", sep: str = ".") -> List[Dict[str, Any]]: + items: List[Dict[str, Any]] = [] for k, v in config.items(): - if k.startswith("_"): # Skip internal keys + if k.startswith("_"): continue - new_key = f"{parent_key}{sep}{k}" if parent_key else k if isinstance(v, dict): items.extend(flatten_config(v, new_key, sep=sep)) else: - items.append( - { - "Key": new_key, - "Value": str(v), - "Type": type(v).__name__, - "_selection_args": [new_key], - } - ) + items.append({ + "key": new_key, + "value": v, + "value_display": str(v), + "type": type(v).__name__, + }) return items @@ -98,53 +93,133 @@ def set_nested_config(config: Dict[str, Any], key: str, value: str) -> bool: return True -def _run(piped_result: Any, args: List[str], config: Dict[str, Any]) -> int: - # Reload config to ensure we have the latest on disk - # We don't use the passed 'config' because we want to edit the file - # and 'config' might contain runtime objects (like worker manager) - # But load_config() returns a fresh dict from disk (or cache) - # We should use load_config() +def _extract_piped_value(result: Any) -> Optional[str]: + if isinstance(result, str): + return result.strip() if result.strip() else None + if isinstance(result, (int, float)): + return str(result) + if isinstance(result, dict): + val = result.get("value") + if val is not None: + return str(val).strip() + return None + +def _extract_value_arg(args: Sequence[str]) -> Optional[str]: + if not args: + return None + tokens = [str(tok) for tok in args if tok is not None] + flags = {"-value", "--value", "-set-value", "--set-value"} + for idx, tok in enumerate(tokens): + text = tok.strip() + if not text: + continue + low = text.lower() + if low in flags and idx + 1 < len(tokens): + candidate = str(tokens[idx + 1]).strip() + if candidate: + return candidate + if "=" in low: + head, val = low.split("=", 1) + if head in flags and val: + return val.strip() + for tok in tokens: + text = str(tok).strip() + if text and not text.startswith("-"): + return text + return None + + +def _get_selected_config_key() -> Optional[str]: + try: + indices = ctx.get_last_selection() or [] + except Exception: + indices = [] + try: + items = ctx.get_last_result_items() or [] + except Exception: + items = [] + if not indices or not items: + return None + idx = indices[0] + if idx < 0 or idx >= len(items): + return None + item = items[idx] + if isinstance(item, dict): + return item.get("key") + return getattr(item, "key", None) + + +def _show_config_table(config_data: Dict[str, Any]) -> int: + items = flatten_config(config_data) + if not items: + print("No configuration entries available.") + return 0 + items.sort(key=lambda x: x.get("key")) + + table = ResultTable("Configuration") + table.set_table("config") + table.set_source_command(".config", []) + + for item in items: + row = table.add_row() + row.add_column("Key", item.get("key", "")) + row.add_column("Value", item.get("value_display", "")) + row.add_column("Type", item.get("type", "")) + + ctx.set_last_result_table_overlay(table, items) + ctx.set_current_stage_table(table) + return 0 + + +def _strip_value_quotes(value: str) -> str: + if not value: + return value + if (value.startswith('"') and value.endswith('"')) or (value.startswith("'") and value.endswith("'")): + return value[1:-1] + return value + + +def _run(piped_result: Any, args: List[str], config: Dict[str, Any]) -> int: current_config = load_config() - # Parse args - # We handle args manually because of the potential for spaces in values - # and the @ expansion logic in CLI.py passing args + selection_key = _get_selected_config_key() + value_from_args = _extract_value_arg(args) if selection_key else None + value_from_pipe = _extract_piped_value(piped_result) + + if selection_key: + new_value = value_from_pipe or value_from_args + if not new_value: + print( + "Provide a new value via pipe or argument: @N | .config " + ) + return 1 + new_value = _strip_value_quotes(new_value) + try: + set_nested_config(current_config, selection_key, new_value) + save_config(current_config) + print(f"Updated '{selection_key}' to '{new_value}'") + return 0 + except Exception as exc: + print(f"Error updating config: {exc}") + return 1 if not args: - # List mode - items = flatten_config(current_config) - # Sort by key - items.sort(key=lambda x: x["Key"]) + return _show_config_table(current_config) - # Emit items for ResultTable - from SYS import pipeline as ctx - - for item in items: - ctx.emit(item) - return 0 - - # Update mode key = args[0] - if len(args) < 2: print(f"Error: Value required for key '{key}'") return 1 - value = " ".join(args[1:]) - - # Remove quotes if present - if (value.startswith('"') and value.endswith('"')) or (value.startswith("'") - and value.endswith("'")): - value = value[1:-1] - + value = _strip_value_quotes(" ".join(args[1:])) try: set_nested_config(current_config, key, value) save_config(current_config) print(f"Updated '{key}' to '{value}'") return 0 - except Exception as e: - print(f"Error updating config: {e}") + except Exception as exc: + print(f"Error updating config: {exc}") return 1 diff --git a/cmdnat/matrix.py b/cmdnat/matrix.py index 670210b..d66bad6 100644 --- a/cmdnat/matrix.py +++ b/cmdnat/matrix.py @@ -15,6 +15,121 @@ from SYS import pipeline as ctx _MATRIX_PENDING_ITEMS_KEY = "matrix_pending_items" _MATRIX_PENDING_TEXT_KEY = "matrix_pending_text" +_MATRIX_MENU_STATE_KEY = "matrix_menu_state" +_MATRIX_SELECTED_SETTING_KEY_KEY = "matrix_selected_setting_key" + + +def _extract_piped_value(result: Any) -> Optional[str]: + """Extract the piped value from result (string, number, or dict with 'value' key).""" + if isinstance(result, str): + return result.strip() if result.strip() else None + if isinstance(result, (int, float)): + return str(result) + if isinstance(result, dict): + # Fallback to value field if it's a dict + val = result.get("value") + if val is not None: + return str(val).strip() + return None + + +def _extract_value_arg(args: Sequence[str]) -> Optional[str]: + """Extract a fallback value from command-line args (value flag or positional).""" + if not args: + return None + tokens = [str(tok) for tok in args if tok is not None] + value_flags = {"-value", "--value", "-set-value", "--set-value"} + for idx, tok in enumerate(tokens): + low = tok.strip() + if not low: + continue + low_lower = low.lower() + if low_lower in value_flags and idx + 1 < len(tokens): + candidate = str(tokens[idx + 1]).strip() + if candidate: + return candidate + if "=" in low_lower: + head, val = low_lower.split("=", 1) + if head in value_flags and val: + return val.strip() + # Fallback to first non-flag token + for tok in tokens: + text = str(tok).strip() + if text and not text.startswith("-"): + return text + return None + + +def _extract_set_value_arg(args: Sequence[str]) -> Optional[str]: + """Extract the value from -set-value flag.""" + if not args: + return None + try: + tokens = list(args) + except Exception: + return None + for i, tok in enumerate(tokens): + try: + if str(tok).lower() == "-set-value" and i + 1 < len(tokens): + return str(tokens[i + 1]).strip() + except Exception: + continue + return None + + +def _update_matrix_config(config: Dict[str, Any], key: str, value: Any) -> bool: + """Update a Matrix config value and write to config file. + + Returns True if successful, False otherwise. + """ + try: + from SYS.config import get_config_path + from configparser import ConfigParser + + if not isinstance(config, dict): + return False + + # Ensure provider.matrix section exists + providers = config.get("provider", {}) + if not isinstance(providers, dict): + providers = {} + config["provider"] = providers + + matrix_conf = providers.get("matrix", {}) + if not isinstance(matrix_conf, dict): + matrix_conf = {} + providers["matrix"] = matrix_conf + + # Update the in-memory config + matrix_conf[key] = value + + # Try to write to config file using configparser + try: + config_path = get_config_path() + if not config_path: + return False + + parser = ConfigParser() + if Path(config_path).exists(): + parser.read(config_path) + + section_name = "provider=matrix" + if not parser.has_section(section_name): + parser.add_section(section_name) + + parser.set(section_name, key, str(value)) + + with open(config_path, "w") as f: + parser.write(f) + + return True + except Exception as exc: + debug(f"[matrix] Failed to write config file: {exc}") + # Config was updated in memory at least + return True + except Exception as exc: + debug(f"[matrix] Failed to update Matrix config: {exc}") + return False def _has_flag(args: Sequence[str], flag: str) -> bool: @@ -466,9 +581,282 @@ def _resolve_upload_path(item: Any, config: Dict[str, Any]) -> Optional[str]: return None +def _show_main_menu() -> int: + """Display main menu: Rooms or Settings.""" + table = ResultTable("Matrix (select with @N)") + table.set_table("matrix") + table.set_source_command(".matrix", []) + + menu_items = [ + { + "title": "Rooms", + "description": "List and select rooms for uploads", + "action": "rooms", + }, + { + "title": "Settings", + "description": "View and modify Matrix configuration", + "action": "settings", + }, + ] + + for item in menu_items: + row = table.add_row() + row.add_column("Action", item["title"]) + row.add_column("Description", item["description"]) + + ctx.set_last_result_table_overlay(table, menu_items) + ctx.set_current_stage_table(table) + ctx.set_pending_pipeline_tail([[".matrix", "-menu-select"]], ".matrix") + return 0 + + +def _show_settings_table(config: Dict[str, Any]) -> int: + """Display Matrix configuration settings as a modifiable table.""" + table = ResultTable("Matrix Settings (select with @N to modify)") + table.set_table("matrix") + table.set_source_command(".matrix", ["-settings"]) + + matrix_conf = {} + try: + if isinstance(config, dict): + providers = config.get("provider") + if isinstance(providers, dict): + matrix_conf = providers.get("matrix") or {} + except Exception: + pass + + settings_items = [] + if isinstance(matrix_conf, dict): + for key in sorted(matrix_conf.keys()): + value = matrix_conf[key] + # Skip sensitive/complex values + if key in ("password",): + value = "***" + settings_items.append({ + "key": key, + "value": str(value), + "original_value": value, + }) + + if not settings_items: + log("No Matrix settings configured. Edit config.conf manually.", file=sys.stderr) + return 0 + + for item in settings_items: + row = table.add_row() + row.add_column("Key", item["key"]) + row.add_column("Value", item["value"]) + + ctx.set_last_result_table_overlay(table, settings_items) + ctx.set_current_stage_table(table) + ctx.set_pending_pipeline_tail([[".matrix", "-settings-edit"]], ".matrix") + return 0 + + +def _handle_menu_selection(selected: Any, config: Dict[str, Any]) -> int: + """Handle main menu selection (rooms or settings).""" + items = _normalize_to_list(selected) + if not items: + return 1 + + item = items[0] # Only consider first selection + action = None + if isinstance(item, dict): + action = item.get("action") + else: + action = getattr(item, "action", None) + + if action == "settings": + return _show_settings_table(config) + else: + return _show_rooms_table(config) + + +def _handle_settings_edit(result: Any, args: Sequence[str], config: Dict[str, Any]) -> int: + """Handle settings modification from selected rows. + + Two-stage flow: + 1. User selects @4 from settings table (@4 stores selection index in context) + 2. User pipes value (@4 | "30" → result becomes "30") + 3. Handler uses both: + - get_last_selection() → get the row index + - result → the piped value + - Retrieve the setting key from stored items + - Update config + + Usage: @4 | "30" + """ + # Get the last selected indices (@4 would give [3]) + selection_indices = [] + try: + selection_indices = ctx.get_last_selection() or [] + except Exception: + pass + + # Get the last result items (the settings table items) + last_items = [] + try: + last_items = ctx.get_last_result_items() or [] + except Exception: + pass + + if not selection_indices or not last_items: + log("No setting selected. Use @N to select a setting first.", file=sys.stderr) + return 1 + + # Get the selected settings item + idx = selection_indices[0] + if idx < 0 or idx >= len(last_items): + log("Invalid selection", file=sys.stderr) + return 1 + + selected_item = last_items[idx] + key = None + if isinstance(selected_item, dict): + key = selected_item.get("key") + else: + key = getattr(selected_item, "key", None) + + if not key: + log("Invalid settings selection", file=sys.stderr) + return 1 + + # Prevent modifying sensitive settings + if key in ("password", "access_token"): + log(f"Cannot modify sensitive setting: {key}", file=sys.stderr) + return 1 + + # Extract the piped value or fallback to CLI args + new_value = _extract_piped_value(result) or _extract_value_arg(args) + if new_value is None: + log(f"To modify '{key}', pipe a literal value (e.g. @N | '30') or pass it as an arg: .matrix -settings-edit 30", file=sys.stderr) + return 1 + + # Update the config with the new value + if _update_matrix_config(config, str(key), new_value): + log(f"✓ Updated {key} = {new_value}") + return 0 + else: + log(f"✗ Failed to update {key}", file=sys.stderr) + return 1 + + +def _show_rooms_table(config: Dict[str, Any]) -> int: + """Display rooms (refactored original behavior).""" + from Provider.matrix import Matrix + + try: + provider = Matrix(config) + except Exception as exc: + log(f"Matrix not available: {exc}", file=sys.stderr) + return 1 + + try: + configured_ids = None + # Use `-all` flag to override room filter + all_rooms_flag = False + if hasattr(ctx, "get_last_args"): + try: + last_args = ctx.get_last_args() or [] + all_rooms_flag = any(str(a).lower() == "-all" for a in last_args) + except Exception: + all_rooms_flag = False + + if not all_rooms_flag: + ids = [ + str(v).strip() for v in _parse_config_room_filter_ids(config) + if str(v).strip() + ] + if ids: + configured_ids = ids + + rooms = provider.list_rooms(room_ids=configured_ids) + except Exception as exc: + log(f"Failed to list Matrix rooms: {exc}", file=sys.stderr) + return 1 + + # Diagnostics if a configured filter yields no rows + if not rooms and not all_rooms_flag: + configured_ids_dbg = [ + str(v).strip() for v in _parse_config_room_filter_ids(config) + if str(v).strip() + ] + if configured_ids_dbg: + try: + joined_ids = provider.list_joined_room_ids() + debug(f"[matrix] Configured room filter IDs: {configured_ids_dbg}") + debug(f"[matrix] Joined room IDs (from Matrix): {joined_ids}") + except Exception: + pass + + if not rooms: + if _parse_config_room_filter_ids(config) and not all_rooms_flag: + log( + "No joined rooms matched the configured Matrix room filter (use: .matrix -all)", + file=sys.stderr, + ) + else: + log("No joined rooms found.", file=sys.stderr) + return 0 + + table = ResultTable("Matrix Rooms (select with @N)") + table.set_table("matrix") + table.set_source_command(".matrix", []) + + for room in rooms: + row = table.add_row() + name = str(room.get("name") or "").strip() if isinstance(room, dict) else "" + room_id = str(room.get("room_id") or "" + ).strip() if isinstance(room, + dict) else "" + row.add_column("Name", name) + row.add_column("Room", room_id) + + # Make selection results clearer + room_items: List[Dict[str, Any]] = [] + for room in rooms: + if not isinstance(room, dict): + continue + room_id = str(room.get("room_id") or "").strip() + name = str(room.get("name") or "").strip() + room_items.append( + { + **room, + "store": "matrix", + "title": name or room_id or "Matrix Room", + } + ) + + ctx.set_last_result_table_overlay(table, room_items) + ctx.set_current_stage_table(table) + ctx.set_pending_pipeline_tail([[".matrix", "-send"]], ".matrix") + return 0 + + def _run(result: Any, args: Sequence[str], config: Dict[str, Any]) -> int: + """Main Matrix cmdlet execution. + + Flow: + 1. First call: Show main menu (Rooms or Settings) + 2. User selects @1 (rooms) or @2 (settings) + 3. -menu-select: Route to appropriate handler + 4. -send: Send files to selected room(s) (when uploading) + 5. -settings-edit: Handle settings modification + """ + # Handle menu selection routing + if _has_flag(args, "-menu-select"): + return _handle_menu_selection(result, config) + + # Handle settings view/edit + if _has_flag(args, "-settings"): + return _show_settings_table(config) + + if _has_flag(args, "-settings-edit"): + return _handle_settings_edit(result, args, config) + # Internal stage: send previously selected items to selected rooms. - if any(str(a).lower() == "-send" for a in (args or [])): + if _has_flag(args, "-send"): # Ensure we don't re-print the rooms picker table on the send stage. try: if hasattr(ctx, "set_last_result_table_overlay"): @@ -596,109 +984,27 @@ def _run(result: Any, args: Sequence[str], config: Dict[str, Any]) -> int: pass return 1 if any_failed else 0 - # Default stage: show rooms, then wait for @N selection to resume sending. + # Default stage: handle piped vs non-piped behavior selected_items = _normalize_to_list(result) - if not selected_items: - log( - "Usage: @N | .matrix (select items first, then pick a room)", - file=sys.stderr - ) - return 1 - ctx.store_value(_MATRIX_PENDING_ITEMS_KEY, selected_items) try: ctx.store_value(_MATRIX_PENDING_TEXT_KEY, _extract_text_arg(args)) except Exception: pass - - from Provider.matrix import Matrix - - try: - provider = Matrix(config) - except Exception as exc: - log(f"Matrix not available: {exc}", file=sys.stderr) - return 1 - - try: - configured_ids = None - if not _has_flag(args, "-all"): - ids = [ - str(v).strip() for v in _parse_config_room_filter_ids(config) - if str(v).strip() - ] - if ids: - configured_ids = ids - - rooms = provider.list_rooms(room_ids=configured_ids) - except Exception as exc: - log(f"Failed to list Matrix rooms: {exc}", file=sys.stderr) - return 1 - - # Diagnostics if a configured filter yields no rows (provider filtered before name lookups for speed). - if not rooms and not _has_flag(args, "-all"): - configured_ids_dbg = [ - str(v).strip() for v in _parse_config_room_filter_ids(config) - if str(v).strip() - ] - if configured_ids_dbg: - try: - joined_ids = provider.list_joined_room_ids() - debug(f"[matrix] Configured room filter IDs: {configured_ids_dbg}") - debug(f"[matrix] Joined room IDs (from Matrix): {joined_ids}") - except Exception: - pass - - if not rooms: - if _parse_config_room_filter_ids(config) and not _has_flag(args, "-all"): - log( - "No joined rooms matched the configured Matrix room filter (use: .matrix -all)", - file=sys.stderr, - ) - else: - log("No joined rooms found.", file=sys.stderr) - return 0 - - table = ResultTable("Matrix Rooms (select with @N)") - table.set_table("matrix") - table.set_source_command(".matrix", []) - - for room in rooms: - row = table.add_row() - name = str(room.get("name") or "").strip() if isinstance(room, dict) else "" - room_id = str(room.get("room_id") or "" - ).strip() if isinstance(room, - dict) else "" - row.add_column("Name", name) - row.add_column("Room", room_id) - - # Make selection results clearer: stash a friendly title/store on the backing items. - # This avoids confusion when the selection handler prints PipeObject debug info. - room_items: List[Dict[str, Any]] = [] - for room in rooms: - if not isinstance(room, dict): - continue - room_id = str(room.get("room_id") or "").strip() - name = str(room.get("name") or "").strip() - room_items.append( - { - **room, - "store": "matrix", - "title": name or room_id or "Matrix Room", - } - ) - - # Overlay table: user selects @N, then we resume with `.matrix -send`. - ctx.set_last_result_table_overlay(table, room_items) - ctx.set_current_stage_table(table) - ctx.set_pending_pipeline_tail([[".matrix", "-send"]], ".matrix") - return 0 + + # When piped (result has data), show rooms directly. + # When not piped (first command), show main menu. + if selected_items: + return _show_rooms_table(config) + else: + return _show_main_menu() CMDLET = Cmdlet( name=".matrix", alias=["matrix", "rooms"], - summary="Send selected items to a Matrix room", + summary="Send selected items to a Matrix room or manage settings", usage="@N | .matrix", arg=[ CmdletArg( @@ -707,6 +1013,30 @@ CMDLET = Cmdlet( description="(internal) Send to selected room(s)", required=False, ), + CmdletArg( + name="menu-select", + type="bool", + description="(internal) Handle menu selection (rooms/settings)", + required=False, + ), + CmdletArg( + name="settings", + type="bool", + description="(internal) Show settings table", + required=False, + ), + CmdletArg( + name="settings-edit", + type="bool", + description="(internal) Handle settings modification", + required=False, + ), + CmdletArg( + name="set-value", + type="string", + description="New value for selected setting (use with -settings-edit)", + required=False, + ), CmdletArg( name="all", type="bool",