Dinakar Sarbada
537 lines
18 KiB
Python
537 lines
18 KiB
Python
"""First-run setup wizard for last30days.
|
|
|
|
Detects first run, performs auto-setup (cookie extraction + yt-dlp check),
|
|
and writes configuration. The actual wizard UI is SKILL.md-driven (the LLM
|
|
presents it), but this module provides the detection and setup actions.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import shutil
|
|
import subprocess
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Optional, Tuple
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.request import Request, urlopen
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def is_first_run(config: Dict[str, Any]) -> bool:
|
|
"""Return True if the setup wizard has not been completed.
|
|
|
|
Checks for SETUP_COMPLETE in the config dict. If it's not set
|
|
(None or empty string), the user hasn't gone through setup yet.
|
|
"""
|
|
return not config.get("SETUP_COMPLETE")
|
|
|
|
|
|
def run_auto_setup(config: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Perform the auto-setup actions.
|
|
|
|
- Runs cookie extraction in auto mode for all registered domains
|
|
- Checks if yt-dlp is installed
|
|
|
|
Returns:
|
|
Dict with keys:
|
|
cookies_found: {source_name: browser_name} for each source where cookies were found
|
|
ytdlp_installed: bool
|
|
env_written: bool (always False here — caller writes config separately)
|
|
"""
|
|
from . import cookie_extract
|
|
from .env import COOKIE_DOMAINS
|
|
|
|
cookies_found: Dict[str, str] = {}
|
|
|
|
for source_name, spec in COOKIE_DOMAINS.items():
|
|
domain = spec["domain"]
|
|
cookie_names = spec["cookies"]
|
|
|
|
try:
|
|
result = cookie_extract.extract_cookies_with_source("auto", domain, cookie_names)
|
|
except Exception as exc:
|
|
logger.debug("Cookie extraction failed for %s: %s", source_name, exc)
|
|
continue
|
|
|
|
if result is not None:
|
|
_cookies, browser_name = result
|
|
cookies_found[source_name] = browser_name
|
|
|
|
# Check yt-dlp availability and install via Homebrew if missing
|
|
ytdlp_action: str
|
|
if shutil.which("yt-dlp") is not None:
|
|
ytdlp_installed = True
|
|
ytdlp_action = "already_installed"
|
|
elif shutil.which("brew") is not None:
|
|
brew_stderr = ""
|
|
try:
|
|
proc = subprocess.run(
|
|
["brew", "install", "yt-dlp"],
|
|
capture_output=True, text=True, timeout=120,
|
|
)
|
|
if proc.returncode == 0:
|
|
ytdlp_installed = True
|
|
ytdlp_action = "installed"
|
|
else:
|
|
ytdlp_installed = False
|
|
ytdlp_action = "install_failed"
|
|
brew_stderr = proc.stderr
|
|
logger.warning("brew install yt-dlp failed: %s", proc.stderr)
|
|
except Exception as exc:
|
|
ytdlp_installed = False
|
|
ytdlp_action = "install_failed"
|
|
brew_stderr = str(exc)
|
|
logger.warning("brew install yt-dlp exception: %s", exc)
|
|
else:
|
|
ytdlp_installed = False
|
|
ytdlp_action = "no_homebrew"
|
|
|
|
results: Dict[str, Any] = {
|
|
"cookies_found": cookies_found,
|
|
"ytdlp_installed": ytdlp_installed,
|
|
"ytdlp_action": ytdlp_action,
|
|
"env_written": False,
|
|
}
|
|
if ytdlp_action == "install_failed":
|
|
results["ytdlp_stderr"] = brew_stderr
|
|
return results
|
|
|
|
|
|
def write_setup_config(env_path: Path, from_browser: str = "auto") -> bool:
|
|
"""Write SETUP_COMPLETE and FROM_BROWSER to the .env file.
|
|
|
|
Creates the file and parent directories if needed.
|
|
Appends to existing file without overwriting existing keys.
|
|
|
|
Args:
|
|
env_path: Path to the .env file (e.g. ~/.config/last30days/.env)
|
|
from_browser: Browser extraction mode to write (default: "auto")
|
|
|
|
Returns:
|
|
True if config was written successfully, False on error.
|
|
"""
|
|
try:
|
|
env_path = Path(env_path)
|
|
env_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Read existing content to avoid overwriting keys
|
|
existing_keys: set = set()
|
|
existing_content = ""
|
|
if env_path.exists():
|
|
existing_content = env_path.read_text(encoding="utf-8")
|
|
for line in existing_content.splitlines():
|
|
stripped = line.strip()
|
|
if stripped and not stripped.startswith("#") and "=" in stripped:
|
|
key = stripped.split("=", 1)[0].strip()
|
|
existing_keys.add(key)
|
|
|
|
lines_to_add = []
|
|
if "SETUP_COMPLETE" not in existing_keys:
|
|
lines_to_add.append("SETUP_COMPLETE=true")
|
|
if "FROM_BROWSER" not in existing_keys:
|
|
lines_to_add.append(f"FROM_BROWSER={from_browser}")
|
|
|
|
if not lines_to_add:
|
|
return True # Nothing to write, already configured
|
|
|
|
# Ensure trailing newline before appending
|
|
with open(env_path, "a", encoding="utf-8") as f:
|
|
if existing_content and not existing_content.endswith("\n"):
|
|
f.write("\n")
|
|
f.write("\n".join(lines_to_add) + "\n")
|
|
|
|
return True
|
|
|
|
except OSError as exc:
|
|
logger.error("Failed to write setup config to %s: %s", env_path, exc)
|
|
return False
|
|
|
|
|
|
def get_setup_status_text(results: Dict[str, Any]) -> str:
|
|
"""Return a human-readable summary of auto-setup results.
|
|
|
|
Args:
|
|
results: Dict from run_auto_setup()
|
|
|
|
Returns:
|
|
Multi-line status text.
|
|
"""
|
|
lines = []
|
|
lines.append("Setup complete! Here's what I found:")
|
|
lines.append("")
|
|
|
|
cookies_found = results.get("cookies_found", {})
|
|
if cookies_found:
|
|
for source, browser in cookies_found.items():
|
|
lines.append(f" - {source.upper()} cookies found in {browser}")
|
|
else:
|
|
lines.append(" - No browser cookies found for X/Twitter")
|
|
|
|
ytdlp_action = results.get("ytdlp_action", "")
|
|
if ytdlp_action == "installed":
|
|
lines.append(" - Installed yt-dlp via Homebrew")
|
|
elif ytdlp_action == "install_failed":
|
|
lines.append(" - yt-dlp install failed \u2014 run `brew install yt-dlp` manually")
|
|
elif ytdlp_action == "no_homebrew":
|
|
lines.append(" - yt-dlp not found. Install Homebrew first, then: brew install yt-dlp")
|
|
elif ytdlp_action == "already_installed":
|
|
lines.append(" - yt-dlp already installed")
|
|
elif results.get("ytdlp_installed", False):
|
|
lines.append(" - yt-dlp is installed (YouTube search ready)")
|
|
else:
|
|
lines.append(" - yt-dlp not found (install with: brew install yt-dlp)")
|
|
|
|
env_written = results.get("env_written", False)
|
|
if env_written:
|
|
lines.append("")
|
|
lines.append("Configuration saved. Future runs will auto-detect your browsers.")
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# OpenClaw server-side setup (no browser, JSON output)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_OPENCLAW_KEY_NAMES = [
|
|
"SCRAPECREATORS_API_KEY",
|
|
"XAI_API_KEY",
|
|
"BRAVE_API_KEY",
|
|
"EXA_API_KEY",
|
|
"SERPER_API_KEY",
|
|
"OPENAI_API_KEY",
|
|
"AUTH_TOKEN",
|
|
]
|
|
|
|
|
|
def run_openclaw_setup(config: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Server-side setup probe: no cookies, just tool + key availability.
|
|
|
|
Returns a dict suitable for JSON output to stdout so that SKILL.md
|
|
can present appropriate options to the user.
|
|
"""
|
|
yt_dlp = shutil.which("yt-dlp") is not None
|
|
node = shutil.which("node") is not None
|
|
python3 = shutil.which("python3") is not None
|
|
|
|
keys: Dict[str, bool] = {}
|
|
for key_name in _OPENCLAW_KEY_NAMES:
|
|
short = key_name.lower().replace("_api_key", "").replace("_key", "").replace("_token", "")
|
|
# Normalize: AUTH_TOKEN -> auth, SCRAPECREATORS_API_KEY -> scrapecreators
|
|
keys[short] = bool(config.get(key_name))
|
|
|
|
# Determine x_method
|
|
if config.get("XAI_API_KEY"):
|
|
x_method: Optional[str] = "xai"
|
|
elif config.get("AUTH_TOKEN") and config.get("CT0"):
|
|
x_method = "cookies"
|
|
else:
|
|
x_method = None
|
|
|
|
return {
|
|
"yt_dlp": yt_dlp,
|
|
"node": node,
|
|
"python3": python3,
|
|
"keys": keys,
|
|
"x_method": x_method,
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# PAT auth flow (GitHub token via ScrapeCreators)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_PAT_BASE = "https://api.scrapecreators.com/v1/github/pat"
|
|
|
|
|
|
def auth_with_pat(github_token: str) -> Optional[Dict[str, Any]]:
|
|
"""Authenticate with ScrapeCreators using a GitHub PAT.
|
|
|
|
POSTs the token to the PAT auth endpoint. ScrapeCreators verifies it
|
|
against GitHub's API, creates/finds the account, and returns an API key.
|
|
|
|
Returns:
|
|
Dict with api_key, github_username, etc. on success, None on failure.
|
|
"""
|
|
try:
|
|
req = Request(f"{_PAT_BASE}/auth", data=b"", method="POST")
|
|
req.add_header("Authorization", f"Bearer {github_token}")
|
|
with urlopen(req, timeout=15) as resp:
|
|
data = json.loads(resp.read())
|
|
except HTTPError as exc:
|
|
if exc.code == 422:
|
|
logger.warning("PAT auth: insufficient scope — user needs user:email")
|
|
else:
|
|
logger.warning("PAT auth failed: %s", exc)
|
|
return None
|
|
except (URLError, OSError) as exc:
|
|
logger.warning("PAT auth request failed: %s", exc)
|
|
return None
|
|
|
|
if not data.get("api_key"):
|
|
logger.warning("PAT auth returned no api_key: %s", data)
|
|
return None
|
|
|
|
return data
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Device auth flow (GitHub OAuth via ScrapeCreators)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_DEVICE_BASE = "https://api.scrapecreators.com/v1/github/device"
|
|
|
|
|
|
def run_device_auth() -> Optional[Tuple[str, str, str, int]]:
|
|
"""Start the device authorization flow.
|
|
|
|
POSTs to the ScrapeCreators device/code endpoint.
|
|
|
|
Returns:
|
|
(device_code, user_code, verification_uri, interval) on success,
|
|
None on failure.
|
|
"""
|
|
try:
|
|
body = json.dumps({}).encode()
|
|
req = Request(f"{_DEVICE_BASE}/code", data=body, method="POST")
|
|
req.add_header("Content-Type", "application/json")
|
|
with urlopen(req, timeout=15) as resp:
|
|
data = json.loads(resp.read())
|
|
except (HTTPError, URLError, OSError) as exc:
|
|
logger.warning("Device auth code request failed: %s", exc)
|
|
return None
|
|
|
|
device_code = data.get("device_code")
|
|
user_code = data.get("user_code")
|
|
verification_uri = data.get("verification_uri")
|
|
interval = data.get("interval", 5)
|
|
|
|
if not device_code or not user_code:
|
|
logger.warning("Device auth returned incomplete response: %s", data)
|
|
return None
|
|
|
|
return (device_code, user_code, verification_uri or "", interval)
|
|
|
|
|
|
def poll_device_auth(
|
|
device_code: str,
|
|
interval: int,
|
|
timeout: int = 300,
|
|
user_code: str = "",
|
|
clipboard_ok: bool = False,
|
|
) -> Optional[str]:
|
|
"""Poll for an access token after the user authorizes the device.
|
|
|
|
Args:
|
|
device_code: The device_code from run_device_auth().
|
|
interval: Polling interval in seconds.
|
|
timeout: Maximum time to poll in seconds.
|
|
user_code: The user code to remind about during polling.
|
|
clipboard_ok: Whether the code was copied to clipboard.
|
|
|
|
Returns:
|
|
access_token on success, None on timeout or failure.
|
|
"""
|
|
import sys
|
|
|
|
started_at = time.time()
|
|
deadline = started_at + timeout
|
|
last_reminder = started_at
|
|
reminder_count = 0
|
|
max_reminders = 4
|
|
reminder_interval = 30 # seconds between reminders
|
|
|
|
while time.time() < deadline:
|
|
time.sleep(interval)
|
|
|
|
# Periodic reminder of the code while waiting
|
|
if (
|
|
user_code
|
|
and reminder_count < max_reminders
|
|
and time.time() - last_reminder >= reminder_interval
|
|
):
|
|
clipboard_hint = " (on your clipboard)" if clipboard_ok else ""
|
|
print(
|
|
f" Still waiting... Your code: {user_code}{clipboard_hint}",
|
|
file=sys.stderr,
|
|
flush=True,
|
|
)
|
|
last_reminder = time.time()
|
|
reminder_count += 1
|
|
|
|
try:
|
|
body = json.dumps({"device_code": device_code}).encode()
|
|
req = Request(f"{_DEVICE_BASE}/token", data=body, method="POST")
|
|
req.add_header("Content-Type", "application/json")
|
|
with urlopen(req, timeout=15) as resp:
|
|
data = json.loads(resp.read())
|
|
except HTTPError as exc:
|
|
if exc.code in (400, 403, 428):
|
|
continue
|
|
logger.warning("Device auth poll error: %s", exc)
|
|
return None
|
|
except (URLError, OSError):
|
|
continue
|
|
|
|
if data.get("access_token"):
|
|
return data["access_token"]
|
|
|
|
error = data.get("error")
|
|
if error == "slow_down":
|
|
interval = min(interval + 2, 30)
|
|
continue
|
|
if error == "authorization_pending":
|
|
continue
|
|
if error in ("expired_token", "access_denied"):
|
|
logger.warning("Device auth failed: %s", error)
|
|
return None
|
|
|
|
return None
|
|
|
|
|
|
def fetch_api_key(access_token: str) -> Optional[str]:
|
|
"""Fetch the ScrapeCreators API key using the GitHub access token.
|
|
|
|
GETs the device/profile endpoint with Bearer auth.
|
|
|
|
Returns:
|
|
api_key string on success, None on failure.
|
|
"""
|
|
try:
|
|
req = Request(f"{_DEVICE_BASE}/profile")
|
|
req.add_header("Authorization", f"Bearer {access_token}")
|
|
with urlopen(req, timeout=15) as resp:
|
|
data = json.loads(resp.read())
|
|
except (HTTPError, URLError, OSError) as exc:
|
|
logger.warning("Failed to fetch API key: %s", exc)
|
|
return None
|
|
|
|
return data.get("api_key")
|
|
|
|
|
|
def run_full_device_auth(timeout: int = 300) -> Dict[str, Any]:
|
|
"""Run the complete GitHub device auth flow and return JSON-serializable result.
|
|
|
|
Chains: start device flow -> open browser -> poll -> fetch API key.
|
|
Designed to be called from the CLI and have its stdout parsed by the LLM.
|
|
|
|
Returns:
|
|
Dict with status and relevant fields:
|
|
- {"status": "success", "api_key": "sc_...", "user_code": "ABCD-1234"}
|
|
- {"status": "error", "message": "..."}
|
|
- {"status": "timeout", "user_code": "ABCD-1234"}
|
|
- {"status": "denied"}
|
|
"""
|
|
import webbrowser
|
|
|
|
# Step 1: Start device flow
|
|
result = run_device_auth()
|
|
if result is None:
|
|
return {"status": "error", "message": "Failed to start device auth flow"}
|
|
|
|
device_code, user_code, verification_uri, interval = result
|
|
|
|
import sys
|
|
|
|
# Step 2: Copy code to clipboard BEFORE opening browser
|
|
clipboard_ok = False
|
|
if sys.platform == "darwin":
|
|
try:
|
|
subprocess.run(
|
|
["pbcopy"], input=user_code.encode(), check=True, timeout=5,
|
|
)
|
|
clipboard_ok = True
|
|
except Exception:
|
|
pass # pbcopy unavailable or failed, fall through
|
|
|
|
# Step 3: Show code prominently, then open browser
|
|
clipboard_hint = " (copied to clipboard)" if clipboard_ok else ""
|
|
code_line = f" Your code: {user_code}{clipboard_hint}"
|
|
action_line = " Paste it on the GitHub page that just opened"
|
|
width = max(len(code_line), len(action_line)) + 2
|
|
border = "-" * width
|
|
print(f"\n+{border}+", file=sys.stderr)
|
|
print(f"|{code_line.ljust(width)}|", file=sys.stderr)
|
|
print(f"|{action_line.ljust(width)}|", file=sys.stderr)
|
|
print(f"+{border}+", file=sys.stderr)
|
|
|
|
if verification_uri:
|
|
try:
|
|
webbrowser.open(verification_uri)
|
|
except Exception:
|
|
print(f"Open: {verification_uri}", file=sys.stderr)
|
|
|
|
print("Waiting for authorization...", file=sys.stderr, flush=True)
|
|
|
|
# Step 4: Poll for token (with periodic code reminders)
|
|
access_token = poll_device_auth(
|
|
device_code, interval, timeout=timeout,
|
|
user_code=user_code, clipboard_ok=clipboard_ok,
|
|
)
|
|
if access_token is None:
|
|
return {"status": "timeout", "user_code": user_code, "clipboard_ok": clipboard_ok}
|
|
|
|
# Step 4: Fetch API key
|
|
api_key = fetch_api_key(access_token)
|
|
if api_key is None:
|
|
return {
|
|
"status": "error",
|
|
"message": "Authorized but failed to fetch API key",
|
|
"clipboard_ok": clipboard_ok,
|
|
}
|
|
|
|
return {"status": "success", "method": "device", "api_key": api_key, "user_code": user_code, "clipboard_ok": clipboard_ok}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Unified GitHub auth: PAT first, device flow fallback
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def run_github_auth(timeout: int = 300) -> Dict[str, Any]:
|
|
"""Try PAT auth via gh CLI, fall back to device flow.
|
|
|
|
1. Check for `gh` CLI
|
|
2. If found, run `gh auth token` to get a PAT
|
|
3. POST PAT to ScrapeCreators — if it works, done
|
|
4. If PAT fails for any reason, fall through to device flow
|
|
|
|
Returns JSON-serializable dict with status, method, and api_key.
|
|
"""
|
|
import sys
|
|
|
|
# Step 1: Try PAT via gh CLI
|
|
gh_path = shutil.which("gh")
|
|
if gh_path:
|
|
try:
|
|
result = subprocess.run(
|
|
["gh", "auth", "token"],
|
|
capture_output=True, text=True, timeout=10,
|
|
)
|
|
if result.returncode == 0 and result.stdout.strip():
|
|
token = result.stdout.strip()
|
|
print("Found gh CLI — trying PAT auth...", file=sys.stderr)
|
|
pat_result = auth_with_pat(token)
|
|
if pat_result and pat_result.get("api_key"):
|
|
return {
|
|
"status": "success",
|
|
"method": "pat",
|
|
"api_key": pat_result["api_key"],
|
|
"github_username": pat_result.get("github_username", ""),
|
|
}
|
|
# PAT failed — might be insufficient scope
|
|
print(
|
|
"PAT auth didn't work (scope or endpoint issue). "
|
|
"Falling back to GitHub device flow...",
|
|
file=sys.stderr,
|
|
)
|
|
except Exception as exc:
|
|
logger.debug("gh auth token failed: %s", exc)
|
|
|
|
# Step 2: Fall back to device flow
|
|
if not gh_path:
|
|
print("gh CLI not found — using GitHub device flow...", file=sys.stderr)
|
|
|
|
return run_full_device_auth(timeout=timeout)
|