Files
firefrost-services/services/modpack-version-checker/src/modpack_checker/cli.py
Claude (The Golden Chronicler #50) 04e9b407d5 feat: Migrate Arbiter and Modpack Version Checker to monorepo
WHAT WAS DONE:
- Migrated Arbiter (discord-oauth-arbiter) code to services/arbiter/
- Migrated Modpack Version Checker code to services/modpack-version-checker/
- Created .env.example for Arbiter with all required environment variables
- Moved systemd service file to services/arbiter/deploy/
- Organized directory structure per Gemini monorepo recommendations

WHY:
- Consolidate all service code in one repository
- Prepare for Gemini code review (Panel v1.12 compatibility check)
- Enable service-prefixed Git tagging (arbiter-v2.1.0, modpack-v1.0.0)
- Support npm workspaces for shared dependencies

SERVICES MIGRATED:
1. Arbiter (Discord OAuth bot) - Originally written by Gemini + Claude
   - Full source code from ops-manual docs/implementation/
   - Created comprehensive .env.example
   - Ready for Panel v1.12 compatibility verification

2. Modpack Version Checker (Python CLI tool)
   - Full source code from ops-manual docs/tasks/
   - Written for Panel v1.11, needs Gemini review for v1.12
   - Never had code review before

STILL TODO:
- Whitelist Manager - Pull from Billing VPS (38.68.14.188)
  - Currently deployed and running
  - Needs Panel v1.12 API compatibility fix (Task #86)
  - Requires SSH access to pull code

NEXT STEPS:
- Gemini code review for Panel v1.12 API compatibility
- Create package.json for each service
- Test npm workspaces integration
- Deploy after verification

FILES:
- services/arbiter/ (25 new files, full application)
- services/modpack-version-checker/ (21 new files, full application)

Signed-off-by: The Golden Chronicler <claude@firefrostgaming.com>
2026-03-31 21:52:42 +00:00

566 lines
18 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Click-based CLI — all user-facing commands."""
from __future__ import annotations
import sys
from typing import Optional
import click
from rich import box
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
from . import __version__
from .config import Config
from .curseforge import (
CurseForgeAuthError,
CurseForgeClient,
CurseForgeError,
CurseForgeNotFoundError,
)
from .database import Database
from .notifier import DiscordNotifier, NotificationError
console = Console()
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _load_db(cfg: Config) -> Database:
return Database(cfg.database_path)
def _require_client(cfg: Config) -> CurseForgeClient:
"""Return a configured API client, or exit with a helpful message."""
if not cfg.curseforge_api_key:
console.print("[red]Error:[/red] No CurseForge API key configured.")
console.print(
"Set one with: [bold]modpack-checker config set-key YOUR_KEY[/bold]"
)
console.print(
"Get a free key at: [dim]https://console.curseforge.com[/dim]"
)
sys.exit(1)
return CurseForgeClient(cfg.curseforge_api_key)
# ---------------------------------------------------------------------------
# Root group
# ---------------------------------------------------------------------------
@click.group(context_settings={"help_option_names": ["-h", "--help"]})
@click.version_option(version=__version__, prog_name="modpack-checker")
def cli() -> None:
"""Modpack Version Checker — monitor CurseForge modpacks for updates.
\b
Quick start:
modpack-checker config set-key YOUR_CURSEFORGE_KEY
modpack-checker add 238222
modpack-checker check
"""
# ---------------------------------------------------------------------------
# config sub-group
# ---------------------------------------------------------------------------
@cli.group("config")
def config_group() -> None:
"""Manage configuration (API key, webhook URL, etc.)."""
@config_group.command("set-key")
@click.argument("api_key")
def config_set_key(api_key: str) -> None:
"""Save your CurseForge API key and validate it."""
cfg = Config.load()
cfg.curseforge_api_key = api_key.strip()
cfg.save()
client = CurseForgeClient(api_key)
with console.status("Validating API key with CurseForge…"):
valid = client.validate_api_key()
if valid:
console.print("[green]✓[/green] API key saved and validated.")
else:
console.print(
"[yellow]⚠[/yellow] API key saved, but validation failed. "
"Double-check at [dim]https://console.curseforge.com[/dim]"
)
@config_group.command("set-webhook")
@click.argument("webhook_url")
def config_set_webhook(webhook_url: str) -> None:
"""Save a Discord webhook URL and send a test message."""
cfg = Config.load()
cfg.discord_webhook_url = webhook_url.strip()
cfg.save()
notifier = DiscordNotifier(webhook_url)
try:
with console.status("Testing webhook…"):
notifier.test()
console.print("[green]✓[/green] Webhook saved and test message sent.")
except NotificationError as exc:
console.print(f"[yellow]⚠[/yellow] Webhook saved, but test failed: {exc}")
@config_group.command("set-interval")
@click.argument("hours", type=int)
def config_set_interval(hours: int) -> None:
"""Set how often the scheduler checks for updates (in hours, 1168)."""
if not 1 <= hours <= 168:
console.print("[red]Error:[/red] Interval must be between 1 and 168 hours.")
sys.exit(1)
cfg = Config.load()
cfg.check_interval_hours = hours
cfg.save()
console.print(f"[green]✓[/green] Check interval set to {hours} hour(s).")
@config_group.command("show")
def config_show() -> None:
"""Display the current configuration."""
cfg = Config.load()
key = cfg.curseforge_api_key
if key and len(key) > 8:
masked = f"{key[:4]}{'*' * (len(key) - 8)}{key[-4:]}"
elif key:
masked = "****"
else:
masked = "[red]Not configured[/red]"
table = Table(title="Configuration", box=box.ROUNDED, show_header=False)
table.add_column("Setting", style="cyan", min_width=22)
table.add_column("Value")
table.add_row("CurseForge API Key", masked)
table.add_row(
"Discord Webhook",
cfg.discord_webhook_url or "[dim]Not configured[/dim]",
)
table.add_row("Database", cfg.database_path)
table.add_row("Check Interval", f"{cfg.check_interval_hours} hour(s)")
table.add_row(
"Notifications",
"[green]On[/green]" if cfg.notification_on_update else "[red]Off[/red]",
)
console.print(table)
# ---------------------------------------------------------------------------
# add
# ---------------------------------------------------------------------------
@cli.command()
@click.argument("modpack_id", type=int)
def add(modpack_id: int) -> None:
"""Add a modpack to the watch list by its CurseForge project ID.
\b
Example:
modpack-checker add 238222 # All The Mods 9
"""
cfg = Config.load()
client = _require_client(cfg)
db = _load_db(cfg)
with console.status(f"Looking up modpack {modpack_id} on CurseForge…"):
try:
name = client.get_mod_name(modpack_id)
except CurseForgeNotFoundError:
console.print(
f"[red]Error:[/red] No modpack found with ID {modpack_id} on CurseForge."
)
sys.exit(1)
except CurseForgeAuthError as exc:
console.print(f"[red]Auth error:[/red] {exc}")
sys.exit(1)
except CurseForgeError as exc:
console.print(f"[red]API error:[/red] {exc}")
sys.exit(1)
try:
db.add_modpack(modpack_id, name)
except ValueError:
console.print(
f"[yellow]Already tracked:[/yellow] [bold]{name}[/bold] (ID: {modpack_id})"
)
return
console.print(
f"[green]✓[/green] Now tracking [bold]{name}[/bold] (ID: {modpack_id})"
)
console.print("Run [bold]modpack-checker check[/bold] to fetch the current version.")
# ---------------------------------------------------------------------------
# remove
# ---------------------------------------------------------------------------
@cli.command()
@click.argument("modpack_id", type=int)
@click.confirmation_option(prompt="Remove this modpack and all its history?")
def remove(modpack_id: int) -> None:
"""Remove a modpack from the watch list."""
cfg = Config.load()
db = _load_db(cfg)
modpack = db.get_modpack(modpack_id)
if modpack is None:
console.print(
f"[red]Error:[/red] Modpack ID {modpack_id} is not in your watch list."
)
sys.exit(1)
db.remove_modpack(modpack_id)
console.print(f"[green]✓[/green] Removed [bold]{modpack.name}[/bold].")
# ---------------------------------------------------------------------------
# list
# ---------------------------------------------------------------------------
@cli.command(name="list")
def list_modpacks() -> None:
"""Show all watched modpacks and their last known versions."""
cfg = Config.load()
db = _load_db(cfg)
modpacks = db.get_all_modpacks()
if not modpacks:
console.print("[dim]Watch list is empty.[/dim]")
console.print(
"Add a modpack with: [bold]modpack-checker add <curseforge-id>[/bold]"
)
return
table = Table(
title=f"Watched Modpacks ({len(modpacks)})",
box=box.ROUNDED,
)
table.add_column("CF ID", style="cyan", justify="right", no_wrap=True)
table.add_column("Name", style="bold white")
table.add_column("Current Version", style="green")
table.add_column("Last Checked", style="dim")
table.add_column("Alerts", justify="center")
for mp in modpacks:
last_checked = (
mp.last_checked.strftime("%Y-%m-%d %H:%M")
if mp.last_checked
else "[dim]Never[/dim]"
)
alerts = "[green]✓[/green]" if mp.notification_enabled else "[red]✗[/red]"
version = mp.current_version or "[dim]—[/dim]"
table.add_row(str(mp.curseforge_id), mp.name, version, last_checked, alerts)
console.print(table)
# ---------------------------------------------------------------------------
# check
# ---------------------------------------------------------------------------
@cli.command()
@click.option(
"--id", "-m", "modpack_id", type=int, default=None,
help="Check a single modpack by CurseForge ID.",
)
@click.option(
"--notify/--no-notify", default=True,
help="Send Discord notifications for updates found (default: on).",
)
def check(modpack_id: Optional[int], notify: bool) -> None:
"""Check all (or one) watched modpack(s) for updates."""
cfg = Config.load()
client = _require_client(cfg)
db = _load_db(cfg)
if modpack_id is not None:
target = db.get_modpack(modpack_id)
if target is None:
console.print(
f"[red]Error:[/red] Modpack ID {modpack_id} is not in your watch list."
)
sys.exit(1)
modpacks = [target]
else:
modpacks = db.get_all_modpacks()
if not modpacks:
console.print("[dim]No modpacks to check.[/dim]")
console.print(
"Add one with: [bold]modpack-checker add <curseforge-id>[/bold]"
)
return
notifier: Optional[DiscordNotifier] = None
if notify and cfg.discord_webhook_url and cfg.notification_on_update:
notifier = DiscordNotifier(cfg.discord_webhook_url)
updates = 0
errors = 0
for mp in modpacks:
with console.status(f"Checking [bold]{mp.name}[/bold]…"):
try:
file_obj = client.get_latest_file(mp.curseforge_id)
except CurseForgeNotFoundError:
console.print(
f" [red]✗[/red] {mp.name}: not found on CurseForge "
f"(ID: {mp.curseforge_id})"
)
errors += 1
continue
except CurseForgeError as exc:
console.print(f" [red]✗[/red] {mp.name}: API error — {exc}")
errors += 1
continue
if file_obj is None:
console.print(f" [yellow]⚠[/yellow] {mp.name}: no files found on CurseForge.")
errors += 1
continue
new_version = client.extract_version(file_obj)
notification_sent = False
if mp.current_version == new_version:
line = f"[green]✓[/green] {mp.name}: up to date ([bold]{new_version}[/bold])"
elif mp.current_version is None:
line = (
f"[cyan]→[/cyan] {mp.name}: "
f"initial version recorded as [bold]{new_version}[/bold]"
)
else:
updates += 1
line = (
f"[yellow]↑[/yellow] {mp.name}: "
f"[dim]{mp.current_version}[/dim] → [bold green]{new_version}[/bold green]"
)
if notifier and mp.notification_enabled:
try:
notifier.send_update(
mp.name, mp.curseforge_id, mp.current_version, new_version
)
notification_sent = True
line += " [dim](notified)[/dim]"
except NotificationError as exc:
line += f" [red](notification failed: {exc})[/red]"
db.update_version(mp.curseforge_id, new_version, notification_sent)
console.print(f" {line}")
# Summary line
console.print()
parts = []
if updates:
parts.append(f"[yellow]{updates} update(s) found[/yellow]")
if errors:
parts.append(f"[red]{errors} error(s)[/red]")
if not updates and not errors:
parts.append("[green]All modpacks are up to date.[/green]")
console.print(" ".join(parts))
# ---------------------------------------------------------------------------
# status
# ---------------------------------------------------------------------------
@cli.command()
@click.argument("modpack_id", type=int)
@click.option("--limit", "-n", default=10, show_default=True, help="History entries to show.")
def status(modpack_id: int, limit: int) -> None:
"""Show detailed status and check history for a modpack."""
cfg = Config.load()
db = _load_db(cfg)
mp = db.get_modpack(modpack_id)
if mp is None:
console.print(
f"[red]Error:[/red] Modpack ID {modpack_id} is not in your watch list."
)
sys.exit(1)
last_checked = (
mp.last_checked.strftime("%Y-%m-%d %H:%M UTC") if mp.last_checked else "Never"
)
notif_str = "[green]Enabled[/green]" if mp.notification_enabled else "[red]Disabled[/red]"
console.print(
Panel(
f"[bold white]{mp.name}[/bold white]\n"
f"CurseForge ID : [cyan]{mp.curseforge_id}[/cyan]\n"
f"Version : [green]{mp.current_version or 'Not checked yet'}[/green]\n"
f"Last Checked : [dim]{last_checked}[/dim]\n"
f"Notifications : {notif_str}",
title="Modpack Status",
border_style="cyan",
)
)
history = db.get_check_history(modpack_id, limit=limit)
if not history:
console.print("[dim]No check history yet.[/dim]")
return
table = Table(title=f"Check History (last {limit})", box=box.SIMPLE)
table.add_column("Timestamp", style="dim")
table.add_column("Version", style="green")
table.add_column("Notified", justify="center")
for entry in history:
notified = "[green]✓[/green]" if entry.notification_sent else "[dim]—[/dim]"
table.add_row(
entry.checked_at.strftime("%Y-%m-%d %H:%M"),
entry.version_found or "[dim]Unknown[/dim]",
notified,
)
console.print(table)
# ---------------------------------------------------------------------------
# notifications (toggle per-modpack alerts)
# ---------------------------------------------------------------------------
@cli.command()
@click.argument("modpack_id", type=int)
@click.option(
"--enable/--disable",
default=True,
help="Enable or disable Discord alerts for this modpack.",
)
def notifications(modpack_id: int, enable: bool) -> None:
"""Enable or disable Discord notifications for a specific modpack."""
cfg = Config.load()
db = _load_db(cfg)
mp = db.get_modpack(modpack_id)
if mp is None:
console.print(
f"[red]Error:[/red] Modpack ID {modpack_id} is not in your watch list."
)
sys.exit(1)
db.toggle_notifications(modpack_id, enable)
state = "[green]enabled[/green]" if enable else "[red]disabled[/red]"
console.print(
f"[green]✓[/green] Notifications {state} for [bold]{mp.name}[/bold]."
)
# ---------------------------------------------------------------------------
# schedule (background daemon)
# ---------------------------------------------------------------------------
@cli.command()
@click.option(
"--hours", "-h", "hours", type=int, default=None,
help="Override the configured check interval (hours).",
)
def schedule(hours: Optional[int]) -> None:
"""Run continuous background checks on a configurable interval.
Requires the [scheduler] extra: pip install modpack-version-checker[scheduler]
"""
try:
from apscheduler.schedulers.blocking import BlockingScheduler
except ImportError:
console.print("[red]Error:[/red] APScheduler is not installed.")
console.print(
"Install it with: [bold]pip install modpack-version-checker[scheduler][/bold]"
)
sys.exit(1)
cfg = Config.load()
interval = hours or cfg.check_interval_hours
def _run_check() -> None:
"""Inner function executed by the scheduler."""
client = _require_client(cfg)
db = _load_db(cfg)
modpacks = db.get_all_modpacks()
notifier: Optional[DiscordNotifier] = None
if cfg.discord_webhook_url and cfg.notification_on_update:
notifier = DiscordNotifier(cfg.discord_webhook_url)
for mp in modpacks:
try:
file_obj = client.get_latest_file(mp.curseforge_id)
if file_obj is None:
continue
new_version = client.extract_version(file_obj)
notification_sent = False
if (
mp.current_version is not None
and mp.current_version != new_version
and notifier
and mp.notification_enabled
):
try:
notifier.send_update(
mp.name, mp.curseforge_id, mp.current_version, new_version
)
notification_sent = True
except NotificationError:
pass
db.update_version(mp.curseforge_id, new_version, notification_sent)
except CurseForgeError:
pass # Log silently in daemon mode; don't crash the scheduler
scheduler = BlockingScheduler()
scheduler.add_job(_run_check, "interval", hours=interval)
console.print(
Panel(
f"Checking every [bold]{interval}[/bold] hour(s).\n"
"Press [bold]Ctrl-C[/bold] to stop.",
title="Modpack Checker — Scheduler Running",
border_style="green",
)
)
# Run immediately so the user gets instant feedback
_run_check()
try:
scheduler.start()
except KeyboardInterrupt:
console.print("\n[dim]Scheduler stopped.[/dim]")
# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------
def main() -> None:
cli()
if __name__ == "__main__":
main()