From 6e65b53f3a8a4e4edfe97e6250a44a44bcdb4d71 Mon Sep 17 00:00:00 2001 From: Timmy Time Date: Mon, 23 Mar 2026 19:34:46 +0000 Subject: [PATCH] [loop-cycle-5] feat: implement 4 TODO stubs in timmyctl/cli.py (#1128) (#1158) --- src/timmyctl/cli.py | 138 +++++++++++++++++++++++++++++++++++++++----- 1 file changed, 124 insertions(+), 14 deletions(-) diff --git a/src/timmyctl/cli.py b/src/timmyctl/cli.py index c9784daa..a96779c7 100644 --- a/src/timmyctl/cli.py +++ b/src/timmyctl/cli.py @@ -9,6 +9,9 @@ Usage: import json import os +import subprocess +import urllib.error +import urllib.request from pathlib import Path from typing import Any @@ -31,6 +34,37 @@ AUTOMATIONS_CONFIG = DEFAULT_CONFIG_DIR / "automations.json" DAILY_RUN_CONFIG = DEFAULT_CONFIG_DIR / "daily_run.json" TRIAGE_RULES_CONFIG = DEFAULT_CONFIG_DIR / "triage_rules.yaml" +GITEA_URL = os.environ.get("GITEA_URL", "http://143.198.27.163:3000") +GITEA_REPO = "rockachopa/Timmy-time-dashboard" + + +def _get_gitea_token() -> str | None: + """Read the Gitea API token from env or config files.""" + token = os.environ.get("GITEA_TOKEN") + if token: + return token.strip() + for candidate in [ + Path("~/.hermes/gitea_token_vps").expanduser(), + Path("~/.hermes/gitea_token").expanduser(), + ]: + try: + return candidate.read_text(encoding="utf-8").strip() + except FileNotFoundError: + continue + return None + + +def _gitea_api_get(endpoint: str) -> Any: + """GET a Gitea API endpoint and return parsed JSON.""" + url = f"{GITEA_URL}/api/v1{endpoint}" + token = _get_gitea_token() + req = urllib.request.Request(url) + if token: + req.add_header("Authorization", f"token {token}") + req.add_header("Accept", "application/json") + with urllib.request.urlopen(req, timeout=15) as resp: + return json.loads(resp.read().decode("utf-8")) + def _load_json_config(path: Path) -> dict[str, Any]: """Load a JSON config file, returning empty dict on error.""" @@ -131,9 +165,43 @@ def daily_run( console.print("[yellow]Dry run mode — no actions executed.[/yellow]") else: console.print("[green]Executing daily run automations...[/green]") - # TODO: Implement actual automation execution - # This would call the appropriate scripts from the automations config - console.print("[dim]Automation execution not yet implemented.[/dim]") + auto_config_path = _get_config_dir() / "automations.json" + auto_config = _load_json_config(auto_config_path) + all_automations = auto_config.get("automations", []) + enabled = [a for a in all_automations if a.get("enabled", False)] + if not enabled: + console.print("[yellow]No enabled automations found.[/yellow]") + for auto in enabled: + cmd = auto.get("command") + name = auto.get("name", auto.get("id", "unnamed")) + if not cmd: + console.print(f"[yellow]Skipping {name} — no command defined.[/yellow]") + continue + console.print(f"[cyan]▶ Running: {name}[/cyan]") + if verbose: + console.print(f"[dim] $ {cmd}[/dim]") + try: + result = subprocess.run( # noqa: S602 + cmd, + shell=True, + capture_output=True, + text=True, + timeout=120, + ) + if result.stdout.strip(): + console.print(result.stdout.strip()) + if result.returncode != 0: + console.print( + f"[red] ✗ {name} exited with code {result.returncode}[/red]" + ) + if result.stderr.strip(): + console.print(f"[red]{result.stderr.strip()}[/red]") + else: + console.print(f"[green] ✓ {name} completed successfully[/green]") + except subprocess.TimeoutExpired: + console.print(f"[red] ✗ {name} timed out after 120s[/red]") + except Exception as exc: + console.print(f"[red] ✗ {name} failed: {exc}[/red]") @app.command() @@ -159,9 +227,14 @@ def log_run( console.print(f"[dim]Message:[/dim] {message}") console.print() - # TODO: Persist to actual logbook file - # This would append to a logbook file (e.g., .loop/logbook.jsonl) - console.print("[green]✓[/green] Entry logged (simulated)") + logbook_path = Path(".loop/logbook.jsonl") + logbook_path.parent.mkdir(parents=True, exist_ok=True) + entry = json.dumps( + {"timestamp": timestamp, "category": category, "message": message} + ) + with open(logbook_path, "a", encoding="utf-8") as f: + f.write(entry + "\n") + console.print(f"[green]✓[/green] Entry logged to {logbook_path}") @app.command() @@ -205,27 +278,64 @@ def inbox( console.print(auto_table) console.print() - # TODO: Fetch actual PRs from Gitea API if include_prs: - pr_table = Table(title="Open Pull Requests (placeholder)") + pr_table = Table(title="Open Pull Requests") pr_table.add_column("#", style="cyan") pr_table.add_column("Title", style="green") pr_table.add_column("Author", style="yellow") pr_table.add_column("Status", style="magenta") - pr_table.add_row("—", "[dim]No PRs fetched (Gitea API not configured)[/dim]", "—", "—") + try: + prs = _gitea_api_get(f"/repos/{GITEA_REPO}/pulls?state=open") + if prs: + for pr in prs[:limit]: + pr_table.add_row( + str(pr.get("number", "")), + pr.get("title", ""), + pr.get("user", {}).get("login", ""), + pr.get("state", ""), + ) + else: + pr_table.add_row("—", "[dim]No open PRs[/dim]", "—", "—") + except Exception as exc: + pr_table.add_row( + "—", f"[red]Error fetching PRs: {exc}[/red]", "—", "—" + ) console.print(pr_table) console.print() - # TODO: Fetch relevant issues from Gitea API if include_issues: - issue_table = Table(title="Issues Calling for Attention (placeholder)") + issue_table = Table(title="Issues Calling for Attention") issue_table.add_column("#", style="cyan") issue_table.add_column("Title", style="green") issue_table.add_column("Type", style="yellow") issue_table.add_column("Priority", style="magenta") - issue_table.add_row( - "—", "[dim]No issues fetched (Gitea API not configured)[/dim]", "—", "—" - ) + try: + issues = _gitea_api_get( + f"/repos/{GITEA_REPO}/issues?state=open&type=issues&limit={limit}" + ) + if issues: + for issue in issues[:limit]: + labels = [lb.get("name", "") for lb in issue.get("labels", [])] + priority = next( + (lb for lb in labels if "priority" in lb.lower()), + "—", + ) + issue_type = next( + (lb for lb in labels if lb.lower() in ("bug", "feature", "refactor", "enhancement")), + "—", + ) + issue_table.add_row( + str(issue.get("number", "")), + issue.get("title", ""), + issue_type, + priority, + ) + else: + issue_table.add_row("—", "[dim]No open issues[/dim]", "—", "—") + except Exception as exc: + issue_table.add_row( + "—", f"[red]Error fetching issues: {exc}[/red]", "—", "—" + ) console.print(issue_table) console.print()