Medios-Macina/cmdnat/config.py

import datetime
import sqlite3
from pathlib import Path
from typing import List, Dict, Any, Optional, Sequence

from SYS.cmdlet_spec import Cmdlet, CmdletArg
from SYS.config import (
    load_config,
    save_config,
    save_config_and_verify,
    set_nested_config_value,
)
from SYS.database import LOG_DB_PATH, db
from SYS.logger import log
from SYS.plugin_config import (
    build_default_plugin_config,
    get_configurable_plugin_types,
    get_configurable_store_types,
)
from SYS import pipeline as ctx
from SYS.result_table import Table
from cmdnat._parsing import (
    VALUE_ARG_FLAGS,
    extract_piped_value as _extract_piped_value,
    extract_arg_value as _extract_arg_value,
    extract_value_arg as _extract_value_arg,
    has_flag as _has_flag,
)


_PREFERENCES_BROWSE_PATH = "__preferences__"
_PLUGINS_BROWSE_PATH = "__plugins__"
_PLUGIN_CATEGORY_KEYS = ("plugin",)
_CREATE_INSTANCE_FLAG = "-create-instance"
_CHOOSE_FLAG = "-choose"
_KNOWN_SECTION_LABELS = {
    "plugin": "Plugins",
}
_KNOWN_SECTION_DESCRIPTIONS = {
    _PREFERENCES_BROWSE_PATH: "Global preferences and simple values",
    _PLUGINS_BROWSE_PATH: "All configured plugins and plugin instances",
    "plugin": "Plugin configuration",
}
_SENSITIVE_CONFIG_KEYS = {
    "access_key",
    "access_token",
    "api",
    "api_key",
    "apikey",
    "authorization",
    "bearer_token",
    "cookie",
    "cookies",
    "password",
    "secret",
    "token",
}
_CONFIG_ITEM_FIELDS = (
    "kind",
    "key",
    "title",
    "browse_path",
    "name",
    "value",
    "value_display",
    "type",
    "display_path",
    "instance_target",
    "choices",
)

CMDLET = Cmdlet(
    name=".config",
    summary="Manage configuration settings",
    usage=".config [key] [value] | .config -log [count]",
    arg=[
        CmdletArg(
            name="key",
            description="Configuration key to update (dot-separated)",
            required=False
        ),
        CmdletArg(
            name="value",
            description="New value for the configuration key",
            required=False
        ),
    ],
)


def _extract_log_limit(args: Sequence[str], default: int = 30) -> int:
    try:
        tokens = [str(arg).strip() for arg in (args or []) if str(arg).strip()]
    except Exception:
        return default

    for idx, token in enumerate(tokens):
        lowered = token.lower()
        if lowered in {"-log", "--log"}:
            if idx + 1 < len(tokens):
                candidate = tokens[idx + 1]
                if candidate and not candidate.startswith("-"):
                    try:
                        return max(1, min(200, int(candidate)))
                    except Exception:
                        return default
            return default
        if lowered.startswith("-log=") or lowered.startswith("--log="):
            _, value = lowered.split("=", 1)
            try:
                return max(1, min(200, int(value)))
            except Exception:
                return default
    return default


def _fallback_log_path() -> Path:
    return Path(db.db_path).with_name("logs") / "log_fallback.txt"


def _load_recent_config_logs(limit: int = 30) -> List[Dict[str, str]]:
    rows: List[Dict[str, str]] = []
    sql = """
        SELECT timestamp, level, module, message
        FROM logs
        WHERE lower(module) LIKE ?
           OR lower(message) LIKE ?
           OR lower(message) LIKE ?
           OR lower(message) LIKE ?
        ORDER BY id DESC
        LIMIT ?
    """
    params = (
        "%config%",
        "%config%",
        "%save failed%",
        "%saving configuration failed%",
        int(limit),
    )

    try:
        with sqlite3.connect(str(LOG_DB_PATH), timeout=5.0) as conn:
            conn.row_factory = sqlite3.Row
            cur = conn.cursor()
            cur.execute(sql, params)
            fetched = cur.fetchall()
            cur.close()
        for row in fetched:
            rows.append(
                {
                    "timestamp": str(row["timestamp"] or ""),
                    "level": str(row["level"] or ""),
                    "module": str(row["module"] or ""),
                    "message": str(row["message"] or ""),
                }
            )
    except Exception:
        rows = []

    if rows:
        return rows

    fallback = _fallback_log_path()
    try:
        if not fallback.exists():
            return []
        lines = fallback.read_text(encoding="utf-8", errors="replace").splitlines()
        matches = [
            line for line in lines
            if any(term in line.lower() for term in ("config", "save failed", "saving configuration failed"))
        ]
        for line in reversed(matches[-limit:]):
            rows.append(
                {
                    "timestamp": "",
                    "level": "FALLBACK",
                    "module": "fallback",
                    "message": line,
                }
            )
    except Exception:
        return []
    return rows


