import click import json import sys import asyncio import os from pathlib import Path from typing import Optional, Dict, Any from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn, TaskProgressColumn from rich.console import Console from .scraper import PorkbunAPI from .cache import DomainCache def load_config() -> Dict[str, Any]: """Load configuration from ~/.config/porkbun/cli.json if it exists.""" config_path = Path.home() / ".config" / "porkbun" / "cli.json" if config_path.exists(): try: with config_path.open('r') as f: return json.load(f) except Exception as e: click.echo(f"Warning: Failed to load config from {config_path}: {e}", err=True) return {} def resolve_credentials(api_key: Optional[str], secret_key: Optional[str]) -> tuple[str, str]: """Resolve API credentials from CLI args, env vars, or config file. Priority: CLI args > env vars > config file """ config = load_config() resolved_api_key = ( api_key or os.getenv('PORKBUN_API_KEY') or config.get('api_key') or config.get('apiKey') ) resolved_secret_key = ( secret_key or os.getenv('PORKBUN_SECRET_KEY') or config.get('secret_key') or config.get('secretKey') ) if not resolved_api_key or not resolved_secret_key: click.echo("Error: API credentials not found!", err=True) click.echo("", err=True) click.echo("Provide credentials via one of:", err=True) click.echo(" 1. Command line: --api-key KEY --secret-key SECRET", err=True) click.echo(" 2. Environment: PORKBUN_API_KEY and PORKBUN_SECRET_KEY", err=True) click.echo(" 3. Config file: ~/.config/porkbun/cli.json", err=True) click.echo("", err=True) click.echo("Example config file:", err=True) click.echo(' {', err=True) click.echo(' "api_key": "pk1_...",', err=True) click.echo(' "secret_key": "sk1_..."', err=True) click.echo(' }', err=True) sys.exit(1) return resolved_api_key, resolved_secret_key async def async_main(domains: list[str], api_key: str, secret_key: str, refresh: bool, clear_cache: bool, debug: bool, verbose: bool): """Async implementation of the main function.""" cache = DomainCache() if clear_cache: cache.clear() if not verbose: click.echo("Cache cleared.", err=True) api = PorkbunAPI(api_key, secret_key, debug=debug) if verbose: # Use rich progress bars for verbose mode console = Console(stderr=True) progress = Progress( SpinnerColumn(), TextColumn("[progress.description]{task.description}"), BarColumn(), TaskProgressColumn(), console=console, ) with progress: overall_task = progress.add_task("[cyan]Processing domains...", total=len(domains)) for domain in domains: task_id = progress.add_task(f"[yellow]{domain}", total=100) try: if not refresh: progress.update(task_id, description=f"[yellow]{domain} - Checking cache...") cached_data = cache.get(domain) if cached_data: progress.update(task_id, description=f"[green]{domain} - ✓ From cache", completed=100) click.echo(json.dumps(cached_data)) progress.update(overall_task, advance=1) continue progress.update(task_id, description=f"[cyan]{domain} - Fetching from API...", completed=50) result = await api.get_domain_pricing(domain) # Always save to cache cache.set(domain, result) progress.update(task_id, description=f"[green]{domain} - ✓ Complete", completed=100) click.echo(json.dumps(result)) except Exception as e: progress.update(task_id, description=f"[red]{domain} - ✗ Error", completed=100) error_result = { "domain": domain, "error": str(e) } click.echo(json.dumps(error_result)) progress.update(overall_task, advance=1) await api.close() else: # Original non-verbose mode try: for domain in domains: try: if not refresh: cached_data = cache.get(domain) if cached_data: if debug: click.echo(f"[CACHE] Using cached result for {domain}", err=True) click.echo(json.dumps(cached_data)) continue if debug: click.echo(f"[API] Fetching pricing for {domain}...", err=True) result = await api.get_domain_pricing(domain) # Always save to cache (updates existing or creates new) cache.set(domain, result) click.echo(json.dumps(result)) except Exception as e: error_result = { "domain": domain, "error": str(e) } click.echo(json.dumps(error_result)) finally: await api.close() @click.command(context_settings=dict(help_option_names=['-h', '--help'])) @click.argument('domains_file', type=click.Path(exists=True, path_type=Path), required=False) @click.option('-d', '--domain', 'domains', multiple=True, help='Domain to query (can be specified multiple times)') @click.option('--api-key', default=None, help='Porkbun API key') @click.option('--secret-key', default=None, help='Porkbun secret key') @click.option('--refresh', is_flag=True, help='Refresh data from API and update cache') @click.option('--clear-cache', is_flag=True, help='Clear the cache before running') @click.option('--debug', is_flag=True, help='Show detailed request/response information') @click.option('-q', '--quiet', is_flag=True, help='Disable progress bars (show only JSON output)') def main(domains_file: Optional[Path], domains: tuple[str], api_key: Optional[str], secret_key: Optional[str], refresh: bool, clear_cache: bool, debug: bool, quiet: bool): """Query domain pricing from Porkbun API. Accepts domains either from a file (one domain per line) or via -d/--domain flags. Examples: porkbun domains.txt porkbun -d example.com -d test.net porkbun --domain example.com Credentials are resolved in this order: 1. Command line arguments (--api-key, --secret-key) 2. Environment variables (PORKBUN_API_KEY, PORKBUN_SECRET_KEY) 3. Config file (~/.config/porkbun/cli.json) """ # Validate input if not domains_file and not domains: click.echo("Error: Must provide either DOMAINS_FILE or use -d/--domain flag", err=True) sys.exit(1) if domains_file and domains: click.echo("Error: Cannot use both DOMAINS_FILE and -d/--domain flag", err=True) sys.exit(1) # Load domains domain_list = [] if domains_file: with domains_file.open('r') as f: domain_list = [line.strip() for line in f if line.strip()] else: domain_list = list(domains) resolved_api_key, resolved_secret_key = resolve_credentials(api_key, secret_key) # Verbose is enabled by default, disabled with -q/--quiet verbose = not quiet asyncio.run(async_main(domain_list, resolved_api_key, resolved_secret_key, refresh, clear_cache, debug, verbose)) if __name__ == '__main__': main()