"""Cayce Ops CLI (Typer). Component sub-apps are imported lazily so the skeleton CLI works before every component module lands, and so one broken component can't take down the CLI. """ from __future__ import annotations import typer from cayceops import __version__ app = typer.Typer(name="cayceops", help="Cayce autonomous operations stack", no_args_is_help=True) db_app = typer.Typer(name="db", help="Schema lifecycle") app.add_typer(db_app) @db_app.command("init") def db_init() -> None: """Create all tables from ORM metadata (dev/test).""" from cayceops.db import init_db init_db() typer.echo("schema created (metadata.create_all)") @db_app.command("upgrade") def db_upgrade() -> None: """Run Alembic migrations to head (production path).""" from cayceops.db import upgrade_db upgrade_db() typer.echo("migrations applied") @app.command() def version() -> None: """Print the version.""" typer.echo(f"cayce-ops {__version__}") def _load_component(name: str): """Import a component module by name (lazy).""" import importlib return importlib.import_module(f"cayceops.components.{name}") # --- scanner ----------------------------------------------------------------- scanner_app = typer.Typer(name="scanner", help="Opportunity discovery (read-only)") app.add_typer(scanner_app) @scanner_app.command("run") def scanner_run() -> None: """Scan configured sources and file proposals. Never executes anything.""" raise typer.Exit(_load_component("scanner").run()) # --- executor ---------------------------------------------------------------- executor_app = typer.Typer(name="executor", help="Execute approved proposals") app.add_typer(executor_app) @executor_app.command("run") def executor_run() -> None: """Process approved proposals through registered action plugins.""" raise typer.Exit(_load_component("executor").run()) @executor_app.command("approve") def executor_approve( proposal_id: int, actor: str = typer.Option(..., help="Human approver id") ) -> None: """Approve a proposal (human action, audited).""" raise typer.Exit(_load_component("executor").approve(proposal_id, actor)) @executor_app.command("reject") def executor_reject(proposal_id: int, actor: str = typer.Option(...)) -> None: """Reject a proposal (human action, audited).""" raise typer.Exit(_load_component("executor").reject(proposal_id, actor)) # --- maild ------------------------------------------------------------------- maild_app = typer.Typer(name="maild", help="Merchant mail negotiation") app.add_typer(maild_app) @maild_app.command("poll") def maild_poll() -> None: """Poll IMAP, classify inbound mail, draft replies.""" raise typer.Exit(_load_component("maild").poll()) @maild_app.command("approve-draft") def maild_approve(msg_id: int, actor: str = typer.Option(...)) -> None: """Approve a drafted reply so SMTP may send it.""" raise typer.Exit(_load_component("maild").approve_draft(msg_id, actor)) @maild_app.command("send-approved") def maild_send() -> None: """Send all approved drafts via SMTP.""" raise typer.Exit(_load_component("maild").send_approved()) # --- publisher --------------------------------------------------------------- publisher_app = typer.Typer(name="publisher", help="Themed content calendar") app.add_typer(publisher_app) @publisher_app.command("plan") def publisher_plan() -> None: """Fill the calendar with drafts for the coming days (per theme rotation).""" raise typer.Exit(_load_component("publisher").plan()) @publisher_app.command("image") def publisher_image(item_id: int) -> None: """Render the brand image for a content item.""" raise typer.Exit(_load_component("publisher").make_image(item_id)) @publisher_app.command("approve") def publisher_approve(item_id: int, actor: str = typer.Option(...)) -> None: """Approve a draft content item.""" raise typer.Exit(_load_component("publisher").approve(item_id, actor)) @publisher_app.command("publish-due") def publisher_publish() -> None: """Publish approved content items that are due.""" raise typer.Exit(_load_component("publisher").publish_due()) # --- monitor ----------------------------------------------------------------- monitor_app = typer.Typer(name="monitor", help="Static status page") app.add_typer(monitor_app) @monitor_app.command("build") def monitor_build() -> None: """Render the static status page (no sensitive data).""" raise typer.Exit(_load_component("monitor").build()) def main() -> None: # pragma: no cover app() if __name__ == "__main__": # pragma: no cover main()