def _format_log_timestamp_local(raw_value: str) -> str:
    text = str(raw_value or "").strip()
    if not text:
        return ""
    for pattern in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M:%S.%f"):
        try:
            parsed = datetime.datetime.strptime(text, pattern).replace(tzinfo=datetime.timezone.utc)
            return parsed.astimezone().strftime("%Y-%m-%d %H:%M:%S")
        except Exception:
            continue
    return text


def _show_config_logs(args: Sequence[str]) -> int:
    limit = _extract_log_limit(args)
    rows = _load_recent_config_logs(limit=limit)
    if not rows:
        print(
            f"No recent config/save logs found in {LOG_DB_PATH.name} or {_fallback_log_path().name}."
        )
        return 0

    table = Table("Configuration Logs")
    table.set_table("config.logs")
    table.set_source_command(".config", ["-log", str(limit)])

    for row_data in rows:
        row = table.add_row()
        row.add_column("Time (local)", _format_log_timestamp_local(row_data.get("timestamp", "")))
        row.add_column("Level", row_data.get("level", ""))
        row.add_column("Module", row_data.get("module", ""))
        row.add_column("Message", row_data.get("message", ""))

    ctx.set_last_result_table_overlay(table, rows)
    ctx.set_current_stage_table(table)
    print(f"Showing {len(rows)} recent configuration log entries.")
    return 0


def set_nested_config(config: Dict[str, Any], key: str, value: str) -> bool:
    return set_nested_config_value(config, key, value, on_error=print)


def _visible_config_entries(config_data: Any) -> List[tuple[str, Any]]:
    if not isinstance(config_data, dict):
        return []
    return [
        (str(key), value)
        for key, value in config_data.items()
        if isinstance(key, str) and not key.startswith("_")
    ]


def _format_config_label(value: Any) -> str:
    text = str(value or "").strip()
    if not text:
        return "Configuration"
    if text == _PREFERENCES_BROWSE_PATH:
        return "Preferences"
    if text == _PLUGINS_BROWSE_PATH:
        return "Plugins"
    return text.replace("_", " ").replace("-", " ").strip().title()


def _format_config_path_label(browse_path: Optional[str]) -> str:
    text = str(browse_path or "").strip()
    if not text:
        return "Root"
    if text == _PREFERENCES_BROWSE_PATH:
        return "Preferences"
    if text == _PLUGINS_BROWSE_PATH:
        return "Plugins"
    parts = [part for part in text.split(".") if part]
    formatted: List[str] = []
    for idx, part in enumerate(parts):
        if idx == 0 and part in _PLUGIN_CATEGORY_KEYS:
            formatted.append("Plugins")
        else:
            formatted.append(_format_config_label(part))
    return " / ".join(formatted)


def _format_config_value(value: Any) -> str:
    if isinstance(value, bool):
        return "true" if value else "false"
    if value is None:
        return "null"
    if isinstance(value, (list, tuple, set)):
        return ", ".join(str(item) for item in value)
    return str(value)


def _is_sensitive_config_key(key_path: str) -> bool:
    leaf = str(key_path or "").split(".")[-1].strip().lower()
    return leaf in _SENSITIVE_CONFIG_KEYS


def _format_config_entry_count(value: Any) -> str:
    count = len(_visible_config_entries(value)) if isinstance(value, dict) else 0
    if count == 1:
        return "1 entry"
    return f"{count} entries"


def _get_configurable_plugin_names() -> List[str]:
    try:
        return [
            str(name).strip().lower()
            for name in (get_configurable_plugin_types() or [])
            if str(name).strip()
        ]
    except Exception:
        return []


def _get_configurable_tool_names() -> List[str]:
    """Retain compatibility with the retired tool configuration category."""
    return []


def _get_multi_instance_plugin_names() -> set[str]:
    try:
        return {
            str(name).strip().lower()
            for name in (get_configurable_store_types() or [])
            if str(name).strip()
        }
    except Exception:
        return set()


def _split_config_path(value: Optional[str]) -> List[str]:
    return [part for part in str(value or "").split(".") if part]


def _is_multi_instance_plugin_name(name: str) -> bool:
    return str(name or "").strip().lower() in _get_multi_instance_plugin_names()


def _is_multi_instance_plugin_root_path(browse_path: Optional[str]) -> bool:
    parts = _split_config_path(browse_path)
    return (
        len(parts) == 2
        and parts[0] == "plugin"
        and _is_multi_instance_plugin_name(parts[1])
    )


