goyimnose/Medios-Macina
1
0
Fork 0
Code Issues Pull Requests Packages Releases 1 Wiki Activity

removed TUI and others

Browse Source
This commit is contained in:
Nose
2026-05-14 20:47:20 -07:00
parent 036977832b
commit 717cb13dda
31 changed files with 378 additions and 7790 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CLI.py
+17 -131
View File
@@ -60,7 +60,6 @@ from SYS.rich_display import (
from cmdnat._status_shared import (
add_startup_check as _shared_add_startup_check,
collect_plugin_startup_checks as _collect_plugin_startup_checks,
has_store_subtype as _has_store_subtype,
has_tool as _has_tool,
)
@@ -94,7 +93,7 @@ from SYS.cmdlet_catalog import (
list_cmdlet_metadata,
list_cmdlet_names,
)
from SYS.config import load_config, resolve_cookies_path
from SYS.config import load_config
from SYS.result_table import Table
from SYS.worker import WorkerManagerRegistry, WorkerStages, WorkerOutputMirror, WorkerStageSession
@@ -2408,7 +2407,7 @@ Come to love it when others take what you share, as there is no greater joy
name: str,
*,
provider: str = "",
store: str = "",
instance: str = "",
files: int | str | None = None,
detail: str = "",
) -> None:
@@ -2417,7 +2416,7 @@ Come to love it when others take what you share, as there is no greater joy
status,
name,
provider=provider,
store=store,
instance=instance,
files=files,
detail=detail,
)
@@ -2439,127 +2438,15 @@ Come to love it when others take what you share, as there is no greater joy
except Exception as exc:
_add_startup_check("DISABLED", "MPV", detail=str(exc))
store_registry = None
if config:
try:
from Store import Store as StoreRegistry
store_registry = StoreRegistry(config=config, suppress_debug=True)
except Exception:
store_registry = None
if _has_store_subtype(config, "hydrusnetwork"):
store_cfg = config.get("store")
hydrus_cfg = (
store_cfg.get("hydrusnetwork",
{}) if isinstance(store_cfg,
dict) else {}
)
if isinstance(hydrus_cfg, dict):
for instance_name, instance_cfg in hydrus_cfg.items():
if not isinstance(instance_cfg, dict):
continue
name_key = str(instance_cfg.get("NAME") or instance_name)
url_val = str(instance_cfg.get("URL") or "").strip()
ok = bool(
store_registry
and store_registry.is_available(name_key)
)
status = "ENABLED" if ok else "DISABLED"
if ok:
total = None
try:
if store_registry:
backend = store_registry[name_key]
total = getattr(backend, "total_count", None)
if total is None:
getter = getattr(
backend,
"get_total_count",
None
)
if callable(getter):
total = getter()
except Exception:
total = None
detail = url_val
files = total if isinstance(
total,
int
) and total >= 0 else None
else:
err = None
if store_registry:
err = store_registry.get_backend_error(
instance_name
) or store_registry.get_backend_error(name_key)
detail = (url_val + (" - " if url_val else "")
) + (err or "Unavailable")
files = None
_add_startup_check(
status,
name_key,
store="hydrusnetwork",
files=files,
detail=detail
)
provider_cfg = None
if isinstance(config, dict):
provider_cfg = config.get("plugin")
if not isinstance(provider_cfg, dict):
provider_cfg = config.get("provider")
if isinstance(provider_cfg, dict) and provider_cfg:
for check in _collect_plugin_startup_checks(config):
_add_startup_check(
str(check.get("status") or "UNKNOWN"),
str(check.get("name") or "Plugin"),
provider=str(check.get("plugin") or ""),
detail=str(check.get("detail") or ""),
files=check.get("files"),
)
if _has_store_subtype(config, "debrid"):
try:
from SYS.config import get_debrid_api_key
from plugins.alldebrid.api import AllDebridClient
api_key = get_debrid_api_key(config)
if not api_key:
_add_startup_check(
"DISABLED",
"Debrid",
store="debrid",
detail="Not configured"
)
else:
client = AllDebridClient(api_key)
base_url = str(getattr(client,
"base_url",
"") or "").strip()
_add_startup_check(
"ENABLED",
"Debrid",
store="debrid",
detail=base_url or "Connected"
)
except Exception as exc:
_add_startup_check(
"DISABLED",
"Debrid",
store="debrid",
detail=str(exc)
)
try:
cookiefile = resolve_cookies_path(config)
if cookiefile is not None:
_add_startup_check("FOUND", "Cookies", detail=str(cookiefile))
else:
_add_startup_check("MISSING", "Cookies", detail="Not found")
except Exception as exc:
_add_startup_check("ERROR", "Cookies", detail=str(exc))
for check in _collect_plugin_startup_checks(config):
_add_startup_check(
str(check.get("status") or "UNKNOWN"),
str(check.get("name") or "Plugin"),
provider=str(check.get("plugin") or ""),
instance=str(check.get("instance") or ""),
detail=str(check.get("detail") or ""),
files=check.get("files"),
)
# Tool checks (configured via [tool=...])
if _has_tool(config, "florencevision"):
@@ -2599,13 +2486,12 @@ Come to love it when others take what you share, as there is no greater joy
provider="tool",
detail=str(exc),
)
if startup_table.rows:
stdout_console().print()
stdout_console().print(startup_table)
except Exception as exc:
if debug_enabled:
debug(f"⚠ Could not check service availability: {exc}")
_add_startup_check("ERROR", "STARTUP", detail=str(exc))
if startup_table.rows:
stdout_console().print()
stdout_console().print(startup_table)
style = Style.from_dict(
{
SYS/config.py
+1 -1
View File
@@ -900,7 +900,7 @@ def _extract_expected_alldebrid_key(config: Dict[str, Any]) -> Optional[str]:
return expected_key
def load_config(*, emit_summary: bool = True) -> Dict[str, Any]:
def load_config(*, emit_summary: bool = False) -> Dict[str, Any]:
global _CONFIG_CACHE, _LAST_SAVED_CONFIG, _CONFIG_SUMMARY_PENDING
if _CONFIG_CACHE:
if emit_summary and _CONFIG_SUMMARY_PENDING:
TUI/pipeline_runner.py → SYS/pipeline_runner.py
+102 -102
View File
@@ -1,7 +1,8 @@
"""Pipeline execution utilities for the Textual UI.
"""Shared pipeline runner utilities.
The TUI is a frontend to the CLI, so it must use the same pipeline executor
implementation as the CLI (`CLI.PipelineExecutor`).
This module wraps the canonical CLI pipeline executor so non-CLI callers can
execute pipelines and capture the resulting table/items without depending on
the discontinued Textual UI package.
"""
from __future__ import annotations
@@ -9,27 +10,21 @@ from __future__ import annotations
import contextlib
import io
import shlex
import sys
import traceback
from pathlib import Path
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Sequence
BASE_DIR = Path(__file__).resolve().parent
ROOT_DIR = BASE_DIR.parent
for path in (ROOT_DIR, BASE_DIR):
str_path = str(path)
if str_path not in sys.path:
sys.path.insert(0, str_path)
from SYS import pipeline as ctx
from CLI import ConfigLoader
from SYS import pipeline as ctx
from SYS.logger import debug, set_debug
from SYS.pipeline import PipelineExecutor
from SYS.worker import WorkerManagerRegistry
from SYS.logger import set_debug, debug
from SYS.rich_display import capture_rich_output
import traceback
from SYS.result_table import Table
from SYS.rich_display import capture_rich_output
from SYS.worker import WorkerManagerRegistry
REPO_ROOT = Path(__file__).resolve().parents[1]
@dataclass(slots=True)
@@ -39,7 +34,7 @@ class PipelineStageResult:
name: str
args: Sequence[str]
emitted: List[Any] = field(default_factory=list)
result_table: Optional[Any] = None # ResultTable object if available
result_table: Optional[Any] = None
status: str = "pending"
error: Optional[str] = None
@@ -52,42 +47,46 @@ class PipelineRunResult:
success: bool
stages: List[PipelineStageResult] = field(default_factory=list)
emitted: List[Any] = field(default_factory=list)
result_table: Optional[Any] = None # Final ResultTable object if available
result_table: Optional[Any] = None
stdout: str = ""
stderr: str = ""
error: Optional[str] = None
def to_summary(self) -> Dict[str, Any]:
"""Provide a JSON-friendly representation for logging or UI."""
return {
"pipeline":
self.pipeline,
"success":
self.success,
"error":
self.error,
"pipeline": self.pipeline,
"success": self.success,
"error": self.error,
"stages": [
{
"name": stage.name,
"status": stage.status,
"error": stage.error,
"emitted": len(stage.emitted),
} for stage in self.stages
}
for stage in self.stages
],
}
class PipelineRunner:
"""TUI wrapper that delegates to the canonical CLI pipeline executor."""
"""Wrapper around the canonical CLI pipeline executor."""
def __init__(self, config_loader: Optional[Any] = None, executor: Optional[Any] = None) -> None:
# Allow dependency injection or lazily construct CLI dependencies so tests
# don't fail due to import-order issues in pytest environments.
self._config_loader = config_loader if config_loader is not None else (ConfigLoader(root=ROOT_DIR) if ConfigLoader else None)
if executor is not None:
self._executor = executor
else:
self._executor = PipelineExecutor(config_loader=self._config_loader)
def __init__(
self,
config_loader: Optional[Any] = None,
executor: Optional[Any] = None,
) -> None:
self._config_loader = (
config_loader
if config_loader is not None
else ConfigLoader(root=REPO_ROOT)
)
self._executor = (
executor
if executor is not None
else PipelineExecutor(config_loader=self._config_loader)
)
self._worker_manager = None
@property
@@ -101,8 +100,7 @@ class PipelineRunner:
seeds: Optional[Any] = None,
seed_table: Optional[Any] = None,
isolate: bool = False,
on_log: Optional[Callable[[str],
None]] = None,
on_log: Optional[Callable[[str], None]] = None,
) -> PipelineRunResult:
snapshot: Optional[Dict[str, Any]] = None
if isolate:
@@ -171,8 +169,8 @@ class PipelineRunner:
try:
with capture_rich_output(stdout=stdout_buffer, stderr=stderr_buffer):
with (
contextlib.redirect_stdout(stdout_buffer),
contextlib.redirect_stderr(stderr_buffer),
contextlib.redirect_stdout(stdout_buffer),
contextlib.redirect_stderr(stderr_buffer),
):
if on_log:
on_log("Executing pipeline via CLI executor...")
@@ -187,11 +185,11 @@ class PipelineRunner:
result.stdout = stdout_buffer.getvalue()
result.stderr = stderr_buffer.getvalue()
# Pull the canonical state out of pipeline context.
table = None
try:
table = (
ctx.get_display_table() or ctx.get_current_stage_table()
ctx.get_display_table()
or ctx.get_current_stage_table()
or ctx.get_last_result_table()
)
except Exception:
@@ -226,7 +224,7 @@ class PipelineRunner:
)
if result.error:
result.success = False
elif any(m in combined for m in failure_markers):
elif any(marker in combined for marker in failure_markers):
result.success = False
if not result.error:
result.error = "Pipeline failed"
@@ -237,84 +235,89 @@ class PipelineRunner:
try:
self._restore_ctx_state(snapshot)
except Exception:
# Best-effort; isolation should never break normal operation.
pass
return result
@staticmethod
def _snapshot_ctx_state() -> Dict[str, Any]:
"""Best-effort snapshot of pipeline context using PipelineState.
This reads from the active PipelineState (ContextVar or global fallback)
to produce a consistent snapshot that can be restored later.
"""
def _copy(val: Any) -> Any:
if isinstance(val, list):
return val.copy()
if isinstance(val, dict):
return val.copy()
return val
def _copy(value: Any) -> Any:
if isinstance(value, list):
return value.copy()
if isinstance(value, dict):
return value.copy()
return value
state = ctx.get_pipeline_state()
snap: Dict[str, Any] = {}
snapshot: Dict[str, Any] = {}
snapshot["live_progress"] = _copy(state.live_progress)
snapshot["current_context"] = state.current_context
snapshot["last_search_query"] = state.last_search_query
snapshot["pipeline_refreshed"] = state.pipeline_refreshed
snapshot["last_items"] = _copy(state.last_items)
snapshot["last_result_table"] = state.last_result_table
snapshot["last_result_items"] = _copy(state.last_result_items)
snapshot["last_result_subject"] = state.last_result_subject
# Simple scalar/list/dict fields
snap["live_progress"] = _copy(state.live_progress)
snap["current_context"] = state.current_context
snap["last_search_query"] = state.last_search_query
snap["pipeline_refreshed"] = state.pipeline_refreshed
snap["last_items"] = _copy(state.last_items)
snap["last_result_table"] = state.last_result_table
snap["last_result_items"] = _copy(state.last_result_items)
snap["last_result_subject"] = state.last_result_subject
# Deep-copy history/forward stacks (copy nested item lists)
def _copy_history(hist: Optional[List[tuple]]) -> List[tuple]:
def _copy_history(history: Optional[List[tuple]]) -> List[tuple]:
out: List[tuple] = []
try:
for (t, items, subj) in list(hist or []):
items_copy = items.copy() if isinstance(items, list) else list(items) if items else []
out.append((t, items_copy, subj))
for table_value, items, subject in list(history or []):
if isinstance(items, list):
items_copy = items.copy()
elif items:
items_copy = list(items)
else:
items_copy = []
out.append((table_value, items_copy, subject))
except Exception:
debug(traceback.format_exc())
return out
snap["result_table_history"] = _copy_history(state.result_table_history)
snap["result_table_forward"] = _copy_history(state.result_table_forward)
snap["current_stage_table"] = state.current_stage_table
snap["display_items"] = _copy(state.display_items)
snap["display_table"] = state.display_table
snap["display_subject"] = state.display_subject
snap["last_selection"] = _copy(state.last_selection)
snap["pipeline_command_text"] = state.pipeline_command_text
snap["current_cmdlet_name"] = state.current_cmdlet_name
snap["current_stage_text"] = state.current_stage_text
snap["pipeline_values"] = _copy(state.pipeline_values) if isinstance(state.pipeline_values, dict) else state.pipeline_values
snap["pending_pipeline_tail"] = [list(stage) for stage in (state.pending_pipeline_tail or [])]
snap["pending_pipeline_source"] = state.pending_pipeline_source
snap["ui_library_refresh_callback"] = state.ui_library_refresh_callback
snap["pipeline_stop"] = state.pipeline_stop
return snap
snapshot["result_table_history"] = _copy_history(state.result_table_history)
snapshot["result_table_forward"] = _copy_history(state.result_table_forward)
snapshot["current_stage_table"] = state.current_stage_table
snapshot["display_items"] = _copy(state.display_items)
snapshot["display_table"] = state.display_table
snapshot["display_subject"] = state.display_subject
snapshot["last_selection"] = _copy(state.last_selection)
snapshot["pipeline_command_text"] = state.pipeline_command_text
snapshot["current_cmdlet_name"] = state.current_cmdlet_name
snapshot["current_stage_text"] = state.current_stage_text
snapshot["pipeline_values"] = (
_copy(state.pipeline_values)
if isinstance(state.pipeline_values, dict)
else state.pipeline_values
)
snapshot["pending_pipeline_tail"] = [
list(stage) for stage in (state.pending_pipeline_tail or [])
]
snapshot["pending_pipeline_source"] = state.pending_pipeline_source
snapshot["ui_library_refresh_callback"] = state.ui_library_refresh_callback
snapshot["pipeline_stop"] = state.pipeline_stop
return snapshot
@staticmethod
def _restore_ctx_state(snapshot: Dict[str, Any]) -> None:
if not snapshot:
return
state = ctx.get_pipeline_state()
# Helper for restoring history-like stacks
def _restore_history(key: str, val: Any) -> None:
def _restore_history(key: str, value: Any) -> None:
try:
if isinstance(val, list):
out: List[tuple] = []
for (t, items, subj) in val:
items_copy = items.copy() if isinstance(items, list) else list(items) if items else []
out.append((t, items_copy, subj))
setattr(state, key, out)
if not isinstance(value, list):
return
restored: List[tuple] = []
for table_value, items, subject in value:
if isinstance(items, list):
items_copy = items.copy()
elif items:
items_copy = list(items)
else:
items_copy = []
restored.append((table_value, items_copy, subject))
setattr(state, key, restored)
except Exception:
debug(traceback.format_exc())
@@ -366,7 +369,4 @@ class PipelineRunner:
if "pipeline_stop" in snapshot:
state.pipeline_stop = snapshot["pipeline_stop"]
except Exception:
# Best-effort; don't break the pipeline runner
pass
pass
TUI.py
+2 -2
View File
@@ -48,7 +48,7 @@ for path in (REPO_ROOT, TUI_DIR):
if str_path not in sys.path:
sys.path.insert(0, str_path)
from TUI.pipeline_runner import PipelineRunResult # type: ignore # noqa: E402
from SYS.pipeline_runner import PipelineRunResult # type: ignore # noqa: E402
from SYS.result_table import Table, extract_hash_value, extract_store_value, get_result_table_row_style # type: ignore # noqa: E402
from SYS.config import load_config # type: ignore # noqa: E402
@@ -57,7 +57,7 @@ from Store.registry import Store as StoreRegistry # type: ignore # noqa: E402
from SYS.cmdlet_catalog import ensure_registry_loaded, list_cmdlet_names # type: ignore # noqa: E402
from SYS.cli_syntax import validate_pipeline_text # type: ignore # noqa: E402
from TUI.pipeline_runner import PipelineRunner # type: ignore # noqa: E402
from SYS.pipeline_runner import PipelineRunner # type: ignore # noqa: E402
def _dedup_preserve_order(items: List[str]) -> List[str]:
TUI/__init__.py
-1
View File
@@ -1 +0,0 @@
"""Medeia-Macina TUI - Terminal User Interface."""
TUI/menu_actions.py
-79
View File
@@ -1,79 +0,0 @@
"""Utilities that drive the modern Textual UI menus and presets."""
from __future__ import annotations
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Dict, Iterable, List, Sequence
BASE_DIR = Path(__file__).resolve().parent
ROOT_DIR = BASE_DIR.parent
for path in (ROOT_DIR, BASE_DIR):
str_path = str(path)
if str_path not in sys.path:
sys.path.insert(0, str_path)
import SYS.metadata as metadata
@dataclass(slots=True)
class PipelinePreset:
"""Simple descriptor for a reusable pipeline."""
label: str
description: str
pipeline: str
PIPELINE_PRESETS: List[PipelinePreset] = [
PipelinePreset(
label="Download → Merge → Local",
description=
"Use file -download with playlist auto-selection, merge the pieces, tag, then import into local storage.",
pipeline=
'file -download "<url>" | file -merge | metadata -add -instance local | file -add -storage local',
),
PipelinePreset(
label="Download → Hydrus",
description="Fetch media, auto-tag, and push directly into Hydrus.",
pipeline=
'file -download "<url>" | file -merge | metadata -add -instance hydrus | file -add -storage hydrus',
),
PipelinePreset(
label="Search Local Library",
description=
"Run file -query against the local library and emit a result table for further piping.",
pipeline='file -library local -query "<keywords>"',
),
]
def load_tags(file_path: Path) -> List[str]:
"""Read tags for a file using metadata.py as the single source of truth."""
try:
return metadata.read_tags_from_file(file_path)
except Exception:
return []
def group_tags_by_namespace(tags: Sequence[str]) -> Dict[str, List[str]]:
"""Return tags grouped by namespace for quick UI summaries."""
grouped: Dict[str,
List[str]] = {}
for tag in metadata.normalize_tags(list(tags)):
namespace, value = metadata.split_tag(tag)
key = namespace or "_untagged"
grouped.setdefault(key, []).append(value)
for items in grouped.values():
items.sort()
return grouped
def normalize_tags(tags: Iterable[str]) -> List[str]:
"""Expose metadata.normalize_tags for callers that imported the old helper."""
return metadata.normalize_tags(list(tags))
TUI/modalscreen/__init__.py
-7
View File
@@ -1,7 +0,0 @@
"""Modal screens for the Downlow Hub UI application."""
from .export import ExportModal
from .search import SearchModal
from .workers import WorkersModal
__all__ = ["ExportModal", "SearchModal", "WorkersModal"]
TUI/modalscreen/access.py
-151
View File
@@ -1,151 +0,0 @@
"""Modal for displaying files/url to access in web mode."""
from textual.screen import ModalScreen
from textual.containers import Container, Vertical, Horizontal
from textual.widgets import Static, Button, Label
from textual.app import ComposeResult
import logging
logger = logging.getLogger(__name__)
class AccessModal(ModalScreen):
"""Modal to display a file/URL that can be accessed from phone browser."""
CSS = """
Screen {
align: center middle;
}
#access-container {
width: 80;
height: auto;
border: thick $primary;
background: $surface;
}
#access-header {
dock: top;
height: 3;
background: $boost;
border-bottom: solid $accent;
content-align: center middle;
}
#access-content {
height: auto;
width: 1fr;
padding: 1 2;
border-bottom: solid $accent;
}
#access-footer {
dock: bottom;
height: 3;
background: $boost;
border-top: solid $accent;
align: center middle;
}
.access-url {
width: 1fr;
height: auto;
margin-bottom: 1;
border: solid $accent;
padding: 1;
}
.access-label {
width: 1fr;
height: auto;
margin-bottom: 1;
}
Button {
margin-right: 1;
}
"""
def __init__(self, title: str, content: str, is_url: bool = False):
"""Initialize access modal.
Args:
title: Title of the item being accessed
content: The URL or file path
is_url: Whether this is a URL (True) or file path (False)
"""
super().__init__()
self.item_title = title
self.item_content = content
self.is_url = is_url
def compose(self) -> ComposeResult:
"""Create the modal layout."""
with Container(id="access-container"):
with Vertical(id="access-header"):
yield Label(f"[bold]{self.item_title}[/bold]")
yield Label("[dim]Click link below to open in your browser[/dim]")
with Vertical(id="access-content"):
if self.is_url:
yield Label("[bold cyan]Link:[/bold cyan]", classes="access-label")
else:
yield Label("[bold cyan]File:[/bold cyan]", classes="access-label")
# Display as clickable link using HTML link element for web mode
# Rich link markup `[link=URL]` has parsing issues with url containing special chars
# Instead, use the HTML link markup that Textual-serve renders as <a> tag
# Format: [link=URL "tooltip"]text[/link] - the quotes help with parsing
link_text = f'[link="{self.item_content}"]Open in Browser[/link]'
content_box = Static(link_text, classes="access-url")
yield content_box
# Also show the URL for reference/copying
yield Label(self.item_content, classes="access-label")
yield Label(
"\n[yellow]↑ Click the link above to open on your device[/yellow]",
classes="access-label",
)
with Horizontal(id="access-footer"):
yield Button("Copy URL", id="copy-btn", variant="primary")
yield Button("Close", id="close-btn", variant="default")
def on_button_pressed(self, event: Button.Pressed) -> None:
"""Handle button presses."""
if event.button.id == "copy-btn":
# Copy to clipboard (optional - not critical if fails)
logger.info(f"Attempting to copy: {self.item_content}")
try:
# Try to use pyperclip if available
try:
import pyperclip
pyperclip.copy(self.item_content)
logger.info("URL copied to clipboard via pyperclip")
except ImportError:
# Fallback: try xclip on Linux or pbcopy on Mac
import subprocess
import sys
if sys.platform == "win32":
# Windows: use clipboard via pyperclip (already tried)
logger.debug(
"Windows clipboard not available without pyperclip"
)
else:
# Linux/Mac
process = subprocess.Popen(
["xclip",
"-selection",
"clipboard"],
stdin=subprocess.PIPE
)
process.communicate(self.item_content.encode("utf-8"))
logger.info("URL copied to clipboard via xclip")
except Exception as e:
logger.debug(f"Clipboard copy not available: {e}")
# Not critical - just informational
elif event.button.id == "close-btn":
self.dismiss()
TUI/modalscreen/config_modal.py
-1933
View File
File diff suppressed because it is too large Load Diff
TUI/modalscreen/download.py
-2153
View File
File diff suppressed because it is too large Load Diff
TUI/modalscreen/download.tcss
-183
View File
@@ -1,183 +0,0 @@
/* Download Modal Screen Stylesheet */
Screen {
background: $surface;
overlay: screen;
}
#download_modal {
width: 100%;
height: 100%;
border: heavy $primary;
background: $boost;
}
#download_title {
dock: top;
height: 1;
content-align: center middle;
background: $primary;
color: $text;
text-style: bold;
padding: 0 1;
}
/* Main horizontal layout: 2 columns left/right split */
#main_layout {
width: 1fr;
height: 1fr;
layout: horizontal;
padding: 1;
border: none;
}
/* Left column */
#left_column {
width: 2fr;
height: 1fr;
layout: vertical;
}
/* Right column */
#right_column {
width: 1fr;
height: 1fr;
layout: vertical;
}
/* All containers styling */
.grid_container {
width: 1fr;
height: 1fr;
padding: 1;
layout: vertical;
margin: 0 0 1 0;
}
#tags_container {
border: mediumpurple;
}
#url_container {
border: solid $accent;
}
#files_container {
border: solid $accent;
}
#playlist_container {
border: solid $accent;
layout: vertical;
height: 0;
}
#playlist_tree {
width: 1fr;
height: auto;
border: none;
padding: 0;
}
#playlist_input {
width: 1fr;
height: 1;
border: none;
padding: 0 1;
margin: 1 0 0 0;
}
#playlist_input_row {
width: 1fr;
height: auto;
layout: horizontal;
margin: 1 0 0 0;
}
.section_title {
width: 1fr;
height: 1;
text-align: left;
color: $text-muted;
text-style: bold;
margin: 0 0 0 0;
padding: 0;
}
/* TextArea widgets in containers */
#tags_textarea {
width: 1fr;
height: 1fr;
border: none;
padding: 0;
}
#paragraph_textarea {
width: 1fr;
height: 1fr;
border: none;
padding: 0;
}
/* Select widgets in containers */
#files_select {
width: 1fr;
height: 1fr;
border: none;
}
/* Footer layout - horizontal: checkboxes left, source middle, buttons right */
#footer_layout {
width: 1fr;
height: auto;
layout: horizontal;
padding: 1;
margin: 0;
background: $boost;
}
#checkbox_row {
width: auto;
height: auto;
layout: horizontal;
align: left middle;
}
#source_select {
width: 30;
height: 1;
border: none;
padding: 0 1;
margin: 0;
}
#button_row {
width: auto;
height: auto;
layout: horizontal;
align: right middle;
}
/* Progress bar - shown during download */
#progress_bar {
width: 1fr;
height: 0;
}
/* Checkbox and Button styling */
Checkbox {
margin: 0 2 0 0;
}
Button {
margin: 0 1 0 0;
width: 12;
}
#cancel_btn {
width: 12;
}
#submit_btn {
width: 12;
}
TUI/modalscreen/export.py
-593
View File
@@ -1,593 +0,0 @@
"""Export modal screen for exporting files with metadata."""
from textual.app import ComposeResult
from textual.screen import ModalScreen
from textual.containers import Container, Horizontal, Vertical
from textual.widgets import Static, Button, Input, TextArea, Select
from textual.binding import Binding
import logging
from typing import Optional
from pathlib import Path
import sys
# Add parent directory to path for imports
sys.path.insert(0, str(Path(__file__).parent.parent))
from SYS.utils import format_metadata_value
logger = logging.getLogger(__name__)
class ExportModal(ModalScreen):
"""Modal screen for exporting files with metadata and tags."""
BINDINGS = [
Binding("escape",
"cancel",
"Cancel"),
]
CSS_PATH = "export.tcss"
def __init__(
self,
result_data: Optional[dict] = None,
hydrus_available: bool = False,
debrid_available: bool = False,
):
"""Initialize the export modal with result data.
Args:
result_data: Dictionary containing:
- title: str - Item title
- tags: str - Comma-separated tags
- metadata: dict - File metadata (source-specific from item.metadata or local DB)
- source: str - Source identifier ('local', 'hydrus', 'debrid', etc)
- current_result: object - The full search result object
hydrus_available: bool - Whether Hydrus API is available
debrid_available: bool - Whether Debrid API is available
"""
super().__init__()
self.result_data = result_data or {}
self.hydrus_available = hydrus_available
self.debrid_available = debrid_available
self.metadata_display: Optional[Static] = None
self.tags_textarea: Optional[TextArea] = None
self.export_to_select: Optional[Select] = None
self.custom_path_input: Optional[Input] = None
self.libraries_select: Optional[Select] = None
self.size_input: Optional[Input] = None
self.format_select: Optional[Select] = None
self.file_ext: Optional[
str] = None # Store the file extension for format filtering
self.file_type: Optional[
str] = None # Store the file type (audio, video, image, document)
self.default_format: Optional[
str] = None # Store the default format to set after mount
def _determine_file_type(self, ext: str) -> tuple[str, list]:
"""Determine file type from extension and return type and format options.
Args:
ext: File extension (e.g., '.mp3', '.mp4', '.jpg')
Returns:
Tuple of (file_type, format_options) where format_options is a list of (label, value) tuples
"""
ext_lower = ext.lower() if ext else ""
from SYS.utils_constant import mime_maps
found_type = "unknown"
# Find type based on extension
for category, formats in mime_maps.items():
for fmt_key, fmt_info in formats.items():
if fmt_info.get("ext") == ext_lower:
found_type = category
break
if found_type != "unknown":
break
# Build format options for the found type
format_options = []
# If unknown, fallback to audio (matching legacy behavior)
target_type = found_type if found_type in mime_maps else "audio"
if target_type in mime_maps:
# Sort formats alphabetically
sorted_formats = sorted(mime_maps[target_type].items())
for fmt_key, fmt_info in sorted_formats:
label = fmt_key.upper()
value = fmt_key
format_options.append((label, value))
return (target_type, format_options)
def _get_library_options(self) -> list:
"""Get available library options from config.conf."""
options = [("Local", "local")]
try:
from SYS.config import (
load_config,
get_hydrus_access_key,
get_hydrus_url,
get_debrid_api_key,
)
config = load_config()
hydrus_url = (get_hydrus_url(config, "home") or "").strip()
hydrus_key = (get_hydrus_access_key(config, "home") or "").strip()
if self.hydrus_available and hydrus_url and hydrus_key:
options.append(("Hydrus Network", "hydrus"))
debrid_api_key = get_debrid_api_key(config)
if self.debrid_available and debrid_api_key:
options.append(("Debrid", "debrid"))
except Exception as e:
logger.error(f"Error loading config for libraries: {e}")
return options
def _get_metadata_text(self) -> str:
"""Format metadata from result data in a consistent display format."""
metadata = self.result_data.get("metadata",
{})
source = self.result_data.get("source", "unknown")
logger.info(
f"_get_metadata_text called - source: {source}, metadata type: {type(metadata)}, keys: {list(metadata.keys()) if metadata else 'empty'}"
)
if not metadata:
logger.info(
"_get_metadata_text - No metadata found, returning 'No metadata available'"
)
return "No metadata available"
lines = []
# Only display these specific fields in this order
display_fields = [
"duration",
"size",
"ext",
"media_type",
"time_imported",
"time_modified",
"hash",
]
# Display fields in a consistent order
for field in display_fields:
if field in metadata:
value = metadata[field]
# Skip complex types and None values
if isinstance(value, (dict, list)) or value is None:
continue
# Use central formatting rule
formatted_value = format_metadata_value(field, value)
# Format: "Field Name: value"
field_label = field.replace("_", " ").title()
lines.append(f"{field_label}: {formatted_value}")
# If we found any fields, display them
if lines:
logger.info(
f"_get_metadata_text - Returning {len(lines)} formatted metadata lines"
)
return "\n".join(lines)
else:
logger.info("_get_metadata_text - No matching fields found in metadata")
return "No metadata available"
def compose(self) -> ComposeResult:
"""Compose the export modal screen."""
with Container(id="export-container"):
yield Static("Export File with Metadata", id="export-title")
# Row 1: Three columns (Tag, Metadata, Export-To Options)
self.tags_textarea = TextArea(
text=self._format_tags(),
id="tags-area",
read_only=False,
)
yield self.tags_textarea
self.tags_textarea.border_title = "Tag"
# Metadata display instead of files tree
self.metadata_display = Static(
self._get_metadata_text(),
id="metadata-display",
)
yield self.metadata_display
self.metadata_display.border = ("solid", "dodgerblue")
# Right column: Export options
with Vertical(id="export-options"):
# Export To selector
self.export_to_select = Select(
[
("0x0",
"0x0"),
("Libraries",
"libraries"),
("Custom Path",
"path")
],
id="export-to-select",
)
yield self.export_to_select
# Libraries selector (initially hidden)
library_options = self._get_library_options()
self.libraries_select = Select(library_options, id="libraries-select")
yield self.libraries_select
# Custom path input (initially hidden)
self.custom_path_input = Input(
placeholder="Enter custom export path",
id="custom-path-input"
)
yield self.custom_path_input
# Get metadata for size and format options
metadata = self.result_data.get("metadata",
{})
original_size = metadata.get("size", "")
ext = metadata.get("ext", "")
# Store the extension and determine file type
self.file_ext = ext
self.file_type, format_options = self._determine_file_type(ext)
# Format size in MB for display
if original_size:
size_mb = (
int(original_size /
(1024 * 1024)) if isinstance(original_size,
(int,
float)) else original_size
)
size_display = f"{size_mb}Mb"
else:
size_display = ""
# Size input
self.size_input = Input(
value=size_display,
placeholder="Size (can reduce)",
id="size-input",
disabled=(
self.file_type == "document"
), # Disable for documents - no resizing needed
)
yield self.size_input
# Determine the default format value (match current extension to format options)
default_format = None
if ext and format_options:
# Map extension to format value (e.g., .flac -> "flac", .mp3 -> "mp3", .m4a -> "m4a")
ext_lower = ext.lower().lstrip(".") # Remove leading dot if present
# Try to find matching format option
for _, value in format_options:
if value and (ext_lower == value or f".{ext_lower}" == ext
or ext.endswith(f".{value}")):
default_format = value
logger.debug(f"Matched extension {ext} to format {value}")
break
# If no exact match, use first option
if not default_format and format_options:
default_format = format_options[0][1]
logger.debug(
f"No format match for {ext}, using first option: {default_format}"
)
# Store the default format to apply after mount
self.default_format = default_format
# Format selector based on file type
self.format_select = Select(
format_options if format_options else [("No conversion", "")],
id="format-select",
disabled=not format_options, # Disable if no format options (e.g., documents)
)
yield self.format_select
# Row 2: Buttons
with Horizontal(id="export-buttons"):
yield Button("Cancel", id="cancel-btn", variant="default")
yield Button("Export", id="export-btn", variant="primary")
def _format_tags(self) -> str:
"""Format tags from result data."""
tags = self.result_data.get("tags", "")
if isinstance(tags, str):
# Split by comma and rejoin with newlines
tags_list = [tag.strip() for tag in tags.split(",") if tag.strip()]
return "\n".join(tags_list)
elif isinstance(tags, list):
return "\n".join(tags)
return ""
def on_button_pressed(self, event: Button.Pressed) -> None:
"""Handle button press events."""
button_id = event.button.id
if button_id == "export-btn":
self._handle_export()
elif button_id == "cancel-btn":
self.action_cancel()
def on_select_changed(self, event: Select.Changed) -> None:
"""Handle select widget changes."""
if event.control.id == "export-to-select":
# Show/hide custom path and libraries based on selection
if self.custom_path_input:
self.custom_path_input.display = event.value == "path"
if self.libraries_select:
self.libraries_select.display = event.value == "libraries"
elif event.control.id == "libraries-select":
# Handle library selection (no special action needed currently)
logger.debug(f"Library selected: {event.value}")
def on_mount(self) -> None:
"""Handle mount event."""
# Initially hide custom path and libraries inputs (default is "0x0")
if self.custom_path_input:
self.custom_path_input.display = False
if self.libraries_select:
self.libraries_select.display = False
# Set the default format value to show it selected instead of "Select"
if self.default_format and self.format_select:
self.format_select.value = self.default_format
logger.debug(f"Set format selector to default value: {self.default_format}")
# Refresh metadata display after mount to ensure data is loaded
if self.metadata_display:
metadata_text = self._get_metadata_text()
self.metadata_display.update(metadata_text)
logger.debug(
f"Updated metadata display on mount: {bool(self.result_data.get('metadata'))}"
)
def _handle_export(self) -> None:
"""Handle the export action."""
try:
tags_text = self.tags_textarea.text.strip()
export_to = self.export_to_select.value if self.export_to_select else "0x0"
custom_path = self.custom_path_input.value.strip(
) if self.custom_path_input else ""
# Get library value - handle Select.BLANK case
library = "local" # default
if self.libraries_select and str(self.libraries_select.value
) != "Select.BLANK":
library = str(self.libraries_select.value)
elif self.libraries_select and self.libraries_select:
# If value is Select.BLANK, try to get from the options
try:
# Get first available library option as fallback
options = self._get_library_options()
if options:
library = options[0][
1] # Get the value part of first option tuple
except Exception:
library = "local"
size = self.size_input.value.strip() if self.size_input else ""
file_format = self.format_select.value if self.format_select else "mp4"
# Parse tags from textarea (one per line)
export_tags = set()
for line in tags_text.split("\n"):
tag = line.strip()
if tag:
export_tags.add(tag)
# For Hydrus export, filter out metadata-only tags (hash:, url:, relationship:)
if export_to == "libraries" and library == "hydrus":
metadata_prefixes = {"hash:",
"url:",
"relationship:"}
export_tags = {
tag
for tag in export_tags if not any(
tag.lower().startswith(prefix) for prefix in metadata_prefixes
)
}
logger.info(
f"Filtered tags for Hydrus - removed metadata tags, {len(export_tags)} tags remaining"
)
# Extract title and add as searchable tags if not already present
title = self.result_data.get("title", "").strip()
if title:
# Add the full title as a tag if not already present
title_tag = f"title:{title}"
if title_tag not in export_tags and not any(t.startswith("title:")
for t in export_tags):
export_tags.add(title_tag)
# Extract individual words from title as searchable tags (if reasonable length)
# Skip very short words and common stop words
if len(title) < 100: # Only for reasonably short titles
stop_words = {
"the",
"a",
"an",
"and",
"or",
"of",
"in",
"to",
"for",
"is",
"it",
"at",
"by",
"from",
"with",
"as",
"be",
"on",
"that",
"this",
"this",
}
words = title.lower().split()
for word in words:
# Clean up word (remove punctuation)
clean_word = "".join(c for c in word if c.isalnum())
# Only add if not a stop word and has some length
if clean_word and len(
clean_word) > 2 and clean_word not in stop_words:
if clean_word not in export_tags:
export_tags.add(clean_word)
logger.info(
f"Extracted {len(words)} words from title, added searchable title tags"
)
# Validate required fields - allow export to continue for Hydrus even with 0 actual tags
# (metadata tags will still be in the sidecar, and tags can be added later)
if not export_tags and export_to != "libraries":
logger.warning("No tags provided for export")
return
if export_to == "libraries" and not export_tags:
logger.warning(
"No actual tags for Hydrus export (only metadata was present)"
)
# Don't return - allow export to continue, file will be added to Hydrus even without tags
# Determine export path
export_path = None
if export_to == "path":
if not custom_path:
logger.warning("Custom path required but not provided")
return
export_path = custom_path
elif export_to == "libraries":
export_path = library # "local", "hydrus", "debrid"
else:
export_path = export_to # "0x0"
# Get metadata from result_data
metadata = self.result_data.get("metadata",
{})
# Extract file source info from result_data (passed by hub-ui)
file_hash = self.result_data.get("hash")
file_url = self.result_data.get("url")
file_path = self.result_data.get("path")
source = self.result_data.get("source", "unknown")
# Prepare export data
export_data = {
"export_to": export_to,
"export_path": export_path,
"library": library if export_to == "libraries" else None,
"tags": export_tags,
"size": size if size else None,
"format": file_format,
"metadata": metadata,
"original_data": self.result_data,
"hash": file_hash,
"url": file_url,
"path": file_path,
"source": source,
}
logger.info(
f"Export initiated: destination={export_path}, format={file_format}, size={size}, tags={export_tags}, source={source}, hash={file_hash}, path={file_path}"
)
# Dismiss the modal and return the export data
self.dismiss(export_data)
except Exception as e:
logger.error(f"Error during export: {e}", exc_info=True)
def action_cancel(self) -> None:
"""Handle cancel action."""
self.dismiss(None)
def create_notes_sidecar(file_path: Path, notes: str) -> None:
"""Create a .notes sidecar file with notes text.
Only creates file if notes are not empty.
Args:
file_path: Path to the exported file
notes: Notes text
"""
if not notes or not notes.strip():
return
notes_path = file_path.with_suffix(file_path.suffix + ".notes")
try:
with open(notes_path, "w", encoding="utf-8") as f:
f.write(notes.strip())
logger.info(f"Created notes sidecar: {notes_path}")
except Exception as e:
logger.error(f"Failed to create notes sidecar: {e}", exc_info=True)
def determine_needs_conversion(current_ext: str, target_format: str) -> bool:
"""Determine if conversion is needed between two formats.
Args:
current_ext: Current file extension (e.g., '.flac')
target_format: Target format name (e.g., 'mp3') or NoSelection object
Returns:
True if conversion is needed, False if it's already the target format
"""
# Handle NoSelection or None
if (not target_format or target_format == ""
or str(target_format.__class__.__name__) == "NoSelection"):
return False # No conversion requested
# Normalize the current extension
current_ext_lower = current_ext.lower().lstrip(".")
target_format_lower = str(target_format).lower()
# Check if they match
return current_ext_lower != target_format_lower
def calculate_size_tolerance(metadata: dict,
user_size_mb: Optional[str]) -> tuple[Optional[int],
Optional[int]]:
"""Calculate target size with 1MB grace period.
Args:
metadata: File metadata containing 'size' in bytes
user_size_mb: User-entered size like "756Mb" or empty string
Returns:
Tuple of (target_bytes, grace_bytes) where grace_bytes is 1MB (1048576),
or (None, None) if no size specified
"""
grace_bytes = 1 * 1024 * 1024 # 1MB grace period
if not user_size_mb or not user_size_mb.strip():
return None, grace_bytes
try:
# Parse the size string (format like "756Mb")
size_str = user_size_mb.strip().lower()
if size_str.endswith("mb"):
size_str = size_str[:-2]
elif size_str.endswith("m"):
size_str = size_str[:-1]
size_mb = float(size_str)
target_bytes = int(size_mb * 1024 * 1024)
return target_bytes, grace_bytes
except (ValueError, AttributeError):
return None, grace_bytes
TUI/modalscreen/export.tcss
-85
View File
@@ -1,85 +0,0 @@
/* Export Modal Screen Styling */
ExportModal {
align: center middle;
}
#export-container {
width: 140;
height: 55;
background: $panel;
border: solid $primary;
layout: grid;
grid-columns: 1fr 1fr 1fr;
grid-rows: auto 1fr auto;
}
#export-title {
height: 1;
text-align: center;
text-style: bold;
color: $accent;
background: $boost;
padding: 1 2;
column-span: 3;
}
/* Row 1: Three columns */
#tags-area {
height: 1fr;
column-span: 1;
border: solid mediumvioletred;
}
#metadata-display {
height: 1fr;
column-span: 1;
border: solid dodgerblue;
overflow: auto;
padding: 1;
}
#export-options {
height: 1fr;
column-span: 1;
border: solid mediumpurple;
layout: vertical;
padding: 1;
}
#export-options Select,
#export-options Input {
height: 3;
margin: 0 0 1 0;
}
#custom-path-input {
height: 3;
margin: 0 0 1 0;
}
#libraries-select {
height: 3;
margin: 0 0 1 0;
}
#size-input {
height: 3;
margin: 0 0 1 0;
}
#format-select {
height: 3;
}
/* Row 2: Buttons */
#export-buttons {
height: auto;
column-span: 3;
layout: horizontal;
}
#export-buttons Button {
width: 1fr;
margin: 0 1;
}
TUI/modalscreen/matrix_room_picker.py
-243
View File
@@ -1,243 +0,0 @@
from __future__ import annotations
from typing import Any, Dict, List, Optional
from textual.app import ComposeResult
from textual.containers import Container, Horizontal, ScrollableContainer
from textual.screen import ModalScreen
from textual.widgets import Static, Button, Checkbox, ListView, ListItem
from textual import work
from rich.text import Text
from ProviderCore.registry import get_plugin
import logging
logger = logging.getLogger(__name__)
class MatrixRoomPicker(ModalScreen[List[str]]):
"""Modal that lists Matrix rooms and returns selected defaults."""
CSS = """
MatrixRoomPicker {
align: center middle;
background: rgba(0, 0, 0, 0.45);
}
#matrix-room-picker {
width: 70%;
height: 70%;
background: $panel;
border: thick $primary;
padding: 1;
}
#matrix-room-picker-hint {
text-style: italic;
color: $text-muted;
margin-bottom: 1;
}
#matrix-room-scroll {
height: 1fr;
border: solid $surface;
padding: 1;
margin-bottom: 1;
}
#matrix-room-list {
padding: 0;
}
.matrix-room-row {
border-bottom: solid $surface;
padding: 1 0;
align: left middle;
}
.matrix-room-meta {
padding-left: 1;
content-align: left middle;
}
#matrix-room-actions {
height: 3;
align: right middle;
}
#matrix-room-status {
height: 3;
text-style: bold;
}
"""
def __init__(
self,
config: Dict[str, Any],
*,
existing: Optional[List[str]] = None,
rooms: Optional[List[Dict[str, Any]]] = None,
) -> None:
super().__init__()
self.config = config
self._prefetched_rooms = rooms
self._existing_ids = {str(r).strip() for r in (existing or []) if str(r).strip()}
self._checkbox_map: Dict[str, str] = {}
self._rooms: List[Dict[str, Any]] = []
self._status_widget: Optional[Static] = None
self._checklist: Optional[Vertical] = None
self._save_button: Optional[Button] = None
def compose(self) -> ComposeResult:
with Container(id="matrix-room-picker"):
yield Static("Matrix Default Rooms", classes="section-title")
yield Static(
"Choose rooms to keep in the sharing defaults and autocomplete.",
id="matrix-room-picker-hint",
)
with ScrollableContainer(id="matrix-room-scroll"):
yield ListView(id="matrix-room-list")
with Horizontal(id="matrix-room-actions"):
yield Button("Cancel", variant="error", id="matrix-room-cancel")
yield Button("Select All", id="matrix-room-select-all")
yield Button("Clear All", id="matrix-room-clear")
yield Button("Save defaults", variant="success", id="matrix-room-save")
yield Static("Loading rooms...", id="matrix-room-status")
def on_mount(self) -> None:
self._status_widget = self.query_one("#matrix-room-status", Static)
self._checklist = self.query_one("#matrix-room-list", ListView)
self._save_button = self.query_one("#matrix-room-save", Button)
if self._save_button:
self._save_button.disabled = True
if self._prefetched_rooms is not None:
self._apply_room_results(self._prefetched_rooms, None)
else:
if self._status_widget:
self._set_status("Loading rooms...")
self._load_rooms_background()
def on_list_view_selected(self, event: ListView.Selected) -> None:
"""Intercept ListView.Selected events and prevent them from bubbling up
to parent components which may react (e.g., the main config modal).
Selecting an item should not implicitly close the picker or change the
outer editor state."""
try:
# Stop propagation so parent handlers (ConfigModal) don't react.
event.stop()
except Exception:
logger.exception("Failed to stop ListView.Selected event propagation")
def _set_status(self, text: str) -> None:
if self._status_widget:
self._status_widget.update(text)
def on_checkbox_changed(self, event: Checkbox.Changed) -> None:
# Enable Save only when at least one checkbox is selected
any_selected = False
for checkbox_id in self._checkbox_map.keys():
try:
cb = self.query_one(f"#{checkbox_id}", Checkbox)
if cb.value:
any_selected = True
break
except Exception:
logger.exception("Error querying checkbox in MatrixRoomPicker; skipping")
continue
if self._save_button:
self._save_button.disabled = not any_selected
def _render_rooms(self, rooms: List[Dict[str, Any]]) -> None:
if not self._checklist:
return
for child in list(self._checklist.children):
child.remove()
self._checkbox_map.clear()
self._rooms = rooms
if not rooms:
self._set_status("Matrix returned no rooms.")
if self._save_button:
self._save_button.disabled = True
return
for idx, room in enumerate(rooms):
room_id = str(room.get("room_id") or "").strip()
name = str(room.get("name") or "").strip()
checkbox_id = f"matrix-room-{idx}"
# Prefer display name; otherwise fall back to room id
label_text = name or room_id or "Matrix Room"
checkbox = Checkbox(
label_text,
id=checkbox_id,
value=bool(room_id and room_id in self._existing_ids),
)
self._checkbox_map[checkbox_id] = room_id
list_item = ListItem(checkbox, classes="matrix-room-row")
self._checklist.mount(list_item)
self._set_status("Loaded rooms. Select one or more and save.")
if self._save_button:
self._save_button.disabled = False
@work(thread=True)
def _load_rooms_background(self) -> None:
try:
provider = get_plugin("matrix", self.config)
if provider is None:
raise RuntimeError("Matrix plugin unavailable")
rooms = provider.list_rooms()
self.app.call_from_thread(self._apply_room_results, rooms, None)
except Exception as exc:
self.app.call_from_thread(self._apply_room_results, [], str(exc))
def _apply_room_results(self, rooms: List[Dict[str, Any]], error: Optional[str]) -> None:
if error:
self._set_status(f"Failed to load Matrix rooms: {error}")
if self._save_button:
self._save_button.disabled = True
return
self._render_rooms(rooms)
# Ensure save button is enabled only if at least one checkbox is selected
any_selected = False
for cbid in self._checkbox_map.keys():
try:
cb = self.query_one(f"#{cbid}", Checkbox)
if cb.value:
any_selected = True
break
except Exception:
continue
if self._save_button:
self._save_button.disabled = not any_selected
def on_button_pressed(self, event: Button.Pressed) -> None:
if event.button.id == "matrix-room-cancel":
self.dismiss([])
elif event.button.id == "matrix-room-select-all":
for checkbox_id in self._checkbox_map.keys():
try:
cb = self.query_one(f"#{checkbox_id}", Checkbox)
cb.value = True
except Exception:
logger.exception("Failed to set checkbox value in MatrixRoomPicker")
if self._save_button:
self._save_button.disabled = False
elif event.button.id == "matrix-room-clear":
for checkbox_id in self._checkbox_map.keys():
try:
cb = self.query_one(f"#{checkbox_id}", Checkbox)
cb.value = False
except Exception:
logger.exception("Failed to set checkbox value to False in MatrixRoomPicker")
if self._save_button:
self._save_button.disabled = True
elif event.button.id == "matrix-room-save":
selected: List[str] = []
for checkbox_id, room_id in self._checkbox_map.items():
try:
cb = self.query_one(f"#{checkbox_id}", Checkbox)
if cb.value and room_id:
selected.append(room_id)
except Exception:
logger.exception("Failed to read checkbox state for '%s' while saving MatrixRoomPicker selection", checkbox_id)
self.dismiss(selected)
TUI/modalscreen/search.py
-433
View File
@@ -1,433 +0,0 @@
"""Search modal screen for OpenLibrary and Soulseek."""
from textual.app import ComposeResult
from textual.screen import ModalScreen
from textual.containers import Horizontal, Vertical
from textual.widgets import Static, Button, Input, Select, DataTable, TextArea
from textual.binding import Binding
from textual.message import Message
import logging
from typing import Optional, Any, List
from pathlib import Path
import sys
import asyncio
# Add parent directory to path for imports
sys.path.insert(0, str(Path(__file__).parent.parent))
from SYS.config import load_config, resolve_output_dir
from SYS.result_table import Table
from ProviderCore.registry import get_plugin_with_capability
logger = logging.getLogger(__name__)
class SearchModal(ModalScreen):
"""Modal screen for searching OpenLibrary and Soulseek."""
BINDINGS = [
Binding("escape",
"cancel",
"Cancel"),
Binding("enter",
"search_focused",
"Search"),
Binding("ctrl+t",
"scrape_tags",
"Scrape Tags"),
]
CSS_PATH = "search.tcss"
class SearchSelected(Message):
"""Posted when user selects a search result."""
def __init__(self, result: dict) -> None:
self.result = result
super().__init__()
def __init__(self, app_instance=None):
"""Initialize the search modal.
Args:
app_instance: Reference to the main App instance for worker creation
"""
super().__init__()
self.app_instance = app_instance
self.source_select: Optional[Select] = None
self.search_input: Optional[Input] = None
self.results_table: Optional[DataTable] = None
self.tags_textarea: Optional[TextArea] = None
self.library_source_select: Optional[Select] = None
self.current_results: List[Any] = [] # List of SearchResult objects
self.current_result_table: Optional[Table] = None
self.is_searching = False
self.current_worker = None # Track worker for search operations
def compose(self) -> ComposeResult:
"""Create child widgets for the search modal."""
with Vertical(id="search-container"):
yield Static("Search Books & Music", id="search-title")
with Horizontal(id="search-controls"):
# Source selector
self.source_select = Select(
[("OpenLibrary",
"openlibrary"),
("Soulseek",
"soulseek")],
value="openlibrary",
id="source-select",
)
yield self.source_select
# Search input
self.search_input = Input(
placeholder="Enter search query...",
id="search-input"
)
yield self.search_input
# Search button
yield Button("Search", id="search-button", variant="primary")
# Results table
self.results_table = DataTable(id="results-table")
yield self.results_table
# Two-column layout: tags on left, source/submit on right
with Horizontal(id="bottom-controls"):
# Left column: Tags textarea
with Vertical(id="tags-column"):
self.tags_textarea = TextArea(
text="",
id="result-tags-textarea",
read_only=False
)
self.tags_textarea.border_title = "Tags [Ctrl+T: Scrape]"
yield self.tags_textarea
# Right column: Library source and submit button
with Vertical(id="source-submit-column"):
# Library source selector (for OpenLibrary results)
self.library_source_select = Select(
[("Local",
"local"),
("Download",
"download")],
value="local",
id="library-source-select",
)
yield self.library_source_select
# Submit button
yield Button("Submit", id="submit-button", variant="primary")
# Buttons at bottom
with Horizontal(id="search-buttons"):
yield Button("Select", id="select-button", variant="primary")
yield Button("Download", id="download-button", variant="primary")
yield Button("Cancel", id="cancel-button", variant="default")
def on_mount(self) -> None:
"""Set up the table columns and focus."""
# Set up results table columns
self.results_table.add_columns(
"Title",
"Author/Artist",
"Year/Album",
"Details"
)
# Focus on search input
self.search_input.focus()
async def _perform_search(self) -> None:
"""Perform the actual search based on selected source."""
if not self.search_input or not self.source_select or not self.results_table:
logger.error("[search-modal] Widgets not initialized")
return
query = self.search_input.value.strip()
if not query:
logger.warning("[search-modal] Empty search query")
return
source = self.source_select.value
if not source or not isinstance(source, str):
logger.warning("[search-modal] No source selected")
return
# Clear existing results
self.results_table.clear(columns=True)
self.current_results = []
self.current_result_table = None
self.is_searching = True
# Create worker for tracking
if self.app_instance and hasattr(self.app_instance, "create_worker"):
self.current_worker = self.app_instance.create_worker(
source,
title=f"{source.capitalize()} Search: {query[:40]}",
description=f"Searching {source} for: {query}",
)
self.current_worker.log_step(f"Connecting to {source}...")
try:
provider = get_plugin_with_capability(source, "search")
if not provider:
logger.error(f"[search-modal] Provider not available: {source}")
if self.current_worker:
self.current_worker.finish(
"error",
f"Provider not available: {source}"
)
return
logger.info(f"[search-modal] Searching {source} for: {query}")
results = provider.search(query, limit=20)
self.current_results = results
if self.current_worker:
self.current_worker.log_step(f"Found {len(results)} results")
# Create ResultTable
table = Table(f"Search Results: {query}")
for res in results:
row = table.add_row()
# Add columns from result.columns
if res.columns:
for name, value in res.columns:
row.add_column(name, value)
else:
# Fallback if no columns defined
row.add_column("Title", res.title)
row.add_column(
"Target",
getattr(res,
"path",
None) or getattr(res,
"url",
None) or getattr(res,
"target",
None) or "",
)
self.current_result_table = table
# Populate UI
if table.rows:
# Add headers
headers = [col.name for col in table.rows[0].columns]
self.results_table.add_columns(*headers)
# Add rows
for row_vals in table.to_datatable_rows():
self.results_table.add_row(*row_vals)
else:
self.results_table.add_columns("Message")
self.results_table.add_row("No results found")
# Finish worker
if self.current_worker:
self.current_worker.finish("completed", f"Found {len(results)} results")
except Exception as e:
logger.error(f"[search-modal] Search error: {e}", exc_info=True)
if self.current_worker:
self.current_worker.finish("error", f"Search failed: {str(e)}")
finally:
self.is_searching = False
def on_button_pressed(self, event: Button.Pressed) -> None:
"""Handle button presses."""
button_id = event.button.id
if button_id == "search-button":
# Run search asynchronously
asyncio.create_task(self._perform_search())
elif button_id == "select-button":
# Get selected row and populate tags textarea
if self.results_table and self.results_table.row_count > 0:
selected_row = self.results_table.cursor_row
if 0 <= selected_row < len(self.current_results):
result = self.current_results[selected_row]
# Populate tags textarea with result metadata
self._populate_tags_from_result(result)
else:
logger.warning("[search-modal] No results to select")
elif button_id == "download-button":
# Download the selected result
if self.current_results and self.results_table.row_count > 0:
selected_row = self.results_table.cursor_row
if 0 <= selected_row < len(self.current_results):
result = self.current_results[selected_row]
if getattr(result, "table", "") == "openlibrary":
asyncio.create_task(self._download_book(result))
else:
logger.warning(
"[search-modal] Download only supported for OpenLibrary results"
)
else:
logger.warning("[search-modal] No result selected for download")
elif button_id == "submit-button":
# Submit the current result with tags and source
if self.current_results and self.results_table.row_count > 0:
selected_row = self.results_table.cursor_row
if 0 <= selected_row < len(self.current_results):
result = self.current_results[selected_row]
# Convert to dict if needed for submission
if hasattr(result, "to_dict"):
result_dict = result.to_dict()
else:
result_dict = result
# Get tags from textarea
tags_text = self.tags_textarea.text if self.tags_textarea else ""
# Get library source (if OpenLibrary)
library_source = (
self.library_source_select.value
if self.library_source_select else "local"
)
# Add tags and source to result
result_dict["tags_text"] = tags_text
result_dict["library_source"] = library_source
# Post message and dismiss
self.post_message(self.SearchSelected(result_dict))
self.dismiss(result_dict)
else:
logger.warning("[search-modal] No result selected for submission")
elif button_id == "cancel-button":
self.dismiss(None)
def _populate_tags_from_result(self, result: Any) -> None:
"""Populate the tags textarea from a selected result."""
if not self.tags_textarea:
return
# Handle both SearchResult objects and dicts
if hasattr(result, "full_metadata"):
metadata = result.full_metadata or {}
source = result.table
title = result.title
else:
# Handle dict (legacy or from to_dict)
if "full_metadata" in result:
metadata = result["full_metadata"] or {}
elif "raw_data" in result:
metadata = result["raw_data"] or {}
else:
metadata = result
source = result.get("table", "")
title = result.get("title", "")
# Format tags based on result source
if source == "openlibrary":
# For OpenLibrary: title, author, year
author = (
", ".join(metadata.get("authors",
[]))
if isinstance(metadata.get("authors"),
list) else metadata.get("authors",
"")
)
year = str(metadata.get("year", ""))
tags = []
if title:
tags.append(title)
if author:
tags.append(author)
if year:
tags.append(year)
tags_text = "\n".join(tags)
elif source == "soulseek":
# For Soulseek: artist, album, title, track
tags = []
if metadata.get("artist"):
tags.append(metadata["artist"])
if metadata.get("album"):
tags.append(metadata["album"])
if metadata.get("track_num"):
tags.append(f"Track {metadata['track_num']}")
if title:
tags.append(title)
tags_text = "\n".join(tags)
else:
# Generic fallback
tags = [title]
tags_text = "\n".join(tags)
self.tags_textarea.text = tags_text
logger.info("[search-modal] Populated tags textarea from result")
async def _download_book(self, result: Any) -> None:
"""Download a book from OpenLibrary using the provider."""
if getattr(result, "table", "") != "openlibrary":
logger.warning(
"[search-modal] Download only supported for OpenLibrary results"
)
return
try:
config = load_config()
output_dir = resolve_output_dir(config)
provider = get_plugin_with_capability("openlibrary", "search", config=config)
if not provider:
logger.error("[search-modal] Provider not available: openlibrary")
return
title = getattr(result, "title", "")
logger.info(f"[search-modal] Starting download for: {title}")
downloaded = await asyncio.to_thread(provider.download, result, output_dir)
if downloaded:
logger.info(f"[search-modal] Download successful: {downloaded}")
else:
logger.warning(f"[search-modal] Download failed for: {title}")
except Exception as e:
logger.error(f"[search-modal] Download error: {e}", exc_info=True)
def action_search_focused(self) -> None:
"""Action for Enter key - only search if search input is focused."""
if self.search_input and self.search_input.has_focus and not self.is_searching:
asyncio.create_task(self._perform_search())
def action_scrape_tags(self) -> None:
"""Action for Ctrl+T - populate tags from selected result."""
if self.current_results and self.results_table and self.results_table.row_count > 0:
try:
selected_row = self.results_table.cursor_row
if 0 <= selected_row < len(self.current_results):
result = self.current_results[selected_row]
self._populate_tags_from_result(result)
logger.info(
f"[search-modal] Ctrl+T: Populated tags from result at row {selected_row}"
)
else:
logger.warning(
f"[search-modal] Ctrl+T: Invalid row index {selected_row}"
)
except Exception as e:
logger.error(f"[search-modal] Ctrl+T error: {e}")
else:
logger.warning("[search-modal] Ctrl+T: No results selected")
def action_cancel(self) -> None:
"""Action for Escape key - close modal."""
self.dismiss(None)
def on_input_submitted(self, event: Input.Submitted) -> None:
"""Handle Enter key in search input - only trigger search here."""
if event.input.id == "search-input":
if not self.is_searching:
asyncio.create_task(self._perform_search())
TUI/modalscreen/search.tcss
-121
View File
@@ -1,121 +0,0 @@
/* Search Modal Screen Styling */
SearchModal {
align: center middle;
}
Screen {
layout: vertical;
}
#search-container {
width: 140;
height: 40;
background: $panel;
border: solid $primary;
layout: vertical;
}
Static#search-title {
height: 3;
dock: top;
text-align: center;
text-style: bold;
color: $accent;
background: $boost;
padding: 1 2;
}
#search-controls {
height: auto;
layout: horizontal;
padding: 1;
border: solid $primary;
}
#source-select {
width: 20;
margin-right: 1;
}
#search-input {
width: 1fr;
margin-right: 1;
}
#search-button {
width: 12;
}
#results-table {
height: 1fr;
border: solid $primary;
}
DataTable {
border: solid $accent;
}
DataTable > .datatable--header {
background: $boost;
color: $accent;
text-style: bold;
}
DataTable > .datatable--cursor-row {
background: $accent;
}
#bottom-controls {
height: auto;
layout: horizontal;
padding: 1;
border: solid $primary;
}
#tags-column {
width: 1fr;
layout: vertical;
padding-right: 1;
height: auto;
}
#result-tags-textarea {
height: 10;
width: 1fr;
border: solid $accent;
}
#source-submit-column {
width: 20;
layout: vertical;
padding-left: 1;
height: auto;
}
#library-source-select {
width: 1fr;
margin-bottom: 1;
}
#submit-button {
width: 1fr;
}
#search-buttons {
height: 3;
dock: bottom;
layout: horizontal;
padding: 1;
border: solid $primary;
align: center middle;
}
#select-button {
width: 12;
margin-right: 2;
}
#cancel-button {
width: 12;
}
TUI/modalscreen/selection_modal.py
-69
View File
@@ -1,69 +0,0 @@
from textual.app import ComposeResult
from textual.screen import ModalScreen
from textual.containers import Container, ScrollableContainer
from textual.widgets import Static, Button
from typing import List
class SelectionModal(ModalScreen[str]):
"""A modal for selecting a type from a list of strings."""
CSS = """
SelectionModal {
align: center middle;
background: rgba(0, 0, 0, 0.5);
}
#selection-container {
width: 40%;
height: 60%;
background: $panel;
border: thick $primary;
padding: 1;
}
.selection-title {
background: $accent;
color: $text;
padding: 0 1;
margin-bottom: 1;
text-align: center;
text-style: bold;
height: 3;
content-align: center middle;
}
.selection-button {
width: 100%;
margin-bottom: 1;
}
#selection-cancel {
width: 100%;
margin-top: 1;
background: $error;
}
"""
def __init__(self, title: str, options: List[str]) -> None:
super().__init__()
self.selection_title = title
self.options = sorted(options)
def compose(self) -> ComposeResult:
with Container(id="selection-container"):
yield Static(self.selection_title, classes="selection-title")
with ScrollableContainer():
for i, opt in enumerate(self.options):
yield Button(opt, id=f"opt-{i}", classes="selection-button")
yield Button("Cancel", id="selection-cancel")
def on_button_pressed(self, event: Button.Pressed) -> None:
if event.button.id == "selection-cancel":
self.dismiss("")
elif event.button.id and event.button.id.startswith("opt-"):
try:
idx = int(event.button.id.replace("opt-", ""))
selection = self.options[idx]
self.dismiss(selection)
except (ValueError, IndexError):
pass
TUI/modalscreen/workers.py
-709
View File
@@ -1,709 +0,0 @@
"""Workers modal screen for monitoring and managing background tasks."""
from textual.app import ComposeResult
from textual.screen import ModalScreen
from textual.containers import Horizontal, Vertical
from textual.widgets import Static, Button, DataTable, TextArea
from textual.binding import Binding
from textual.message import Message
import logging
from typing import Optional, Dict, List, Any
from pathlib import Path
import sys
# Add parent directory to path for imports
sys.path.insert(0, str(Path(__file__).parent.parent))
logger = logging.getLogger(__name__)
class WorkersModal(ModalScreen):
"""Modal screen for monitoring running and finished workers."""
BINDINGS = [
Binding("escape",
"cancel",
"Cancel"),
]
CSS_PATH = "workers.tcss"
class WorkerUpdated(Message):
"""Posted when worker list is updated."""
def __init__(self, workers: List[Dict[str, Any]]) -> None:
self.workers = workers
super().__init__()
class WorkerCancelled(Message):
"""Posted when user cancels a worker."""
def __init__(self, worker_id: str) -> None:
self.worker_id = worker_id
super().__init__()
def __init__(self, app_instance=None):
"""Initialize the workers modal.
Args:
app_instance: Reference to the hub app for accessing worker info
"""
super().__init__()
self.app_instance = app_instance
self.running_table: Optional[DataTable] = None
self.finished_table: Optional[DataTable] = None
self.stdout_display: Optional[TextArea] = None
self.running_workers: List[Dict[str, Any]] = []
self.finished_workers: List[Dict[str, Any]] = []
self.selected_worker_id: Optional[str] = None
self.show_running = False # Start with finished tab
def compose(self) -> ComposeResult:
"""Create child widgets for the workers modal."""
with Vertical(id="workers-container"):
# Title with toggle buttons
with Horizontal(id="workers-title-bar"):
yield Static("Workers Monitor", id="workers-title")
yield Button("Running", id="toggle-running-btn", variant="primary")
yield Button("Finished", id="toggle-finished-btn", variant="default")
# Running tab content (initially hidden)
with Vertical(id="running-section"):
self.running_table = DataTable(id="running-table")
yield self.running_table
with Horizontal(id="running-controls"):
yield Button("Refresh", id="running-refresh-btn", variant="primary")
yield Button(
"Stop Selected",
id="running-stop-btn",
variant="warning"
)
yield Button("Stop All", id="running-stop-all-btn", variant="error")
# Finished tab content (initially visible)
with Vertical(id="finished-section"):
self.finished_table = DataTable(id="finished-table")
yield self.finished_table
with Horizontal(id="finished-controls"):
yield Button(
"Refresh",
id="finished-refresh-btn",
variant="primary"
)
yield Button(
"Clear Selected",
id="finished-clear-btn",
variant="warning"
)
yield Button(
"Clear All",
id="finished-clear-all-btn",
variant="error"
)
# Shared textarea for displaying worker logs
with Vertical(id="logs-section"):
yield Static("Worker Logs:", id="logs-label")
self.stdout_display = TextArea(id="stdout-display", read_only=True)
yield self.stdout_display
with Horizontal(id="workers-buttons"):
yield Button("Close", id="close-btn", variant="primary")
def on_mount(self) -> None:
"""Set up the tables and load worker data."""
# Set up running workers table
if self.running_table:
self.running_table.add_columns(
"ID",
"Type",
"Status",
"Pipe",
"Progress",
"Started",
"Details"
)
self.running_table.zebra_stripes = True
# Set up finished workers table
if self.finished_table:
self.finished_table.add_columns(
"ID",
"Type",
"Result",
"Pipe",
"Started",
"Completed",
"Duration",
"Details"
)
self.finished_table.zebra_stripes = True
# Set initial view (show finished by default)
self._update_view_visibility()
# Load initial data
self.refresh_workers()
# Don't set up periodic refresh - it was causing issues with stdout display
# Users can click the Refresh button to update manually
def refresh_workers(self) -> None:
"""Refresh the workers data from app instance."""
try:
if not self.app_instance:
logger.warning("[workers-modal] No app instance provided")
return
# Get running workers from app instance
# This assumes the app has a get_running_workers() method
if hasattr(self.app_instance, "get_running_workers"):
self.running_workers = self.app_instance.get_running_workers()
else:
self.running_workers = []
# Get finished workers from app instance
if hasattr(self.app_instance, "get_finished_workers"):
self.finished_workers = self.app_instance.get_finished_workers()
if self.finished_workers:
logger.info(
f"[workers-modal-refresh] Got {len(self.finished_workers)} finished workers from app"
)
# Log the keys in the first worker to verify structure
if isinstance(self.finished_workers[0], dict):
logger.info(
f"[workers-modal-refresh] First worker keys: {list(self.finished_workers[0].keys())}"
)
logger.info(
f"[workers-modal-refresh] First worker: {self.finished_workers[0]}"
)
else:
logger.warning(
f"[workers-modal-refresh] First worker is not a dict: {type(self.finished_workers[0])}"
)
else:
self.finished_workers = []
# Update tables
self._update_running_table()
self._update_finished_table()
logger.info(
f"[workers-modal] Refreshed: {len(self.running_workers)} running, {len(self.finished_workers)} finished"
)
except Exception as e:
logger.error(f"[workers-modal] Error refreshing workers: {e}")
def _update_view_visibility(self) -> None:
"""Toggle visibility between running and finished views."""
try:
running_section = self.query_one("#running-section", Vertical)
finished_section = self.query_one("#finished-section", Vertical)
toggle_running_btn = self.query_one("#toggle-running-btn", Button)
toggle_finished_btn = self.query_one("#toggle-finished-btn", Button)
if self.show_running:
running_section.display = True
finished_section.display = False
toggle_running_btn.variant = "primary"
toggle_finished_btn.variant = "default"
logger.debug("[workers-modal] Switched to Running view")
else:
running_section.display = False
finished_section.display = True
toggle_running_btn.variant = "default"
toggle_finished_btn.variant = "primary"
logger.debug("[workers-modal] Switched to Finished view")
except Exception as e:
logger.error(f"[workers-modal] Error updating view visibility: {e}")
def _update_running_table(self) -> None:
"""Update the running workers table."""
try:
if not self.running_table:
logger.error("[workers-modal] Running table not initialized")
return
self.running_table.clear()
if not self.running_workers:
self.running_table.add_row(
"---",
"---",
"---",
"---",
"---",
"---",
"No workers running"
)
logger.debug("[workers-modal] No running workers to display")
return
logger.debug(
f"[workers-modal] Updating running table with {len(self.running_workers)} workers"
)
for idx, worker_info in enumerate(self.running_workers):
try:
worker_id = worker_info.get("id", "unknown")
worker_type = worker_info.get("type", "unknown")
status = worker_info.get("status", "running")
progress = worker_info.get("progress", "")
started = worker_info.get("started", "")
details = worker_info.get("details", "")
pipe = worker_info.get("pipe", "")
# Ensure values are strings
worker_id = str(worker_id) if worker_id else "unknown"
worker_type = str(worker_type) if worker_type else "unknown"
status = str(status) if status else "running"
progress = str(progress) if progress else "---"
started = str(started) if started else "---"
details = str(details) if details else "---"
pipe_display = self._summarize_pipe(pipe)
# Truncate long strings
progress = progress[:20]
started = started[:19]
details = details[:30]
pipe_display = pipe_display[:40]
self.running_table.add_row(
worker_id[:8],
worker_type[:15],
status[:10],
pipe_display,
progress,
started,
details,
)
if idx == 0: # Log first entry
logger.debug(
f"[workers-modal] Added running row {idx}: {worker_id[:8]} {worker_type[:15]} {status}"
)
except Exception as row_error:
logger.error(
f"[workers-modal] Error adding running row {idx}: {row_error}",
exc_info=True,
)
logger.debug(
f"[workers-modal] Updated running table with {len(self.running_workers)} workers"
)
except Exception as e:
logger.error(
f"[workers-modal] Error updating running table: {e}",
exc_info=True
)
def _update_finished_table(self) -> None:
"""Update the finished workers table."""
try:
if not self.finished_table:
logger.error("[workers-modal] Finished table not initialized")
return
self.finished_table.clear()
if not self.finished_workers:
self.finished_table.add_row(
"---",
"---",
"---",
"---",
"---",
"---",
"---",
"No finished workers"
)
logger.debug("[workers-modal] No finished workers to display")
return
logger.info(
f"[workers-modal-update] STARTING to update finished table with {len(self.finished_workers)} workers"
)
added_count = 0
error_count = 0
for idx, worker_info in enumerate(self.finished_workers):
try:
worker_id = worker_info.get("id", "unknown")
worker_type = worker_info.get("type", "unknown")
result = worker_info.get("result", "unknown")
completed = worker_info.get("completed", "")
duration = worker_info.get("duration", "")
details = worker_info.get("details", "")
pipe = worker_info.get("pipe", "")
started = worker_info.get("started", "")
# Ensure values are strings
worker_id = str(worker_id) if worker_id else "unknown"
worker_type = str(worker_type) if worker_type else "unknown"
result = str(result) if result else "unknown"
completed = str(completed) if completed else "---"
duration = str(duration) if duration else "---"
details = str(details) if details else "---"
started = str(started) if started else "---"
pipe_display = self._summarize_pipe(pipe)
# Truncate long strings
result = result[:15]
completed = completed[:19]
started = started[:19]
duration = duration[:10]
details = details[:30]
pipe_display = pipe_display[:40]
self.finished_table.add_row(
worker_id[:8],
worker_type[:15],
result,
pipe_display,
started,
completed,
duration,
details,
)
added_count += 1
except Exception as row_error:
error_count += 1
logger.error(
f"[workers-modal-update] Error adding finished row {idx}: {row_error}",
exc_info=True,
)
logger.info(
f"[workers-modal-update] COMPLETED: Added {added_count}/{len(self.finished_workers)} finished workers (errors: {error_count})"
)
logger.debug(
f"[workers-modal-update] Finished table row_count after update: {self.finished_table.row_count}"
)
except Exception as e:
logger.error(
f"[workers-modal] Error updating finished table: {e}",
exc_info=True
)
def on_data_table_row_highlighted(self, event: DataTable.RowHighlighted) -> None:
"""Handle row highlight in tables - display stdout."""
try:
logger.info(
f"[workers-modal] Row highlighted, cursor_row: {event.cursor_row}"
)
# Get the selected worker from the correct table
workers_list = None
if event.control == self.running_table:
workers_list = self.running_workers
logger.debug("[workers-modal] Highlighted in running table")
elif event.control == self.finished_table:
workers_list = self.finished_workers
logger.debug(
f"[workers-modal] Highlighted in finished table, list size: {len(workers_list)}"
)
else:
logger.warning(f"[workers-modal] Unknown table: {event.control}")
return
# Get the worker at this row
if workers_list and 0 <= event.cursor_row < len(workers_list):
worker = workers_list[event.cursor_row]
worker_id = worker.get("id", "")
logger.info(f"[workers-modal] Highlighted worker: {worker_id}")
if worker_id:
self.selected_worker_id = worker_id
# Display the stdout
self._update_stdout_display(worker_id, worker)
else:
logger.warning(
f"[workers-modal] Row {event.cursor_row} out of bounds for list of size {len(workers_list) if workers_list else 0}"
)
except Exception as e:
logger.error(
f"[workers-modal] Error handling row highlight: {e}",
exc_info=True
)
def on_data_table_cell_highlighted(self, event: DataTable.CellHighlighted) -> None:
"""Handle cell highlight in tables - display stdout (backup for row selection)."""
try:
# CellHighlighted has coordinate (row, column) not cursor_row
cursor_row = event.coordinate.row
logger.debug(
f"[workers-modal] Cell highlighted, row: {cursor_row}, column: {event.coordinate.column}"
)
# Get the selected worker from the correct table
workers_list = None
if event.data_table == self.running_table:
workers_list = self.running_workers
logger.debug("[workers-modal] Cell highlighted in running table")
elif event.data_table == self.finished_table:
workers_list = self.finished_workers
logger.debug(
f"[workers-modal] Cell highlighted in finished table, list size: {len(workers_list)}"
)
else:
return
# Get the worker at this row
if workers_list and 0 <= cursor_row < len(workers_list):
worker = workers_list[cursor_row]
worker_id = worker.get("id", "")
if worker_id and worker_id != self.selected_worker_id:
logger.info(f"[workers-modal] Cell-highlighted worker: {worker_id}")
self.selected_worker_id = worker_id
# Display the stdout
self._update_stdout_display(worker_id, worker)
except Exception as e:
logger.debug(f"[workers-modal] Error handling cell highlight: {e}")
def _update_stdout_display(
self,
worker_id: str,
worker: Optional[Dict[str,
Any]] = None
) -> None:
"""Update the stdout textarea with logs from the selected worker."""
try:
if not self.stdout_display:
logger.error("[workers-modal] stdout_display not initialized")
return
logger.debug(
f"[workers-modal] Updating stdout display for worker: {worker_id}"
)
worker_data = worker or self._locate_worker(worker_id)
stdout_text = self._resolve_worker_stdout(worker_id, worker_data)
pipe_text = self._resolve_worker_pipe(worker_id, worker_data)
events = self._get_worker_events(worker_id)
timeline_text = self._format_worker_timeline(events)
sections = []
if pipe_text:
sections.append(f"Pipe:\n{pipe_text}")
if timeline_text:
sections.append("Timeline:\n" + timeline_text)
logs_body = (stdout_text or "").strip()
sections.append(
"Logs:\n" + (logs_body if logs_body else "(no logs recorded)")
)
combined_text = "\n\n".join(sections)
logger.debug(
f"[workers-modal] Setting textarea to {len(combined_text)} chars (stdout_len={len(stdout_text or '')})"
)
self.stdout_display.text = combined_text
if len(combined_text) > 10:
try:
self.stdout_display.cursor_location = (len(combined_text) - 1, 0)
except Exception:
logger.exception("Failed to set stdout_display cursor location")
logger.info("[workers-modal] Updated stdout display successfully")
except Exception as e:
logger.error(
f"[workers-modal] Error updating stdout display: {e}",
exc_info=True
)
def _locate_worker(self, worker_id: str) -> Optional[Dict[str, Any]]:
for worker in self.running_workers or []:
if isinstance(worker, dict) and worker.get("id") == worker_id:
return worker
for worker in self.finished_workers or []:
if isinstance(worker, dict) and worker.get("id") == worker_id:
return worker
return None
def _resolve_worker_stdout(
self,
worker_id: str,
worker: Optional[Dict[str,
Any]]
) -> str:
if worker and worker.get("stdout"):
return worker.get("stdout", "") or ""
manager = getattr(self.app_instance, "worker_manager", None)
if manager:
try:
return manager.get_stdout(worker_id) or ""
except Exception as exc:
logger.debug(
f"[workers-modal] Could not fetch stdout for {worker_id}: {exc}"
)
return ""
def _resolve_worker_pipe(
self,
worker_id: str,
worker: Optional[Dict[str,
Any]]
) -> str:
if worker and worker.get("pipe"):
return str(worker.get("pipe"))
record = self._fetch_worker_record(worker_id)
if record and record.get("pipe"):
return str(record.get("pipe"))
return ""
def _fetch_worker_record(self, worker_id: str) -> Optional[Dict[str, Any]]:
manager = getattr(self.app_instance, "worker_manager", None)
if not manager:
return None
try:
return manager.get_worker(worker_id)
except Exception as exc:
logger.debug(
f"[workers-modal] Could not fetch worker record {worker_id}: {exc}"
)
return None
def _get_worker_events(self,
worker_id: str,
limit: int = 250) -> List[Dict[str,
Any]]:
manager = getattr(self.app_instance, "worker_manager", None)
if not manager:
return []
try:
return manager.get_worker_events(worker_id, limit=limit)
except Exception as exc:
logger.debug(
f"[workers-modal] Could not fetch worker events {worker_id}: {exc}"
)
return []
def _format_worker_timeline(self, events: List[Dict[str, Any]]) -> str:
if not events:
return ""
lines: List[str] = []
for event in events:
timestamp = self._format_event_timestamp(event.get("created_at"))
label = (event.get("event_type") or "").upper() or "EVENT"
channel = (event.get("channel") or "").upper()
if channel and channel not in label:
label = f"{label}/{channel}"
step = event.get("step") or ""
message = event.get("message") or ""
prefix = ""
if event.get("event_type") == "step" and step:
prefix = f"{step} :: "
elif step and step not in message:
prefix = f"{step} :: "
formatted_message = self._format_message_block(message)
lines.append(f"[{timestamp}] {label}: {prefix}{formatted_message}")
return "\n".join(lines)
def _format_event_timestamp(self, raw_timestamp: Any) -> str:
if not raw_timestamp:
return "--:--:--"
text = str(raw_timestamp)
if "T" in text:
time_part = text.split("T", 1)[1]
elif " " in text:
time_part = text.split(" ", 1)[1]
else:
time_part = text
return time_part[:8] if len(time_part) >= 8 else time_part
def _format_message_block(self, message: str) -> str:
clean = (message or "").strip()
if not clean:
return "(empty)"
lines = clean.splitlines()
if len(lines) == 1:
return lines[0]
head, *rest = lines
indented = "\n".join(f" {line}" for line in rest)
return f"{head}\n{indented}"
def _summarize_pipe(self, pipe_value: Any, limit: int = 40) -> str:
text = str(pipe_value or "").strip()
if not text:
return "(none)"
return text if len(text) <= limit else text[:limit - 3] + "..."
def on_button_pressed(self, event: Button.Pressed) -> None:
"""Handle button presses."""
button_id = event.button.id
try:
if button_id == "toggle-running-btn":
self.show_running = True
self._update_view_visibility()
return
elif button_id == "toggle-finished-btn":
self.show_running = False
self._update_view_visibility()
return
if button_id == "running-refresh-btn":
self.refresh_workers()
elif button_id == "running-stop-btn":
# Stop selected running worker
if self.running_table and self.running_table.row_count > 0:
try:
selected_row = self.running_table.cursor_row
if 0 <= selected_row < len(self.running_workers):
worker = self.running_workers[selected_row]
worker_id = worker.get("id")
if self.app_instance and hasattr(self.app_instance,
"stop_worker"):
self.app_instance.stop_worker(worker_id)
logger.info(
f"[workers-modal] Stopped worker: {worker_id}"
)
self.refresh_workers()
except Exception as e:
logger.error(f"[workers-modal] Error stopping worker: {e}")
elif button_id == "running-stop-all-btn":
# Stop all running workers
if self.app_instance and hasattr(self.app_instance, "stop_all_workers"):
self.app_instance.stop_all_workers()
logger.info("[workers-modal] Stopped all workers")
self.refresh_workers()
elif button_id == "finished-refresh-btn":
self.refresh_workers()
elif button_id == "finished-clear-btn":
# Clear selected finished worker
if self.finished_table and self.finished_table.row_count > 0:
try:
selected_row = self.finished_table.cursor_row
if 0 <= selected_row < len(self.finished_workers):
worker = self.finished_workers[selected_row]
worker_id = worker.get("id")
if self.app_instance and hasattr(self.app_instance,
"clear_finished_worker"):
self.app_instance.clear_finished_worker(worker_id)
logger.info(
f"[workers-modal] Cleared worker: {worker_id}"
)
self.refresh_workers()
except Exception as e:
logger.error(f"[workers-modal] Error clearing worker: {e}")
elif button_id == "finished-clear-all-btn":
# Clear all finished workers
if self.app_instance and hasattr(self.app_instance,
"clear_all_finished_workers"):
self.app_instance.clear_all_finished_workers()
logger.info("[workers-modal] Cleared all finished workers")
self.refresh_workers()
elif button_id == "close-btn":
self.dismiss(None)
except Exception as e:
logger.error(f"[workers-modal] Error in on_button_pressed: {e}")
def action_cancel(self) -> None:
"""Action for Escape key - close modal."""
self.dismiss(None)
TUI/modalscreen/workers.tcss
-119
View File
@@ -1,119 +0,0 @@
/* Workers Modal Stylesheet */
Screen {
background: $surface;
color: $text;
}
#workers-container {
width: 100%;
height: 100%;
layout: vertical;
background: $panel;
}
#workers-title-bar {
dock: top;
height: 3;
layout: horizontal;
background: $boost;
border: solid $accent;
padding: 0 1;
}
#workers-title {
width: 1fr;
height: 100%;
content-align-vertical: middle;
color: $text;
text-style: bold;
}
#toggle-running-btn,
#toggle-finished-btn {
width: auto;
height: 100%;
margin: 0;
}
#running-section,
#finished-section {
width: 100%;
height: 40%;
layout: vertical;
border: solid $accent;
}
#running-table,
#finished-table {
width: 100%;
height: 1fr;
border: solid $accent;
}
#running-controls,
#finished-controls {
width: 100%;
height: auto;
min-height: 3;
layout: horizontal;
background: $boost;
padding: 1;
border-top: solid $accent;
}
#running-controls Button,
#finished-controls Button {
margin-right: 1;
min-width: 15;
}
#logs-label {
height: 1;
margin: 0 1;
text-style: bold;
}
#logs-section {
width: 100%;
height: 1fr;
layout: vertical;
border: solid $accent;
background: $panel;
}
#stdout-display {
width: 100%;
height: 1fr;
border: solid $accent;
margin: 1;
}
#workers-buttons {
dock: bottom;
height: auto;
min-height: 3;
layout: horizontal;
border: solid $accent;
padding: 1;
}
#workers-buttons Button {
margin-right: 1;
min-width: 15;
}
DataTable {
border: solid $accent;
}
DataTable > .datatable--header {
background: $boost;
color: $text;
text-style: bold;
}
DataTable > .datatable--cursor {
background: $accent;
color: $panel;
}
TUI/tui.tcss
-249
View File
@@ -1,249 +0,0 @@
#app-shell {
width: 100%;
height: 100%;
padding: 1 2;
background: $surface;
layout: vertical;
}
#command-pane {
width: 100%;
height: auto;
background: $boost;
padding: 1;
border: round $primary;
}
#command-row {
width: 100%;
height: auto;
}
#pipeline-input {
width: 1fr;
min-height: 3;
padding: 0 1;
background: $surface;
color: $text;
border: round $primary;
}
#pipeline-input:focus {
border: double $primary;
background: $surface;
}
#status-panel {
width: auto;
max-width: 25;
height: 3;
text-style: bold;
content-align: center middle;
padding: 0 1;
border: solid $panel-darken-1;
}
#cmd-suggestions {
width: 100%;
height: auto;
max-height: 8;
margin-top: 1;
background: $surface;
border: round $panel-darken-2;
}
#results-pane {
width: 100%;
height: 2fr;
padding: 0 1 1 1;
background: $panel;
border: round $panel-darken-2;
margin-top: 1;
}
#results-pane .section-title {
margin-top: 0;
margin-bottom: 0;
}
#results-layout {
width: 100%;
height: 1fr;
}
#results-list-pane {
width: 2fr;
height: 1fr;
padding-right: 1;
}
#results-tags-pane {
width: 1fr;
height: 1fr;
padding: 0 1;
border-left: solid $panel-darken-2;
}
#results-meta-pane {
width: 1fr;
height: 1fr;
padding-left: 1;
border-left: solid $panel-darken-2;
}
#store-select {
width: 24;
margin-right: 2;
height: 3;
}
#output-path {
width: 1fr;
height: 3;
}
#bottom-pane {
width: 100%;
height: 1fr;
padding: 1;
background: $panel;
border: round $panel-darken-2;
}
#store-row {
width: 100%;
height: auto;
}
#logs-workers-row {
width: 100%;
height: 1fr;
margin-top: 1;
}
#logs-pane,
#workers-pane {
width: 1fr;
height: 100%;
padding: 0 1;
}
.section-title {
text-style: bold;
color: $text-muted;
margin-top: 1;
}
#log-output {
height: 1fr;
}
#workers-table {
height: 1fr;
}
#results-table {
height: 1fr;
border: solid #ffffff;
background: #ffffff;
color: #000000;
padding: 0;
}
#results-table > .datatable--header {
background: #ffffff;
color: #000000;
text-style: bold;
}
#inline-tags-output {
height: 1fr;
border: solid #ffffff;
background: #ffffff;
color: #000000;
}
#metadata-tree {
height: 1fr;
border: solid #ffffff;
background: #ffffff;
color: #000000;
}
.status-info {
background: $boost;
color: $text;
}
.status-success {
background: $success 20%;
color: $success;
}
.status-error {
background: $error 20%;
color: $error;
}
#run-button {
width: auto;
min-width: 10;
margin: 0 1;
}
#tags-button,
#actions-button,
#metadata-button,
#relationships-button {
width: auto;
min-width: 12;
margin: 0 1;
}
#popup-title {
width: 100%;
height: 3;
text-style: bold;
content-align: center middle;
border: round $panel-darken-2;
background: $boost;
}
#popup-text,
#tags-editor {
height: 1fr;
border: round $panel-darken-2;
}
#tags-buttons {
width: 100%;
height: auto;
margin-top: 1;
}
#actions-list {
width: 100%;
height: auto;
margin-top: 1;
}
#actions-list Button {
width: 100%;
margin-bottom: 1;
}
#actions-footer {
width: 100%;
height: auto;
margin-top: 1;
}
#tags-status {
width: 1fr;
height: 3;
content-align: left middle;
}
cmdlet/file/search.py
+1 -1
View File
@@ -207,7 +207,7 @@ class search_file(Cmdlet):
"search-file -query 'site:example.com filetype:epub history' # Web: site-scoped search",
"",
"Plugin search (-plugin):",
"search-file -plugin youtube 'tutorial' # Search YouTube plugin",
"search-file -plugin ytdlp -query 'search:tutorial' # Search YouTube via yt-dlp",
"search-file -plugin ftp -instance work '*' # Search a named FTP/SCP plugin instance",
"search-file -plugin alldebrid '*' # List AllDebrid magnets",
"search-file -plugin alldebrid -open 123 '*' # Show files for a magnet",
cmdnat/_status_shared.py
+120 -37
View File
@@ -16,7 +16,7 @@ def add_startup_check(
name: str,
*,
provider: str = "",
store: str = "",
instance: str = "",
files: int | str | None = None,
detail: str = "",
) -> None:
@@ -24,11 +24,82 @@ def add_startup_check(
row.add_column("STATUS", upper_text(status))
row.add_column("NAME", upper_text(name))
row.add_column("PLUGIN", upper_text(provider or ""))
row.add_column("STORE", upper_text(store or ""))
row.add_column("INSTANCE", upper_text(instance or ""))
row.add_column("FILES", "" if files is None else str(files))
row.add_column("DETAIL", upper_text(detail or ""))
def _provider_config_map(config: dict) -> dict[str, Any]:
if not isinstance(config, dict):
return {}
provider_cfg = config.get("plugin")
if not isinstance(provider_cfg, dict):
provider_cfg = config.get("provider")
return provider_cfg if isinstance(provider_cfg, dict) else {}
def _iter_registered_plugin_infos() -> tuple[Any, ...]:
try:
from ProviderCore.registry import REGISTRY
return tuple(
sorted(
REGISTRY.iter_plugins(),
key=lambda info: str(
getattr(info, "canonical_name", "") or ""
).lower(),
)
)
except Exception:
return ()
def _extract_configured_instance_names(raw_entry: Any) -> list[str]:
if not isinstance(raw_entry, dict) or not raw_entry:
return []
if not all(isinstance(value, dict) for value in raw_entry.values()):
return []
names: list[str] = []
for key in raw_entry.keys():
name = str(key or "").strip()
if not name or name.lower() == "default":
continue
names.append(name)
return names
def _resolve_startup_instance_text(
plugin: Any,
summary: dict[str, Any],
configured_entry: Any,
) -> str:
instance_text = str(summary.get("instance") or "").strip()
if instance_text:
return instance_text
raw_instances = summary.get("instances")
if isinstance(raw_instances, (list, tuple, set)):
values = [str(value).strip() for value in raw_instances if str(value).strip()]
if values:
return ", ".join(values)
elif raw_instances is not None:
instance_text = str(raw_instances).strip()
if instance_text:
return instance_text
try:
configured_instances = plugin.configured_instances() if plugin is not None else []
except Exception:
configured_instances = []
if configured_instances:
return ", ".join(str(value).strip() for value in configured_instances if str(value).strip())
return ", ".join(_extract_configured_instance_names(configured_entry))
def has_store_subtype(cfg: dict, subtype: str) -> bool:
store_cfg = cfg.get("store")
if not isinstance(store_cfg, dict):
@@ -113,61 +184,73 @@ def ping_first(urls: list[str]) -> tuple[bool, str]:
def collect_plugin_startup_checks(config: dict) -> list[dict[str, Any]]:
provider_cfg = None
if isinstance(config, dict):
provider_cfg = config.get("plugin")
if not isinstance(provider_cfg, dict):
provider_cfg = config.get("provider")
if not isinstance(provider_cfg, dict) or not provider_cfg:
return []
try:
from ProviderCore.registry import get_plugin_class
except Exception:
return []
provider_cfg = _provider_config_map(config)
checks: list[dict[str, Any]] = []
for plugin_name in provider_cfg.keys():
plugin_key = str(plugin_name or "").strip().lower()
seen_plugin_keys: set[str] = set()
for info in _iter_registered_plugin_infos():
plugin_key = str(getattr(info, "canonical_name", "") or "").strip().lower()
if not plugin_key:
continue
seen_plugin_keys.add(plugin_key)
plugin_class = None
try:
plugin_class = get_plugin_class(plugin_key)
except Exception:
plugin_class = None
if plugin_class is None:
checks.append(
{
"status": "UNKNOWN",
"name": provider_display_name(plugin_key),
"plugin": plugin_key,
"detail": "Not registered",
}
)
continue
plugin = None
summary: dict[str, Any]
display_name = provider_display_name(plugin_key)
configured_entry: Any = None
try:
plugin = plugin_class(config)
plugin = info.plugin_class(config)
configured_entry = plugin.plugin_config_root()
summary = plugin.status_summary()
except Exception as exc:
summary = {
"status": "DISABLED",
"name": provider_display_name(plugin_key),
"name": display_name,
"plugin": plugin_key,
"detail": str(exc),
}
status = str(summary.get("status") or "UNKNOWN").strip().upper() or "UNKNOWN"
name = str(summary.get("name") or display_name)
detail = str(summary.get("detail") or "").strip()
if detail.lower() == "configured" and not configured_entry:
detail = "Available"
if not detail:
if status == "ENABLED":
detail = "Configured" if configured_entry else "Available"
else:
detail = "Not configured" if not configured_entry else "Unavailable"
checks.append(
{
"status": str(summary.get("status") or "UNKNOWN"),
"name": str(summary.get("name") or provider_display_name(plugin_key)),
"status": status,
"name": name,
"plugin": str(summary.get("plugin") or plugin_key),
"detail": str(summary.get("detail") or ""),
"instance": _resolve_startup_instance_text(
plugin,
summary,
configured_entry if configured_entry else provider_cfg.get(plugin_key),
),
"detail": detail,
"files": summary.get("files"),
}
)
for plugin_name, raw_entry in sorted(provider_cfg.items()):
plugin_key = str(plugin_name or "").strip().lower()
if not plugin_key or plugin_key in seen_plugin_keys:
continue
checks.append(
{
"status": "UNKNOWN",
"name": provider_display_name(plugin_key),
"plugin": plugin_key,
"instance": ", ".join(_extract_configured_instance_names(raw_entry)),
"detail": "Not registered",
"files": None,
}
)
return checks
cmdnat/config.py
+4 -40
View File
@@ -291,47 +291,11 @@ def _run(piped_result: Any, args: List[str], config: Dict[str, Any]) -> int:
return 1
if not args:
# Check if we're in an interactive terminal and can launch a Textual modal
if sys.stdin.isatty() and not piped_result:
try:
from textual.app import App
from TUI.modalscreen.config_modal import ConfigModal
class ConfigApp(App):
def on_mount(self) -> None:
self.title = "Config Editor"
# We push the modal screen. It will sit on top of the main (blank) screen.
# Using a callback to exit the app when the modal is dismissed.
self.push_screen(ConfigModal(), callback=self.exit_on_close)
def exit_on_close(self, result: Any = None) -> None:
self.exit()
with ctx.suspend_live_progress():
app = ConfigApp()
app.run()
# After modal exits, show the new status table if possible
try:
from cmdlet._shared import SharedArgs
from cmdnat.status import CMDLET as STATUS_CMDLET
# We reload the config one more time because it might have changed on disk
fresh_config = load_config()
# Force refresh of shared caches (especially stores)
SharedArgs._refresh_store_choices_cache(fresh_config)
# Update the global SharedArgs choices so cmdlets pick up new stores
SharedArgs.STORE.choices = SharedArgs.get_store_choices(fresh_config, force=True)
return STATUS_CMDLET.exec(None, [], fresh_config)
except Exception:
pass
return 0
except Exception as exc:
# Fall back to table display if Textual modal fails
print(f"Note: Could not launch interactive editor ({exc}). Showing configuration table:")
return _show_config_table(current_config)
print(
"Interactive TUI config editor has been discontinued. "
"Showing configuration table instead."
)
return _show_config_table(current_config)
key = args[0]
cmdnat/status.py
+3 -57
View File
@@ -4,14 +4,12 @@ import shutil
from typing import Any, Dict, List
from SYS.cmdlet_spec import Cmdlet
from SYS.config import resolve_cookies_path
from SYS import pipeline as ctx
from SYS.result_table import Table
from SYS.logger import set_debug, debug
from SYS.logger import set_debug
from cmdnat._status_shared import (
add_startup_check as _add_startup_check,
collect_plugin_startup_checks as _collect_plugin_startup_checks,
has_store_subtype as _has_store_subtype,
)
CMDLET = Cmdlet(
@@ -35,7 +33,6 @@ def _run(result: Any, args: List[str], config: Dict[str, Any]) -> int:
set_debug(debug_enabled)
except Exception:
pass
debug(f"Status check: debug_enabled={debug_enabled}")
_add_startup_check(startup_table, "ENABLED" if debug_enabled else "DISABLED", "DEBUGGING")
try:
@@ -45,51 +42,8 @@ def _run(result: Any, args: List[str], config: Dict[str, Any]) -> int:
MPV()
mpv_path = shutil.which("mpv")
_add_startup_check(startup_table, "ENABLED", "MPV", detail=mpv_path or "Available")
debug(f"MPV check OK: path={mpv_path or 'Available'}")
except Exception as exc:
_add_startup_check(startup_table, "DISABLED", "MPV", detail=str(exc))
debug(f"MPV check failed: {exc}")
# Store Registry
store_registry = None
try:
from Store import Store as StoreRegistry
store_registry = StoreRegistry(config=config, suppress_debug=True)
try:
backends = store_registry.list_backends()
except Exception:
backends = []
debug(f"StoreRegistry initialized. backends={backends}")
except Exception as exc:
debug(f"StoreRegistry initialization failed: {exc}")
store_registry = None
# Hydrus
if _has_store_subtype(config, "hydrusnetwork"):
hcfg = config.get("store", {}).get("hydrusnetwork", {})
for iname, icfg in hcfg.items():
if not isinstance(icfg, dict): continue
nkey = str(icfg.get("NAME") or iname)
uval = str(icfg.get("URL") or "").strip()
debug(f"Hydrus network check: name={nkey}, url={uval}")
ok = bool(store_registry and store_registry.is_available(nkey))
status = "ENABLED" if ok else "DISABLED"
files = None
detail = uval
if ok and store_registry:
try:
backend = store_registry[nkey]
files = getattr(backend, "total_count", None)
if files is None and hasattr(backend, "get_total_count"):
files = backend.get_total_count()
debug(f"Hydrus backend '{nkey}' available: files={files}")
except Exception as exc:
debug(f"Hydrus backend '{nkey}' check failed: {exc}")
else:
err = store_registry.get_backend_error(iname) if store_registry else None
debug(f"Hydrus backend '{nkey}' not available: {err}")
detail = f"{uval} - {err or 'Unavailable'}"
_add_startup_check(startup_table, status, nkey, store="hydrusnetwork", files=files, detail=detail)
for check in _collect_plugin_startup_checks(config):
_add_startup_check(
@@ -97,27 +51,19 @@ def _run(result: Any, args: List[str], config: Dict[str, Any]) -> int:
str(check.get("status") or "UNKNOWN"),
str(check.get("name") or "Plugin"),
provider=str(check.get("plugin") or ""),
instance=str(check.get("instance") or ""),
files=check.get("files"),
detail=str(check.get("detail") or ""),
)
# Cookies
try:
cf = resolve_cookies_path(config)
_add_startup_check(startup_table, "FOUND" if cf else "MISSING", "Cookies", detail=str(cf) if cf else "Not found")
debug(f"Cookies: resolved cookiefile={cf}")
except Exception as exc:
debug(f"Cookies check failed: {exc}")
except Exception as exc:
debug(f"Status check failed: {exc}")
_add_startup_check(startup_table, "ERROR", "STATUS", detail=str(exc))
if startup_table.rows:
# Mark as rendered to prevent CLI.py from auto-printing it to stdout
# (avoiding duplication in TUI logs, while keeping it in TUI Results)
setattr(startup_table, "_rendered_by_cmdlet", True)
ctx.set_current_stage_table(startup_table)
debug(f"Status check completed: {len(startup_table.rows)} checks recorded")
return 0
plugins/mpv/LUA/main.lua
+27
View File
@@ -4608,6 +4608,25 @@ function M._raw_format_selection_id(fmt)
return display_id
end
function M._raw_info_has_audio_variant_selectors(raw)
if type(raw) ~= 'table' or type(raw.formats) ~= 'table' then
return false
end
for _, fmt in ipairs(raw.formats) do
if type(fmt) == 'table' then
local format_id = trim(tostring(fmt.format_id or ''))
local vcodec = tostring(fmt.vcodec or 'none')
local acodec = tostring(fmt.acodec or 'none')
if format_id:match('^%d+%-%w+$') and vcodec == 'none' and acodec ~= 'none' then
return true
end
end
end
return false
end
function M._raw_format_picker_score(fmt)
local note = trim(tostring(fmt and (fmt.format_note or fmt.format) or '')):lower()
local format_id = trim(tostring(fmt and fmt.format_id or '')):lower()
@@ -4715,6 +4734,14 @@ local function _cache_formats_from_raw_info(url, raw, source_label)
return nil, 'missing url'
end
if raw == nil then
raw = mp.get_property_native('ytdl-raw-info')
end
if M._raw_info_has_audio_variant_selectors(raw) then
return nil, 'raw info requires validated probe'
end
local tbl, err = _build_formats_table_from_raw_info(url, raw)
if type(tbl) ~= 'table' or type(tbl.rows) ~= 'table' then
return nil, err or 'raw format conversion failed'
plugins/mpv/mpv_ipc.py
+1 -1
View File
@@ -477,7 +477,7 @@ class MPV:
pipeline += f" | file -add -plugin local -instance {_q(path or '')}"
try:
from TUI.pipeline_runner import PipelineRunner # noqa: WPS433
from SYS.pipeline_runner import PipelineRunner # noqa: WPS433
runner = PipelineRunner()
result = runner.run_pipeline(pipeline)
plugins/mpv/pipeline_helper.py
+1 -1
View File
@@ -305,7 +305,7 @@ def _run_pipeline(
json_output: bool = False,
) -> Dict[str, Any]:
# Import after sys.path fix.
from TUI.pipeline_runner import PipelineRunner # noqa: WPS433
from SYS.pipeline_runner import PipelineRunner # noqa: WPS433
def _json_safe(value: Any) -> Any:
if value is None or isinstance(value, (str, int, float, bool)):
plugins/youtube/__init__.py
-218
View File
@@ -1,218 +0,0 @@
from __future__ import annotations
import sys
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(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 -plugin 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"],
}
# If the user provides extra args on the selection stage, forward them to download-file.
AUTO_STAGE_USE_SELECTION_ARGS = True
@property
def preserve_order(self) -> bool:
return True
def search(
self,
query: str,
limit: int = 10,
filters: Optional[Dict[str,
Any]] = None,
**kwargs: Any,
) -> List[SearchResult]:
# Use the yt_dlp Python module (installed via requirements.txt).
try:
import yt_dlp # type: ignore
ydl_opts: Dict[str,
Any] = {
"quiet": True,
"skip_download": True,
"extract_flat": True
}
with yt_dlp.YoutubeDL(ydl_opts) as ydl: # type: ignore[arg-type]
search_query = f"ytsearch{limit}:{query}"
info = ydl.extract_info(search_query, download=False)
entries = info.get("entries") or []
results: List[SearchResult] = []
for video_data in entries[:limit]:
title = video_data.get("title", "Unknown")
video_id = video_data.get("id", "")
url = video_data.get(
"url"
) or f"https://youtube.com/watch?v={video_id}"
uploader = video_data.get("uploader", "Unknown")
duration = video_data.get("duration", 0)
view_count = video_data.get("view_count", 0)
duration_str = (
f"{int(duration // 60)}:{int(duration % 60):02d}"
if duration else ""
)
views_str = f"{view_count:,}" if view_count else ""
results.append(
SearchResult(
table="youtube",
title=title,
path=url,
detail=f"By: {uploader}",
annotations=[duration_str,
f"{views_str} views"],
media_kind="video",
columns=[
("Title",
title),
("Uploader",
uploader),
("Duration",
duration_str),
("Views",
views_str),
],
full_metadata={
"video_id": video_id,
"uploader": uploader,
"duration": duration,
"view_count": view_count,
# Selection metadata for table system and @N expansion
"_selection_args": ["-url", url],
},
)
)
return results
except Exception:
log("[youtube] yt_dlp import failed", file=sys.stderr)
return []
def validate(self) -> bool:
try:
return True
except Exception:
return False
# Minimal provider registration for the new table system
try:
from SYS.result_table_adapters import register_plugin
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_plugin(
"youtube",
_adapter,
columns=_columns_factory,
selection_fn=_selection_fn,
metadata={"description": "YouTube video search using yt_dlp"},
)
except Exception:
# best-effort registration
pass
plugins/ytdlp/__init__.py
+53 -5
View File
@@ -13,7 +13,7 @@ from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple
from urllib.parse import urlparse
from ProviderCore.base import Provider, SearchResult
from ProviderCore.base import Provider, SearchResult, parse_inline_query_arguments
from SYS.provider_helpers import TableProviderMixin
from SYS.logger import debug, log
from SYS.models import DownloadError, DownloadMediaResult, DownloadOptions
@@ -32,6 +32,7 @@ from tool.ytdlp import (
_read_text_file,
collapse_picker_formats,
format_for_table_selection,
get_display_format_id,
get_selection_format_id,
is_browseable_format,
is_url_supported_by_ytdlp,
@@ -357,6 +358,12 @@ def _format_id_for_query_index(
return normalized or s_val
candidate_formats = collapse_picker_formats(fmts, video_audio_suffix="bestaudio")
if s_val and not s_val.startswith("#"):
for item in candidate_formats:
if get_display_format_id(item) == s_val:
normalized = get_selection_format_id(item, video_audio_suffix="bestaudio")
return normalized or s_val
filtered_formats = candidate_formats if candidate_formats else list(fmts)
if idx <= 0 or idx > len(filtered_formats):
raise ValueError(f"Format index {idx} out of range")
@@ -497,6 +504,10 @@ def _build_pipe_objects(
class ytdlp(TableProviderMixin, Provider):
"""yt-dlp-backed search and direct download plugin."""
PLUGIN_NAME = "ytdlp"
PLUGIN_ALIASES = ("youtube",)
SEARCH_QUERY_KEYS = ("search", "q")
@classmethod
def url_patterns(cls) -> Tuple[str, ...]:
try:
@@ -529,10 +540,47 @@ class ytdlp(TableProviderMixin, Provider):
TABLE_AUTO_STAGES = {
"ytdlp.formatlist": ["download-file"],
"ytdlp.search": ["download-file"],
"youtube": ["download-file"],
}
AUTO_STAGE_USE_SELECTION_ARGS = True
def extract_query_arguments(self, query: str) -> Tuple[str, Dict[str, Any]]:
normalized_query, inline_args = parse_inline_query_arguments(query)
search_parts: List[str] = []
for key in self.SEARCH_QUERY_KEYS:
value = str(inline_args.pop(key, "") or "").strip()
if value:
search_parts.append(value)
if normalized_query:
search_parts.append(normalized_query)
resolved_query = " ".join(part for part in search_parts if part).strip()
if not resolved_query:
resolved_query = str(query or "").strip()
filters: Dict[str, Any] = dict(inline_args or {})
filters.setdefault("search_provider", "youtube")
return resolved_query, filters
def get_table_type(
self,
query: str,
filters: Optional[Dict[str, Any]] = None,
) -> str:
_ = query, filters
return "youtube"
def get_table_title(
self,
query: str,
filters: Optional[Dict[str, Any]] = None,
) -> str:
_ = filters
q = str(query or "").strip() or "*"
return f"YouTube: {q}"
def search(
self,
query: str,
@@ -567,7 +615,7 @@ class ytdlp(TableProviderMixin, Provider):
results.append(
SearchResult(
table="ytdlp.search",
table="youtube",
title=title,
path=url,
detail=f"By: {uploader}",
@@ -1443,11 +1491,11 @@ try:
return ["-title", row.title or ""]
_register_table_plugin_once(
"ytdlp.search",
"youtube",
_search_adapter,
columns=_search_columns_factory,
selection_fn=_search_selection_fn,
metadata={"description": "ytdlp video search using yt-dlp"},
metadata={"description": "YouTube search using yt-dlp"},
)
except Exception as exc:
debug(f"[ytdlp] Provider registration note: {exc}")
scripts/cli_entry.py
+7 -66
View File
@@ -222,72 +222,13 @@ def _run_cli(clean_args: List[str]) -> int:
def _run_gui(clean_args: List[str]) -> int:
"""Run the TUI runner (PipelineHubApp).
The TUI is imported lazily; if Textual or the TUI code is unavailable we
give a helpful error message and exit non‑zero.
"""
try:
repo_root = _ensure_repo_root_on_sys_path()
tui_mod: Optional[ModuleType] = None
selected_module: Optional[str] = None
last_exc: Optional[Exception] = None
def _load_from_file(path: Path) -> ModuleType:
spec = importlib.util.spec_from_file_location(
"medeia_tui_entry",
path,
submodule_search_locations=[str(path.parent)],
)
if spec is None or spec.loader is None:
raise ModuleNotFoundError(f"Cannot load TUI from {path}")
module = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = module
spec.loader.exec_module(module) # type: ignore[attr-defined]
return module
if repo_root is not None:
tui_file = repo_root / "TUI.py"
if tui_file.exists():
try:
tui_mod = _load_from_file(tui_file)
selected_module = str(tui_file)
except Exception as exc:
last_exc = exc
if tui_mod is None:
for candidate in ("TUI.tui", "TUI"):
try:
tui_mod = importlib.import_module(candidate)
selected_module = candidate
break
except ModuleNotFoundError as exc:
last_exc = exc
if tui_mod is None:
raise last_exc or ModuleNotFoundError("No TUI module could be imported")
except Exception as exc:
print(
"Error: Unable to import TUI (Textual may not be installed):",
exc,
file=sys.stderr,
)
return 2
try:
PipelineHubApp = getattr(tui_mod, "PipelineHubApp")
except AttributeError:
module_hint = selected_module or "TUI"
print(
f"Error: '{module_hint}' does not expose 'PipelineHubApp'",
file=sys.stderr,
)
return 2
try:
app = PipelineHubApp()
app.run()
return 0
except SystemExit as exc:
return int(getattr(exc, "code", 0) or 0)
"""Report that the discontinued GUI/TUI mode is no longer available."""
_ = clean_args
print(
"Error: GUI/TUI mode has been discontinued and is no longer available.",
file=sys.stderr,
)
return 2
def main(argv: Optional[List[str]] = None) -> int:
tool/ytdlp.py
+39
View File
@@ -386,6 +386,25 @@ def is_url_supported_by_ytdlp(url: str) -> bool:
_FORMATS_CACHE: Dict[str, tuple[float, List[Dict[str, Any]]]] = {}
def _audio_variant_base_selector(fmt: Any) -> Optional[str]:
if not isinstance(fmt, dict):
return None
format_id = str(fmt.get("format_id") or "").strip()
if not format_id:
return None
vcodec = str(fmt.get("vcodec", "none"))
acodec = str(fmt.get("acodec", "none"))
if vcodec != "none" or acodec == "none":
return None
match = re.fullmatch(r"(?P<base>\d+)-[A-Za-z0-9]+", format_id)
if not match:
return None
return match.group("base")
def list_formats(
url: str,
*,
@@ -449,9 +468,26 @@ def list_formats(
result_container[0] = None
return
selector_cache: Dict[str, bool] = {}
def _selector_has_matches(selector: str) -> bool:
cached = selector_cache.get(selector)
if cached is not None:
return cached
try:
matches = ydl.build_format_selector(selector)(info)
cached = any(True for _ in matches)
except Exception:
cached = False
selector_cache[selector] = cached
return cached
out: List[Dict[str, Any]] = []
for fmt in formats:
if isinstance(fmt, dict):
base_selector = _audio_variant_base_selector(fmt)
if base_selector and not _selector_has_matches(base_selector):
fmt["_medios_selector_valid"] = False
out.append(fmt)
result_container[0] = out
except Exception as exc:
@@ -596,6 +632,9 @@ def is_browseable_format(fmt: Any) -> bool:
"""
if not isinstance(fmt, dict):
return False
if fmt.get("_medios_selector_valid") is False:
return False
format_id = str(fmt.get("format_id") or "").strip()
if not format_id: