| 1 | """First-run setup wizard for last30days. |
| 2 | |
| 3 | Detects first run, performs auto-setup (cookie extraction + yt-dlp check), |
| 4 | and writes configuration. The actual wizard UI is SKILL.md-driven (the LLM |
| 5 | presents it), but this module provides the detection and setup actions. |
| 6 | """ |
| 7 | |
| 8 | import json |
| 9 | import logging |
| 10 | import os |
| 11 | import re |
| 12 | import shutil |
| 13 | import subprocess |
| 14 | import time |
| 15 | from pathlib import Path |
| 16 | from typing import Any, Dict, Optional, Tuple |
| 17 | from urllib.error import HTTPError, URLError |
| 18 | from urllib.request import Request, urlopen |
| 19 | |
| 20 | logger = logging.getLogger(__name__) |
| 21 | |
| 22 | |
| 23 | def is_first_run(config: Dict[str, Any]) -> bool: |
| 24 | """Return True if the setup wizard has not been completed. |
| 25 | |
| 26 | Checks for SETUP_COMPLETE in the config dict. If it's not set |
| 27 | (None or empty string), the user hasn't gone through setup yet. |
| 28 | """ |
| 29 | return not config.get("SETUP_COMPLETE") |
| 30 | |
| 31 | |
| 32 | _WELCOME_TEXT = """Welcome to /last30days! I research any topic across Reddit, X, YouTube, TikTok, Digg, arXiv, Techmeme, HN, Polymarket & more - what people actually said in the last 30 days. Let's get you set up (~30s). |
| 33 | |
| 34 | I synthesize what people are actually saying right now across social, news, and market sources. |
| 35 | |
| 36 | Auto setup gives you the core sources free in about 30 seconds: |
| 37 | - X/Twitter - reads your browser cookies to authenticate (read live each run, never saved to disk). I check Chrome first (fastest - a one-time macOS Keychain prompt may appear; click Always Allow), then Firefox and Safari. |
| 38 | - Reddit with comments - free keyless discovery (RSS + shreddit), no API key needed. |
| 39 | - YouTube search + transcripts - installs yt-dlp (open source, 190K+ GitHub stars). |
| 40 | - Digg - trending news, GitHub stars, and pipeline feeds - installs the free, keyless Digg CLI. |
| 41 | - arXiv (papers) + Techmeme (tech-news) - install free, keyless Printing Press CLIs and run on any topic (arXiv is relevance + recency gated to research topics). |
| 42 | - StockTwits - retail trader sentiment - auto-on when your topic is a ticker or crypto (e.g. "$NVDA earnings", "bitcoin"), off for everything else. |
| 43 | - Trustpilot - brand/company review sentiment - opt-in (add trustpilot to INCLUDE_SOURCES), off by default. |
| 44 | - Hacker News + Polymarket + GitHub (auto-on if the gh CLI is installed) - always on, zero config. |
| 45 | |
| 46 | Want TikTok and Instagram too? ScrapeCreators adds those (10,000 free calls, scrapecreators.com). No kickbacks, no affiliation. |
| 47 | |
| 48 | Power users can turn on more sources in the Manual Setup guide (LinkedIn, Bluesky, Perplexity, and others) - each needs its own credential, so they are off by default.""" |
| 49 | |
| 50 | |
| 51 | def render_welcome() -> str: |
| 52 | """Return the first-run welcome text. |
| 53 | |
| 54 | Owned by the engine (single source of truth) so the model relays it rather |
| 55 | than re-authoring it -- authored prose gets skipped, relayed command output |
| 56 | does not. Mirrors the SKILL.md welcome; keep in sync if the source set |
| 57 | changes. |
| 58 | """ |
| 59 | return _WELCOME_TEXT |
| 60 | |
| 61 | |
| 62 | def run_auto_setup(config: Dict[str, Any], *, allow_browser_cookies: bool = False) -> Dict[str, Any]: |
| 63 | """Perform the auto-setup actions. |
| 64 | |
| 65 | - Optionally runs cookie extraction for all registered domains, trying the |
| 66 | browsers from ``env.cookie_extraction_browsers()``. Browser reads are off |
| 67 | unless ``allow_browser_cookies`` is true. |
| 68 | - Checks if yt-dlp is installed |
| 69 | - Best-effort install of digg-pp-cli (Printing Press library) |
| 70 | |
| 71 | Returns: |
| 72 | Dict with keys: |
| 73 | cookies_found: {source_name: browser_name} for each source where cookies were found |
| 74 | ytdlp_installed: bool |
| 75 | ytdlp_action: already_installed | installed | install_failed | no_homebrew |
| 76 | digg_installed: bool (True when the engine can resolve digg-pp-cli on PATH) |
| 77 | digg_action: already_installed | installed | installed_off_path | install_failed | no_npx |
| 78 | env_written: bool (always False here — caller writes config separately) |
| 79 | ytdlp_stderr: present when ytdlp_action is install_failed |
| 80 | digg_stderr: present when digg_action is install_failed |
| 81 | digg_path: present when digg_action is installed_off_path (binary on disk, not on PATH) |
| 82 | """ |
| 83 | from .env import COOKIE_DOMAINS, cookie_extraction_browsers |
| 84 | |
| 85 | cookies_found: Dict[str, str] = {} |
| 86 | |
| 87 | if allow_browser_cookies: |
| 88 | from . import cookie_extract |
| 89 | |
| 90 | cookie_config = dict(config) |
| 91 | if not (cookie_config.get("FROM_BROWSER") or "").strip(): |
| 92 | # Chromium-first: Chrome/Brave/etc. read cookies via the Keychain |
| 93 | # with no Full Disk Access, so try them before Safari, whose |
| 94 | # binarycookies read requires FDA (the dead-end most users hit). |
| 95 | # firefox/safari stay as the silent fallbacks. Note: an explicit |
| 96 | # comma list preserves this order (cookie_extraction_browsers); |
| 97 | # "auto" would put the silent browsers first, so do not use it here. |
| 98 | cookie_config["FROM_BROWSER"] = "chrome,brave,edge,vivaldi,arc,chromium,firefox,safari" |
| 99 | browsers = cookie_extraction_browsers(cookie_config) |
| 100 | |
| 101 | for source_name, spec in COOKIE_DOMAINS.items(): |
| 102 | domain = spec["domain"] |
| 103 | cookie_names = spec["cookies"] |
| 104 | |
| 105 | for browser in browsers: |
| 106 | try: |
| 107 | result = cookie_extract.extract_cookies_with_source(browser, domain, cookie_names) |
| 108 | except Exception as exc: |
| 109 | logger.debug("Cookie extraction failed for %s via %s: %s", source_name, browser, exc) |
| 110 | continue |
| 111 | if result is not None and result[0]: |
| 112 | cookies_found[source_name] = result[1] |
| 113 | break # Found cookies for this service, stop trying browsers |
| 114 | |
| 115 | # Check yt-dlp availability and install via Homebrew if missing. Windows |
| 116 | # has no Homebrew, and its working install path is `pip install yt-dlp` |
| 117 | # (see #904), so it gets its own no-op-install guidance branch instead of |
| 118 | # falling into the Homebrew-oriented no_homebrew outcome. |
| 119 | ytdlp_action: str |
| 120 | if shutil.which("yt-dlp") is not None: |
| 121 | ytdlp_installed = True |
| 122 | ytdlp_action = "already_installed" |
| 123 | elif os.name == "nt": |
| 124 | ytdlp_installed = False |
| 125 | ytdlp_action = "no_pip_windows" |
| 126 | elif shutil.which("brew") is not None: |
| 127 | brew_stderr = "" |
| 128 | try: |
| 129 | proc = subprocess.run( |
| 130 | ["brew", "install", "yt-dlp"], |
| 131 | capture_output=True, text=True, timeout=120, |
| 132 | ) |
| 133 | if proc.returncode == 0: |
| 134 | ytdlp_installed = True |
| 135 | ytdlp_action = "installed" |
| 136 | else: |
| 137 | ytdlp_installed = False |
| 138 | ytdlp_action = "install_failed" |
| 139 | brew_stderr = proc.stderr |
| 140 | logger.warning("brew install yt-dlp failed: %s", proc.stderr) |
| 141 | except Exception as exc: |
| 142 | ytdlp_installed = False |
| 143 | ytdlp_action = "install_failed" |
| 144 | brew_stderr = str(exc) |
| 145 | logger.warning("brew install yt-dlp exception: %s", exc) |
| 146 | else: |
| 147 | ytdlp_installed = False |
| 148 | ytdlp_action = "no_homebrew" |
| 149 | |
| 150 | digg_installed, digg_action, digg_stderr, digg_path = _install_digg_cli() |
| 151 | pp_sources = install_default_pp_sources() |
| 152 | |
| 153 | results: Dict[str, Any] = { |
| 154 | "cookies_found": cookies_found, |
| 155 | "ytdlp_installed": ytdlp_installed, |
| 156 | "ytdlp_action": ytdlp_action, |
| 157 | "digg_installed": digg_installed, |
| 158 | "digg_action": digg_action, |
| 159 | # Per-CLI status for the additional default-on Printing Press sources |
| 160 | # (arxiv, techmeme, trustpilot): {source: {installed, action, ...}}. |
| 161 | "pp_sources": pp_sources, |
| 162 | "env_written": False, |
| 163 | } |
| 164 | if ytdlp_action == "install_failed": |
| 165 | results["ytdlp_stderr"] = brew_stderr |
| 166 | if digg_action == "install_failed": |
| 167 | results["digg_stderr"] = digg_stderr |
| 168 | if digg_path: |
| 169 | results["digg_path"] = digg_path |
| 170 | return results |
| 171 | |
| 172 | |
| 173 | # Generous timeout: the install shells out to `npx`, which may download the |
| 174 | # Printing Press package and build the Go binary over the network. |
| 175 | DIGG_INSTALL_TIMEOUT = 300 |
| 176 | DIGG_CLI_BIN = "digg-pp-cli" |
| 177 | # Pin the catalog installer; matches printing-press-library npm 0.1.16 default |
| 178 | # ($HOME/.local/bin on macOS/Linux). |
| 179 | PRINTING_PRESS_NPM = "@mvanhorn/printing-press-library@0.1.16" |
| 180 | DIGG_INSTALL_CMD = f"npx -y {PRINTING_PRESS_NPM} install digg --cli-only" |
| 181 | |
| 182 | |
| 183 | def _digg_bin_candidate_paths() -> list[Path]: |
| 184 | """Known install locations for digg-pp-cli (Printing Press library defaults). |
| 185 | |
| 186 | Order: current installer default (~/.local/bin), legacy Go bins, Windows |
| 187 | managed dir. The directory list is ``health.installer_bin_dirs()`` — the |
| 188 | shared single source — with the Digg filename variants appended (plain |
| 189 | name for Unix-style dirs, ``.exe`` in the Windows managed dir). |
| 190 | ``pipeline.available_sources()`` only activates Digg when |
| 191 | ``shutil.which`` resolves on PATH — probing these dirs is for setup |
| 192 | verification and honest off-PATH messaging, not engine activation. |
| 193 | """ |
| 194 | from . import health |
| 195 | |
| 196 | win_dir = health.windows_printing_press_bin_dir() |
| 197 | candidates: list[Path] = [] |
| 198 | for directory in health.installer_bin_dirs(): |
| 199 | if win_dir is not None and directory == win_dir: |
| 200 | candidates.append(directory / f"{DIGG_CLI_BIN}.exe") |
| 201 | else: |
| 202 | candidates.append(directory / DIGG_CLI_BIN) |
| 203 | return candidates |
| 204 | |
| 205 | |
| 206 | def _digg_on_path() -> Optional[str]: |
| 207 | """Return digg-pp-cli when the engine would activate Digg (PATH-resolvable).""" |
| 208 | return shutil.which(DIGG_CLI_BIN) |
| 209 | |
| 210 | |
| 211 | def _digg_off_path_binary() -> Optional[str]: |
| 212 | """Return digg-pp-cli path from known install dirs when not on PATH.""" |
| 213 | for candidate in _digg_bin_candidate_paths(): |
| 214 | if candidate.is_file() and os.access(candidate, os.X_OK): |
| 215 | return str(candidate) |
| 216 | return None |
| 217 | |
| 218 | |
| 219 | def _digg_bin_dir_hint(digg_path: str) -> str: |
| 220 | """Return a copy-pasteable PATH directory for the given binary path.""" |
| 221 | parent = os.path.dirname(os.path.expanduser(digg_path)) |
| 222 | if os.name == "nt": |
| 223 | # Windows PATH edits use absolute dirs; $HOME is a Unix shell convention. |
| 224 | return parent |
| 225 | home = str(Path.home()) |
| 226 | if parent == home: |
| 227 | return "$HOME" |
| 228 | prefix = home + os.sep |
| 229 | if parent.startswith(prefix): |
| 230 | rel = parent[len(prefix):].replace(os.sep, "/") |
| 231 | return f"$HOME/{rel}" if rel else "$HOME" |
| 232 | return parent |
| 233 | |
| 234 | |
| 235 | def _run_npx_install(slug: str) -> Tuple[str, str]: |
| 236 | """Resolve ``npx`` and run the Printing Press catalog install for ``slug``. |
| 237 | |
| 238 | Shared by ``_install_digg_cli`` and ``_install_pp_cli`` -- this is only the |
| 239 | "resolve npx, run the install, interpret no_npx/exception/nonzero-rc" |
| 240 | slice; each caller keeps its own on-path/off-path re-verification |
| 241 | (``_digg_bin_candidate_paths`` vs ``_pp_bin_candidate_paths`` already use |
| 242 | different candidate-directory sources, so merging them here would change |
| 243 | off-path detection behavior beyond this fix's scope). |
| 244 | |
| 245 | Fixes the Windows PATHEXT mismatch: ``shutil.which("npx")`` resolves |
| 246 | ``npx.CMD`` via PATHEXT, but ``subprocess.run`` given the bare string |
| 247 | ``"npx"`` as argv[0] does not do that resolution and fails with |
| 248 | ``WinError 2``. Passing the resolved path is a no-op on macOS/Linux, where |
| 249 | ``shutil.which`` already returns the exact path ``CreateProcess``/``execve`` |
| 250 | would resolve. |
| 251 | |
| 252 | Returns ``(action, stderr)``: ``action`` is ``"no_npx"``, |
| 253 | ``"install_failed"``, or ``""`` when the subprocess ran and returned |
| 254 | ``rc=0`` (in which case ``stderr`` carries any non-fatal stderr output for |
| 255 | the caller's own off-path logging). |
| 256 | """ |
| 257 | npx = shutil.which("npx") |
| 258 | if npx is None: |
| 259 | return "no_npx", "" |
| 260 | try: |
| 261 | proc = subprocess.run( |
| 262 | [npx, "-y", PRINTING_PRESS_NPM, "install", slug, "--cli-only"], |
| 263 | capture_output=True, text=True, timeout=DIGG_INSTALL_TIMEOUT, |
| 264 | ) |
| 265 | except Exception as exc: |
| 266 | logger.warning("npx install %s exception: %s", slug, exc) |
| 267 | return "install_failed", str(exc) |
| 268 | if proc.returncode != 0: |
| 269 | stderr = proc.stderr or f"npx install {slug} exited {proc.returncode}" |
| 270 | logger.warning("npx install %s failed (rc=%s): %s", slug, proc.returncode, stderr) |
| 271 | return "install_failed", stderr |
| 272 | return "", (proc.stderr or "") |
| 273 | |
| 274 | |
| 275 | def _install_digg_cli() -> Tuple[bool, str, str, str]: |
| 276 | """Best-effort install of the digg-pp-cli binary. |
| 277 | |
| 278 | Mirrors the yt-dlp/brew auto-install: it never raises, and degrades to a |
| 279 | recommend-only outcome when the installer is unavailable. Uses |
| 280 | ``@mvanhorn/printing-press-library`` (``--cli-only``) — the same catalog |
| 281 | installer as pp-digg; Hermes/OpenClaw skill wiring is irrelevant here. |
| 282 | |
| 283 | Returns ``(engine_active, action, stderr, off_path_binary)`` where |
| 284 | ``engine_active`` is True only when ``shutil.which`` resolves the binary |
| 285 | (matching ``pipeline.available_sources()``). ``action`` is one of: |
| 286 | already_installed | installed | installed_off_path | install_failed | no_npx |
| 287 | ``stderr`` is populated on ``install_failed``. ``off_path_binary`` is set |
| 288 | when the binary exists on disk but is not PATH-visible to this process. |
| 289 | """ |
| 290 | on_path = _digg_on_path() |
| 291 | if on_path: |
| 292 | return True, "already_installed", "", "" |
| 293 | off_path = _digg_off_path_binary() |
| 294 | if off_path: |
| 295 | return False, "installed_off_path", "", off_path |
| 296 | action, stderr = _run_npx_install("digg") |
| 297 | if action: |
| 298 | return False, action, stderr, "" |
| 299 | on_path = _digg_on_path() |
| 300 | if on_path: |
| 301 | return True, "installed", "", "" |
| 302 | off_path = _digg_off_path_binary() |
| 303 | if off_path: |
| 304 | combined = stderr.strip() |
| 305 | if combined: |
| 306 | logger.warning("digg-pp-cli installed off PATH: %s", combined) |
| 307 | return False, "installed_off_path", combined, off_path |
| 308 | stderr_msg = stderr or "install completed but digg-pp-cli was not found" |
| 309 | logger.warning("npx install digg failed verification: %s", stderr_msg) |
| 310 | return False, "install_failed", stderr_msg, "" |
| 311 | |
| 312 | |
| 313 | # Additional default-on Printing Press sources installed the same way as Digg: |
| 314 | # (engine source key, slug for `install <slug>`, binary name). These activate in |
| 315 | # ``pipeline.available_sources()`` when ``shutil.which`` resolves the binary. |
| 316 | # Trustpilot is intentionally NOT here: it is opt-in (INCLUDE_SOURCES=trustpilot) |
| 317 | # because of its headless-Chrome cookie harvest, so auto-installing its binary |
| 318 | # for a source that stays off by default would be wasted work. Opting in installs |
| 319 | # it on demand via `npx ... install trustpilot --cli-only` (see CONFIGURATION.md). |
| 320 | PP_DEFAULT_SOURCES: list[tuple[str, str, str]] = [ |
| 321 | ("arxiv", "arxiv", "arxiv-pp-cli"), |
| 322 | ("techmeme", "techmeme", "techmeme-pp-cli"), |
| 323 | ] |
| 324 | |
| 325 | |
| 326 | def _pp_bin_candidate_paths(bin_name: str) -> list[Path]: |
| 327 | """Known install locations for a Printing Press CLI binary (slug-parameterized |
| 328 | mirror of ``_digg_bin_candidate_paths``).""" |
| 329 | home = Path.home() |
| 330 | candidates: list[Path] = [home / ".local" / "bin" / bin_name] |
| 331 | gopath = os.environ.get("GOPATH") |
| 332 | if gopath: |
| 333 | candidates.append(Path(gopath) / "bin" / bin_name) |
| 334 | candidates.append(home / "go" / "bin" / bin_name) |
| 335 | if os.name == "nt": |
| 336 | local_app = os.environ.get("LOCALAPPDATA") or os.environ.get("LocalAppData") |
| 337 | if local_app: |
| 338 | candidates.append( |
| 339 | Path(local_app) / "Programs" / "PrintingPress" / "bin" / f"{bin_name}.exe" |
| 340 | ) |
| 341 | return candidates |
| 342 | |
| 343 | |
| 344 | def _pp_off_path_binary(bin_name: str) -> Optional[str]: |
| 345 | for candidate in _pp_bin_candidate_paths(bin_name): |
| 346 | if candidate.is_file() and os.access(candidate, os.X_OK): |
| 347 | return str(candidate) |
| 348 | return None |
| 349 | |
| 350 | |
| 351 | def _install_pp_cli(slug: str, bin_name: str) -> Tuple[bool, str, str, str]: |
| 352 | """Best-effort install of a Printing Press CLI binary. |
| 353 | |
| 354 | Slug-parameterized mirror of ``_install_digg_cli``: never raises, degrades |
| 355 | to recommend-only when the installer is unavailable. Returns |
| 356 | ``(engine_active, action, stderr, off_path_binary)`` with the same action |
| 357 | taxonomy: already_installed | installed | installed_off_path | |
| 358 | install_failed | no_npx. |
| 359 | """ |
| 360 | on_path = shutil.which(bin_name) |
| 361 | if on_path: |
| 362 | return True, "already_installed", "", "" |
| 363 | off_path = _pp_off_path_binary(bin_name) |
| 364 | if off_path: |
| 365 | return False, "installed_off_path", "", off_path |
| 366 | action, stderr = _run_npx_install(slug) |
| 367 | if action: |
| 368 | return False, action, stderr, "" |
| 369 | on_path = shutil.which(bin_name) |
| 370 | if on_path: |
| 371 | return True, "installed", "", "" |
| 372 | off_path = _pp_off_path_binary(bin_name) |
| 373 | if off_path: |
| 374 | combined = stderr.strip() |
| 375 | if combined: |
| 376 | logger.warning("%s installed off PATH: %s", bin_name, combined) |
| 377 | return False, "installed_off_path", combined, off_path |
| 378 | stderr_msg = stderr or f"install completed but {bin_name} was not found" |
| 379 | logger.warning("npx install %s failed verification: %s", slug, stderr_msg) |
| 380 | return False, "install_failed", stderr_msg, "" |
| 381 | |
| 382 | |
| 383 | def install_default_pp_sources() -> Dict[str, Dict[str, Any]]: |
| 384 | """Best-effort install of every additional default-on Printing Press source. |
| 385 | |
| 386 | Returns ``{source_key: {installed, action, stderr?, path?}}`` so the wizard |
| 387 | can report per-CLI status alongside Digg without raising on any single |
| 388 | failure. |
| 389 | """ |
| 390 | out: Dict[str, Dict[str, Any]] = {} |
| 391 | for source_key, slug, bin_name in PP_DEFAULT_SOURCES: |
| 392 | installed, action, stderr, off_path = _install_pp_cli(slug, bin_name) |
| 393 | entry: Dict[str, Any] = {"installed": installed, "action": action} |
| 394 | if action == "install_failed" and stderr: |
| 395 | entry["stderr"] = stderr |
| 396 | if off_path: |
| 397 | entry["path"] = off_path |
| 398 | out[source_key] = entry |
| 399 | return out |
| 400 | |
| 401 | |
| 402 | def _open_secret_append(path: Path): |
| 403 | """Open ``path`` for appending as a 0o600 secret file with no readable window. |
| 404 | |
| 405 | ``os.open`` with ``O_CREAT|O_WRONLY|O_APPEND`` and mode ``0o600`` sets |
| 406 | restrictive permissions at creation (umask can only further restrict, never |
| 407 | widen, so the file is never world-readable even transiently). An explicit |
| 408 | ``chmod`` afterwards also tightens a pre-existing loose file. This matters |
| 409 | because the .env stores API keys, cookies, and tokens. |
| 410 | """ |
| 411 | fd = os.open(path, os.O_CREAT | os.O_WRONLY | os.O_APPEND, 0o600) |
| 412 | try: |
| 413 | os.chmod(path, 0o600) |
| 414 | except OSError: |
| 415 | pass |
| 416 | return os.fdopen(fd, "a", encoding="utf-8") |
| 417 | |
| 418 | |
| 419 | def _format_env_value(value: str) -> str: |
| 420 | """Quote a value so it round-trips through env.load_env_file. |
| 421 | |
| 422 | env.load_env_file strips a single layer of matching surrounding quotes but |
| 423 | does NOT process backslash escapes, so we wrap (never escape): |
| 424 | - plain tokens (no whitespace, no leading quote): returned unchanged; |
| 425 | - values with whitespace/leading quote and no double-quote: double-quoted; |
| 426 | - values containing a double-quote but no single-quote: single-quoted; |
| 427 | - values containing both quote types: returned as-is (best effort; no |
| 428 | wrapper round-trips through the loader, and tokens never hit this). |
| 429 | Newlines are not valid in a single-line env value and are stripped. |
| 430 | """ |
| 431 | value = value.replace("\r", "").replace("\n", " ") |
| 432 | needs_quoting = (not value) or value[0] in ("'", '"') or any(c.isspace() for c in value) |
| 433 | if not needs_quoting: |
| 434 | return value |
| 435 | if '"' not in value: |
| 436 | return f'"{value}"' |
| 437 | if "'" not in value: |
| 438 | return f"'{value}'" |
| 439 | return value |
| 440 | |
| 441 | |
| 442 | def write_setup_config(env_path: Path, from_browser: str | None = None) -> bool: |
| 443 | """Write SETUP_COMPLETE and FROM_BROWSER to the .env file. |
| 444 | |
| 445 | Creates the file and parent directories if needed. |
| 446 | Appends to existing file without overwriting existing keys. |
| 447 | |
| 448 | Args: |
| 449 | env_path: Path to the .env file (e.g. ~/.config/last30days/.env) |
| 450 | from_browser: Browser extraction mode to persist. Pass the browser that |
| 451 | actually yielded cookies (e.g. "firefox") to fast-path future runs. |
| 452 | Pass None (default) to NOT pin FROM_BROWSER — the steady-state |
| 453 | default (Firefox/Safari, no Keychain prompt) then applies. We avoid |
| 454 | persisting "auto" because it makes every later run probe Chrome and |
| 455 | re-trigger the Keychain prompt. |
| 456 | |
| 457 | Returns: |
| 458 | True if config was written successfully, False on error. |
| 459 | """ |
| 460 | try: |
| 461 | env_path = Path(env_path) |
| 462 | env_path.parent.mkdir(parents=True, exist_ok=True) |
| 463 | |
| 464 | # Read existing content to avoid overwriting keys |
| 465 | existing_keys: set = set() |
| 466 | existing_content = "" |
| 467 | if env_path.exists(): |
| 468 | existing_content = env_path.read_text(encoding="utf-8") |
| 469 | for line in existing_content.splitlines(): |
| 470 | stripped = line.strip() |
| 471 | if stripped and not stripped.startswith("#") and "=" in stripped: |
| 472 | key = stripped.split("=", 1)[0].strip() |
| 473 | existing_keys.add(key) |
| 474 | |
| 475 | lines_to_add = [] |
| 476 | if "SETUP_COMPLETE" not in existing_keys: |
| 477 | lines_to_add.append("SETUP_COMPLETE=true") |
| 478 | if from_browser and "FROM_BROWSER" not in existing_keys: |
| 479 | lines_to_add.append(f"FROM_BROWSER={_format_env_value(from_browser)}") |
| 480 | |
| 481 | if not lines_to_add: |
| 482 | return True # Nothing to write, already configured |
| 483 | |
| 484 | # Create/append as a 0o600 secret file: the .env holds tokens and keys, |
| 485 | # so it must never be created world-readable. |
| 486 | with _open_secret_append(env_path) as f: |
| 487 | if existing_content and not existing_content.endswith("\n"): |
| 488 | f.write("\n") |
| 489 | f.write("\n".join(lines_to_add) + "\n") |
| 490 | |
| 491 | return True |
| 492 | |
| 493 | except OSError as exc: |
| 494 | logger.error("Failed to write setup config to %s: %s", env_path, exc) |
| 495 | return False |
| 496 | |
| 497 | |
| 498 | def write_api_key(env_path: Path, api_key: str, key_name: str = "SCRAPECREATORS_API_KEY") -> bool: |
| 499 | """Append an API key to the .env file as a 0o600 secret. |
| 500 | |
| 501 | Reuses the same secret-safe write path as ``write_setup_config`` so the |
| 502 | value lands with restrictive permissions and round-trips through |
| 503 | ``env.load_env_file``. Idempotent: if ``key_name`` is already present in |
| 504 | the file, nothing is written and the existing value is preserved (we never |
| 505 | clobber a key the user may have set by hand). |
| 506 | |
| 507 | Args: |
| 508 | env_path: Path to the .env file (e.g. ~/.config/last30days/.env). |
| 509 | api_key: The raw key value to persist. |
| 510 | key_name: The env var name to write (default SCRAPECREATORS_API_KEY). |
| 511 | |
| 512 | Returns: |
| 513 | True if the key was written or already present, False on error or when |
| 514 | ``api_key`` is empty. |
| 515 | """ |
| 516 | if not api_key: |
| 517 | return False |
| 518 | try: |
| 519 | env_path = Path(env_path) |
| 520 | env_path.parent.mkdir(parents=True, exist_ok=True) |
| 521 | |
| 522 | existing_content = "" |
| 523 | if env_path.exists(): |
| 524 | existing_content = env_path.read_text(encoding="utf-8") |
| 525 | for line in existing_content.splitlines(): |
| 526 | stripped = line.strip() |
| 527 | if stripped and not stripped.startswith("#") and "=" in stripped: |
| 528 | if stripped.split("=", 1)[0].strip() == key_name: |
| 529 | return True # Already configured; do not duplicate |
| 530 | |
| 531 | line = f"{key_name}={_format_env_value(api_key)}\n" |
| 532 | with _open_secret_append(env_path) as f: |
| 533 | if existing_content and not existing_content.endswith("\n"): |
| 534 | f.write("\n") |
| 535 | f.write(line) |
| 536 | |
| 537 | return True |
| 538 | |
| 539 | except OSError as exc: |
| 540 | logger.error("Failed to write API key to %s: %s", env_path, exc) |
| 541 | return False |
| 542 | |
| 543 | |
| 544 | def mask_api_key(api_key: str) -> str: |
| 545 | """Return a non-secret display form of an API key (prefix + last 4). |
| 546 | |
| 547 | Used so the key never appears verbatim in stdout the host model captures. |
| 548 | Short or empty keys collapse to a fixed placeholder. |
| 549 | """ |
| 550 | if not api_key or len(api_key) <= 8: |
| 551 | return "sc_…" |
| 552 | return f"{api_key[:3]}…{api_key[-4:]}" |
| 553 | |
| 554 | |
| 555 | def get_setup_status_text(results: Dict[str, Any]) -> str: |
| 556 | """Return a human-readable summary of auto-setup results. |
| 557 | |
| 558 | Args: |
| 559 | results: Dict from run_auto_setup() |
| 560 | |
| 561 | Returns: |
| 562 | Multi-line status text. |
| 563 | """ |
| 564 | lines = [] |
| 565 | lines.append("Setup complete! Here's what I found:") |
| 566 | lines.append("") |
| 567 | |
| 568 | cookies_found = results.get("cookies_found", {}) |
| 569 | if cookies_found: |
| 570 | for source, browser in cookies_found.items(): |
| 571 | lines.append(f" - {source.upper()} cookies found in {browser}") |
| 572 | else: |
| 573 | lines.append(" - No browser cookies found for X/Twitter") |
| 574 | |
| 575 | ytdlp_action = results.get("ytdlp_action", "") |
| 576 | if ytdlp_action == "installed": |
| 577 | lines.append(" - Installed yt-dlp via Homebrew") |
| 578 | elif ytdlp_action == "install_failed": |
| 579 | lines.append(" - yt-dlp install failed \u2014 run `brew install yt-dlp` manually") |
| 580 | elif ytdlp_action == "no_homebrew": |
| 581 | lines.append(" - yt-dlp not found. Install Homebrew first, then: brew install yt-dlp") |
| 582 | elif ytdlp_action == "no_pip_windows": |
| 583 | lines.append( |
| 584 | " - yt-dlp not found. Install with: pip install yt-dlp " |
| 585 | "(it may install to a Scripts directory not on PATH -- add it to PATH if YouTube search stays inactive)" |
| 586 | ) |
| 587 | elif ytdlp_action == "already_installed": |
| 588 | lines.append(" - yt-dlp already installed") |
| 589 | elif results.get("ytdlp_installed", False): |
| 590 | lines.append(" - yt-dlp is installed (YouTube search ready)") |
| 591 | else: |
| 592 | lines.append(" - yt-dlp not found (install with: brew install yt-dlp)") |
| 593 | |
| 594 | digg_action = results.get("digg_action", "") |
| 595 | if digg_action == "installed": |
| 596 | lines.append(" - Installed Digg CLI (free AI-news clusters source now active)") |
| 597 | elif digg_action == "already_installed": |
| 598 | lines.append(" - Digg CLI already installed (AI-news clusters active)") |
| 599 | elif digg_action == "installed_off_path": |
| 600 | digg_path = results.get("digg_path", "") |
| 601 | if digg_path: |
| 602 | bin_dir = _digg_bin_dir_hint(digg_path) |
| 603 | lines.append( |
| 604 | f" - Digg CLI found at {digg_path} but not on PATH — add " |
| 605 | f"{bin_dir} to PATH and restart your agent session/gateway " |
| 606 | "for Digg to activate" |
| 607 | ) |
| 608 | else: |
| 609 | lines.append( |
| 610 | " - Digg CLI is installed but not on PATH — add its install " |
| 611 | "directory to PATH and restart your agent session/gateway for " |
| 612 | "Digg to activate" |
| 613 | ) |
| 614 | elif digg_action == "install_failed": |
| 615 | lines.append(f" - Digg CLI install failed — run `{DIGG_INSTALL_CMD}` manually") |
| 616 | elif digg_action == "no_npx": |
| 617 | lines.append( |
| 618 | " - Digg CLI not installed (free, optional). Install Node/npx, then: " |
| 619 | f"{DIGG_INSTALL_CMD}" |
| 620 | ) |
| 621 | |
| 622 | pp_sources = results.get("pp_sources", {}) |
| 623 | pp_name: dict[str, str] = {"arxiv": "arXiv", "techmeme": "Techmeme"} |
| 624 | for source_key, entry in sorted(pp_sources.items()): |
| 625 | name = pp_name.get(source_key, source_key.title()) |
| 626 | action = entry.get("action", "") |
| 627 | if action == "installed": |
| 628 | lines.append(f" - Installed {name} CLI ({name} source now active)") |
| 629 | elif action == "already_installed": |
| 630 | lines.append(f" - {name} CLI already installed ({name} active)") |
| 631 | elif action == "installed_off_path": |
| 632 | path = entry.get("path", "") |
| 633 | if path: |
| 634 | lines.append( |
| 635 | f" - {name} CLI at {path} but not on PATH — add " |
| 636 | f"{os.path.dirname(os.path.expanduser(path))} to PATH and " |
| 637 | f"restart your agent session/gateway for {name} to activate" |
| 638 | ) |
| 639 | else: |
| 640 | lines.append( |
| 641 | f" - {name} CLI installed but not on PATH — add its install " |
| 642 | "directory to PATH and restart your agent session/gateway for " |
| 643 | f"{name} to activate" |
| 644 | ) |
| 645 | elif action == "install_failed": |
| 646 | lines.append( |
| 647 | f" - {name} CLI install failed — run " |
| 648 | f"`npx -y {PRINTING_PRESS_NPM} install {source_key} --cli-only` manually" |
| 649 | ) |
| 650 | elif action == "no_npx": |
| 651 | lines.append( |
| 652 | f" - {name} CLI not installed (free, optional). Install Node/npx, " |
| 653 | f"then: `npx -y {PRINTING_PRESS_NPM} install {source_key} --cli-only`" |
| 654 | ) |
| 655 | |
| 656 | env_written = results.get("env_written", False) |
| 657 | if env_written: |
| 658 | lines.append("") |
| 659 | lines.append("Configuration saved. Future runs will auto-detect your browsers.") |
| 660 | |
| 661 | return "\n".join(lines) |
| 662 | |
| 663 | |
| 664 | # --------------------------------------------------------------------------- |
| 665 | # OpenClaw server-side setup (no browser, JSON output) |
| 666 | # --------------------------------------------------------------------------- |
| 667 | |
| 668 | _OPENCLAW_KEY_NAMES = [ |
| 669 | "SCRAPECREATORS_API_KEY", |
| 670 | "XAI_API_KEY", |
| 671 | "BRAVE_API_KEY", |
| 672 | "EXA_API_KEY", |
| 673 | "SERPER_API_KEY", |
| 674 | "OPENAI_API_KEY", |
| 675 | "AUTH_TOKEN", |
| 676 | ] |
| 677 | |
| 678 | |
| 679 | def run_openclaw_setup(config: Dict[str, Any]) -> Dict[str, Any]: |
| 680 | """Server-side setup probe: no cookies, tool + key availability, Digg CLI. |
| 681 | |
| 682 | Best-effort installs digg-pp-cli when npx is available (same as desktop |
| 683 | ``run_auto_setup``). Returns a dict suitable for JSON output to stdout so |
| 684 | that SKILL.md can present appropriate options to the user. |
| 685 | """ |
| 686 | yt_dlp = shutil.which("yt-dlp") is not None |
| 687 | node = shutil.which("node") is not None |
| 688 | python3 = shutil.which("python3") is not None |
| 689 | |
| 690 | digg_installed, digg_action, digg_stderr, digg_path = _install_digg_cli() |
| 691 | |
| 692 | keys: Dict[str, bool] = {} |
| 693 | for key_name in _OPENCLAW_KEY_NAMES: |
| 694 | short = key_name.lower().replace("_api_key", "").replace("_key", "").replace("_token", "") |
| 695 | # Normalize: AUTH_TOKEN -> auth, SCRAPECREATORS_API_KEY -> scrapecreators |
| 696 | keys[short] = bool(config.get(key_name)) |
| 697 | |
| 698 | # Determine x_method |
| 699 | if config.get("XAI_API_KEY"): |
| 700 | x_method: Optional[str] = "xai" |
| 701 | elif config.get("AUTH_TOKEN") and config.get("CT0"): |
| 702 | x_method = "cookies" |
| 703 | else: |
| 704 | x_method = None |
| 705 | |
| 706 | payload: Dict[str, Any] = { |
| 707 | "yt_dlp": yt_dlp, |
| 708 | "node": node, |
| 709 | "python3": python3, |
| 710 | "digg_cli": digg_installed, |
| 711 | "digg_action": digg_action, |
| 712 | "keys": keys, |
| 713 | "x_method": x_method, |
| 714 | } |
| 715 | if digg_path: |
| 716 | payload["digg_path"] = digg_path |
| 717 | if digg_action == "install_failed" and digg_stderr: |
| 718 | payload["digg_stderr"] = digg_stderr |
| 719 | return payload |
| 720 | |
| 721 | |
| 722 | # --------------------------------------------------------------------------- |
| 723 | # Device auth flow (GitHub OAuth via ScrapeCreators) |
| 724 | # --------------------------------------------------------------------------- |
| 725 | |
| 726 | _DEVICE_BASE = "https://api.scrapecreators.com/v1/github/device" |
| 727 | |
| 728 | # A GitHub device code is always XXXX-XXXX (uppercase alphanumerics). We validate |
| 729 | # user_code against this before copying, labeling, or emitting it so a malformed |
| 730 | # or key-shaped value (e.g. a returning-account server response) is never |
| 731 | # mislabeled as a device code or leaked to stdout/clipboard. |
| 732 | _DEVICE_CODE_RE = re.compile(r"^[0-9A-Z]{4}-[0-9A-Z]{4}$") |
| 733 | |
| 734 | |
| 735 | def _existing_scrapecreators_key() -> Optional[str]: |
| 736 | """Return the SCRAPECREATORS_API_KEY already saved in the .env, if any.""" |
| 737 | try: |
| 738 | from . import env as _env |
| 739 | |
| 740 | if _env.CONFIG_FILE and _env.CONFIG_FILE.exists(): |
| 741 | return _env.load_env_file(_env.CONFIG_FILE).get("SCRAPECREATORS_API_KEY") or None |
| 742 | except Exception as exc: # never let a config-read failure block auth |
| 743 | logger.debug("Could not read existing ScrapeCreators key: %s", exc) |
| 744 | return None |
| 745 | |
| 746 | |
| 747 | def run_device_auth() -> Optional[Tuple[str, str, str, int]]: |
| 748 | """Start the device authorization flow. |
| 749 | |
| 750 | POSTs to the ScrapeCreators device/code endpoint. |
| 751 | |
| 752 | Returns: |
| 753 | (device_code, user_code, verification_uri, interval) on success, |
| 754 | None on failure. |
| 755 | """ |
| 756 | try: |
| 757 | body = json.dumps({}).encode() |
| 758 | req = Request(f"{_DEVICE_BASE}/code", data=body, method="POST") |
| 759 | req.add_header("Content-Type", "application/json") |
| 760 | with urlopen(req, timeout=15) as resp: |
| 761 | data = json.loads(resp.read()) |
| 762 | except (HTTPError, URLError, OSError) as exc: |
| 763 | logger.warning("Device auth code request failed: %s", exc) |
| 764 | return None |
| 765 | |
| 766 | device_code = data.get("device_code") |
| 767 | user_code = data.get("user_code") |
| 768 | verification_uri = data.get("verification_uri") |
| 769 | interval = data.get("interval", 5) |
| 770 | |
| 771 | if not device_code or not user_code: |
| 772 | # Log only the response's key names, never its values — a returning |
| 773 | # account's response could carry a raw API key we must not write to logs. |
| 774 | logger.warning( |
| 775 | "Device auth returned incomplete response (keys: %s)", sorted(data.keys()) |
| 776 | ) |
| 777 | return None |
| 778 | |
| 779 | return (device_code, user_code, verification_uri or "", interval) |
| 780 | |
| 781 | |
| 782 | def poll_device_auth( |
| 783 | device_code: str, |
| 784 | interval: int, |
| 785 | timeout: int = 300, |
| 786 | user_code: str = "", |
| 787 | clipboard_ok: bool = False, |
| 788 | ) -> Optional[str]: |
| 789 | """Poll for an access token after the user authorizes the device. |
| 790 | |
| 791 | Args: |
| 792 | device_code: The device_code from run_device_auth(). |
| 793 | interval: Polling interval in seconds. |
| 794 | timeout: Maximum time to poll in seconds. |
| 795 | user_code: The user code to remind about during polling. |
| 796 | clipboard_ok: Whether the code was copied to clipboard. |
| 797 | |
| 798 | Returns: |
| 799 | access_token on success, None on timeout or failure. |
| 800 | """ |
| 801 | import sys |
| 802 | |
| 803 | started_at = time.time() |
| 804 | deadline = started_at + timeout |
| 805 | last_reminder = started_at |
| 806 | reminder_count = 0 |
| 807 | max_reminders = 4 |
| 808 | reminder_interval = 30 # seconds between reminders |
| 809 | |
| 810 | while time.time() < deadline: |
| 811 | time.sleep(interval) |
| 812 | |
| 813 | # Periodic reminder of the code while waiting |
| 814 | if ( |
| 815 | user_code |
| 816 | and reminder_count < max_reminders |
| 817 | and time.time() - last_reminder >= reminder_interval |
| 818 | ): |
| 819 | clipboard_hint = " (on your clipboard)" if clipboard_ok else "" |
| 820 | print( |
| 821 | f" Still waiting... Your code: {user_code}{clipboard_hint}", |
| 822 | file=sys.stderr, |
| 823 | flush=True, |
| 824 | ) |
| 825 | last_reminder = time.time() |
| 826 | reminder_count += 1 |
| 827 | |
| 828 | try: |
| 829 | body = json.dumps({"device_code": device_code}).encode() |
| 830 | req = Request(f"{_DEVICE_BASE}/token", data=body, method="POST") |
| 831 | req.add_header("Content-Type", "application/json") |
| 832 | with urlopen(req, timeout=15) as resp: |
| 833 | data = json.loads(resp.read()) |
| 834 | except HTTPError as exc: |
| 835 | if exc.code in (400, 403, 428): |
| 836 | continue |
| 837 | logger.warning("Device auth poll error: %s", exc) |
| 838 | return None |
| 839 | except (URLError, OSError): |
| 840 | continue |
| 841 | |
| 842 | if data.get("access_token"): |
| 843 | return data["access_token"] |
| 844 | |
| 845 | error = data.get("error") |
| 846 | if error == "slow_down": |
| 847 | interval = min(interval + 2, 30) |
| 848 | continue |
| 849 | if error == "authorization_pending": |
| 850 | continue |
| 851 | if error in ("expired_token", "access_denied"): |
| 852 | logger.warning("Device auth failed: %s", error) |
| 853 | return None |
| 854 | |
| 855 | return None |
| 856 | |
| 857 | |
| 858 | def fetch_api_key(access_token: str) -> Optional[str]: |
| 859 | """Fetch the ScrapeCreators API key using the GitHub access token. |
| 860 | |
| 861 | GETs the device/profile endpoint with Bearer auth. |
| 862 | |
| 863 | Returns: |
| 864 | api_key string on success, None on failure. |
| 865 | """ |
| 866 | try: |
| 867 | req = Request(f"{_DEVICE_BASE}/profile") |
| 868 | req.add_header("Authorization", f"Bearer {access_token}") |
| 869 | with urlopen(req, timeout=15) as resp: |
| 870 | data = json.loads(resp.read()) |
| 871 | except (HTTPError, URLError, OSError) as exc: |
| 872 | logger.warning("Failed to fetch API key: %s", exc) |
| 873 | return None |
| 874 | |
| 875 | api_key = data.get("api_key") |
| 876 | if not api_key: |
| 877 | # The /profile response parsed but carried no api_key — the common case |
| 878 | # for a GitHub account already linked to a ScrapeCreators account. Log |
| 879 | # the response's FIELD NAMES only (never values — the body may contain a |
| 880 | # key under a different field) so the already-registered response shape |
| 881 | # can be handled in a follow-up (see plan OQ1). |
| 882 | logger.warning( |
| 883 | "Device auth /profile returned no api_key (fields: %s)", sorted(data.keys()) |
| 884 | ) |
| 885 | return None |
| 886 | return api_key |
| 887 | |
| 888 | |
| 889 | def _device_handle_path() -> Path: |
| 890 | """Where run_github_start persists the device_code/interval for run_github_poll. |
| 891 | |
| 892 | Kept next to the .env in the config dir; falls back to the OS temp dir when |
| 893 | no config dir is resolvable (clean/no-config mode). |
| 894 | """ |
| 895 | try: |
| 896 | from . import env as _env |
| 897 | |
| 898 | if _env.CONFIG_FILE: |
| 899 | return _env.CONFIG_FILE.parent / ".github-device-handle.json" |
| 900 | except Exception: |
| 901 | pass |
| 902 | import tempfile |
| 903 | |
| 904 | return Path(tempfile.gettempdir()) / "last30days-github-device-handle.json" |
| 905 | |
| 906 | |
| 907 | def _start_device_flow() -> "Tuple[Dict[str, Any], Optional[Dict[str, Any]]]": |
| 908 | """Submit the GitHub device flow and surface the code, without polling. |
| 909 | |
| 910 | Returns ``(public_result, handle)``. ``handle`` is None for the |
| 911 | already-registered and error cases (nothing to poll); otherwise it carries |
| 912 | the private poll state (``device_code``/``interval``/``user_code``/ |
| 913 | ``clipboard_ok``) that never belongs in the public, stdout-printed result. |
| 914 | Callers either persist the handle to a file (``run_github_start``, for a |
| 915 | separate poll process) or hand it straight to ``run_github_poll`` in-memory |
| 916 | (``run_full_device_auth``, so a failed file write can't strand the one-shot). |
| 917 | """ |
| 918 | import sys |
| 919 | import webbrowser |
| 920 | |
| 921 | # Already-registered short-circuit: a saved key means no device dance. The |
| 922 | # key is returned raw here and masked at the CLI boundary before print. |
| 923 | existing = _existing_scrapecreators_key() |
| 924 | if existing: |
| 925 | return ( |
| 926 | { |
| 927 | "status": "already_registered", |
| 928 | "method": "existing", |
| 929 | "api_key": existing, |
| 930 | "persisted": True, |
| 931 | }, |
| 932 | None, |
| 933 | ) |
| 934 | |
| 935 | result = run_device_auth() |
| 936 | if result is None: |
| 937 | return ({"status": "error", "message": "Failed to start device auth flow"}, None) |
| 938 | |
| 939 | device_code, user_code, verification_uri, interval = result |
| 940 | |
| 941 | # Validate the code shape BEFORE copying, labeling, or emitting it. A |
| 942 | # non-conforming user_code (e.g. a key-shaped value) is never surfaced as a |
| 943 | # GitHub device code; we stop rather than instruct the user to paste garbage. |
| 944 | if not _DEVICE_CODE_RE.match(user_code): |
| 945 | logger.warning("Device auth returned a non-device-shaped user_code; aborting.") |
| 946 | return ( |
| 947 | { |
| 948 | "status": "error", |
| 949 | "message": "ScrapeCreators returned an unexpected device-code format.", |
| 950 | }, |
| 951 | None, |
| 952 | ) |
| 953 | |
| 954 | # Structured stdout line for machine consumers. |
| 955 | print( |
| 956 | json.dumps( |
| 957 | { |
| 958 | "event": "device_code_ready", |
| 959 | "user_code": user_code, |
| 960 | "verification_uri": verification_uri, |
| 961 | } |
| 962 | ), |
| 963 | flush=True, |
| 964 | ) |
| 965 | |
| 966 | # Copy the code to the clipboard BEFORE opening the browser. |
| 967 | clipboard_ok = False |
| 968 | if sys.platform == "darwin": |
| 969 | try: |
| 970 | subprocess.run(["pbcopy"], input=user_code.encode(), check=True, timeout=5) |
| 971 | clipboard_ok = True |
| 972 | except Exception: |
| 973 | pass # pbcopy unavailable or failed, fall through |
| 974 | |
| 975 | # Print the code as a plain HUMAN line on stdout too, so a foreground caller |
| 976 | # sees it in the returned output even without reading the JSON. The clipboard |
| 977 | # claim is only made when pbcopy actually succeeded (else: type it). |
| 978 | if clipboard_ok: |
| 979 | print( |
| 980 | f"Your GitHub code: {user_code} (already on your clipboard - just paste it, Cmd+V)", |
| 981 | flush=True, |
| 982 | ) |
| 983 | else: |
| 984 | print(f"Your GitHub code: {user_code} (type it on the GitHub page)", flush=True) |
| 985 | |
| 986 | # Human box on stderr for direct-terminal users. |
| 987 | clipboard_hint = " (copied to clipboard)" if clipboard_ok else "" |
| 988 | code_line = f" Your code: {user_code}{clipboard_hint}" |
| 989 | action_line = " Paste it on the GitHub page that just opened" |
| 990 | width = max(len(code_line), len(action_line)) + 2 |
| 991 | border = "-" * width |
| 992 | print(f"\n+{border}+", file=sys.stderr) |
| 993 | print(f"|{code_line.ljust(width)}|", file=sys.stderr) |
| 994 | print(f"|{action_line.ljust(width)}|", file=sys.stderr) |
| 995 | print(f"+{border}+", file=sys.stderr) |
| 996 | |
| 997 | if verification_uri: |
| 998 | try: |
| 999 | webbrowser.open(verification_uri) |
| 1000 | except Exception: |
| 1001 | print(f"Open: {verification_uri}", file=sys.stderr) |
| 1002 | |
| 1003 | public = { |
| 1004 | "status": "awaiting_authorization", |
| 1005 | "user_code": user_code, |
| 1006 | "verification_uri": verification_uri, |
| 1007 | "clipboard_ok": clipboard_ok, |
| 1008 | } |
| 1009 | handle = { |
| 1010 | "device_code": device_code, |
| 1011 | "interval": interval, |
| 1012 | "user_code": user_code, |
| 1013 | "clipboard_ok": clipboard_ok, |
| 1014 | } |
| 1015 | return (public, handle) |
| 1016 | |
| 1017 | |
| 1018 | def run_github_start() -> Dict[str, Any]: |
| 1019 | """Start the device flow and persist the poll handle for a later |
| 1020 | ``run_github_poll`` process. Returns the public result (never the private |
| 1021 | device_code). See ``_start_device_flow`` for the returned statuses.""" |
| 1022 | public, handle = _start_device_flow() |
| 1023 | if handle is not None: |
| 1024 | # Persist the poll handle (0o600) so a separate --github-poll process can |
| 1025 | # resume it. Best-effort: the in-memory one-shot path does not depend on |
| 1026 | # this write succeeding. |
| 1027 | path = _device_handle_path() |
| 1028 | try: |
| 1029 | path.parent.mkdir(parents=True, exist_ok=True) |
| 1030 | path.write_text(json.dumps(handle), encoding="utf-8") |
| 1031 | os.chmod(path, 0o600) |
| 1032 | except Exception as exc: |
| 1033 | logger.warning("Could not persist device handle: %s", exc) |
| 1034 | return public |
| 1035 | |
| 1036 | |
| 1037 | def run_github_poll(timeout: int = 300, *, _handle: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: |
| 1038 | """Poll for authorization using the handle from start. |
| 1039 | |
| 1040 | ``_handle`` (in-memory, from the one-shot) takes precedence over the |
| 1041 | persisted handle file. Returns success (with the fetched key), timeout, or |
| 1042 | the honest "Authorized but failed to fetch API key" branch. Deletes the |
| 1043 | persisted handle when the flow terminates. |
| 1044 | """ |
| 1045 | import sys |
| 1046 | |
| 1047 | if _handle is not None: |
| 1048 | data = _handle |
| 1049 | else: |
| 1050 | try: |
| 1051 | data = json.loads(_device_handle_path().read_text(encoding="utf-8")) |
| 1052 | except Exception: |
| 1053 | return { |
| 1054 | "status": "error", |
| 1055 | "message": "No pending GitHub device flow; run setup --github-start first.", |
| 1056 | } |
| 1057 | |
| 1058 | device_code = data["device_code"] |
| 1059 | interval = int(data.get("interval", 5)) |
| 1060 | user_code = data.get("user_code", "") |
| 1061 | # Read the real clipboard state so the polling reminder never falsely claims |
| 1062 | # the code is on the clipboard (non-macOS, or a failed pbcopy). Missing key |
| 1063 | # (older handle) defaults to False -- don't overstate. |
| 1064 | clipboard_ok = bool(data.get("clipboard_ok", False)) |
| 1065 | |
| 1066 | print("Waiting for authorization...", file=sys.stderr, flush=True) |
| 1067 | access_token = poll_device_auth( |
| 1068 | device_code, interval, timeout=timeout, user_code=user_code, clipboard_ok=clipboard_ok |
| 1069 | ) |
| 1070 | |
| 1071 | def _cleanup() -> None: |
| 1072 | try: |
| 1073 | _device_handle_path().unlink() |
| 1074 | except Exception: |
| 1075 | pass |
| 1076 | |
| 1077 | if access_token is None: |
| 1078 | _cleanup() |
| 1079 | return {"status": "timeout", "user_code": user_code} |
| 1080 | |
| 1081 | api_key = fetch_api_key(access_token) |
| 1082 | _cleanup() |
| 1083 | if api_key is None: |
| 1084 | return { |
| 1085 | "status": "error", |
| 1086 | "message": "Authorized but failed to fetch API key", |
| 1087 | } |
| 1088 | |
| 1089 | return {"status": "success", "method": "device", "api_key": api_key, "user_code": user_code} |
| 1090 | |
| 1091 | |
| 1092 | def run_full_device_auth(timeout: int = 300) -> Dict[str, Any]: |
| 1093 | """Back-compat one-shot: start the device flow, then poll to completion. |
| 1094 | |
| 1095 | Passes the poll handle to ``run_github_poll`` IN MEMORY, so a failed handle- |
| 1096 | file write can't strand the one-shot. Kept so callers of ``setup --github`` / |
| 1097 | ``--device-auth`` still work; the model-driven wizard uses the two-command |
| 1098 | split (start then poll) instead. |
| 1099 | """ |
| 1100 | public, handle = _start_device_flow() |
| 1101 | if handle is None: |
| 1102 | return public # already_registered or error |
| 1103 | return run_github_poll(timeout=timeout, _handle=handle) |
| 1104 | |
| 1105 | |
| 1106 | # --------------------------------------------------------------------------- |
| 1107 | # Unified GitHub auth |
| 1108 | # --------------------------------------------------------------------------- |
| 1109 | |
| 1110 | |
| 1111 | def run_github_auth(timeout: int = 300) -> Dict[str, Any]: |
| 1112 | """Run the --github setup path via device auth (one-shot, back-compat). |
| 1113 | |
| 1114 | The existing-key short-circuit now lives in run_github_start; this delegates |
| 1115 | to the start+poll chain. This path must not read or forward local GitHub CLI |
| 1116 | tokens. |
| 1117 | """ |
| 1118 | return run_full_device_auth(timeout=timeout) |
| 1119 |