def _plugin_schema_field_keys(plugin_name: str) -> set[str]:
    defaults = build_default_plugin_config(plugin_name)
    if not isinstance(defaults, dict):
        return set()
    return {
        str(key or "").strip().lower()
        for key in defaults.keys()
        if str(key or "").strip()
    }


def _looks_like_single_instance_branch(plugin_name: str, branch: Any) -> bool:
    if not isinstance(branch, dict) or not branch:
        return False

    schema_keys = _plugin_schema_field_keys(plugin_name)
    entry_keys = {str(key or "").strip().lower() for key in branch.keys()}
    looks_like_single = bool(schema_keys and entry_keys.intersection(schema_keys))
    if not looks_like_single:
        looks_like_single = not all(isinstance(value, dict) for value in branch.values())
    return looks_like_single


def _normalize_multi_instance_branch(plugin_name: str, branch: Any) -> Dict[str, Any]:
    if not isinstance(branch, dict):
        return {}
    if _looks_like_single_instance_branch(plugin_name, branch):
        return {"default": dict(branch)}
    return {
        str(key): value
        for key, value in _visible_config_entries(branch)
        if isinstance(value, dict)
    }


def _build_create_instance_item(category: str, plugin_name: str) -> Dict[str, Any]:
    target = f"{category}.{plugin_name}"
    return {
        "kind": "create_instance",
        "key": f"{target}.__new_instance__",
        "title": "Add Instance",
        "name": "add_instance",
        "value": None,
        "value_display": "Create with @N | .config <name>",
        "display_path": f"{_format_config_path_label(target)} / Add Instance",
        "type": "action",
        "instance_target": target,
    }


def _build_synthetic_plugin_branch(category: str, name: str) -> Optional[Dict[str, Any]]:
    normalized_category = str(category or "").strip().lower()
    normalized_name = str(name or "").strip().lower()
    if not normalized_name:
        return None

    branch = build_default_plugin_config(normalized_name)
    if not isinstance(branch, dict):
        return None
    if normalized_name in _get_multi_instance_plugin_names():
        return {"default": dict(branch)}
    return dict(branch)


def _find_configured_plugin_branch(
    config_data: Dict[str, Any],
    category: str,
    name: str,
) -> Optional[tuple[str, Dict[str, Any]]]:
    category_block = config_data.get(category)
    if not isinstance(category_block, dict):
        return None

    target = str(name or "").strip().lower()
    for raw_name, raw_value in _visible_config_entries(category_block):
        if str(raw_name or "").strip().lower() != target or not isinstance(raw_value, dict):
            continue
        return raw_name, raw_value
    return None


def _resolve_plugin_branch(
    config_data: Dict[str, Any],
    category: str,
    name: str,
) -> Optional[tuple[str, Dict[str, Any], bool]]:
    found = _find_configured_plugin_branch(config_data, category, name)
    if found is not None:
        resolved_name, resolved_value = found
        return resolved_name, resolved_value, True

    normalized_category = str(category or "").strip().lower()
    normalized_name = str(name or "").strip().lower()
    if not normalized_name:
        return None

    if normalized_name not in _get_configurable_plugin_names():
        return None

    synthetic = _build_synthetic_plugin_branch(normalized_category, normalized_name)
    if synthetic is None:
        return None
    return normalized_name, synthetic, False


def _iter_plugin_branches(config_data: Dict[str, Any]) -> List[tuple[str, str, Any]]:
    branches: List[tuple[str, str, Any]] = []
    if not isinstance(config_data, dict):
        return branches

    for category in _PLUGIN_CATEGORY_KEYS:
        category_block = config_data.get(category)
        if not isinstance(category_block, dict):
            continue
        for name, value in _visible_config_entries(category_block):
            branches.append((category, name, value))
    return branches


def _iter_available_plugin_branches(config_data: Dict[str, Any]) -> List[tuple[str, str, Any, bool]]:
    branches: List[tuple[str, str, Any, bool]] = []
    seen: set[str] = set()

    for category, name, value in _iter_plugin_branches(config_data):
        normalized_name = str(name or "").strip().lower()
        if not normalized_name:
            continue
        branches.append((category, name, value, True))
        seen.add(normalized_name)

    for name in _get_configurable_plugin_names():
        if name in seen:
            continue
        synthetic = _build_synthetic_plugin_branch("plugin", name)
        if synthetic is None:
            continue
        branches.append(("plugin", name, synthetic, False))
        seen.add(name)

    for name in _get_configurable_tool_names():
        if name in seen:
            continue
        synthetic = _build_synthetic_plugin_branch("tool", name)
        if synthetic is None:
            continue
        branches.append(("tool", name, synthetic, False))
        seen.add(name)

    return branches


