OpenBB/openbb_platform/providers/sec/openbb_sec/cli.py

"""Standalone launcher and cache manager for the OpenBB SEC provider.

``openbb-sec`` resolves the SEC disk-cache folder — from, highest priority
first, the ``--cache-dir`` argument, the ``OPENBB_SEC_CACHE_DIR`` environment
variable, the ``cache_directory`` preference discovered from openbb.toml / user
settings, then the OpenBB default — and serves the provider as a self-contained
OpenBB Workspace backend by delegating to ``openbb-api`` with a SEC app factory.

Commands
--------
serve (default)
    Launch the SEC backend (``openbb-api --app openbb_sec.cli:create_app
    --factory``), honoring ``--host``, ``--port`` and ``--reload``.
info
    Print the resolved cache folder, size limit and disk usage.
path
    Print the resolved cache folder.
clear
    Empty the cache.
"""

import argparse
import os
import sys

from openbb_sec.utils import cache

_APP_FACTORY = "openbb_sec.cli:create_app"


def _abs(path: str) -> str:
    """Return an absolute, user-expanded path."""
    return os.path.abspath(os.path.expanduser(path))


def resolve_cache_directory(
    cache_dir: "str | None" = None, *, discover: bool = True
) -> str:
    """Resolve the SEC cache folder from arg, env var, openbb.toml, or default.

    Parameters
    ----------
    cache_dir : str | None
        An explicit folder from the command line; wins when provided.
    discover : bool
        When True, load the layered openbb-core config (``.env`` files and
        openbb.toml) so the resolved folder reflects those layers.

    Returns
    -------
    str
        The absolute SEC cache folder.
    """
    if cache_dir:
        return _abs(cache_dir)
    if discover:
        from openbb_core.app.config.loader import load_layered_config

        load_layered_config()
    override = os.environ.get(cache.CACHE_DIR_ENV_VAR)
    if override:
        return _abs(override)
    from openbb_core.app.service.user_service import UserService

    parent = UserService().default_user_settings.preferences.cache_directory
    return os.path.join(_abs(parent), "sec")


def create_app():
    """Build the SEC-focused FastAPI app (factory for ``openbb-api --app``).

    Mounts the OpenBB REST app and serves the SEC ``widgets.json`` / ``apps.json``
    at the backend root so the OpenBB Workspace discovers the SEC dashboards.
    """
    from fastapi.responses import JSONResponse
    from openbb_core.api.rest_api import app

    from openbb_sec import sec_router

    paths = {getattr(route, "path", "") for route in app.routes}
    headers = {"X-Backend-Type": "OpenBB Platform"}

    if "/widgets.json" not in paths:

        @app.get("/widgets.json", include_in_schema=False)
        async def _widgets_json():
            """Serve the SEC widgets.json at the backend root."""
            return JSONResponse(await sec_router.get_widgets_json(), headers=headers)

    if "/apps.json" not in paths:

        @app.get("/apps.json", include_in_schema=False)
        async def _apps_json():
            """Serve the SEC apps.json at the backend root."""
            return JSONResponse(await sec_router.get_apps_json(), headers=headers)

    return app


def _print_info(cache_dir: str) -> None:
    """Print the resolved cache folder, size limit and current disk usage."""
    print(f"SEC cache directory: {cache_dir}")
    print(f"Size limit:          {cache.get_size_limit():,} bytes")
    if not os.path.isdir(cache_dir):
        print("Disk usage:          (not created yet)")
        return
    total = 0
    for root, _dirs, files in os.walk(cache_dir):
        for name in files:
            file_path = os.path.join(root, name)
            if os.path.isfile(file_path):
                total += os.path.getsize(file_path)
    print(f"Disk usage:          {total:,} bytes")


def serve(passthrough: "list[str]") -> int:
    """Launch the SEC backend through ``openbb-api`` with the SEC app factory."""
    import importlib.util
    import subprocess

    if importlib.util.find_spec("openbb_platform_api") is None:
        print(
            "openbb-api is required to serve. Install it with:"
            " pip install openbb-platform-api"
        )
        return 1

    command = [
        sys.executable,
        "-m",
        "openbb_platform_api.main",
        "--app",
        _APP_FACTORY,
        "--factory",
        *passthrough,
    ]
    return subprocess.run(command, check=False).returncode


def main(argv: "list[str] | None" = None) -> int:
    """Entry point for the ``openbb-sec`` console script."""
    parser = argparse.ArgumentParser(
        prog="openbb-sec",
        description="Serve the OpenBB SEC provider and manage its disk cache.",
    )
    parser.add_argument(
        "command",
        nargs="?",
        choices=["serve", "info", "path", "clear"],
        default="serve",
        help="Action to run. Defaults to 'serve'.",
    )
    parser.add_argument(
        "-c", "--cache-dir", help="SEC cache folder. Overrides env var and config."
    )
    parser.add_argument("--host", help="Host to serve on. Defaults to 127.0.0.1.")
    parser.add_argument("--port", help="Port to serve on. Defaults to 6900.")
    parser.add_argument(
        "--reload", action="store_true", help="Enable auto-reload while serving."
    )
    args = parser.parse_args(argv)

    cache_dir = resolve_cache_directory(args.cache_dir)
    os.environ[cache.CACHE_DIR_ENV_VAR] = cache_dir

    if args.command == "path":
        print(cache_dir)
        return 0
    if args.command == "info":
        _print_info(cache_dir)
        return 0
    if args.command == "clear":
        print(f"Cleared {cache.clear_cache()} cache entries from {cache_dir}")
        return 0

    passthrough: list[str] = []
    if args.host:
        passthrough += ["--host", args.host]
    if args.port:
        passthrough += ["--port", str(args.port)]
    if args.reload:
        passthrough += ["--reload", "true"]
    return serve(passthrough)


if __name__ == "__main__":  # pragma: no cover - module CLI execution guard
    sys.exit(main())