96 lines
3.4 KiB
Python
96 lines
3.4 KiB
Python
import logging
|
|
import asyncio
|
|
import json
|
|
from argparse import ArgumentParser
|
|
from pathlib import Path
|
|
|
|
from .functions import get_all_financials
|
|
from .constants import MAIN_LOGGER_NAME, DEFAULT_CONCURRENT_BATCH_SIZE
|
|
|
|
|
|
log = logging.getLogger(MAIN_LOGGER_NAME)
|
|
|
|
TICKER_SYMBOL = 'ticker_symbol'
|
|
QUARTERLY = 'quarterly'
|
|
BATCH_SIZE = 'concurrent_batch_size'
|
|
TO_FILE = 'to_file'
|
|
JSON_INDENT = 'json_indent'
|
|
VERBOSE = 'verbose'
|
|
|
|
|
|
def parse_cli() -> dict:
|
|
"""Returns the parsed command line arguments for the program as a dictionary."""
|
|
parser = ArgumentParser(description="Scrape company financials")
|
|
parser.add_argument(
|
|
TICKER_SYMBOL,
|
|
type=str,
|
|
nargs='+',
|
|
help="Stock ticker symbol of the company to be scraped the financials of"
|
|
)
|
|
parser.add_argument(
|
|
'-Q', f'--{QUARTERLY}',
|
|
action='store_true',
|
|
help="If set, the financial data for the last quarters is returned; otherwise yearly data is returned."
|
|
)
|
|
parser.add_argument(
|
|
'-b', f'--{BATCH_SIZE.replace("_", "-")}',
|
|
type=int,
|
|
default=DEFAULT_CONCURRENT_BATCH_SIZE,
|
|
help="If multiple ticker symbols are passed, the company financials can be scraped concurrently. "
|
|
"This argument determines how many companies are scraped concurrently. "
|
|
"By default, they are scraped sequentially (i.e. a batch size of 1)."
|
|
)
|
|
parser.add_argument(
|
|
'-f', f'--{TO_FILE.replace("_", "-")}',
|
|
type=Path,
|
|
help="Writes results to the specified destination file. If omitted results are printed to stdout."
|
|
)
|
|
parser.add_argument(
|
|
f'--{JSON_INDENT.replace("_", "-")}',
|
|
type=int,
|
|
help="If set to a positive integer and the output format is JSON (default), the resulting JSON document is "
|
|
"indented accordingly for more readability; if omitted, output is returned in one line."
|
|
)
|
|
parser.add_argument(
|
|
'-v', f'--{VERBOSE}',
|
|
action='count',
|
|
default=0,
|
|
help="Verbose mode. Reduces the log level and thus prints out more status messages while running. "
|
|
"Using this flag multiple times increases verbosity further."
|
|
)
|
|
return vars(parser.parse_args())
|
|
|
|
|
|
def configure_logging(verbosity: int) -> None:
|
|
"""
|
|
Sets up logging by adding a stdout-handler and setting the root logger's level according to the specified verbosity.
|
|
A verbosity of 0 (or less) sets the log level to CRITICAL.
|
|
"""
|
|
root_logger = logging.getLogger()
|
|
root_logger.addHandler(logging.StreamHandler())
|
|
root_logger.setLevel(logging.CRITICAL)
|
|
if verbosity > 2:
|
|
root_logger.setLevel(logging.DEBUG)
|
|
elif verbosity == 2:
|
|
root_logger.setLevel(logging.INFO)
|
|
elif verbosity == 1:
|
|
root_logger.setLevel(logging.WARNING)
|
|
|
|
|
|
async def main() -> None:
|
|
"""Parses CLI arguments, configures logging to stderr, performs the scraping, and prints/saves the data."""
|
|
args = parse_cli()
|
|
configure_logging(args[VERBOSE])
|
|
data = await get_all_financials(*args[TICKER_SYMBOL], quarterly=args[QUARTERLY],
|
|
concurrent_batch_size=args[BATCH_SIZE])
|
|
path: Path = args[TO_FILE]
|
|
if path is None:
|
|
print(json.dumps(data, indent=args[JSON_INDENT]))
|
|
return
|
|
with open(path, 'w') as f:
|
|
json.dump(data, f, indent=args[JSON_INDENT])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
asyncio.run(main())
|