def _collect_plugin_root_items(config_data: Dict[str, Any]) -> List[Dict[str, Any]]:
    plugin_items: Dict[str, Dict[str, Any]] = {}
    for category, name, value, is_configured in _iter_available_plugin_branches(config_data):
        key = str(name or "").strip().lower()
        if not key:
            continue
        existing = plugin_items.get(key)
        if existing is None:
            plugin_items[key] = {
                "kind": "section",
                "title": _format_config_label(name),
                "browse_path": f"{category}.{name}",
                "summary": _format_config_entry_count(value),
                "type": "section",
                "description": "Plugin configuration" if is_configured else "Plugin configuration (available to configure)",
            }
            continue

        if str(category) == "plugin" and not str(existing.get("browse_path") or "").startswith("plugin."):
            existing["browse_path"] = f"{category}.{name}"
        try:
            current_count = int(str(existing.get("summary") or "0").split()[0])
        except Exception:
            current_count = 0
        extra_count = len(_visible_config_entries(value)) if isinstance(value, dict) else 0
        merged_count = current_count + extra_count
        existing["summary"] = "1 entry" if merged_count == 1 else f"{merged_count} entries"

    return sorted(plugin_items.values(), key=lambda item: str(item.get("title") or "").lower())


def _resolve_config_branch(
    config_data: Dict[str, Any],
    browse_path: Optional[str],
) -> Optional[Dict[str, Any]]:
    text = str(browse_path or "").strip()
    if not text:
        return config_data if isinstance(config_data, dict) else None

    if text == _PREFERENCES_BROWSE_PATH:
        return {
            key: value
            for key, value in _visible_config_entries(config_data)
            if not isinstance(value, dict)
        }

    if text == _PLUGINS_BROWSE_PATH:
        return {
            str(item.get("title") or ""): item
            for item in _collect_plugin_root_items(config_data)
        }

    parts = [part for part in text.split(".") if part]
    if len(parts) >= 2 and parts[0] in _PLUGIN_CATEGORY_KEYS:
        resolved = _resolve_plugin_branch(config_data, parts[0], parts[1])
        if resolved is None:
            return None
        _, current, _ = resolved
        if parts[0] == "plugin" and _is_multi_instance_plugin_name(parts[1]):
            current = _normalize_multi_instance_branch(parts[1], current)
        for part in parts[2:]:
            if not isinstance(current, dict):
                return None
            current = current.get(part)
        return current if isinstance(current, dict) else None

    current: Any = config_data
    for part in parts:
        if not isinstance(current, dict):
            return None
        current = current.get(part)
    return current if isinstance(current, dict) else None


def _build_section_item(
    *,
    title: str,
    browse_path: str,
    value: Any,
    description: Optional[str] = None,
) -> Dict[str, Any]:
    return {
        "kind": "section",
        "title": title,
        "browse_path": browse_path,
        "summary": _format_config_entry_count(value),
        "type": "section",
        "description": str(description or "").strip() or _KNOWN_SECTION_DESCRIPTIONS.get(browse_path, ""),
    }


def _build_value_item(
    *,
    key_path: str,
    name: str,
    value: Any,
    choices: Any = None,
) -> Dict[str, Any]:
    display_value = "***" if _is_sensitive_config_key(key_path) else _format_config_value(value)
    path_parts = [part for part in str(key_path or "").split(".") if part]
    display_path = " / ".join(
        [_format_config_path_label(".".join(path_parts[:-1]))] if len(path_parts) > 1 else []
        + [_format_config_label(path_parts[-1])] if path_parts else [_format_config_label(name)]
    )
    choice_list = list(choices) if isinstance(choices, (list, tuple)) else []
    return {
        "kind": "value",
        "key": key_path,
        "name": name,
        "title": _format_config_label(name),
        "value": value,
        "value_display": display_value,
        "display_path": display_path,
        "type": type(value).__name__,
        "choices": choice_list,
    }


def _create_or_get_plugin_instance(
    config_data: Dict[str, Any],
    instance_target: str,
    instance_name: str,
) -> tuple[str, bool]:
    parts = _split_config_path(instance_target)
    if len(parts) != 2 or parts[0] != "plugin":
        raise ValueError(f"Unsupported instance target '{instance_target}'")

    category, plugin_name = parts
    raw_instance_name = str(instance_name or "").strip()
    if not raw_instance_name:
        raise ValueError("Instance name is required")
    if raw_instance_name.startswith("_"):
        raise ValueError("Instance names cannot start with '_' characters")

    category_block = config_data.get(category)
    if not isinstance(category_block, dict):
        category_block = {}
        config_data[category] = category_block

    plugin_block = category_block.get(plugin_name)
    if not isinstance(plugin_block, dict):
        plugin_block = {}
        category_block[plugin_name] = plugin_block

    if _looks_like_single_instance_branch(plugin_name, plugin_block):
        existing_default = dict(plugin_block)
        plugin_block.clear()
        plugin_block["default"] = existing_default

    target_key = None
    lowered_target = raw_instance_name.lower()
    for existing_key in plugin_block.keys():
        if str(existing_key or "").strip().lower() == lowered_target:
            target_key = str(existing_key)
            break

    if target_key is not None and isinstance(plugin_block.get(target_key), dict):
        return f"{category}.{plugin_name}.{target_key}", False

    plugin_block[raw_instance_name] = dict(build_default_plugin_config(plugin_name))
    return f"{category}.{plugin_name}.{raw_instance_name}", True


