Medios-Macina/cmdlet/search_provider.py

"""search-provider cmdlet: Search external providers (bandcamp, libgen, soulseek, youtube, alldebrid)."""

from __future__ import annotations

from typing import Any, Dict, List, Sequence, Optional
import sys
import json
import uuid
import importlib

from SYS.logger import log, debug
from ProviderCore.registry import get_search_provider, list_search_providers

from . import _shared as sh

Cmdlet, CmdletArg, should_show_help = (
    sh.Cmdlet,
    sh.CmdletArg,
    sh.should_show_help,
)
from SYS import pipeline as ctx

# Optional dependencies
try:
    from SYS.config import get_local_storage_path
except Exception:  # pragma: no cover
    get_local_storage_path = None  # type: ignore


class Search_Provider(Cmdlet):
    """Search external content providers."""

    def __init__(self):
        super().__init__(
            name="search-provider",
            summary=
            "Search external providers (bandcamp, libgen, soulseek, youtube, alldebrid, loc, internetarchive)",
            usage="search-provider -provider <provider> <query> [-limit N] [-open ID]",
            arg=[
                CmdletArg(
                    "provider",
                    type="string",
                    required=True,
                    description=
                    "Provider name: bandcamp, libgen, soulseek, youtube, alldebrid, loc, internetarchive",
                ),
                CmdletArg(
                    "query",
                    type="string",
                    required=True,
                    description="Search query (supports provider-specific syntax)",
                ),
                CmdletArg(
                    "limit",
                    type="int",
                    description="Maximum results to return (default: 50)"
                ),
                CmdletArg(
                    "open",
                    type="int",
                    description=
                    "(alldebrid) Open folder/magnet by ID and list its files",
                ),
            ],
            detail=[
                "Search external content providers:",
                "- alldebrid: List your AllDebrid account folders (magnets). Select @N to view files.",
                '  Example: search-provider -provider alldebrid "*"',
                '  Example: search-provider -provider alldebrid -open 123 "*"',
                "- bandcamp: Search for music albums/tracks",
                '  Example: search-provider -provider bandcamp "artist:altrusian grace"',
                "- libgen: Search Library Genesis for books",
                '  Example: search-provider -provider libgen "python programming"',
                "- loc: Search Library of Congress (Chronicling America)",
                '  Example: search-provider -provider loc "lincoln"',
                "- soulseek: Search P2P network for music",
                '  Example: search-provider -provider soulseek "pink floyd"',
                "- youtube: Search YouTube for videos",
                '  Example: search-provider -provider youtube "tutorial"',
                "- internetarchive: Search archive.org items (advancedsearch syntax)",
                '  Example: search-provider -provider internetarchive "title:(lincoln) AND mediatype:texts"',
                "",
                "Query syntax:",
                "- bandcamp: Use 'artist:Name' to search by artist",
                "- libgen: Supports isbn:, author:, title: prefixes",
                "- soulseek: Plain text search",
                "- youtube: Plain text search",
                "- internetarchive: Archive.org advancedsearch query syntax",
                "",
                "Results can be piped to other cmdlet:",
                '  search-provider -provider bandcamp "artist:grace" | @1 | download-file',
            ],
            exec=self.run,
        )
        self.register()

    def run(self, result: Any, args: Sequence[str], config: Dict[str, Any]) -> int:
        """Execute search-provider cmdlet."""
        if should_show_help(args):
            ctx.emit(self.__dict__)
            return 0

        args_list = [str(a) for a in (args or [])]

        # Dynamic flag variants from cmdlet arg definitions.
        flag_registry = self.build_flag_registry()
        provider_flags = {
            f.lower()
            for f in (flag_registry.get("provider") or {"-provider", "--provider"})
        }
        query_flags = {
            f.lower()
            for f in (flag_registry.get("query") or {"-query", "--query"})
        }
        limit_flags = {
            f.lower()
            for f in (flag_registry.get("limit") or {"-limit", "--limit"})
        }
        open_flags = {
            f.lower()
            for f in (flag_registry.get("open") or {"-open", "--open"})
        }

        provider_name: Optional[str] = None
        query: Optional[str] = None
        limit = 50
        open_id: Optional[int] = None
        positionals: List[str] = []

        i = 0
        while i < len(args_list):
            token = args_list[i]
            low = token.lower()
            if low in provider_flags and i + 1 < len(args_list):
                provider_name = args_list[i + 1]
                i += 2
            elif low in query_flags and i + 1 < len(args_list):
                query = args_list[i + 1]
                i += 2
            elif low in limit_flags and i + 1 < len(args_list):
                try:
                    limit = int(args_list[i + 1])
                except ValueError:
                    log(
                        f"Warning: Invalid limit value '{args_list[i + 1]}', using default 50",
                        file=sys.stderr,
                    )
                    limit = 50
                i += 2
            elif low in open_flags and i + 1 < len(args_list):
                try:
                    open_id = int(args_list[i + 1])
                except ValueError:
                    log(
                        f"Warning: Invalid open value '{args_list[i + 1]}', ignoring",
                        file=sys.stderr,
                    )
                    open_id = None
                i += 2
            elif not token.startswith("-"):
                positionals.append(token)
                i += 1
            else:
                i += 1

        # Backwards-compatible positional form: search-provider <provider> <query>
        if provider_name is None and positionals:
            provider_name = positionals[0]
            positionals = positionals[1:]

        if query is None and positionals:
            query = " ".join(positionals).strip() or None

        if not provider_name or not query:
            log("Error: search-provider requires a provider and query", file=sys.stderr)
            log(f"Usage: {self.usage}", file=sys.stderr)
            log("Available providers:", file=sys.stderr)
            providers = list_search_providers(config)
            for name, available in sorted(providers.items()):
                status = "✓" if available else "✗"
                log(f"  {status} {name}", file=sys.stderr)
            return 1

        debug(
            f"[search-provider] provider={provider_name}, query={query}, limit={limit}"
        )

        # Get provider
        provider = get_search_provider(provider_name, config)
        if not provider:
            log(f"Error: Provider '{provider_name}' is not available", file=sys.stderr)
            log("Available providers:", file=sys.stderr)
            providers = list_search_providers(config)
            for name, available in sorted(providers.items()):
                if available:
                    log(f"  - {name}", file=sys.stderr)
            return 1

        worker_id = str(uuid.uuid4())
        library_root = get_local_storage_path(
            config or {}
        ) if get_local_storage_path else None

        db = None
        if library_root:
            try:
                from API.folder import API_folder_store

                db = API_folder_store(library_root)
            except Exception:
                db = None

        try:
            # Use the worker DB if available; otherwise, run as a stateless one-off.
            if db is not None:
                db.__enter__()
                db.insert_worker(
                    worker_id,
                    "search-provider",
                    title=f"Search: {query}",
                    description=f"Provider: {provider_name}, Query: {query}",
                    pipe=ctx.get_current_command_text(),
                )

            results_list = []
            import result_table

            importlib.reload(result_table)
            from SYS.result_table import ResultTable

            provider_text = str(provider_name or "").strip()
            provider_lower = provider_text.lower()
            if provider_lower == "youtube":
                provider_label = "Youtube"
            elif provider_lower == "openlibrary":
                provider_label = "OpenLibrary"
            elif provider_lower == "loc":
                provider_label = "LoC"
            else:
                provider_label = (
                    provider_text[:1].upper() +
                    provider_text[1:] if provider_text else "Provider"
                )

            if provider_lower == "alldebrid" and open_id is not None:
                table_title = f"{provider_label} Files: {open_id}".strip().rstrip(":")
            else:
                table_title = f"{provider_label}: {query}".strip().rstrip(":")
            preserve_order = provider_name.lower() in ("youtube", "openlibrary", "loc")
            table = ResultTable(table_title).set_preserve_order(preserve_order)
            table.set_table(provider_name)
            table.set_source_command("search-provider", list(args))

            debug(f"[search-provider] Calling {provider_name}.search()")
            if provider_lower == "alldebrid":
                if open_id is not None:
                    # Second-stage: show files for selected folder/magnet.
                    results = provider.search(
                        query,
                        limit=limit,
                        filters={
                            "view": "files",
                            "magnet_id": open_id
                        }
                    )
                else:
                    # Default: show folders (magnets) so user can select @N.
                    results = provider.search(
                        query,
                        limit=limit,
                        filters={
                            "view": "folders"
                        }
                    )
            else:
                results = provider.search(query, limit=limit)
            debug(f"[search-provider] Got {len(results)} results")

            if not results:
                log(f"No results found for query: {query}", file=sys.stderr)
                if db is not None:
                    db.append_worker_stdout(worker_id, json.dumps([], indent=2))
                    db.update_worker_status(worker_id, "completed")
                return 0

            # Emit results for pipeline
            for search_result in results:
                item_dict = (
                    search_result.to_dict()
                    if hasattr(search_result,
                               "to_dict") else dict(search_result)
                )

                # Ensure table field is set (should be by provider, but just in case)
                if "table" not in item_dict:
                    item_dict["table"] = provider_name

                row_index = len(table.rows)
                table.add_result(
                    search_result
                )  # ResultTable handles SearchResult objects

                # For AllDebrid folder rows, allow @N to open and show files.
                try:
                    if (provider_lower == "alldebrid" and getattr(search_result,
                                                                  "media_kind",
                                                                  "") == "folder"):
                        magnet_id = None
                        meta = getattr(search_result, "full_metadata", None)
                        if isinstance(meta, dict):
                            magnet_id = meta.get("magnet_id")
                        if magnet_id is not None:
                            table.set_row_selection_args(
                                row_index,
                                ["-open",
                                 str(magnet_id),
                                 "-query",
                                 "*"]
                            )
                except Exception:
                    pass
                results_list.append(item_dict)
                ctx.emit(item_dict)

            ctx.set_last_result_table(table, results_list)
            # Ensure @N selection expands against this newly displayed table.
            ctx.set_current_stage_table(table)
            if db is not None:
                db.append_worker_stdout(worker_id, json.dumps(results_list, indent=2))
                db.update_worker_status(worker_id, "completed")

            return 0

        except Exception as e:
            log(f"Error searching {provider_name}: {e}", file=sys.stderr)
            import traceback

            debug(traceback.format_exc())
            if db is not None:
                try:
                    db.update_worker_status(worker_id, "error")
                except Exception:
                    pass
            return 1
        finally:
            if db is not None:
                try:
                    db.__exit__(None, None, None)
                except Exception:
                    pass


# Register cmdlet instance (catalog + REPL autocomplete expects module-level CMDLET)
CMDLET = Search_Provider()

# Backwards-compatible alias
Search_Provider_Instance = CMDLET