import asyncio import argparse import json import os from dataclasses import asdict, is_dataclass from typing import Any from . import unified_api from .browser.auth import login_to_schwab from .core.config import load_config, get_schwab_credentials, set_config_path, set_cookies_path def _to_serializable(obj: Any) -> Any: if is_dataclass(obj): return asdict(obj) if isinstance(obj, list): return [_to_serializable(item) for item in obj] if isinstance(obj, dict): return {key: _to_serializable(value) for key, value in obj.items()} return obj def _print_envelope(envelope): payload = dict(envelope) payload["data"] = _to_serializable(payload.get("data")) print(json.dumps(payload, indent=2, default=str)) async def test_scraper(ticker: str, debug: bool): """Test the get_morningstar_data function.""" print(f"Running scraper test for ticker: {ticker}") data = await unified_api.get_morningstar_data(ticker, debug=debug) _print_envelope(data) async def async_main(): parser = argparse.ArgumentParser(description="Schwab Morningstar Scraper CLI") parser.add_argument("ticker", nargs='?', help="Stock ticker to scrape") parser.add_argument("--debug", action="store_true", help="Enable debug output") parser.add_argument("--login", action="store_true", help="Login only (don't scrape)") parser.add_argument("--test", action="store_true", help="Test mode") parser.add_argument("--phase1", action="store_true", help="Extract Phase 1 enhanced equity data (quote, dividends, earnings, valuation ratios)") # Configuration file paths parser.add_argument("--config-path", metavar="PATH", help="Custom path for config.json file") parser.add_argument("--cookies-path", metavar="PATH", help="Custom path for cookies.json file") # Session commands parser.add_argument("--session-status", action="store_true", help="Display current session status") parser.add_argument("--export-cookies", metavar="PATH", help="Export cookies to file") parser.add_argument("--set-cookies", metavar="PATH", help="Load cookies from file") # Transactions + accounts parser.add_argument("--transactions", action="store_true", help="Export and parse transaction history") parser.add_argument("--list-accounts", action="store_true", help="List available accounts") parser.add_argument("--account", help="Account identifier (ending digits like 604 or name like Joint)") parser.add_argument("--start-date", help="Start date for custom range (YYYY-MM-DD)") parser.add_argument("--end-date", help="End date for custom range (YYYY-MM-DD)") parser.add_argument("--time-period", help="Preset period (e.g., 'Current Month', 'Last 6 Months')") # Accounts & positions parser.add_argument("--account-overview", nargs='?', const="", help="Show balances for account or aggregate if omitted") parser.add_argument("--positions", nargs='?', const="", help="Show positions for account or aggregate if omitted") parser.add_argument("--portfolio-snapshot", nargs='?', const="", help="Show portfolio snapshot for account or aggregate if omitted") parser.add_argument("--include-non-equity", action="store_true", help="Include non-equity positions") parser.add_argument("--no-aggregate", action="store_true", help="Disable symbol aggregation in portfolio snapshot") args = parser.parse_args() # Apply custom path overrides if provided if args.config_path: if not os.path.exists(args.config_path): print(f"Error: Config file not found: {args.config_path}") return set_config_path(args.config_path) if args.cookies_path: # Note: cookies.json may not exist yet (created on first login) # so we don't validate existence, only that parent directory exists cookies_dir = os.path.dirname(args.cookies_path) if cookies_dir and not os.path.exists(cookies_dir): print(f"Error: Directory for cookies file does not exist: {cookies_dir}") return set_cookies_path(args.cookies_path) if args.login: # Set up debug logging when --debug is used if args.debug: import logging logging.basicConfig(level=logging.DEBUG, format='%(asctime)s %(levelname)s %(name)s: %(message)s') print("Debug logging enabled") config = load_config() username, password = get_schwab_credentials(config) if username and password: print("Attempting to log in...") if args.debug: print(f"Using browserless server: {config.get('playwright', {}).get('url', 'default')}") cookies = await login_to_schwab(username, password) if cookies: print("Login successful and cookies saved.") print(f"Saved {len(cookies)} cookies to cookies.json") else: print("Login failed.") else: print("Schwab username and password not found in config.json.") return if args.session_status: envelope = await unified_api.get_session_status(debug=args.debug) _print_envelope(envelope) return if args.set_cookies: envelope = await unified_api.set_cookies(args.set_cookies, debug=args.debug) _print_envelope(envelope) return if args.export_cookies: envelope = await unified_api.export_cookies(args.export_cookies, debug=args.debug) _print_envelope(envelope) return if args.list_accounts: envelope = await unified_api.list_accounts(debug=args.debug) _print_envelope(envelope) return if args.account_overview is not None: account_arg = args.account_overview or None envelope = await unified_api.get_account_overview(account=account_arg, debug=args.debug) _print_envelope(envelope) return if args.positions is not None: account_arg = args.positions or None envelope = await unified_api.get_positions( account=account_arg, include_non_equity=args.include_non_equity, debug=args.debug, ) _print_envelope(envelope) return if args.portfolio_snapshot is not None: account_arg = args.portfolio_snapshot or None envelope = await unified_api.get_portfolio_snapshot( account=account_arg, aggregate_by_symbol=not args.no_aggregate, include_non_equity=args.include_non_equity, debug=args.debug, ) _print_envelope(envelope) return if args.transactions: envelope = await unified_api.get_transaction_history( account=args.account, start_date=args.start_date, end_date=args.end_date, time_period=args.time_period, debug=args.debug, ) _print_envelope(envelope) return if args.ticker: if args.test: await test_scraper(args.ticker, args.debug) elif args.phase1: print(f"Extracting Phase 1 enhanced equity data for {args.ticker}...") envelope = await unified_api.get_equity_phase1_data(args.ticker, debug=args.debug) _print_envelope(envelope) else: print(f"Scraping Morningstar data for {args.ticker}...") envelope = await unified_api.get_morningstar_data(args.ticker, debug=args.debug) _print_envelope(envelope) return parser.print_help() def main(): """Entry point for console script""" asyncio.run(async_main()) if __name__ == "__main__": main()