def _resolve_update_key(config_data: Dict[str, Any], selection_key: str) -> str:
    parts = _split_config_path(selection_key)
    if (
        len(parts) >= 4
        and parts[0] == "plugin"
        and parts[2].lower() == "default"
        and _is_multi_instance_plugin_name(parts[1])
    ):
        category_block = config_data.get(parts[0])
        plugin_block = category_block.get(parts[1]) if isinstance(category_block, dict) else None
        if _looks_like_single_instance_branch(parts[1], plugin_block):
            return ".".join([parts[0], parts[1], *parts[3:]])
    return selection_key


def _build_root_config_items(config_data: Dict[str, Any]) -> List[Dict[str, Any]]:
    items: List[Dict[str, Any]] = []
    visible_entries = _visible_config_entries(config_data)

    preferences = {
        key: value
        for key, value in visible_entries
        if not isinstance(value, dict)
    }
    if preferences:
        items.append(
            _build_section_item(
                title="Preferences",
                browse_path=_PREFERENCES_BROWSE_PATH,
                value=preferences,
            )
        )

    plugin_items = _collect_plugin_root_items(config_data)
    if plugin_items:
        items.append(
            _build_section_item(
                title="Plugins",
                browse_path=_PLUGINS_BROWSE_PATH,
                value={item["title"]: item for item in plugin_items},
            )
        )

    other_sections: List[Dict[str, Any]] = []
    for key, value in visible_entries:
        if key in set(_PLUGIN_CATEGORY_KEYS) or not isinstance(value, dict):
            continue
        other_sections.append(
            _build_section_item(
                title=_format_config_label(key),
                browse_path=key,
                value=value,
            )
        )

    other_sections.sort(key=lambda item: str(item.get("title") or "").lower())
    items.extend(other_sections)
    return items


def _lookup_plugin_choices(browse_path: str, key_name: str) -> Optional[List[str]]:
    parts = _split_config_path(browse_path)
    if len(parts) < 2 or parts[0] != "plugin":
        return None
    plugin_name = parts[1]
    try:
        from SYS.plugin_config import get_plugin_schema
        schema = get_plugin_schema(plugin_name)
    except Exception:
        return None
    for field in schema:
        if field.get("key") == key_name:
            choices = field.get("choices")
            if isinstance(choices, list):
                return [str(c) for c in choices]
            break
    return None


def _build_nested_config_items(
    config_data: Dict[str, Any],
    browse_path: str,
) -> List[Dict[str, Any]]:
    if browse_path == _PLUGINS_BROWSE_PATH:
        return _collect_plugin_root_items(config_data)

    branch = _resolve_config_branch(config_data, browse_path)
    if branch is None:
        return []

    section_items: List[Dict[str, Any]] = []
    value_items: List[Dict[str, Any]] = []
    action_items: List[Dict[str, Any]] = []
    is_preferences_view = browse_path == _PREFERENCES_BROWSE_PATH
    parts = _split_config_path(browse_path)
    is_multi_instance_root = _is_multi_instance_plugin_root_path(browse_path)

    if is_multi_instance_root:
        branch = _normalize_multi_instance_branch(parts[1], branch)

    for key, value in _visible_config_entries(branch):
        full_key = key if is_preferences_view else f"{browse_path}.{key}"
        if isinstance(value, dict):
            section_items.append(
                _build_section_item(
                    title=_format_config_label(key),
                    browse_path=full_key,
                    value=value,
                )
            )
        else:
            value_items.append(
                _build_value_item(
                    key_path=full_key,
                    name=key,
                    value=value,
                    choices=_lookup_plugin_choices(browse_path, key),
                )
            )

    section_items.sort(key=lambda item: str(item.get("title") or "").lower())
    value_items.sort(key=lambda item: str(item.get("name") or "").lower())
    if is_multi_instance_root:
        action_items.append(_build_create_instance_item(parts[0], parts[1]))
    return section_items + value_items + action_items


def _build_config_items(
    config_data: Dict[str, Any],
    browse_path: Optional[str] = None,
) -> List[Dict[str, Any]]:
    text = str(browse_path or "").strip()
    if not text:
        return _build_root_config_items(config_data)
    return _build_nested_config_items(config_data, text)


