| 1 | """Bright Data CLI adapter for last30days. |
| 2 | |
| 3 | Shells out to the ``brightdata`` CLI (``@brightdata/cli``) to run Bright |
| 4 | Data Pipelines. The CLI owns authentication end to end -- ``brightdata |
| 5 | login`` does a gh-style zero-click browser flow and stores credentials in |
| 6 | a platform config directory -- so this module never handles a login, and |
| 7 | never reads credential *contents*: the auth probe is presence-only. |
| 8 | |
| 9 | Activation gate: two-way, mirroring the digg CLI-gated precedent but with |
| 10 | an auth dimension the digg source does not have. |
| 11 | |
| 12 | 1. ``shutil.which("brightdata")`` must resolve on the **agent subprocess |
| 13 | PATH** (not merely exist on disk -- Hermes/OpenClaw gateways often drop |
| 14 | ``~/.local/bin``). |
| 15 | 2. A credential signal must be present: either ``BRIGHTDATA_API_KEY`` |
| 16 | resolved through the normal config layering, or the CLI's own |
| 17 | credentials file in the platform config dir. |
| 18 | |
| 19 | The second check is deliberately offline. A stale token passes it and |
| 20 | then 401s fast at call time; that path degrades to empty results with the |
| 21 | CLI's own error line preserved in the envelope, which is the AE2 contract. |
| 22 | |
| 23 | Metering note (R13): no pricing logic lives here. One pipeline request |
| 24 | costs one credit against the account's monthly free tier regardless of how |
| 25 | many records come back, so caps in the calling adapter bound *records* |
| 26 | (paid-tier cost), not credits. Credit and auth warnings from the CLI are |
| 27 | passed through verbatim rather than interpreted. |
| 28 | """ |
| 29 | |
| 30 | from __future__ import annotations |
| 31 | |
| 32 | import json |
| 33 | import os |
| 34 | import shutil |
| 35 | import sys |
| 36 | from pathlib import Path |
| 37 | from typing import Any, Dict, List, Optional, Sequence |
| 38 | |
| 39 | from . import log, subproc |
| 40 | |
| 41 | |
| 42 | CLI_BIN = "brightdata" |
| 43 | |
| 44 | # Env var carrying an explicit API key. Registered in env.py so `.env` file |
| 45 | # and keychain users pass the gate the same way process-env users do; when |
| 46 | # it resolves from a non-process-env layer we hand it to the CLI via -k. |
| 47 | API_KEY_ENV = "BRIGHTDATA_API_KEY" |
| 48 | |
| 49 | # Credentials filename written by `brightdata login`. Probed for existence |
| 50 | # only -- never opened, parsed, or logged. |
| 51 | _CREDENTIALS_FILENAME = "credentials.json" |
| 52 | _CONFIG_DIRNAME = "brightdata-cli" |
| 53 | |
| 54 | # The CLI's own polling timeout sits below our subprocess timeout so the CLI |
| 55 | # exits cleanly with its own error rather than being SIGTERM'd mid-poll. Its |
| 56 | # timeout path throws with zero records (verified in its polling module -- |
| 57 | # never partial output), so a timed-out pull is a clean parseable failure. |
| 58 | _CLI_TIMEOUT_MARGIN = 10 |
| 59 | |
| 60 | |
| 61 | def _log(msg: str) -> None: |
| 62 | log.source_log("BrightData", msg, tty_only=False) |
| 63 | |
| 64 | |
| 65 | def _config_dir() -> Path: |
| 66 | """Platform config directory the Bright Data CLI stores credentials in. |
| 67 | |
| 68 | Mirrors the CLI's own credentials module: APPDATA on Windows, the |
| 69 | Application Support tree on macOS, XDG_CONFIG_HOME (or ~/.config) on |
| 70 | everything else. |
| 71 | """ |
| 72 | if sys.platform == "win32": |
| 73 | base = os.environ.get("APPDATA") |
| 74 | root = Path(base) if base else Path.home() / "AppData" / "Roaming" |
| 75 | elif sys.platform == "darwin": |
| 76 | root = Path.home() / "Library" / "Application Support" |
| 77 | else: |
| 78 | base = os.environ.get("XDG_CONFIG_HOME") |
| 79 | root = Path(base) if base else Path.home() / ".config" |
| 80 | return root / _CONFIG_DIRNAME |
| 81 | |
| 82 | |
| 83 | def is_installed() -> bool: |
| 84 | """True when the brightdata binary resolves on the agent subprocess PATH.""" |
| 85 | return shutil.which(CLI_BIN) is not None |
| 86 | |
| 87 | |
| 88 | def _api_key(config: Optional[Dict[str, Any]]) -> str: |
| 89 | if not config: |
| 90 | return "" |
| 91 | return str(config.get(API_KEY_ENV) or "").strip() |
| 92 | |
| 93 | |
| 94 | def has_credentials(config: Optional[Dict[str, Any]] = None) -> bool: |
| 95 | """True when some credential signal exists, without reading any secret. |
| 96 | |
| 97 | Presence-only by design: an explicit API key resolved through config |
| 98 | layering, or the existence of the CLI's credentials file. The file is |
| 99 | never opened. This cannot distinguish a live token from an expired one |
| 100 | -- that is what the fast 401 at call time is for. |
| 101 | """ |
| 102 | if _api_key(config): |
| 103 | return True |
| 104 | try: |
| 105 | return (_config_dir() / _CREDENTIALS_FILENAME).exists() |
| 106 | except OSError: |
| 107 | return False |
| 108 | |
| 109 | |
| 110 | def is_available(config: Optional[Dict[str, Any]] = None) -> bool: |
| 111 | """The full activation gate: binary on PATH *and* a credential signal.""" |
| 112 | return is_installed() and has_credentials(config) |
| 113 | |
| 114 | |
| 115 | def gate_status(config: Optional[Dict[str, Any]] = None) -> Dict[str, bool]: |
| 116 | """Two-field probe for ``pipeline.diagnose`` (bird_installed precedent). |
| 117 | |
| 118 | Network-free, so it is safe on the ``--diagnose`` / doctor path. |
| 119 | """ |
| 120 | installed = is_installed() |
| 121 | return { |
| 122 | "brightdata_installed": installed, |
| 123 | "brightdata_authenticated": installed and has_credentials(config), |
| 124 | } |
| 125 | |
| 126 | |
| 127 | def _build_args( |
| 128 | pipeline_type: str, |
| 129 | params: Sequence[str], |
| 130 | *, |
| 131 | cli_timeout: int, |
| 132 | ) -> List[str]: |
| 133 | """Assemble the CLI invocation. |
| 134 | |
| 135 | The API key is deliberately **absent** here -- it travels in the child's |
| 136 | environment instead (see ``_child_env``). Process arguments are not a |
| 137 | secret channel: ``/proc/<pid>/cmdline`` is world-readable under the |
| 138 | default ``hidepid=0``, and a review pull lives for up to 180s, so a key |
| 139 | on the command line is readable by any other local user and is captured |
| 140 | verbatim by execve auditing, process accounting, and any monitoring |
| 141 | agent that snapshots ``ps``. Mirrors the ``bird_x`` cookie-injection |
| 142 | precedent. |
| 143 | |
| 144 | Positional params are fenced behind ``--`` so a keyword that happens to |
| 145 | begin with a dash is parsed as a search term rather than as an option. |
| 146 | """ |
| 147 | return [ |
| 148 | CLI_BIN, |
| 149 | "pipelines", |
| 150 | pipeline_type, |
| 151 | "--json", |
| 152 | "--timeout", |
| 153 | str(cli_timeout), |
| 154 | "--", |
| 155 | *(str(p) for p in params), |
| 156 | ] |
| 157 | |
| 158 | |
| 159 | def _child_env(api_key: str) -> Optional[Dict[str, str]]: |
| 160 | """Environment for the child process, carrying the key when we have one. |
| 161 | |
| 162 | Returns None when there is nothing to inject, so the child simply |
| 163 | inherits the parent environment (the common case: the CLI owns its own |
| 164 | credentials file, or the key is already exported). |
| 165 | """ |
| 166 | if not api_key: |
| 167 | return None |
| 168 | return {**os.environ, API_KEY_ENV: api_key} |
| 169 | |
| 170 | |
| 171 | def _scrub(text: str, secret: str) -> str: |
| 172 | """Remove a secret from text before it is logged or returned. |
| 173 | |
| 174 | Defense in depth for the passthrough paths: the stderr lines this |
| 175 | module deliberately surfaces are auth and quota failures, which are |
| 176 | exactly the messages a CLI is most likely to echo the rejected |
| 177 | credential back in. |
| 178 | """ |
| 179 | if not secret or not text: |
| 180 | return text |
| 181 | return text.replace(secret, "***") |
| 182 | |
| 183 | |
| 184 | def _extract_records(payload: Any) -> List[Dict[str, Any]]: |
| 185 | """Pull the record list out of a parsed CLI payload. |
| 186 | |
| 187 | Verified live (2026-08-13): both amazon pipelines return a **bare JSON |
| 188 | array** of flat record dicts, not the ``{"results": [...]}`` envelope the |
| 189 | digg CLI uses. The dict branches below are defensive against CLI churn, |
| 190 | which is a live risk on a package this young. |
| 191 | """ |
| 192 | if isinstance(payload, list): |
| 193 | return [r for r in payload if isinstance(r, dict)] |
| 194 | if isinstance(payload, dict): |
| 195 | for key in ("records", "results", "data"): |
| 196 | value = payload.get(key) |
| 197 | if isinstance(value, list): |
| 198 | return [r for r in value if isinstance(r, dict)] |
| 199 | return [] |
| 200 | |
| 201 | |
| 202 | def run_pipeline( |
| 203 | pipeline_type: str, |
| 204 | params: Sequence[str], |
| 205 | *, |
| 206 | timeout: int, |
| 207 | config: Optional[Dict[str, Any]] = None, |
| 208 | ) -> Dict[str, Any]: |
| 209 | """Run one Bright Data pipeline and return ``{"records", "error"}``. |
| 210 | |
| 211 | Never raises. Every failure mode -- missing binary, spawn failure, |
| 212 | subprocess timeout, non-zero exit, unparseable stdout -- returns empty |
| 213 | records plus a one-line ``error`` string, so callers can record the |
| 214 | failure in ``errors_by_source`` without branching on exception types. |
| 215 | |
| 216 | The CLI's first stderr line is preserved verbatim as the error (auth |
| 217 | 401s and low-credit warnings are the cases that matter), and also |
| 218 | mirrored to ``source_log`` so the failure is visible in non-TTY hosts. |
| 219 | |
| 220 | Args: |
| 221 | pipeline_type: pipeline name, e.g. ``amazon_product_search``. |
| 222 | params: positional pipeline params, passed through in order. |
| 223 | timeout: subprocess timeout in seconds. The CLI's own polling |
| 224 | timeout is set just below this so it can fail cleanly first. |
| 225 | config: resolved config dict, consulted only for the API key. |
| 226 | |
| 227 | Returns: |
| 228 | ``{"records": [...]}`` on success, else ``{"records": [], "error": str}``. |
| 229 | """ |
| 230 | if not is_installed(): |
| 231 | return {"records": [], "error": f"{CLI_BIN} not on PATH"} |
| 232 | |
| 233 | cli_timeout = max(5, int(timeout) - _CLI_TIMEOUT_MARGIN) |
| 234 | key = _api_key(config) |
| 235 | cmd = _build_args(pipeline_type, params, cli_timeout=cli_timeout) |
| 236 | |
| 237 | try: |
| 238 | result = subproc.run_with_timeout(cmd, timeout=timeout, env=_child_env(key)) |
| 239 | except subproc.SubprocTimeout as exc: |
| 240 | _log(f"Timeout: {exc}") |
| 241 | return {"records": [], "error": str(exc)} |
| 242 | except FileNotFoundError as exc: |
| 243 | _log(f"Binary missing: {exc}") |
| 244 | return {"records": [], "error": str(exc)} |
| 245 | except OSError as exc: |
| 246 | _log(f"Spawn failed: {exc}") |
| 247 | return {"records": [], "error": str(exc)} |
| 248 | |
| 249 | stderr = _scrub(result.stderr or "", key) |
| 250 | _passthrough_warnings(stderr) |
| 251 | |
| 252 | if result.returncode != 0: |
| 253 | lines = [ln.strip() for ln in stderr.strip().splitlines() if ln.strip()] |
| 254 | # The CLI narrates polling progress on stderr, so the *last* line is |
| 255 | # the actual failure; the first line is "Triggering pipeline...". |
| 256 | first = lines[-1] if lines else f"exit {result.returncode}" |
| 257 | _log(f"CLI exit {result.returncode}: {first}") |
| 258 | return {"records": [], "error": first} |
| 259 | |
| 260 | stdout = result.stdout or "" |
| 261 | if not stdout.strip(): |
| 262 | return {"records": []} |
| 263 | try: |
| 264 | payload = json.loads(stdout) |
| 265 | except json.JSONDecodeError as exc: |
| 266 | _log(f"JSON decode failed: {exc}") |
| 267 | return {"records": [], "error": f"json decode: {exc}"} |
| 268 | |
| 269 | return {"records": _extract_records(payload)} |
| 270 | |
| 271 | |
| 272 | # Substrings that mark a stderr line worth surfacing even on a successful |
| 273 | # run -- credit exhaustion and auth trouble are the two the user must see. |
| 274 | # Matched case-insensitively against the CLI's own wording, and echoed |
| 275 | # verbatim rather than reworded (R13: no pricing logic, no interpretation). |
| 276 | _WARNING_MARKERS = ("credit", "quota", "balance", "unauthor", "401", "expired", "login") |
| 277 | |
| 278 | |
| 279 | def _passthrough_warnings(stderr: str) -> None: |
| 280 | """Echo credit/auth warning lines from the CLI verbatim. |
| 281 | |
| 282 | Skips the routine polling narration so a normal run stays quiet. |
| 283 | """ |
| 284 | for line in (stderr or "").splitlines(): |
| 285 | text = line.strip() |
| 286 | if not text or text.lower().startswith(("status:", "triggering", "triggered", "data received")): |
| 287 | continue |
| 288 | lowered = text.lower() |
| 289 | if any(marker in lowered for marker in _WARNING_MARKERS): |
| 290 | _log(text) |
| 291 |