192 lines
8.9 KiB
Python
192 lines
8.9 KiB
Python
from __future__ import annotations
|
|
|
|
import shutil
|
|
from typing import Any, Dict, List
|
|
|
|
from SYS.cmdlet_spec import Cmdlet
|
|
from SYS import pipeline as ctx
|
|
from SYS.result_table import Table
|
|
from SYS.logger import set_debug, debug
|
|
from cmdnat._status_shared import (
|
|
add_startup_check as _add_startup_check,
|
|
default_provider_ping_targets as _default_provider_ping_targets,
|
|
has_provider as _has_provider,
|
|
has_store_subtype as _has_store_subtype,
|
|
has_tool as _has_tool,
|
|
ping_first as _ping_first,
|
|
ping_url as _ping_url,
|
|
provider_display_name as _provider_display_name,
|
|
)
|
|
|
|
CMDLET = Cmdlet(
|
|
name=".status",
|
|
summary="Check and display service/provider status",
|
|
usage=".status",
|
|
arg=[],
|
|
)
|
|
|
|
def _run(result: Any, args: List[str], config: Dict[str, Any]) -> int:
|
|
startup_table = Table(
|
|
"*********<IGNITIO>*********<NOUSEMPEH>*********<RUGRAPOG>*********<OMEGHAU>*********"
|
|
)
|
|
startup_table._interactive(True)._perseverance(True)
|
|
startup_table.set_value_case("upper")
|
|
|
|
debug_enabled = bool(config.get("debug", False))
|
|
try:
|
|
# Ensure global debug state follows config so HTTPClient and other helpers
|
|
# emit debug-level information during the status check.
|
|
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:
|
|
# MPV check
|
|
try:
|
|
from MPV.mpv_ipc import MPV
|
|
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)
|
|
|
|
# Providers
|
|
pcfg = config.get("provider", {})
|
|
if isinstance(pcfg, dict) and pcfg:
|
|
from ProviderCore.registry import list_providers, list_search_providers, list_file_providers
|
|
from Provider.metadata_provider import list_metadata_providers
|
|
|
|
p_avail = list_providers(config) or {}
|
|
s_avail = list_search_providers(config) or {}
|
|
f_avail = list_file_providers(config) or {}
|
|
m_avail = list_metadata_providers(config) or {}
|
|
debug(f"Provider registries: providers={list(p_avail.keys())}, search={list(s_avail.keys())}, file={list(f_avail.keys())}, metadata={list(m_avail.keys())}")
|
|
|
|
already = {"matrix"}
|
|
for pname in pcfg.keys():
|
|
prov = str(pname).lower()
|
|
if prov in already: continue
|
|
display = _provider_display_name(prov)
|
|
|
|
if prov == "alldebrid":
|
|
try:
|
|
from Provider.alldebrid import _get_debrid_api_key
|
|
from API.alldebrid import AllDebridClient
|
|
api_key = _get_debrid_api_key(config)
|
|
debug(f"AllDebrid configured: api_key_present={bool(api_key)}")
|
|
if not api_key:
|
|
_add_startup_check(startup_table, "DISABLED", display, provider=prov, detail="Not configured")
|
|
else:
|
|
client = AllDebridClient(api_key)
|
|
_add_startup_check(startup_table, "ENABLED", display, provider=prov, detail=getattr(client, "base_url", "Connected"))
|
|
debug(f"AllDebrid client connected: base_url={getattr(client, 'base_url', 'unknown')}")
|
|
except Exception as exc:
|
|
_add_startup_check(startup_table, "DISABLED", display, provider=prov, detail=str(exc))
|
|
debug(f"AllDebrid check failed: {exc}")
|
|
already.add(prov)
|
|
continue
|
|
|
|
is_known = prov in p_avail or prov in s_avail or prov in f_avail or prov in m_avail
|
|
if not is_known:
|
|
_add_startup_check(startup_table, "UNKNOWN", display, provider=prov, detail="Not registered")
|
|
debug(f"Provider {prov} not registered")
|
|
else:
|
|
ok_val = p_avail.get(prov) or s_avail.get(prov) or f_avail.get(prov) or m_avail.get(prov)
|
|
detail = "Configured" if ok_val else "Not configured"
|
|
ping_targets = _default_provider_ping_targets(prov)
|
|
if ping_targets:
|
|
debug(f"Provider {prov} ping targets: {ping_targets}")
|
|
pok, pdet = _ping_first(ping_targets)
|
|
debug(f"Provider {prov} ping result: ok={pok}, detail={pdet}")
|
|
detail = pdet if ok_val else f"{detail} | {pdet}"
|
|
_add_startup_check(startup_table, "ENABLED" if ok_val else "DISABLED", display, provider=prov, detail=detail)
|
|
already.add(prov)
|
|
|
|
# Matrix
|
|
if _has_provider(config, "matrix"):
|
|
try:
|
|
from Provider.matrix import Matrix
|
|
m_prov = Matrix(config)
|
|
mcfg = config.get("provider", {}).get("matrix", {})
|
|
hs = str(mcfg.get("homeserver") or "").strip()
|
|
rid = str(mcfg.get("room_id") or "").strip()
|
|
detail = f"{hs} room:{rid}"
|
|
valid = False
|
|
try:
|
|
valid = bool(m_prov.validate())
|
|
except Exception as exc:
|
|
debug(f"Matrix validate failed: {exc}")
|
|
_add_startup_check(startup_table, "ENABLED" if valid else "DISABLED", "Matrix", provider="matrix", detail=detail)
|
|
debug(f"Matrix check: homeserver={hs}, room_id={rid}, validate={valid}")
|
|
except Exception as exc:
|
|
_add_startup_check(startup_table, "DISABLED", "Matrix", provider="matrix", detail=str(exc))
|
|
debug(f"Matrix instantiation failed: {exc}")
|
|
|
|
# Cookies
|
|
try:
|
|
from tool.ytdlp import YtDlpTool
|
|
cf = YtDlpTool(config).resolve_cookiefile()
|
|
_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}")
|
|
|
|
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
|
|
|
|
CMDLET.exec = _run
|