def _build_config_table_title(browse_path: Optional[str]) -> str:
    text = str(browse_path or "").strip()
    if not text:
        return "Configuration"
    return f"Configuration: {_format_config_path_label(text)}"


def _build_config_header_lines(browse_path: Optional[str]) -> List[str]:
    text = str(browse_path or "").strip()
    if not text:
        return [
            "Use @N on a section to drill in. Use @.. to go back.",
        ]
    if _is_multi_instance_plugin_root_path(text):
        return [
            f"Path: {_format_config_path_label(text)}",
            "Use @N on an instance to drill in. Use @N | .config <name> on Add Instance to create a new instance, then update its fields in the table that opens. Use @.. to go back.",
        ]
    parts = _split_config_path(text)
    if (
        len(parts) == 3
        and parts[0] == "plugin"
        and _is_multi_instance_plugin_name(parts[1])
    ):
        return [
            f"Path: {_format_config_path_label(text)}",
            "Use @N | .config <value> to update a setting. After creating an instance, set its path, credentials, or other fields here. Use @.. to go back.",
        ]
    return [
        f"Path: {_format_config_path_label(text)}",
        "Use @N on a section to drill in. Use @N | .config <value> to update a setting. Use @.. to go back.",
    ]


def _extract_create_instance_target(args: Sequence[str]) -> Optional[str]:
    return _extract_arg_value(args, flags={_CREATE_INSTANCE_FLAG, "--create-instance"}, allow_positional=False)


def _extract_browse_arg(args: Sequence[str]) -> Optional[str]:
    return _extract_arg_value(args, flags={"-browse", "--browse"}, allow_positional=False)


def _extract_choose_arg(args: Sequence[str]) -> Optional[str]:
    return _extract_arg_value(args, flags={_CHOOSE_FLAG}, allow_positional=False)


def _resolve_config_value_raw(config: Dict[str, Any], key_path: str) -> Any:
    parts = [p for p in str(key_path or "").split(".") if p]
    current: Any = config
    for part in parts:
        if isinstance(current, dict):
            current = current.get(part)
        else:
            return None
    return current


def _extract_selected_update_value(args: Sequence[str]) -> Optional[str]:
    explicit = _extract_arg_value(args, flags=VALUE_ARG_FLAGS, allow_positional=False)
    if explicit is not None:
        return explicit

    tokens = [str(arg).strip() for arg in (args or []) if str(arg).strip()]
    positional = [token for token in tokens if not token.startswith("-")]
    if len(positional) == 1:
        return positional[0]
    return None


def _get_selected_config_item() -> Optional[Dict[str, Any]]:
    try:
        indices = ctx.get_last_selection() or []
    except Exception:
        indices = []
    try:
        items = ctx.get_last_result_items() or []
    except Exception:
        items = []
    if not indices or not items:
        return None
    idx = indices[0]
    if idx < 0 or idx >= len(items):
        return None
    return _normalize_config_item(items[idx])


def _normalize_config_item(candidate: Any) -> Optional[Dict[str, Any]]:
    if candidate is None:
        return None

    normalized: Dict[str, Any] = {}
    sources: List[Any] = [candidate]

    if isinstance(candidate, dict):
        extra = candidate.get("extra")
        if isinstance(extra, dict):
            sources.append(extra)
    else:
        try:
            extra = getattr(candidate, "extra", None)
        except Exception:
            extra = None
        if isinstance(extra, dict):
            sources.append(extra)

    for source in sources:
        if isinstance(source, dict):
            getter = source.get
            for key in _CONFIG_ITEM_FIELDS:
                if key in normalized:
                    continue
                value = getter(key)
                if value is not None:
                    normalized[key] = value
            continue

        for key in _CONFIG_ITEM_FIELDS:
            if key in normalized:
                continue
            try:
                value = getattr(source, key, None)
            except Exception:
                value = None
            if value is not None:
                normalized[key] = value

    return normalized or None


def _show_choice_table(
    target_key: str,
    choices: List[str],
    current_value: str,
    display_name: str,
) -> int:
    items: List[Dict[str, Any]] = []
    for choice in choices:
        choice_str = str(choice)
        marker = "  ← current" if choice_str == current_value else ""
        items.append({
            "kind": "choice",
            "key": target_key,
            "title": choice_str,
            "value": choice_str,
            "display": f"{choice_str}{marker}",
        })

    table = Table(f"Configuration: {display_name}", preserve_order=True)
    table.set_table("config")
    table.set_source_command(".config", [target_key])
    table.set_header_lines([
        f"Available values for {display_name}. Current: {current_value}",
        "Use @N to select a value. Use @.. to go back.",
    ])

    for idx, item in enumerate(items):
        row = table.add_row()
        row.add_column("Choice", item.get("display", item.get("title", "")))
        table.set_row_selection_action(idx, [".config", str(item.get("key")), str(item.get("value"))])

    ctx.set_last_result_table(table, items)
    ctx.set_current_stage_table(table)
    print()
    print(table)
    return 0


