72 lines
2.1 KiB
Python
72 lines
2.1 KiB
Python
import asyncio
|
|
import csv
|
|
import json
|
|
from argparse import ArgumentParser
|
|
from pathlib import Path
|
|
from typing import Dict
|
|
|
|
from aiohttp import ClientSession
|
|
|
|
from .functions import get_company_financials, ResultDict, log
|
|
from .constants import END_DATE
|
|
|
|
|
|
JSON_EXT, CSV_EXT = '.json', '.csv'
|
|
|
|
TICKER_SYMBOL = 'ticker_symbol'
|
|
QUARTERLY = 'quarterly'
|
|
TO_FILE = 'to_file'
|
|
|
|
|
|
def parse_cli() -> dict:
|
|
parser = ArgumentParser(description="Scrape company financials")
|
|
parser.add_argument(
|
|
TICKER_SYMBOL,
|
|
type=str,
|
|
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(
|
|
'-f', f'--{TO_FILE.replace("_", "-")}',
|
|
type=Path,
|
|
help="Writes results to the specified destination file. If omitted results are printed to stdout."
|
|
)
|
|
return vars(parser.parse_args())
|
|
|
|
|
|
def write_to_csv(data: Dict[str, ResultDict], file_obj) -> None:
|
|
writer = csv.writer(file_obj)
|
|
for statement_key, statement_dict in data.items():
|
|
end_dates = statement_dict.pop(END_DATE)
|
|
writer.writerow([statement_key] + list(end_dates))
|
|
for key, values in statement_dict.items():
|
|
writer.writerow([key] + list(str(val) for val in values))
|
|
|
|
|
|
async def main() -> None:
|
|
args = parse_cli()
|
|
session = ClientSession()
|
|
try:
|
|
data = await get_company_financials(args[TICKER_SYMBOL], quarterly=args[QUARTERLY], session=session)
|
|
finally:
|
|
await session.close()
|
|
path: Path = args[TO_FILE]
|
|
if path is None:
|
|
print(json.dumps(data, indent=2))
|
|
return
|
|
with open(path, 'w') as f:
|
|
if path.suffix.lower() == CSV_EXT:
|
|
write_to_csv(data, f)
|
|
return
|
|
if not path.suffix.lower() == JSON_EXT:
|
|
log.warning(f"Extension '{path.suffix}' unknown; using JSON format")
|
|
json.dump(data, f, indent=2)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
asyncio.run(main())
|