mirror of
https://github.com/OpenBB-finance/OpenBB.git
synced 2026-10-28 13:01:38 +00:00
18caebf116
* allow to execute reports url from reports menu * lint for James * ruff * ruff --------- Co-authored-by: James Maslek <jmaslek11@gmail.com>
378 lines
14 KiB
Python
378 lines
14 KiB
Python
"""Reports Controller Module."""
|
|
__docformat__ = "numpy"
|
|
|
|
import argparse
|
|
import logging
|
|
import os
|
|
|
|
# pylint: disable=R1732, R0912
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
import requests
|
|
|
|
from openbb_terminal.core.session.current_user import get_current_user
|
|
from openbb_terminal.custom_prompt_toolkit import NestedCompleter
|
|
from openbb_terminal.decorators import log_start_end
|
|
from openbb_terminal.helper_funcs import parse_and_split_input
|
|
from openbb_terminal.menu import session
|
|
from openbb_terminal.parent_classes import BaseController
|
|
from openbb_terminal.reports import reports_model
|
|
from openbb_terminal.rich_config import MenuText, console
|
|
|
|
# from openbb_terminal.terminal_helper import is_packaged_application
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ReportController(BaseController):
|
|
"""Report Controller class."""
|
|
|
|
CHOICES_COMMANDS: List[str] = [
|
|
"crypto",
|
|
"economy",
|
|
"equity",
|
|
"etf",
|
|
"forecast",
|
|
"forex",
|
|
"portfolio",
|
|
"run",
|
|
"load",
|
|
"exe",
|
|
]
|
|
PATH = "/reports/"
|
|
|
|
REPORTS: List[str] = reports_model.get_reports_available()
|
|
REPORTS = sorted(list(set(CHOICES_COMMANDS).intersection(REPORTS)))
|
|
PARAMETERS_DICT: Dict[str, Any] = {}
|
|
|
|
def __init__(self, queue: Optional[List[str]] = None):
|
|
"""Constructor"""
|
|
|
|
super().__init__(queue)
|
|
self.update_choices()
|
|
|
|
def update_choices(self):
|
|
"""Update controller choices with reports available under templates folder."""
|
|
|
|
for report_name in self.REPORTS:
|
|
# Extract report parameters do display on menu
|
|
self.PARAMETERS_DICT[report_name] = reports_model.extract_parameters(
|
|
str(reports_model.REPORTS_FOLDER / report_name)
|
|
)
|
|
|
|
if session and get_current_user().preferences.USE_PROMPT_TOOLKIT:
|
|
self.choices: dict = {c: {} for c in self.controller_choices} # type: ignore
|
|
self.choices["run"] = {
|
|
"--file": {c: None for c in reports_model.USER_REPORTS},
|
|
"-f": "--file",
|
|
"--parameters": {},
|
|
"-p": "--parameters",
|
|
}
|
|
for report_name in self.REPORTS:
|
|
# Completer with limited user choices to avoid unforeseen problems
|
|
self.choices[report_name] = {}
|
|
for arg in self.PARAMETERS_DICT[report_name]:
|
|
if report_name in reports_model.REPORT_CHOICES:
|
|
self.choices[report_name][
|
|
"--" + arg
|
|
] = reports_model.REPORT_CHOICES[report_name]["--" + arg]
|
|
|
|
self.choices["support"] = self.SUPPORT_CHOICES
|
|
self.choices["about"] = self.ABOUT_CHOICES
|
|
|
|
self.completer = NestedCompleter.from_nested_dict(self.choices)
|
|
|
|
def parse_input(self, an_input: str) -> List:
|
|
"""Overwrite the BaseController parse_input for `askobb` and 'exe'
|
|
|
|
This will allow us to search for something like "P/E" ratio
|
|
"""
|
|
raw_url = r"(exe (--url |-u )?(https?://)?raw\.githubusercontent\.(com)/.*)"
|
|
github_url = r"(exe (--url |-u )?(https?://)?github\.(com)/.*)"
|
|
custom_filters = [raw_url, github_url]
|
|
return parse_and_split_input(an_input=an_input, custom_filters=custom_filters)
|
|
|
|
def print_help(self):
|
|
"""Print help."""
|
|
|
|
self.REPORTS = reports_model.get_reports_available()
|
|
self.REPORTS = sorted(
|
|
list(set(self.CHOICES_COMMANDS).intersection(self.REPORTS))
|
|
)
|
|
self.update_choices()
|
|
|
|
mt = MenuText("reports/")
|
|
mt.add_raw("\n")
|
|
mt.add_info("_OpenBB_reports_")
|
|
MAX_LEN_NAME = max(len(name) for name in self.REPORTS) + 2
|
|
templates_string = ""
|
|
for report_name in self.REPORTS:
|
|
parameters_names = list(self.PARAMETERS_DICT[report_name].keys())
|
|
|
|
if len(parameters_names) > 1 or not parameters_names:
|
|
args = f"<{'> <'.join(parameters_names)}>"
|
|
else:
|
|
args = f"<{parameters_names[0]}>"
|
|
|
|
templates_string += (
|
|
f" [cmds]{report_name}[/cmds]"
|
|
+ f"{(MAX_LEN_NAME-len(report_name))*' '} "
|
|
+ f"[param]{args if args != '<>' else ''}[/param]\n"
|
|
)
|
|
mt.add_raw(f"{templates_string}")
|
|
mt.add_raw("\n")
|
|
mt.add_info("_Custom_reports_")
|
|
mt.add_cmd("run")
|
|
mt.add_cmd("exe")
|
|
console.print(text=mt.menu_text, menu="Reports")
|
|
|
|
@log_start_end(log=logger)
|
|
def call_etf(self, other_args: List[str]):
|
|
self.run_report("etf", other_args)
|
|
|
|
@log_start_end(log=logger)
|
|
def call_forex(self, other_args: List[str]):
|
|
self.run_report("forex", other_args)
|
|
|
|
@log_start_end(log=logger)
|
|
def call_portfolio(self, other_args: List[str]):
|
|
self.run_report("portfolio", other_args)
|
|
|
|
@log_start_end(log=logger)
|
|
def call_economy(self, other_args: List[str]):
|
|
self.run_report("economy", other_args)
|
|
|
|
@log_start_end(log=logger)
|
|
def call_equity(self, other_args: List[str]):
|
|
self.run_report("equity", other_args)
|
|
|
|
@log_start_end(log=logger)
|
|
def call_crypto(self, other_args: List[str]):
|
|
self.run_report("crypto", other_args)
|
|
|
|
@log_start_end(log=logger)
|
|
def call_forecast(self, other_args: List[str]):
|
|
try:
|
|
# noqa: F401, E501 # pylint: disable=C0415,W0611
|
|
import darts # noqa # pyright: reportMissingImports=false
|
|
|
|
FORECASTING_TOOLKIT_ENABLED = True
|
|
except ImportError:
|
|
FORECASTING_TOOLKIT_ENABLED = False
|
|
console.print(
|
|
"[yellow]"
|
|
"Forecasting Toolkit is disabled. "
|
|
"To use the Forecasting features please install the toolkit following the "
|
|
"instructions here: https://my.openbb.co/app/sdk/installation"
|
|
"\n"
|
|
"[/yellow]"
|
|
)
|
|
if FORECASTING_TOOLKIT_ENABLED:
|
|
self.run_report("forecast", other_args)
|
|
|
|
@log_start_end(log=logger)
|
|
def run_report(self, report_name: str, other_args: List[str]):
|
|
parser = argparse.ArgumentParser(
|
|
add_help=False,
|
|
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
|
|
prog=report_name,
|
|
description=f"Run {report_name} report.",
|
|
)
|
|
|
|
if report_name in self.PARAMETERS_DICT:
|
|
# Assign respective parameters as arguments
|
|
for arg_name, arg_default in self.PARAMETERS_DICT[report_name].items():
|
|
choices = reports_model.get_arg_choices(report_name, arg_name)
|
|
|
|
getattr(parser, "add_argument")(
|
|
"--" + arg_name,
|
|
type=str,
|
|
default=arg_default,
|
|
choices=choices,
|
|
dest=arg_name,
|
|
help=arg_name,
|
|
)
|
|
|
|
if (
|
|
other_args
|
|
and "-" not in other_args[0][0]
|
|
and len(self.PARAMETERS_DICT[report_name]) == 1
|
|
):
|
|
other_args.insert(
|
|
0, "--" + list(self.PARAMETERS_DICT[report_name].keys())[0]
|
|
)
|
|
ns_parser = self.parse_simple_args(parser, other_args)
|
|
|
|
if ns_parser:
|
|
if report_name == "equity" and "." in ns_parser.symbol:
|
|
console.print(
|
|
"[red]Cannot currently handle tickers with a '.' in them[/red]\n"
|
|
)
|
|
return
|
|
parameters = vars(ns_parser)
|
|
parameters.pop("help")
|
|
reports_model.render_report(
|
|
input_path=str(reports_model.REPORTS_FOLDER / report_name),
|
|
args_dict=parameters,
|
|
)
|
|
|
|
else:
|
|
console.print("[red]Notebook not found![/red]\n")
|
|
|
|
@log_start_end(log=logger)
|
|
def call_run(self, other_args: List[str]):
|
|
"""Process run command"""
|
|
parser = argparse.ArgumentParser(
|
|
add_help=False,
|
|
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
|
|
prog="run",
|
|
description="Run a notebook from this folder: '"
|
|
f"{str(get_current_user().preferences.USER_CUSTOM_REPORTS_DIRECTORY)}'.",
|
|
)
|
|
parser.add_argument(
|
|
"-f",
|
|
"--file",
|
|
dest="file",
|
|
choices=reports_model.USER_REPORTS,
|
|
required="-h" not in other_args,
|
|
help="The file to be loaded",
|
|
)
|
|
parser.add_argument(
|
|
"-p",
|
|
"--parameters",
|
|
nargs="+",
|
|
dest="parameters",
|
|
help="Report parameters with format 'name:value'.",
|
|
)
|
|
|
|
ns_parser = self.parse_known_args_and_warn(parser, other_args)
|
|
|
|
if ns_parser:
|
|
# Validate parameter inputs
|
|
parameters_dict = {}
|
|
if ns_parser.parameters:
|
|
for p in ns_parser.parameters:
|
|
if ":" in p:
|
|
item = p.split(":")
|
|
if item[1]:
|
|
parameters_dict[item[0]] = item[1]
|
|
else:
|
|
console.print(
|
|
f"[red]Bad format '{p}': empty value.[/red]\nExecuting with defaults.\n"
|
|
)
|
|
else:
|
|
console.print(
|
|
f"[red]Bad format '{p}': use format 'name:value'.[/red]\nExecuting with defaults.\n"
|
|
)
|
|
|
|
if ns_parser.file:
|
|
complete_file_path = str(
|
|
get_current_user().preferences.USER_CUSTOM_REPORTS_DIRECTORY
|
|
/ ns_parser.file
|
|
)
|
|
if os.path.exists(complete_file_path):
|
|
reports_model.render_report(
|
|
input_path=complete_file_path, args_dict=parameters_dict
|
|
)
|
|
else:
|
|
console.print(
|
|
f"[red]Notebook '{ns_parser.file}' not found![/red]\n"
|
|
)
|
|
|
|
def call_exe(self, other_args: List[str]):
|
|
"""Process exe command"""
|
|
parser = argparse.ArgumentParser(
|
|
add_help=False,
|
|
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
|
|
prog="exe",
|
|
description="Run a notebook from a url that contains the ipynb contents",
|
|
)
|
|
parser.add_argument(
|
|
"-u",
|
|
"--url",
|
|
dest="url",
|
|
required="-h" not in other_args,
|
|
help="The url of the file to be loaded",
|
|
)
|
|
parser.add_argument(
|
|
"-p",
|
|
"--parameters",
|
|
nargs="+",
|
|
dest="parameters",
|
|
help="Report parameters with format 'name:value'.",
|
|
)
|
|
|
|
# If first argument is a url, insert the -u flag
|
|
if other_args[0].startswith("http"):
|
|
other_args.insert(0, "--url")
|
|
|
|
ns_parser = self.parse_known_args_and_warn(parser, other_args)
|
|
|
|
if ns_parser:
|
|
# Validate parameter inputs
|
|
parameters_dict = {}
|
|
if ns_parser.parameters:
|
|
for p in ns_parser.parameters:
|
|
if ":" in p:
|
|
item = p.split(":")
|
|
if item[1]:
|
|
parameters_dict[item[0]] = item[1]
|
|
else:
|
|
console.print(
|
|
f"[red]Bad format '{p}': empty value.[/red]\nExecuting with defaults.\n"
|
|
)
|
|
else:
|
|
console.print(
|
|
f"[red]Bad format '{p}': use format 'name:value'.[/red]\nExecuting with defaults.\n"
|
|
)
|
|
|
|
if "raw.githubusercontent" in ns_parser.url:
|
|
url = ns_parser.url
|
|
else:
|
|
url = ns_parser.url.replace(
|
|
"github.com", "raw.githubusercontent.com"
|
|
).replace("/blob", "")
|
|
|
|
if url:
|
|
try:
|
|
# Send an HTTP GET request to fetch the raw notebook
|
|
response = requests.get(url, timeout=20)
|
|
|
|
if response.status_code == 200:
|
|
temporary_folder = os.path.join(
|
|
get_current_user().preferences.USER_REPORTS_DIRECTORY,
|
|
"temporary",
|
|
)
|
|
|
|
# Does the temp folder exist? if not create it
|
|
if not os.path.exists(temporary_folder):
|
|
os.makedirs(temporary_folder)
|
|
|
|
# Local file path where you want to save the notebook
|
|
local_file_path = os.path.join(
|
|
temporary_folder,
|
|
url.split(".com/")[1].replace(
|
|
"/", "_"
|
|
), # .replace(".ipynb", "")
|
|
)
|
|
|
|
# Save the template notebook locally
|
|
with open(local_file_path, "wb") as notebook_file:
|
|
notebook_file.write(response.content)
|
|
|
|
# To confirm that the notebook is saved locally
|
|
if os.path.exists(local_file_path):
|
|
reports_model.render_report(
|
|
input_path=local_file_path, args_dict=parameters_dict
|
|
)
|
|
else:
|
|
console.print(f"[red]Notebook '{url}' not found![/red]\n")
|
|
|
|
else:
|
|
console.print(
|
|
f"Failed to fetch notebook from {url}. Status code: {response.status_code}"
|
|
)
|
|
|
|
except Exception as e:
|
|
console.print(f"An error occurred: {str(e)}")
|