def _show_config_table(
    config_data: Dict[str, Any],
    *,
    browse_path: Optional[str] = None,
) -> int:
    items = _build_config_items(config_data, browse_path=browse_path)
    if not items:
        path_text = _format_config_path_label(browse_path)
        print(f"No configuration entries available for {path_text}.")
        return 0

    table = Table(_build_config_table_title(browse_path), preserve_order=True)
    table.set_table("config")
    if browse_path:
        table.set_source_command(".config", ["-browse", str(browse_path)])
    else:
        table.set_source_command(".config", [])
    table.set_header_lines(_build_config_header_lines(browse_path))

    for idx, item in enumerate(items):
        row = table.add_row()
        row.add_column("Name", item.get("title", ""))
        value_text = item.get("summary") or item.get("value_display", "")
        choices = item.get("choices")
        if choices and isinstance(choices, list) and len(choices) > 0:
            choices_repr = ", ".join(str(c) for c in choices)
            value_text = f"{value_text}  [{choices_repr}]"
        row.add_column("Value", value_text)
        row.add_column("Type", item.get("type", ""))
        if item.get("kind") == "section" and item.get("browse_path"):
            table.set_row_selection_action(
                idx,
                [".config", "-browse", str(item.get("browse_path"))],
            )
        elif item.get("kind") == "value" and choices and isinstance(choices, list) and len(choices) > 0 and item.get("key"):
            table.set_row_selection_action(
                idx,
                [".config", "-choose", str(item.get("key"))],
            )
        elif item.get("kind") == "create_instance" and item.get("instance_target"):
            table.set_row_selection_action(
                idx,
                [".config", _CREATE_INSTANCE_FLAG, str(item.get("instance_target"))],
            )

    ctx.set_last_result_table(table, items)
    ctx.set_current_stage_table(table)
    return 0


def _save_updated_config(config_data: Dict[str, Any], key_path: str) -> None:
    try:
        key_l = str(key_path or "").lower()
    except Exception:
        key_l = ""
    if "alldebrid" in key_l or "all-debrid" in key_l:
        save_config_and_verify(config_data)
        return
    save_config(config_data)


def _resolve_direct_browse_path(
    config_data: Dict[str, Any],
    token: str,
) -> Optional[str]:
    text = str(token or "").strip()
    if not text:
        return None
    lowered = text.lower()
    if lowered in {"preferences", "prefs"}:
        return _PREFERENCES_BROWSE_PATH
    if lowered in {"plugins", "plugin"}:
        return _PLUGINS_BROWSE_PATH

    plugin_branch = _resolve_plugin_branch(config_data, "plugin", lowered)
    if plugin_branch is not None:
        return f"plugin.{plugin_branch[0]}"

    branch = _resolve_config_branch(config_data, text)
    if isinstance(branch, dict):
        return text
    return None


def _strip_value_quotes(value: str) -> str:
    if not value:
        return value
    if (value.startswith('"') and value.endswith('"')) or (value.startswith("'") and value.endswith("'")):
        return value[1:-1]
    return value


