Coverage for src / osiris_cli / utils.py: 0%
120 statements
« prev ^ index » next coverage.py v7.13.0, created at 2025-12-28 16:44 +0200
1from __future__ import annotations
3import json
4import os
5import sys
6import base64
7import shutil
8from datetime import datetime
9from pathlib import Path
10from typing import Any, Dict, List, Optional
12from .config import CONFIG_DIR
14# --- CLIPBOARD UTILS ---
16def copy_to_clipboard(text: str) -> tuple[bool, str]:
17 """
18 Attempts to copy text to clipboard using multiple methods.
19 Returns (success: bool, method_used: str).
20 """
21 # 1. Local (Pyperclip / xclip / wl-copy)
22 # Only try this if we suspect we have access to a clipboard (DISPLAY set or similar)
23 # But on a VPS via SSH, DISPLAY is usually empty.
24 # We try it anyway as it's the "standard" way if dependencies exist.
25 try:
26 import pyperclip
27 pyperclip.copy(text)
28 return True, "System Clipboard"
29 except Exception:
30 pass
32 # 2. Command Line Tools (Fallback if pyperclip failed but tools exist)
33 # (xclip, wl-copy, pbcopy) - typically pyperclip handles these, but explicit check helps debug
35 # 3. OSC 52 (Remote / SSH)
36 # This is the most reliable method for modern terminals (iTerm2, Alacritty, Windows Terminal, WezTerm)
37 # over SSH.
38 try:
39 data = base64.b64encode(text.encode("utf-8")).decode("utf-8")
40 osc = f"\x1b]52;c;{data}\x07"
42 # Tmux wrapping
43 if "tmux" in os.environ.get("TERM", ""):
44 osc = f"\x1bPtmux;\x1b{osc}\x1b\\"
46 # Write directly to stdout. In Textual, this might interfere with UI,
47 # but since it's an escape sequence, it should be invisible if handled by terminal.
48 # Using sys.__stdout__ to bypass any potential redirections.
49 stdout = sys.__stdout__
50 if stdout:
51 stdout.write(osc)
52 stdout.flush()
53 return True, "OSC 52 (Remote)"
54 except Exception:
55 pass
57 return False, "None"
60# --- EXISTING UTILS ---
62def _activity_path() -> Path:
63 return CONFIG_DIR / "activity.log"
66def log_activity(
67 kind: str,
68 detail: Optional[Dict[str, Any]] = None,
69) -> None:
70 """Append a single JSON line to the Osiris activity log."""
71 try:
72 payload: Dict[str, Any] = {
73 "ts": datetime.utcnow().isoformat() + "Z",
74 "kind": kind,
75 }
76 if detail:
77 payload["detail"] = detail
78 path = _activity_path()
79 path.parent.mkdir(parents=True, exist_ok=True)
80 with path.open("a", encoding="utf-8") as f:
81 f.write(json.dumps(payload, ensure_ascii=False) + "\n")
82 except Exception:
83 # Logging must never break core flows.
84 return
87def read_activity(limit: int = 12) -> List[Dict[str, Any]]:
88 """Return the most recent activity entries."""
89 path = _activity_path()
90 if not path.exists():
91 return []
92 try:
93 with path.open("r", encoding="utf-8") as f:
94 lines = [line.strip() for line in f if line.strip()]
95 entries = []
96 for line in lines[-limit:]:
97 try:
98 entries.append(json.loads(line))
99 except Exception:
100 entries.append({"raw": line})
101 return entries
102 except Exception:
103 return []
106def last_recipe_path() -> Optional[Path]:
107 """Return the path to the most recent recipe run log."""
108 recipe_dir = CONFIG_DIR / "recipes"
109 if not recipe_dir.exists():
110 return None
111 try:
112 # Sort by modification time (newest first) or name if timestamped
113 files = sorted(recipe_dir.glob("*.md"), key=lambda p: p.stat().st_mtime, reverse=True)
114 return files[0] if files else None
115 except Exception:
116 return None
118def last_recipe_summary() -> Optional[str]:
119 path = last_recipe_path()
120 if not path:
121 return None
122 return f"{path.name} ({datetime.fromtimestamp(path.stat().st_mtime).isoformat()})"
124def get_git_status(path: str) -> str | None:
125 """
126 Get the short git status of a file (e.g., 'M', 'A', '??').
127 Returns None if clean or error.
128 """
129 import subprocess
130 try:
131 # -z for machine readable, --porcelain for stability
132 res = subprocess.run(
133 ["git", "status", "--porcelain", path],
134 capture_output=True,
135 text=True,
136 timeout=1
137 )
138 if res.returncode != 0 or not res.stdout:
139 return None
141 # Output format is "XY path". We want X or Y.
142 # e.g. " M src/app.py" -> Modified
143 # "?? new.py" -> Untracked
144 line = res.stdout.strip()
145 if not line:
146 return None
148 # First two chars are status
149 status_chars = line[:2]
150 if "M" in status_chars: return "M"
151 if "A" in status_chars: return "A"
152 if "?" in status_chars: return "?"
153 if "D" in status_chars: return "D"
154 return status_chars.strip()
156 except Exception:
157 return None
159# --- OUTPUT / SANITIZATION HELPERS ---
161def mask_secret(val: Any) -> Any:
162 try:
163 # pydantic SecretStr compatibility
164 if hasattr(val, "get_secret_value"):
165 return "****"
166 except Exception:
167 pass
168 return val
171def settings_to_public_dict(obj: Any) -> Dict[str, Any]:
172 try:
173 data = obj.model_dump(exclude_none=True)
174 except Exception:
175 # Fallback to dict()
176 try:
177 data = dict(obj)
178 except Exception:
179 return {}
180 for k, v in list(data.items()):
181 data[k] = mask_secret(v)
182 return data
185def format_for_output(payload: Any, fmt: str = "text") -> str:
186 """
187 Return a string rendered as JSON or text.
188 - If fmt == json and payload is not a string, serialize as JSON.
189 - If payload is a string, return it unchanged for text; for json, wrap in an object.
190 """
191 try:
192 f = (fmt or "text").lower()
193 except Exception:
194 f = "text"
195 if f == "json":
196 if isinstance(payload, str):
197 return json.dumps({"message": payload}, indent=2, ensure_ascii=False)
198 return json.dumps(payload, indent=2, ensure_ascii=False)
199 # text
200 if isinstance(payload, (dict, list)):
201 return json.dumps(payload, indent=2, ensure_ascii=False)
202 return str(payload)