"""CLI entry point for sb-scout.""" from __future__ import annotations import argparse import json import os import re import sys import time from pathlib import Path from typing import Any from sb_scout.alerts import check_alerts, fire_notify, send_msmtp_email from sb_scout.config import ( LoadedConfig, dump_settings, load_config, print_example_config, resolve_settings, ) from sb_scout.fetch import default_cache_dir, fetch_feed from sb_scout.history import append_snapshot, diff_snapshots, load_last_snapshot from sb_scout.models import ( SORT_KEYS, AlertConfig, AppConfig, BonusConfig, DiskRequirements, FilterConfig, ScoreWeights, BUILTIN_CPU_SPECS, ) from sb_scout.output import ( build_markdown_report, build_summary, build_text_report, write_exports, ) from sb_scout.scoring import apply_scores, build_rows, sort_rows # ─── Settings that participate in config layering ───────────────────────────── _LAYERED_KEYS = [ "url", "currency", "top", "format", "sort_by", "active_profiles", "ecc_only", "price_cap", "min_ram_gb", "max_ram_gb", "min_storage_gb", "max_storage_gb", "min_ssd_drives", "min_hdd_drives", "ssd_only", "datacenter", "exclude_datacenter", "cpu_regex", "exclude_cpu_regex", "only_gpu", "price_per_thread_cap", "inic_bonus", "gpu_bonus", "cpu_rank_compute_weight", "storage_rank_storage_weight", "overall_weights", "gpu_hints", "history_file", "cache_dir", "alert_min_overall_score", "alert_min_cpu_score", "alert_min_storage_score", "alert_max_price", "alert_notify_command", "alert_email_to", "alert_email_from", "alert_email_subject", "alert_email_account", ] # ─── Argument parsing ──────────────────────────────────────────────────────── def parse_args(argv: list[str] | None = None) -> argparse.Namespace: p = argparse.ArgumentParser( prog="sb-scout", description="Rank Hetzner Serverbörse (Server Auction) listings by value.", formatter_class=argparse.RawDescriptionHelpFormatter, ) # Actions ag = p.add_argument_group("actions") ag.add_argument("--init-config", action="store_true", help="Print example config and exit") ag.add_argument("--list-profiles", action="store_true", help="List profiles and exit") ag.add_argument("--show-config", action="store_true", help="Show resolved config and exit") # Config cg = p.add_argument_group("config") cg.add_argument("--config", "-c", dest="config_path", help="Path to TOML config") cg.add_argument( "--currency", choices=("USD", "EUR"), default=None, help="Currency feed to use when --url is not provided (default: USD)", ) cg.add_argument( "--profile", "-p", dest="active_profiles", action="append", default=None, help="Named profile; repeat or use commas to combine multiple profiles", ) cg.add_argument( "--cultscale", dest="active_profiles", action="append_const", const="cultscale", help="Compatibility shorthand for --profile cultscale (ecc + mixed-storage)", ) # Feed p.add_argument("--url", default=None, help="Auction feed URL") p.add_argument("--top", type=int, default=None, help="Rows per ranking") # Filters fg = p.add_argument_group("filters") eg = fg.add_mutually_exclusive_group() eg.add_argument("--include-non-ecc", dest="ecc_only", action="store_const", const=False) eg.add_argument("--only-ecc", dest="ecc_only", action="store_const", const=True) fg.add_argument("--price-cap", "--budget-max", dest="price_cap", type=float, default=None) fg.add_argument("--min-ram-gb", type=int, default=None) fg.add_argument("--max-ram-gb", type=int, default=None) fg.add_argument("--min-storage", "--min-storage-gb", dest="min_storage_gb", type=int, default=None) fg.add_argument("--max-storage", "--max-storage-gb", dest="max_storage_gb", type=int, default=None) fg.add_argument("--min-ssd-drives", type=int, default=None) fg.add_argument("--min-hdd-drives", type=int, default=None) fg.add_argument("--ssd-only", action="store_true", default=None) fg.add_argument("--datacenter", default=None) fg.add_argument("--exclude-datacenter", default=None) fg.add_argument("--cpu-regex", default=None) fg.add_argument("--exclude-cpu-regex", default=None) fg.add_argument("--only-gpu", action="store_true", default=None) fg.add_argument("--price-per-thread-cap", type=float, default=None) # Scoring sg = p.add_argument_group("scoring") sg.add_argument("--inic-bonus", type=float, default=None) sg.add_argument("--gpu-bonus", type=float, default=None) sg.add_argument("--cpu-rank-compute-weight", type=float, default=None) sg.add_argument("--storage-rank-storage-weight", type=float, default=None) sg.add_argument("--overall-weights", default=None) sg.add_argument("--sort-by", choices=tuple(SORT_KEYS), default=None) # Alerts alg = p.add_argument_group("alerts") alg.add_argument("--alert-min-overall", dest="alert_min_overall_score", type=float, default=None, help="Alert when overall score >= threshold") alg.add_argument("--alert-min-cpu", dest="alert_min_cpu_score", type=float, default=None) alg.add_argument("--alert-min-storage", dest="alert_min_storage_score", type=float, default=None) alg.add_argument("--alert-max-price", dest="alert_max_price", type=float, default=None, help="Alert when price <= threshold") alg.add_argument("--alert-notify", dest="alert_notify_command", default=None, help="Command to run on alert (gets JSON on stdin)") alg.add_argument("--alert-email-to", action="append", default=None, help="Alert email recipient; repeat or use commas for multiple recipients") alg.add_argument("--alert-email-from", default=None, help="Alert email From header for msmtp") alg.add_argument("--alert-email-subject", default=None, help="Alert email subject line") alg.add_argument("--alert-email-account", default=None, help="Optional msmtp account name (-a ACCOUNT)") # Output og = p.add_argument_group("output") og.add_argument("--format", choices=("text", "markdown"), default=None) og.add_argument("--json", dest="json_stdout", action="store_true", default=False, help="Output structured JSON to stdout (overrides --format)") og.add_argument("--out-dir", default=str(Path(os.environ.get("TMPDIR", "/tmp")) / "sb-scout")) og.add_argument("--cache-dir", default=None, help="XDG cache dir for feed snapshots") og.add_argument("--csv-out", default=None) og.add_argument("--json-out", default=None) og.add_argument("--save-markdown", default=None) og.add_argument("--no-write", action="store_true") og.add_argument("--history-file", default=None, help="JSONL file for price history tracking") og.add_argument("--diff", action="store_true", help="Show diff against last history snapshot") # Watch og.add_argument("--watch", type=int, default=None, metavar="SECONDS", help="Re-fetch every N seconds") p.add_argument("--strict-cpu-map", action="store_true") return p.parse_args(argv) def extract_cli_overrides(args: argparse.Namespace) -> dict[str, Any]: """Extract only explicitly-set CLI values.""" overrides: dict[str, Any] = {} for key in _LAYERED_KEYS: val = getattr(args, key, None) if val is not None: overrides[key] = val return overrides # ─── Settings → AppConfig ───────────────────────────────────────────────────── def _parse_overall_weights(value: Any) -> tuple[float, float, float]: if isinstance(value, (list, tuple)): parts = [float(v) for v in value] else: try: parts = [float(x.strip()) for x in str(value).split(",")] except ValueError as e: raise SystemExit("overall_weights must look like 0.40,0.30,0.30") from e if len(parts) != 3: raise SystemExit("overall_weights must have exactly 3 values") c, r, s = parts if min(c, r, s) < 0: raise SystemExit("overall weights must be non-negative") if c + r + s <= 0: raise SystemExit("overall weights must sum to a positive number") return c, r, s def _compile_regex(pattern: str | None, label: str) -> re.Pattern[str] | None: if not pattern: return None try: return re.compile(pattern, re.IGNORECASE) except re.error as e: raise SystemExit(f"Invalid {label}: {e}") from e def _normalize_multi_value(value: Any) -> tuple[str, ...]: if value is None: return () if isinstance(value, str): return tuple(part.strip() for part in value.split(",") if part.strip()) if isinstance(value, (list, tuple)): parts: list[str] = [] for item in value: if isinstance(item, str): parts.extend(part.strip() for part in item.split(",") if part.strip()) return tuple(parts) return () def build_app_config( settings: dict[str, Any], *, profiles: list[str], out_dir: str, csv_out: str | None, json_out: str | None, no_write: bool, strict_cpu_map: bool, save_markdown: str | None, json_stdout: bool, watch_interval: int | None, ) -> AppConfig: """Validate and build AppConfig from resolved settings.""" currency = str(settings.get("currency", "USD")).upper() url = str(settings["url"]) top = int(settings["top"]) fmt = str(settings.get("format", "text")) sort_by = str(settings.get("sort_by", "overall")) ecc_only = bool(settings.get("ecc_only", True)) price_cap = settings.get("price_cap") min_ram = int(settings.get("min_ram_gb", 0)) max_ram = settings.get("max_ram_gb") min_sto = int(settings.get("min_storage_gb", 0)) max_sto = settings.get("max_storage_gb") min_ssd = int(settings.get("min_ssd_drives", 0)) min_hdd = int(settings.get("min_hdd_drives", 0)) ssd_only = bool(settings.get("ssd_only", False)) dc = settings.get("datacenter") exc_dc = settings.get("exclude_datacenter") cpu_re = settings.get("cpu_regex") exc_cpu_re = settings.get("exclude_cpu_regex") only_gpu = bool(settings.get("only_gpu", False)) ppt_cap = settings.get("price_per_thread_cap") history_file = settings.get("history_file") cache_dir = settings.get("cache_dir") inic_bonus = float(settings.get("inic_bonus", 0.10)) gpu_bonus = float(settings.get("gpu_bonus", 0.15)) crw = float(settings.get("cpu_rank_compute_weight", 0.75)) srw = float(settings.get("storage_rank_storage_weight", 0.75)) oc, or_, os_ = _parse_overall_weights(settings.get("overall_weights", "0.40,0.30,0.30")) gpu_hints_raw = settings.get("gpu_hints") if isinstance(gpu_hints_raw, (list, tuple)): gpu_hints = tuple(str(h).upper() for h in gpu_hints_raw) else: from sb_scout.models import DEFAULT_GPU_HINTS gpu_hints = DEFAULT_GPU_HINTS # Validation if top <= 0: raise SystemExit("top must be > 0") if sort_by not in SORT_KEYS: raise SystemExit(f"sort_by must be one of: {', '.join(SORT_KEYS)}") if fmt not in ("text", "markdown"): raise SystemExit("format must be 'text' or 'markdown'") if min_ram < 0: raise SystemExit("min_ram_gb must be >= 0") if max_ram is not None: max_ram = int(max_ram) if max_ram < min_ram: raise SystemExit("max_ram_gb must be >= min_ram_gb") if min_sto < 0: raise SystemExit("min_storage_gb must be >= 0") if max_sto is not None: max_sto = int(max_sto) if max_sto < min_sto: raise SystemExit("max_storage_gb must be >= min_storage_gb") if price_cap is not None: price_cap = float(price_cap) if price_cap <= 0: raise SystemExit("price_cap must be > 0") if ppt_cap is not None: ppt_cap = float(ppt_cap) if ppt_cap <= 0: raise SystemExit("price_per_thread_cap must be > 0") if not 0 <= crw <= 1: raise SystemExit("cpu_rank_compute_weight must be between 0 and 1") if not 0 <= srw <= 1: raise SystemExit("storage_rank_storage_weight must be between 0 and 1") if ssd_only: min_hdd = 0 min_ssd = max(1, min_ssd) alerts = AlertConfig( min_overall_score=settings.get("alert_min_overall_score"), min_cpu_score=settings.get("alert_min_cpu_score"), min_storage_score=settings.get("alert_min_storage_score"), max_price=settings.get("alert_max_price"), notify_command=settings.get("alert_notify_command"), email_to=_normalize_multi_value(settings.get("alert_email_to")), email_from=settings.get("alert_email_from"), email_subject=str(settings.get("alert_email_subject") or "sb-scout alert"), email_account=settings.get("alert_email_account"), ) profile_label = ", ".join(profiles) if profiles else None return AppConfig( url=url, currency=currency, top=top, output_format=fmt, out_dir=Path(out_dir), cache_dir=Path(cache_dir) if cache_dir else default_cache_dir(), csv_out=csv_out, json_out=json_out, no_write=no_write, strict_cpu_map=strict_cpu_map, save_markdown=save_markdown, sort_by=sort_by, profiles=tuple(profiles), profile=profile_label, bonuses=BonusConfig(inic_bonus=inic_bonus, gpu_bonus=gpu_bonus), weights=ScoreWeights( cpu_rank_compute=crw, storage_rank_storage=srw, overall_compute=oc, overall_ram=or_, overall_storage=os_, ), filters=FilterConfig( ecc_only=ecc_only, price_cap=price_cap, min_ram_gb=min_ram, max_ram_gb=max_ram, min_storage_gb=min_sto, max_storage_gb=max_sto, disk=DiskRequirements(min_ssd_drives=min_ssd, min_hdd_drives=min_hdd, ssd_only=ssd_only), datacenter=dc, exclude_datacenter=exc_dc, cpu_regex_raw=cpu_re, exclude_cpu_regex_raw=exc_cpu_re, cpu_regex=_compile_regex(cpu_re, "cpu_regex"), exclude_cpu_regex=_compile_regex(exc_cpu_re, "exclude_cpu_regex"), only_gpu=only_gpu, price_per_thread_cap=ppt_cap, ), alerts=alerts, gpu_hints=gpu_hints, history_file=history_file, watch_interval=watch_interval, json_stdout=json_stdout, ) # ─── Single run ─────────────────────────────────────────────────────────────── def run_once( config: AppConfig, cpu_specs: dict[str, dict[str, int]], *, show_diff: bool = False, ) -> int: """Execute a single fetch-score-report cycle. Returns exit code.""" payload = fetch_feed(config.url, cache_dir=config.cache_dir) servers = payload["server"] rows, unknown = build_rows(servers, config, cpu_specs) if unknown: msg = ( "Unknown CPU models skipped (add to [cpu_specs] in config):\n- " + "\n- ".join(sorted(unknown)) ) if config.strict_cpu_map: raise SystemExit(msg) print(msg, file=sys.stderr) if not rows: raise SystemExit("No servers left after filtering") apply_scores(rows, config.weights) sort_key = SORT_KEYS[config.sort_by] tops = { "selected": sort_rows(rows, sort_key)[:config.top], "cpu": sort_rows(rows, "score_cpu_rank")[:config.top], "storage": sort_rows(rows, "score_storage_rank")[:config.top], "overall": sort_rows(rows, "score_overall_rank")[:config.top], } export_rows = sort_rows(rows, sort_key) # History and diff if config.history_file: hpath = Path(config.history_file).expanduser() if show_diff: prev = load_last_snapshot(hpath) if prev: d = diff_snapshots(prev, rows) print(json.dumps(d, indent=2)) if not d["new"] and not d["removed"] and not d["price_changes"]: print("No changes since last snapshot.", file=sys.stderr) else: print("No previous snapshot found.", file=sys.stderr) append_snapshot(hpath, rows, profile=config.profile, currency=config.currency) # Alerts hits = check_alerts(rows, config.alerts) if hits and config.alerts.notify_command: fire_notify(config.alerts.notify_command, hits) if hits and config.alerts.email_to: send_msmtp_email(config.alerts, hits) # Output summary = build_summary( config=config, feed_count=len(servers), rows=rows, tops=tops, export_rows=export_rows, ) if config.json_stdout: print(json.dumps(summary, indent=2)) else: if config.output_format == "markdown" or config.save_markdown: md = build_markdown_report( config=config, feed_count=len(servers), rows=rows, sort_key=sort_key, tops=tops, ) if config.output_format == "markdown": print(md, end="") else: print(build_text_report( config=config, feed_count=len(servers), rows=rows, sort_key=sort_key, tops=tops, ), end="") if config.save_markdown: p = Path(config.save_markdown) p.parent.mkdir(parents=True, exist_ok=True) p.write_text(md) # noqa: F821 — md is defined when save_markdown is set if not config.no_write: csv_p, json_p = write_exports( export_rows, summary, config.out_dir, csv_out=config.csv_out, json_out=config.json_out, ) if not config.json_stdout: print(f"\nCSV: {csv_p}") print(f"JSON: {json_p}") if config.save_markdown and not config.json_stdout: print(f"Markdown: {Path(config.save_markdown)}") if hits and not config.json_stdout: print(f"\n🔔 Alert: {len(hits)} server(s) matched alert thresholds", file=sys.stderr) return 0 # ─── Main ───────────────────────────────────────────────────────────────────── def main(argv: list[str] | None = None) -> int: args = parse_args(argv) if args.init_config: print_example_config() return 0 loaded = load_config(args.config_path) if args.list_profiles: profiles = loaded.list_profiles() if not profiles: print("No profiles available.") else: print("Available profiles:\n") for name, desc in profiles.items(): print(f" {name:20s} {desc}") if loaded.source_path: print(f"\nConfig: {loaded.source_path}") return 0 cli_overrides = extract_cli_overrides(args) settings, resolved_profiles = resolve_settings(loaded, args.active_profiles, cli_overrides) if args.show_config: print(dump_settings(settings, profiles=resolved_profiles, source=loaded.source_path)) return 0 config = build_app_config( settings, profiles=resolved_profiles, out_dir=args.out_dir, csv_out=args.csv_out, json_out=args.json_out, no_write=args.no_write, strict_cpu_map=args.strict_cpu_map, save_markdown=args.save_markdown, json_stdout=args.json_stdout, watch_interval=args.watch, ) cpu_specs = dict(BUILTIN_CPU_SPECS) cpu_specs.update(loaded.extra_cpu_specs) if config.watch_interval: try: while True: run_once(config, cpu_specs, show_diff=args.diff) print(f"\n--- waiting {config.watch_interval}s ---\n", file=sys.stderr) time.sleep(config.watch_interval) except KeyboardInterrupt: print("\nStopped.", file=sys.stderr) return 0 else: return run_once(config, cpu_specs, show_diff=args.diff) if __name__ == "__main__": raise SystemExit(main())