def _run(piped_result: Any, args: List[str], config: Dict[str, Any]) -> int:
    import sys

    if _has_flag(args, "-log") or _has_flag(args, "--log"):
        return _show_config_logs(args)

    # Load configuration from the database
    current_config = load_config()

    browse_path = _extract_browse_arg(args)
    if browse_path:
        return _show_config_table(current_config, browse_path=browse_path)

    choose_key = _extract_choose_arg(args)
    if choose_key:
        choices = _lookup_plugin_choices(choose_key, choose_key.split(".")[-1]) or []
        if not choices:
            nested_branch = _resolve_config_branch(current_config, choose_key)
            if isinstance(nested_branch, dict):
                return _show_config_table(current_config, browse_path=choose_key)
            print(f"No choices available for '{_format_config_path_label(choose_key)}'")
            return 0
        try:
            current_value = str(_resolve_config_value_raw(current_config, choose_key) or "").strip()
        except Exception:
            current_value = ""
        display_name = _format_config_path_label(choose_key)
        return _show_choice_table(choose_key, choices, current_value, display_name)

    selection_item = _get_selected_config_item() or _normalize_config_item(piped_result)

    create_instance_target = _extract_create_instance_target(args)
    if create_instance_target:
        print(
            f"Use @N | .config <instance_name> to create a new instance under '{_format_config_path_label(create_instance_target)}', then set its fields in the table that opens."
        )
        return 0

    value_from_pipe = _extract_piped_value(piped_result)
    selection_kind = str((selection_item or {}).get("kind") or "").strip().lower()
    selection_key = str((selection_item or {}).get("key") or "").strip() or None
    selection_browse_path = str((selection_item or {}).get("browse_path") or "").strip() or None
    selection_display_path = str((selection_item or {}).get("display_path") or selection_key or "").strip() or selection_key
    selection_instance_target = str((selection_item or {}).get("instance_target") or "").strip() or None

    if selection_kind == "section" and selection_browse_path and not args and value_from_pipe is None:
        return _show_config_table(current_config, browse_path=selection_browse_path)

    if selection_kind == "create_instance" and selection_instance_target:
        new_instance_name = value_from_pipe or _extract_selected_update_value(args)
        if new_instance_name is None:
            print(
                f"Use @N | .config <instance_name> to create a new instance under '{_format_config_path_label(selection_instance_target)}', then set its fields in the table that opens."
            )
            return 0
        new_instance_name = _strip_value_quotes(new_instance_name)
        try:
            new_browse_path, created = _create_or_get_plugin_instance(
                current_config,
                selection_instance_target,
                new_instance_name,
            )
            _save_updated_config(current_config, new_browse_path)
            status_text = "Created" if created else "Using existing"
            print(
                f"{status_text} instance '{new_instance_name}' at '{_format_config_path_label(new_browse_path)}'. "
                "Configure its fields in the table below."
            )
            return _show_config_table(current_config, browse_path=new_browse_path)
        except Exception as exc:
            log(f"Error creating config instance '{selection_instance_target}': {exc}")
            print(f"Error creating config instance: {exc}")
            return 1

    if selection_kind == "value" and selection_key:
        new_value = value_from_pipe or _extract_selected_update_value(args)
        if new_value is not None:
            new_value = _strip_value_quotes(new_value)
            target_key = _resolve_update_key(current_config, selection_key)
            try:
                set_nested_config(current_config, target_key, new_value)
                _save_updated_config(current_config, target_key)
                print(f"Updated '{selection_display_path}' to '{new_value}'")
                return 0
            except Exception as exc:
                log(f"Error updating config '{target_key}': {exc}")
                print(f"Error updating config: {exc}")
                return 1
        choices = (selection_item or {}).get("choices")
        if choices and isinstance(choices, list) and len(choices) > 0:
            current_value = str((selection_item or {}).get("value") or "").strip()
            display_name = selection_display_path or selection_key
            return _show_choice_table(selection_key, choices, current_value, display_name)

    if selection_kind == "choice" and selection_key:
        choice_value = str((selection_item or {}).get("value") or "").strip()
        if choice_value:
            try:
                target_key = _resolve_update_key(current_config, selection_key)
                set_nested_config(current_config, target_key, choice_value)
                _save_updated_config(current_config, target_key)
                print(f"Updated '{_format_config_path_label(selection_key)}' to '{choice_value}'")
                current_config = load_config()
                key_parts = [p for p in str(selection_key).split(".") if p]
                if len(key_parts) >= 2:
                    parent_path = ".".join(key_parts[:-1])
                    if _resolve_config_branch(current_config, parent_path):
                        return _show_config_table(current_config, browse_path=parent_path)
                return 0
            except Exception as exc:
                log(f"Error updating config '{selection_key}': {exc}")
                print(f"Error updating config: {exc}")
                return 1

    if not args:
        if sys.stdin.isatty() and not piped_result:
            print(
                "Interactive TUI config editor has been discontinued. "
                "Showing configuration table instead."
            )
        return _show_config_table(current_config)

    key = args[0]
    if len(args) < 2:
        browse_target = _resolve_direct_browse_path(current_config, key)
        if browse_target:
            return _show_config_table(current_config, browse_path=browse_target)
        print(f"Error: Value required for key '{key}'")
        return 1

    value = _strip_value_quotes(" ".join(args[1:]))
    try:
        set_nested_config(current_config, key, value)
        _save_updated_config(current_config, key)
        print(f"Updated '{key}' to '{value}'")
        current_config = load_config()
        parent_path = _resolve_direct_browse_path(current_config, key)
        if not parent_path:
            key_parts = [p for p in str(key).split(".") if p]
            if len(key_parts) >= 2:
                candidate = ".".join(key_parts[:-1])
                if _resolve_config_branch(current_config, candidate):
                    parent_path = candidate
        if parent_path:
            return _show_config_table(current_config, browse_path=parent_path)
        return 0
    except Exception as exc:
        log(f"Error updating config '{key}': {exc}")
        print(f"Error updating config: {exc}")
        return 1


CMDLET.exec = _run