env.py - last30days-skill

env.py

1	"""Environment and API key management for last30days skill."""
2
3	from __future__ import annotations
4
5	import datetime
6	import json
7	import locale
8	import os
9	import sys
10	from dataclasses import dataclass
11	from pathlib import Path
12	from typing import Any, Literal
13
14
15	def read_secret_env(name: str, default: str \| None = None) -> str \| None:
16	"""Read a possibly-secret environment variable by name.
17
18	Call sites pass the variable name as an argument here instead of reading a
19	secret-shaped literal environment key inline at the call site. That keeps
20	those literals out of direct env-get calls, which an install-time skill
21	scanner flags as credential exfiltration. Behaviour is identical to a plain
22	environment lookup of ``name`` with ``default``.
23	"""
24	return os.environ.get(name, default)
25
26
27	# Allow override via environment variable for testing
28	# Set LAST30DAYS_CONFIG_DIR="" for clean/no-config mode
29	# Set LAST30DAYS_CONFIG_DIR="/path/to/dir" for custom config location
30	_config_override = os.environ.get('LAST30DAYS_CONFIG_DIR')
31	if _config_override == "":
32	# Empty string = no config file (clean mode)
33	CONFIG_DIR = None
34	CONFIG_FILE = None
35	elif _config_override:
36	CONFIG_DIR = Path(_config_override)
37	CONFIG_FILE = CONFIG_DIR / ".env"
38	else:
39	CONFIG_DIR = Path.home() / ".config" / "last30days"
40	CONFIG_FILE = CONFIG_DIR / ".env"
41
42	# macOS Keychain integration: items stored with this service prefix are picked
43	# up automatically on Darwin as the lowest-priority credential source.
44	# Example: `security add-generic-password -a "$USER" -s last30days-XAI_API_KEY -w "xai-..."`.
45	KEYCHAIN_SERVICE_PREFIX = "last30days-"
46
47	# Optional non-secret aliases for users who already store API keys under a
48	# different Keychain naming convention. Configure as JSON in
49	# LAST30DAYS_KEYCHAIN_ALIASES, for example:
50	# {"XAI_API_KEY":{"account":"keychain-user","service":"existing-xai-api-key"}}
51	# A string value is shorthand for {"service": "..."} with the current user.
52	KEYCHAIN_ALIASES_ENV = "LAST30DAYS_KEYCHAIN_ALIASES"
53
54	# Single source of truth for which credentials the Keychain loader looks up.
55	# The setup-keychain.sh helper mirrors this list and is held in sync via
56	# tests/test_env_keychain.py::test_keychain_keys_match_setup_script.
57	KEYCHAIN_KEYS = (
58	"OPENAI_API_KEY", "XAI_API_KEY", "GOOGLE_API_KEY", "GEMINI_API_KEY",
59	"GOOGLE_GENAI_API_KEY", "SCRAPECREATORS_API_KEY", "APIFY_API_TOKEN",
60	"AUTH_TOKEN", "CT0", "BSKY_HANDLE", "BSKY_APP_PASSWORD",
61	"TRUTHSOCIAL_TOKEN", "BRAVE_API_KEY", "EXA_API_KEY", "SERPER_API_KEY",
62	"OPENROUTER_API_KEY", "PERPLEXITY_API_KEY", "PARALLEL_API_KEY", "XQUIK_API_KEY",
63	"XIAOHONGSHU_API_BASE", "GITHUB_TOKEN",
64	)
65
66	# pass(1) integration: Linux/Unix analog of the Keychain source. Each key in
67	# KEYCHAIN_KEYS is looked up at pass path f"{prefix}{KEY}", the direct analog of
68	# Keychain's "last30days-<KEY>" service-name convention, so any user stores keys
69	# under one namespace without editing code. The prefix is resolved at call time
70	# (in get_config) from LAST30DAYS_PASS_PREFIX in the process env or a config
71	# file, falling back to this default; included verbatim, so keep the trailing
72	# separator. Honors PASSWORD_STORE_DIR.
73	DEFAULT_PASS_PATH_PREFIX = "last30days/"
74
75	AuthSource = Literal["api_key", "none"]
76	AuthStatus = Literal["ok", "missing"]
77
78	AUTH_SOURCE_API_KEY: AuthSource = "api_key"
79	AUTH_SOURCE_NONE: AuthSource = "none"
80
81	AUTH_STATUS_OK: AuthStatus = "ok"
82	AUTH_STATUS_MISSING: AuthStatus = "missing"
83
84	XIAOHONGSHU_DEFAULT_API_BASES = (
85	"http://localhost:18060",
86	"http://host.docker.internal:18060",
87	)
88	XIAOHONGSHU_RESOLVED_API_BASE_KEY = "_XIAOHONGSHU_API_BASE_RESOLVED"
89
90
91	@dataclass(frozen=True)
92	class OpenAIAuth:
93	token: str \| None
94	source: AuthSource
95	status: AuthStatus
96
97
98	BrowserCookieMode = Literal["off", "read", "plan_only"]
99
100
101	@dataclass(frozen=True)
102	class ConfigLoadPolicy:
103	"""Local-read gates for configuration loading.
104
105	Bare library calls use the safe default: no browser-cookie extraction and no
106	project-scoped config. CLI entry points can opt into narrower behavior after
107	parsing command intent.
108	"""
109
110	browser_cookies: BrowserCookieMode = "off"
111	allow_project_config: bool = False
112	inspect_ignored_project_config: bool = False
113
114
115	def _truthy(value: Any) -> bool:
116	if value is None:
117	return False
118	return str(value).strip().lower() in {"1", "true", "yes", "on"}
119
120
121	def is_timestamp_fresh(timestamp_value: Any, ttl_seconds: int) -> bool:
122	"""True when ``timestamp_value`` (ISO-8601 string) is within ``ttl_seconds``.
123
124	Shared freshness gate for the doctor cache and the report cache. The guard
125	order is load-bearing: a non-positive TTL disables caching entirely, a
126	non-string or empty timestamp is stale, a malformed timestamp is stale,
127	naive timestamps are treated as UTC, and a future timestamp (negative age)
128	counts as fresh.
129	"""
130	if ttl_seconds <= 0:
131	return False
132	if not isinstance(timestamp_value, str) or not timestamp_value:
133	return False
134	try:
135	created_at = datetime.datetime.fromisoformat(timestamp_value)
136	except ValueError:
137	return False
138	if created_at.tzinfo is None:
139	created_at = created_at.replace(tzinfo=datetime.timezone.utc)
140	age = datetime.datetime.now(datetime.timezone.utc) - created_at.astimezone(
141	datetime.timezone.utc
142	)
143	return age.total_seconds() <= ttl_seconds
144
145
146	def _project_config_trusted(policy: ConfigLoadPolicy, file_env: dict[str, Any]) -> bool:
147	if policy.allow_project_config:
148	return True
149	process_value = os.environ.get("LAST30DAYS_TRUST_PROJECT_CONFIG")
150	if process_value is not None:
151	return _truthy(process_value)
152	return _truthy(file_env.get("LAST30DAYS_TRUST_PROJECT_CONFIG"))
153
154
155	def _check_file_permissions(path: Path) -> None:
156	"""Warn to stderr if a secrets file has overly permissive permissions."""
157	if os.name == "nt":
158	# Windows reports synthesized POSIX mode bits that do not reflect NTFS ACLs.
159	return
160
161	try:
162	mode = path.stat().st_mode
163	# Check if group or other can read (bits 0o044)
164	if mode & 0o044:
165	sys.stderr.write(
166	f"[last30days] WARNING: {path} is readable by other users. "
167	f"Run: chmod 600 {path}\n"
168	)
169	sys.stderr.flush()
170	except OSError as exc:
171	sys.stderr.write(f"[last30days] WARNING: could not stat {path}: {exc}\n")
172	sys.stderr.flush()
173
174
175	def load_env_file(path: Path) -> dict[str, str]:
176	"""Load environment variables from a file."""
177	env = {}
178	if not path or not path.exists():
179	return env
180	_check_file_permissions(path)
181
182	# Prefer UTF-8 (utf-8-sig transparently strips a BOM written by Windows
183	# editors like Notepad). Fall back to the locale decoder for a genuinely
184	# locale-encoded .env (e.g. cp1252) so an existing file that loaded before
185	# keeps loading. If it decodes as neither, let UnicodeDecodeError surface
186	# rather than corrupting keys/secrets with replacement characters.
187	try:
188	text = path.read_text(encoding='utf-8-sig')
189	except UnicodeDecodeError:
190	text = path.read_text(encoding=locale.getpreferredencoding(False))
191
192	for line in text.splitlines():
193	line = line.strip()
194	if not line or line.startswith('#'):
195	continue
196	if '=' in line:
197	key, _, value = line.partition('=')
198	key = key.strip()
199	value = value.strip()
200	# Remove quotes if present
201	if value and value[0] in ('"', "'") and value[-1] == value[0]:
202	value = value[1:-1]
203	if key and value:
204	env.update({key: value})
205	return env
206
207
208	def _parse_keychain_aliases(raw: str \| None) -> dict[str, list[dict[str, str]]]:
209	"""Parse non-secret Keychain alias metadata from JSON.
210
211	Supported forms:
212	{"XAI_API_KEY": "existing-xai-api-key"}
213	{"XAI_API_KEY": {"service": "existing-xai-api-key", "account": "keychain-user"}}
214	{"XAI_API_KEY": [{"service": "primary"}, {"service": "fallback"}]}
215
216	Invalid entries are ignored so a typo never blocks canonical
217	`last30days-<KEY>` lookups; malformed JSON emits a warning.
218	"""
219	if not raw:
220	return {}
221	try:
222	parsed = json.loads(raw)
223	except json.JSONDecodeError as exc:
224	sys.stderr.write(
225	f"[last30days] WARNING: {KEYCHAIN_ALIASES_ENV} is not valid JSON; "
226	f"ignoring Keychain aliases while keeping canonical lookups enabled: {exc}\n"
227	)
228	sys.stderr.flush()
229	return {}
230	if not isinstance(parsed, dict):
231	return {}
232
233	allowed = set(KEYCHAIN_KEYS)
234	aliases: dict[str, list[dict[str, str]]] = {}
235	for key, spec in parsed.items():
236	if key not in allowed:
237	continue
238	specs = spec if isinstance(spec, list) else [spec]
239	clean_specs: list[dict[str, str]] = []
240	for item in specs:
241	if isinstance(item, str):
242	service = item.strip()
243	account = ""
244	elif isinstance(item, dict):
245	service = str(item.get("service", "")).strip()
246	account = str(item.get("account", "")).strip()
247	else:
248	continue
249	if service:
250	clean_specs.append({"service": service, "account": account})
251	if clean_specs:
252	aliases[key] = clean_specs
253	return aliases
254
255
256	def _load_keychain(keys: list[str], aliases: dict[str, list[dict[str, str]]] \| None = None) -> dict[str, str]:
257	"""Load credentials from macOS Keychain (no-op on other platforms).
258
259	Each key is looked up as a generic password with service name
260	``f"{KEYCHAIN_SERVICE_PREFIX}{key}"`` for the current user. Missing items
261	then fall back to optional alias metadata from
262	``LAST30DAYS_KEYCHAIN_ALIASES``. Lookup failures are silent — Keychain is
263	the lowest-priority source and is meant to be additive over `.env` files
264	and process environment.
265	"""
266	import platform
267	if platform.system() != "Darwin":
268	return {}
269
270	import shutil
271	security = shutil.which("security")
272	if not security:
273	return {}
274
275	import subprocess
276	# USER can be unset under sudo, in Docker without --env USER, or in some CI
277	# runners; fall back to the OS user record so lookups still match items
278	# stored by setup-keychain.sh (which uses $USER).
279	user = os.environ.get("USER")
280	if not user:
281	try:
282	import pwd
283	except ImportError:
284	pwd = None
285
286	if pwd is not None:
287	try:
288	user = pwd.getpwuid(os.getuid()).pw_name
289	except AttributeError:
290	user = "unknown"
291	else:
292	user = "unknown"
293	env: dict[str, str] = {}
294
295	def lookup(account: str, service: str) -> str:
296	try:
297	result = subprocess.run(
298	[security, "find-generic-password",
299	"-a", account,
300	"-s", service,
301	"-w"],
302	capture_output=True, text=True, timeout=5,
303	)
304	except (subprocess.TimeoutExpired, OSError):
305	return ""
306	if result.returncode == 0 and result.stdout.strip():
307	return result.stdout.strip()
308	return ""
309
310	for key in keys:
311	value = lookup(user, f"{KEYCHAIN_SERVICE_PREFIX}{key}")
312	if not value and aliases:
313	for alias in aliases.get(key, []):
314	alias_account = alias.get("account") or user
315	value = lookup(alias_account, alias["service"])
316	if value:
317	break
318	if value:
319	env.update({key: value})
320	return env
321
322
323	def _load_pass(keys: list[str], prefix: str) -> dict[str, str]:
324	"""Load credentials from a pass(1) store (no-op if `pass` is absent).
325
326	The Linux/Unix analog of the macOS Keychain source. Each env-var name is
327	looked up at pass path ``f"{prefix}{key}"`` — mirroring Keychain's
328	``last30days-<key>`` service-name convention — so any user stores keys under
329	that namespace without editing code (prefix overridable via
330	``LAST30DAYS_PASS_PREFIX``). The secret is decrypted in a subprocess and
331	read from stdout's first line (pass keeps the secret there; any metadata
332	follows) — never written to disk, never logged. Honors ``PASSWORD_STORE_DIR``.
333	Missing entries and failures are silent: pass is a lowest-priority, additive
334	source like Keychain, so an explicit .env or process-env value still wins.
335	"""
336	import shutil
337	pass_bin = shutil.which("pass")
338	if not pass_bin:
339	return {}
340
341	import subprocess
342	env: dict[str, str] = {}
343	for key in keys:
344	try:
345	result = subprocess.run(
346	[pass_bin, "show", f"{prefix}{key}"],
347	capture_output=True, text=True, timeout=5,
348	encoding="utf-8", errors="replace",
349	)
350	except (subprocess.TimeoutExpired, OSError):
351	# A timeout (GPG/pinentry hanging) or exec failure isn't a per-key
352	# condition — it means the store is unusable right now. Stop instead
353	# of paying the timeout once per key; otherwise a locked store would
354	# stall every config load by 5s x len(keys). A genuinely missing key
355	# returns fast with a non-zero exit and is handled below.
356	break
357	if result.returncode == 0 and result.stdout.strip():
358	env.update({key: result.stdout.strip().splitlines()[0]})
359	return env
360
361
362	def get_openai_auth(file_env: dict[str, str]) -> OpenAIAuth:
363	"""Resolve OpenAI API auth from explicit user-provided API keys."""
364	api_key = read_secret_env('OPENAI_API_KEY') or file_env.get('OPENAI_API_KEY')
365	if api_key:
366	return OpenAIAuth(
367	token=api_key,
368	source=AUTH_SOURCE_API_KEY,
369	status=AUTH_STATUS_OK,
370	)
371
372	return OpenAIAuth(
373	token=None,
374	source=AUTH_SOURCE_NONE,
375	status=AUTH_STATUS_MISSING,
376	)
377
378
379	def _find_project_env() -> Path \| None:
380	"""Find per-project .env by walking up from cwd.
381
382	Searches for .claude/last30days.env in each parent directory,
383	stopping at the git root, user's home directory, or filesystem root.
384	"""
385	cwd = Path.cwd()
386	for parent in [cwd, *cwd.parents]:
387	candidate = parent / '.claude' / 'last30days.env'
388	if candidate.exists():
389	return candidate
390	if (parent / ".git").exists():
391	break
392	# Stop at filesystem root or home
393	if parent == Path.home() or parent == parent.parent:
394	break
395	return None
396
397
398	def get_config(policy: ConfigLoadPolicy \| None = None) -> dict[str, Any]:
399	"""Load configuration from multiple sources.
400
401	Priority (highest wins):
402	1. Environment variables (os.environ)
403	2. Trusted .claude/last30days.env (per-project config)
404	3. ~/.config/last30days/.env (global config)
405	4. macOS Keychain items prefixed ``last30days-`` (Darwin only)
406	"""
407	policy = policy or ConfigLoadPolicy()
408	# Load from global config file
409	file_env = load_env_file(CONFIG_FILE) if CONFIG_FILE else {}
410
411	# Load per-project config only when trust comes from process env, global
412	# user config, or an explicit policy. A project file cannot grant trust to
413	# itself because it is not parsed until after this decision.
414	project_config_trusted = _project_config_trusted(policy, file_env)
415	project_env_path = _find_project_env() if project_config_trusted else None
416	project_env = load_env_file(project_env_path) if project_env_path else {}
417	ignored_project_env_path = None
418	ignored_project_keys: list[str] = []
419	if not project_config_trusted and policy.inspect_ignored_project_config:
420	ignored_project_env_path = _find_project_env()
421	if ignored_project_env_path:
422	ignored_project_keys = sorted(load_env_file(ignored_project_env_path).keys())
423
424	# Merge file sources: project > global
425	merged_env = {file_env, project_env}
426
427	# Keychain is the lowest-priority source (Darwin only; no-op elsewhere).
428	# Loaded before openai_auth so OPENAI_API_KEY can come from Keychain too.
429	keychain_aliases_raw = os.environ.get(KEYCHAIN_ALIASES_ENV) or merged_env.get(KEYCHAIN_ALIASES_ENV)
430	keychain_aliases = _parse_keychain_aliases(keychain_aliases_raw)
431	keychain_env = _load_keychain(list(KEYCHAIN_KEYS), keychain_aliases)
432	merged_env = {keychain_env, merged_env}
433	# pass(1) store: Linux/Unix analog of Keychain at convention path
434	# {prefix}<KEY>. Decrypts transiently so secrets stay encrypted at rest (no
435	# plaintext .env). Lowest priority: Keychain, the config files, and process
436	# env all win over it. Two efficiency guards so a user who merely has `pass`
437	# on PATH doesn't pay for it: resolve the prefix from the loaded config/env
438	# (not import time, so a .env-set LAST30DAYS_PASS_PREFIX is honored), and
439	# probe ONLY keys still unset after the higher-priority sources — an empty
440	# list short-circuits with no gpg/pinentry calls at all.
441	pass_prefix = (
442	os.environ.get("LAST30DAYS_PASS_PREFIX")
443	or merged_env.get("LAST30DAYS_PASS_PREFIX")
444	or DEFAULT_PASS_PATH_PREFIX
445	)
446	pass_missing = [k for k in KEYCHAIN_KEYS if k not in os.environ and not merged_env.get(k)]
447	pass_env = _load_pass(pass_missing, pass_prefix)
448	merged_env = {pass_env, merged_env}
449
450	openai_auth = get_openai_auth(merged_env)
451
452	# Build config: Codex/OpenAI auth + process.env > project .env > global .env
453	config = {
454	'OPENAI_API_KEY': openai_auth.token,
455	'OPENAI_AUTH_SOURCE': openai_auth.source,
456	'OPENAI_AUTH_STATUS': openai_auth.status,
457	}
458
459	keys = [
460	# Debug flag; also exported to os.environ below so log.py's lazy
461	# os.environ.get() picks up .env values after get_config() runs.
462	('LAST30DAYS_DEBUG', None),
463	('XAI_API_KEY', None),
464	('GOOGLE_API_KEY', None),
465	('GEMINI_API_KEY', None),
466	('GOOGLE_GENAI_API_KEY', None),
467	('XIAOHONGSHU_API_BASE', None),
468	('LAST30DAYS_REASONING_PROVIDER', 'auto'),
469	('LAST30DAYS_PLANNER_MODEL', None),
470	('LAST30DAYS_RERANK_MODEL', None),
471	('LAST30DAYS_X_MODEL', None),
472	('LAST30DAYS_X_BACKEND', None),
473	('LAST30DAYS_REDDIT_BACKEND', None),
474	# Doctor cache freshness window in seconds (doctor --cached).
475	('LAST30DAYS_DOCTOR_TTL', None),
476	# Per-source deadline (seconds) for doctor --probe live checks.
477	('LAST30DAYS_DOCTOR_PROBE_TIMEOUT', None),
478	('LAST30DAYS_REDDIT_SC_MIN_ITEMS', None),
479	('LAST30DAYS_STORE', None),
480	# Discovery topic queue (podcast/X-article pipeline memory). Default
481	# ON; the literal value "off" disables queue writes and annotations.
482	('LAST30DAYS_DISCOVERY_QUEUE', None),
483	# Wall-clock budget (seconds) for the deep-tier enrichment batch on
484	# the discovery resume leg (--discover --judgments). Read from the
485	# resolved config only (pipeline._resume_enrich_budget_seconds);
486	# unset/invalid falls back to 450s. The one-shot --discover path
487	# keeps its fixed 240s quick budget regardless.
488	('LAST30DAYS_ENRICH_BUDGET_SECONDS', None),
489	# Opt-in strict exit: truthy -> CLI exits 3 when any source outcome is
490	# degraded (neither ok, no-results, nor skipped-unconfigured). #384.
491	('LAST30DAYS_STRICT_EXIT', None),
492	('LAST30DAYS_MEMORY_DIR', None),
493	# Optional local-only evidence source. Paths are separated with the
494	# platform path separator (":" on macOS/Linux, ";" on Windows).
495	('LAST30DAYS_CORPUS_DIRS', None),
496	# Corpus evidence is omitted from the stable agent JSON export unless
497	# this explicit privacy opt-in is truthy.
498	('LAST30DAYS_CORPUS_IN_EXPORT', None),
499	('LAST30DAYS_LIBRARY_OWNER', None),
500	('LAST30DAYS_LIBRARY_CONTEXT', 'on'),
501	('LAST30DAYS_PUBLISH_PASSWORD', None),
502	('OPENAI_MODEL_PIN', None),
503	('XAI_MODEL_PIN', None),
504	('OPENAI_BASE_URL', None),
505	('XAI_BASE_URL', None),
506	('OPENROUTER_BASE_URL', None),
507	('SCRAPECREATORS_API_KEY', None),
508	('APIFY_API_TOKEN', None),
509	('AUTH_TOKEN', None),
510	('CT0', None),
511	('BSKY_HANDLE', None),
512	('BSKY_APP_PASSWORD', None),
513	('BSKY_SEARCH_HOST', None),
514	('TRUTHSOCIAL_TOKEN', None),
515	('BRAVE_API_KEY', None),
516	('EXA_API_KEY', None),
517	('SERPER_API_KEY', None),
518	('OPENROUTER_API_KEY', None),
519	('PERPLEXITY_API_KEY', None),
520	('LAST30DAYS_PERPLEXITY_MODE', 'sonar'),
521	('LAST30DAYS_PERPLEXITY_MODEL', None),
522	('LAST30DAYS_PERPLEXITY_MAX_RESULTS', None),
523	('LAST30DAYS_PERPLEXITY_SEARCH_CONTEXT_SIZE', None),
524	('LAST30DAYS_PERPLEXITY_SEARCH_MODE', None),
525	('LAST30DAYS_PERPLEXITY_DOMAIN_FILTER', None),
526	('LAST30DAYS_PERPLEXITY_LANGUAGE_FILTER', None),
527	('LAST30DAYS_PERPLEXITY_COUNTRY', None),
528	('LAST30DAYS_PERPLEXITY_RECENCY_FILTER', None),
529	('LAST30DAYS_PERPLEXITY_REASONING_EFFORT', None),
530	('LAST30DAYS_PERPLEXITY_DEEP_TIMEOUT_SECONDS', '600'),
531	('PARALLEL_API_KEY', None),
532	('XQUIK_API_KEY', None),
533	# Host-native search signal: set by the SKILL.md agent-host path when the
534	# invoking runtime has its own (better) web-search tool, so the engine's
535	# keyless search floor stays off there. Defaults unset -> floor allowed.
536	('LAST30DAYS_NATIVE_SEARCH', None),
537	# Optional SearXNG instance for the keyless-search fallback rung.
538	('LAST30DAYS_SEARXNG_URL', None),
539	# Truthy -> disable Trustpilot's headless-Chrome WAF-cookie harvest in
540	# automated contexts (cron/CI/eval). Read by trustpilot._harvest_allowed.
541	('LAST30DAYS_TRUSTPILOT_NO_BROWSER', None),
542	('FROM_BROWSER', None),
543	('LAST30DAYS_TRUST_PROJECT_CONFIG', None),
544	('SETUP_COMPLETE', None),
545	('INCLUDE_SOURCES', ''),
546	('EXCLUDE_SOURCES', ''),
547	('LAST30DAYS_DEFAULT_SEARCH', ''),
548	# Resolve the user-facing default in last30days.py so an absent value
549	# stays distinguishable from an explicit `default`. That distinction
550	# lets the new key override legacy ELI5_MODE=true configurations.
551	('LAST30DAYS_REGISTER', None),
552	('FUN_LEVEL', 'medium'),
553	# Backward compatibility for configs written by the original `eli5 on`
554	# follow-up command. New writes use LAST30DAYS_REGISTER=eli5.
555	('ELI5_MODE', None),
556	('LAST30DAYS_YOUTUBE_SSH_HOST', None),
557	('LAST30DAYS_REPORT_CACHE_TTL_SECONDS', None),
558	('LAST30DAYS_VERIFY_FRESHNESS', None),
559	('LAST30DAYS_TRANSCRIPT_TIMEOUT', None),
560	('DEGRADED_TRANSCRIPT_THRESHOLD', None),
561	(KEYCHAIN_ALIASES_ENV, None),
562	# Whisper transcription provider for caption-free audio/video. Groq's
563	# free tier is preferred; OPENAI_API_KEY is the paid backstop (already
564	# resolved above via openai_auth).
565	('GROQ_API_KEY', None),
566	('LAST30DAYS_YT_SUB_LANGS', 'en,es,pt'),
567	('LAST30DAYS_YT_TRANSCRIPT_FAST_TIMEOUT', None),
568	('LAST30DAYS_YT_SEARCH_TIMEOUT', None),
569	('GITHUB_TOKEN', None),
570	]
571
572	for key, default in keys:
573	config[key] = os.environ.get(key) or merged_env.get(key, default)
574
575	# Export debug flag to os.environ so log.py's lazy os.environ.get()
576	# picks up .env values. setdefault ensures a shell-exported value is
577	# never overwritten by the (lower-priority) .env value.
578	if config.get('LAST30DAYS_DEBUG'):
579	os.environ.setdefault('LAST30DAYS_DEBUG', config['LAST30DAYS_DEBUG'])
580
581	# youtube_yt reads these tuning knobs lazily from os.environ, so values
582	# loaded from .env must be exported into the current engine process.
583	for key in (
584	'LAST30DAYS_YT_SUB_LANGS',
585	'LAST30DAYS_YT_TRANSCRIPT_FAST_TIMEOUT',
586	'LAST30DAYS_YT_SEARCH_TIMEOUT',
587	):
588	if config.get(key):
589	os.environ.setdefault(key, config[key])
590
591	# Backward-compat: ScrapeCreators' own examples and tutorials use the
592	# SCRAPE_CREATORS_API_KEY spelling (with underscore between SCRAPE and
593	# CREATORS). Accept that form too so users who follow the vendor's docs
594	# don't silently end up with has_scrapecreators=False. Canonical name
595	# wins when both are set.
596	if not config.get('SCRAPECREATORS_API_KEY'):
597	legacy = read_secret_env('SCRAPE_CREATORS_API_KEY') or merged_env.get('SCRAPE_CREATORS_API_KEY')
598	if legacy:
599	config['SCRAPECREATORS_API_KEY'] = legacy
600
601	# Multi-key rotation: comma-separated SCRAPECREATORS_API_KEY round-robins
602	# via random.choice per run. Originally added in #268, accidentally dropped
603	# in v3.0.6, restored here.
604	sc_key_raw = config.get('SCRAPECREATORS_API_KEY') or ''
605	if ',' in sc_key_raw:
606	import random
607	sc_keys = [k.strip() for k in sc_key_raw.split(',') if k.strip()]
608	config['SCRAPECREATORS_API_KEY'] = random.choice(sc_keys) if sc_keys else ''
609
610	# Track which config source was used (highest-priority file source wins
611	# the label; keychain is only reported when nothing else is configured).
612	if project_env_path:
613	config['_CONFIG_SOURCE'] = f'project:{project_env_path}'
614	elif CONFIG_FILE and CONFIG_FILE.exists():
615	config['_CONFIG_SOURCE'] = f'global:{CONFIG_FILE}'
616	elif keychain_env:
617	config['_CONFIG_SOURCE'] = 'keychain'
618	elif pass_env:
619	config['_CONFIG_SOURCE'] = 'pass'
620	else:
621	config['_CONFIG_SOURCE'] = 'env_only'
622	if ignored_project_env_path:
623	config['_IGNORED_PROJECT_CONFIG'] = str(ignored_project_env_path)
624	config['_IGNORED_PROJECT_CONFIG_KEYS'] = ignored_project_keys
625	config['_BROWSER_COOKIE_MODE'] = policy.browser_cookies
626	config['_BROWSER_COOKIE_BROWSERS'] = cookie_extraction_browsers(config)
627
628	if policy.browser_cookies == "read":
629	browser_creds = extract_browser_credentials(config)
630	for key, value in browser_creds.items():
631	if not config.get(key):
632	config[key] = value
633	config[f"_{key}_SOURCE"] = "browser"
634
635	return config
636
637
638	# ---------------------------------------------------------------------------
639	# Browser cookie extraction
640	# ---------------------------------------------------------------------------
641
642	COOKIE_DOMAINS: dict[str, dict[str, Any]] = {
643	"x": {
644	"domain": ".x.com",
645	"cookies": ["auth_token", "ct0"],
646	"mapping": {"auth_token": "AUTH_TOKEN", "ct0": "CT0"},
647	},
648	"truthsocial": {
649	"domain": ".truthsocial.com",
650	"cookies": ["_session_id"],
651	"mapping": {"_session_id": "TRUTHSOCIAL_TOKEN"},
652	},
653	}
654
655
656	def cookie_extraction_browsers(config: dict[str, Any]) -> list[str]:
657	"""Browsers to try for cookie extraction, honoring FROM_BROWSER.
658
659	Default (FROM_BROWSER unset): no browser-cookie reads. The Chromium family
660	(Chrome, Brave, Edge, Vivaldi, Opera, Arc, Chromium) is available only when
661	explicitly selected because reading their cookies on macOS requires the
662	browser's Safe Storage Keychain key, which triggers a system password prompt
663	that cannot be reliably suppressed. On Windows only Firefox cookie
664	extraction is supported; Chrome and Edge use DPAPI-encrypted cookie stores
665	that are not yet supported.
666
667	- ``FROM_BROWSER=<name>`` - a single browser (e.g. ``firefox``, ``brave``,
668	``edge``, ``arc``).
669	- ``FROM_BROWSER=firefox,safari`` - a comma-separated explicit browser list.
670	- ``FROM_BROWSER=auto`` - also try every Chromium browser (user accepts the
671	Keychain dialog when needed).
672	- ``FROM_BROWSER=off`` - returns [] (extraction disabled).
673
674	Returning the browser list from one place keeps the setup wizard and the
675	steady-state path on the same policy, so neither surprises the user with an
676	unrequested Keychain prompt.
677	"""
678	silent_browsers = ["firefox", "safari"]
679	chromium_browsers = ["chrome", "brave", "edge", "vivaldi", "opera", "arc", "chromium"]
680	known_browsers = silent_browsers + chromium_browsers
681	from_browser = (config.get("FROM_BROWSER") or "").strip().lower()
682	if not from_browser:
683	return []
684	if from_browser == "off":
685	return []
686	if from_browser == "auto":
687	return silent_browsers + chromium_browsers
688	if "," in from_browser:
689	requested = [b.strip() for b in from_browser.split(",") if b.strip()]
690	resolved = [b for b in requested if b in known_browsers]
691	unknown = [b for b in requested if b not in known_browsers]
692	if unknown:
693	sys.stderr.write(
694	"[last30days] WARNING: FROM_BROWSER ignored unrecognized browser(s): "
695	f"{', '.join(unknown)} (known: {', '.join(known_browsers)})\n"
696	)
697	sys.stderr.flush()
698	return resolved
699	if from_browser in known_browsers:
700	return [from_browser]
701	# Non-empty, not off/auto, not a known browser, not a list: unrecognized.
702	# Warn rather than fail silently so a typo (FROM_BROWSER=chrme) is visible
703	# instead of looking like "no cookies found".
704	sys.stderr.write(
705	f"[last30days] WARNING: FROM_BROWSER='{from_browser}' is not a recognized "
706	f"browser; no cookies will be read (known: {', '.join(known_browsers)}, "
707	"or 'auto'/'off')\n"
708	)
709	sys.stderr.flush()
710	return []
711
712
713
714	def extract_browser_credentials(config: dict[str, Any]) -> dict[str, str]:
715	"""Extract auth cookies from local browsers.
716
717	Browser selection (and the Chrome-prompt caveat) is handled by
718	``cookie_extraction_browsers``; this function just runs the extraction for
719	each configured cookie domain.
720	"""
721	browsers = cookie_extraction_browsers(config)
722	if not browsers:
723	return {}
724	try:
725	from . import cookie_extract
726	except ImportError:
727	return {}
728	extracted: dict[str, str] = {}
729	for _service, spec in COOKIE_DOMAINS.items():
730	if all(config.get(env_key) for env_key in spec["mapping"].values()):
731	continue
732	for browser in browsers:
733	try:
734	cookies = cookie_extract.extract_cookies(browser, spec["domain"], spec["cookies"])
735	except Exception:
736	continue
737	if cookies:
738	for cookie_name, env_key in spec["mapping"].items():
739	if cookie_name in cookies and not config.get(env_key):
740	extracted[env_key] = cookies[cookie_name]
741	break # Found cookies for this service, stop trying browsers
742	return extracted
743
744
745	def get_x_source_with_method(config: dict[str, Any]) -> tuple[str \| None, str]:
746	"""Return (source, method) for X search, where method describes the auth origin."""
747	if config.get("XAI_API_KEY"):
748	return "xai", "xai"
749	if config.get("AUTH_TOKEN") and config.get("CT0"):
750	method = config.get("_AUTH_TOKEN_SOURCE", "env")
751	return "bird", method
752	# Fall back to xurl CLI (official X API v2, OAuth2, free developer app)
753	from . import xurl_x
754	if xurl_x.is_available():
755	return "xurl", "oauth2"
756	return None, "none"
757
758
759	def config_exists(policy: ConfigLoadPolicy \| None = None) -> bool:
760	"""Check if any configuration source exists."""
761	policy = policy or ConfigLoadPolicy()
762	file_env = load_env_file(CONFIG_FILE) if CONFIG_FILE and CONFIG_FILE.exists() else {}
763	if _project_config_trusted(policy, file_env) and _find_project_env():
764	return True
765	if CONFIG_FILE:
766	return CONFIG_FILE.exists()
767	return False
768
769
770	def get_reddit_source(config: dict[str, Any]) -> str \| None:
771	"""Determine which Reddit backend to use.
772
773	Returns: 'scrapecreators' or None
774	"""
775	if config.get('SCRAPECREATORS_API_KEY'):
776	return 'scrapecreators'
777	return None
778
779
780	# Default X backend priority. The first available backend is the primary X
781	# source; the rest are ordered failover backups, tried only if the one before
782	# returns nothing or errors. There is one X source ("x"); these are its
783	# interchangeable backends, never run in parallel.
784	# xai — xAI/Grok live search (XAI_API_KEY)
785	# bird — X GraphQL scrape via the user's browser cookies (AUTH_TOKEN/CT0)
786	# xurl — official X API v2 (xurl CLI, OAuth2)
787	# xquik — key-based REST X search (XQUIK_API_KEY); keyless of browser cookies
788	_X_BACKEND_ORDER = ("xai", "bird", "xurl", "xquik")
789
790	# Public routing definitions for the doctor/backend-descriptor layer
791	# (lib/backends.py). These are aliases for knowledge this module already
792	# owns — the declared X chain order and the pin/floor env var names — so
793	# descriptors import one source of truth instead of restating it.
794	X_BACKEND_ORDER = _X_BACKEND_ORDER
795	X_BACKEND_PIN_VAR = 'LAST30DAYS_X_BACKEND'
796	REDDIT_BACKEND_PIN_VAR = 'LAST30DAYS_REDDIT_BACKEND'
797	REDDIT_SC_MIN_ITEMS_VAR = 'LAST30DAYS_REDDIT_SC_MIN_ITEMS'
798
799
800	def _x_backend_available(
801	backend: str,
802	config: dict[str, Any],
803	has_bird_creds: bool,
804	local_only: bool = False,
805	) -> bool:
806	if backend == 'xai':
807	return bool(config.get('XAI_API_KEY'))
808	if backend == 'bird':
809	from . import bird_x
810	return has_bird_creds and bird_x.is_bird_installed()
811	if backend == 'xurl':
812	from . import xurl_x
813	if local_only:
814	# Doctor/safe-diagnose path: local evidence only (PATH lookup +
815	# token store) — never the live `xurl whoami` network call.
816	return xurl_x.has_stored_auth()
817	return xurl_x.is_available()
818	if backend == 'xquik':
819	return is_xquik_available(config)
820	return False
821
822
823	def x_backend_chain(config: dict[str, Any], local_only: bool = False) -> list[str]:
824	"""Ordered list of available X backends.
825
826	``chain[0]`` is the default X source; the remaining entries are failover
827	backups, used only when the one before yields no items or errors. There is
828	exactly one X source — these are its backends, never fetched in parallel.
829
830	A ``LAST30DAYS_X_BACKEND`` pin forces a single backend (no failover): the
831	user explicitly chose it. Browser-cookie probing is intentionally avoided
832	(automatic Keychain access causes popups); bird counts as available only
833	when AUTH_TOKEN and CT0 are present explicitly.
834
835	``local_only=True`` is the doctor/safe-diagnose flavor: availability is
836	answered from local evidence only (no subprocess spawns that reach the
837	network — xurl's live `whoami` check is replaced by its on-disk token
838	store). Research-time callers keep the default live semantics.
839	"""
840	from . import bird_x
841	has_bird_creds = bool(config.get('AUTH_TOKEN') and config.get('CT0'))
842	if has_bird_creds:
843	bird_x.set_credentials(config.get('AUTH_TOKEN'), config.get('CT0'))
844
845	preferred = (config.get(X_BACKEND_PIN_VAR) or '').lower()
846	if preferred in _X_BACKEND_ORDER:
847	if _x_backend_available(preferred, config, has_bird_creds, local_only):
848	return [preferred]
849	return []
850
851	return [
852	b for b in _X_BACKEND_ORDER
853	if _x_backend_available(b, config, has_bird_creds, local_only)
854	]
855
856
857	def get_x_source(config: dict[str, Any], local_only: bool = False) -> str \| None:
858	"""The default (primary) X backend, or None if no X source is available.
859
860	Thin wrapper over ``x_backend_chain`` returning the first/primary backend;
861	callers that want failover should use ``x_backend_chain`` directly.
862	``local_only`` is forwarded (see ``x_backend_chain``).
863	"""
864	chain = x_backend_chain(config, local_only=local_only)
865	return chain[0] if chain else None
866
867
868	def x_pending_browser_auth(config: dict[str, Any], local_only: bool = False) -> bool:
869	"""True when X is not available now but ``FROM_BROWSER`` will authenticate it at run time.
870
871	``--diagnose`` / ``--preflight`` load config in ``plan_only`` mode, which
872	deliberately skips browser-cookie extraction (no Keychain popup,
873	``reads_values: false``). As a result ``get_x_source`` returns None and X is
874	dropped from ``available_sources`` even though a normal run would extract the
875	same cookies and authenticate X fine. This predicate reports that
876	"available pending browser auth" state without reading a single cookie — it
877	keys only on the already-resolved browser list (``cookie_extraction_browsers``
878	derives it from ``FROM_BROWSER`` alone, no secrets), bird being installed, and
879	X having a cookie-domain mapping. Side-effect free, so the safe-inspection
880	contract of diagnose/preflight is preserved.
881
882	Returns False whenever X is already available outright (static AUTH_TOKEN/CT0,
883	or xAI/xurl/xquik backend), and in ``read`` mode (a real run has already
884	extracted creds, so its status must be unchanged — never "pending").
885	"""
886	# Already available via a static backend (bird creds, xAI, xurl, xquik).
887	# local_only (doctor/safe-diagnose) answers the xurl leg from the token
888	# store instead of the live `xurl whoami` network call.
889	if get_x_source(config, local_only=local_only):
890	return False
891	# Only meaningful in inspection modes that skip extraction; a real ``read``
892	# run has already attempted extraction and must report its true state.
893	if config.get('_BROWSER_COOKIE_MODE') == 'read':
894	return False
895	if 'x' not in COOKIE_DOMAINS:
896	return False
897	if not cookie_extraction_browsers(config):
898	return False
899	from . import bird_x
900	return bird_x.is_bird_installed()
901
902
903	def is_ytdlp_available() -> bool:
904	"""Check if yt-dlp is installed for YouTube search."""
905	from . import youtube_yt
906	return youtube_yt.is_ytdlp_installed()
907
908
909	def is_youtube_comments_available(config: dict[str, Any]) -> bool:
910	"""Check if YouTube comment enrichment is available.
911
912	yt-dlp fetches YouTube comments free and keyless, so when it is installed
913	comments need no credential and no ``INCLUDE_SOURCES`` opt-in — the opt-in
914	only ever existed to gate ScrapeCreators credit spend, and there is none to
915	gate. ``EXCLUDE_SOURCES=youtube_comments`` remains the off-switch.
916
917	Without yt-dlp, the legacy ScrapeCreators path still applies: it requires
918	SCRAPECREATORS_API_KEY AND ``youtube_comments`` in ``INCLUDE_SOURCES``
919	(mirroring ``is_tiktok_comments_available``), bounded by
920	``enrich_with_comments(max_videos=3)`` at ~3 credits per run.
921	"""
922	if 'youtube_comments' in _parse_exclude_sources(config):
923	return False
924	if is_ytdlp_available():
925	return True
926	if not config.get('SCRAPECREATORS_API_KEY'):
927	return False
928	return 'youtube_comments' in _parse_include_sources(config)
929
930
931	def is_tiktok_comments_available(config: dict[str, Any]) -> bool:
932	"""Check if TikTok comment enrichment is available.
933
934	Requires SCRAPECREATORS_API_KEY AND tiktok_comments in INCLUDE_SOURCES.
935	Mirrors the youtube_comments opt-in pattern.
936	"""
937	if not config.get('SCRAPECREATORS_API_KEY'):
938	return False
939	include = _parse_include_sources(config)
940	return 'tiktok_comments' in include
941
942
943	def is_instagram_comments_available(config: dict[str, Any]) -> bool:
944	"""Check if Instagram comment enrichment is available.
945
946	Requires SCRAPECREATORS_API_KEY AND instagram_comments in INCLUDE_SOURCES.
947	Mirrors the youtube_comments / tiktok_comments opt-in pattern. Comments are
948	fetched via ScrapeCreators (GET /v2/instagram/post/comments) with each
949	comment's ``comment_like_count`` used as its vote for ranking. Part of the
950	default onboarding tier (posts on -> comments on for TikTok/Instagram/YouTube).
951	"""
952	if not config.get('SCRAPECREATORS_API_KEY'):
953	return False
954	return 'instagram_comments' in _parse_include_sources(config)
955
956
957	def is_youtube_sc_available(config: dict[str, Any]) -> bool:
958	"""Check if ScrapeCreators YouTube search fallback is available.
959
960	Used when yt-dlp is not installed or fails.
961	"""
962	return bool(config.get('SCRAPECREATORS_API_KEY'))
963
964
965	def is_hackernews_available() -> bool:
966	"""Check if Hacker News source is available.
967
968	Always returns True - HN uses free Algolia API, no key needed.
969	"""
970	return True
971
972
973	def is_native_search(config: dict[str, Any]) -> bool:
974	"""Whether the invoking host has its own (better) native web search.
975
976	Defined by capability, not host identity: the SKILL.md agent-host path sets
977	``LAST30DAYS_NATIVE_SEARCH`` when the runtime actually has a native web-search
978	tool (e.g. Claude Code's WebSearch). When true, the engine's keyless search
979	floor is suppressed so a worse free search never preempts the model's own.
980	Defaults False (unset), so headless/cron and hosts without native search fall
981	to the keyless floor.
982	"""
983	raw = config.get('LAST30DAYS_NATIVE_SEARCH')
984	if raw is None:
985	return False
986	return str(raw).strip().lower() in ('1', 'true', 'yes', 'on')
987
988
989	def keyless_web_allowed(config: dict[str, Any]) -> bool:
990	"""Whether the engine may use its keyless web-search floor for this run.
991
992	Allowed only when the host does NOT have native search. Independent of
993	whether a paid key is set (the grounding dispatcher prefers paid first and
994	falls to keyless on empty/error for non-native runs).
995	"""
996	return not is_native_search(config)
997
998
999	def transcription_providers(config: dict[str, Any]) -> list[tuple[str, str]]:
1000	"""Ordered (name, api_key) Whisper providers for caption-free transcription.
1001
1002	Groq (free tier) first, OpenAI (paid) as the backstop. Empty when neither
1003	key is set, in which case transcription degrades rather than runs.
1004	"""
1005	providers: list[tuple[str, str]] = []
1006	if config.get('GROQ_API_KEY'):
1007	providers.append(('groq', config['GROQ_API_KEY']))
1008	if config.get('OPENAI_API_KEY'):
1009	providers.append(('openai', config['OPENAI_API_KEY']))
1010	return providers
1011
1012
1013	def is_bluesky_available(config: dict[str, Any]) -> bool:
1014	"""Check if Bluesky source is available.
1015
1016	Requires BSKY_HANDLE and BSKY_APP_PASSWORD (app password from bsky.app/settings).
1017	"""
1018	return bool(config.get('BSKY_HANDLE') and config.get('BSKY_APP_PASSWORD'))
1019
1020
1021	def is_truthsocial_available(config: dict[str, Any]) -> bool:
1022	"""Check if Truth Social source is available.
1023
1024	Requires TRUTHSOCIAL_TOKEN (bearer token from browser dev tools).
1025	"""
1026	return bool(config.get('TRUTHSOCIAL_TOKEN'))
1027
1028
1029	def is_polymarket_available() -> bool:
1030	"""Check if Polymarket source is available.
1031
1032	Always returns True - Gamma API is free, no key needed.
1033	"""
1034	return True
1035
1036
1037	def is_tiktok_available(config: dict[str, Any]) -> bool:
1038	"""Check if TikTok source is available (ScrapeCreators or legacy Apify).
1039
1040	Returns True if SCRAPECREATORS_API_KEY or APIFY_API_TOKEN is set.
1041	"""
1042	return bool(config.get('SCRAPECREATORS_API_KEY') or config.get('APIFY_API_TOKEN'))
1043
1044
1045	def get_tiktok_token(config: dict[str, Any]) -> str:
1046	"""Get TikTok API token, preferring ScrapeCreators over legacy Apify."""
1047	return config.get('SCRAPECREATORS_API_KEY') or config.get('APIFY_API_TOKEN') or ''
1048
1049
1050	def _parse_include_sources(config: dict[str, Any]) -> set[str]:
1051	"""Parse INCLUDE_SOURCES config value into a set of lowercase source names."""
1052	raw = config.get('INCLUDE_SOURCES') or ''
1053	return {s.strip().lower() for s in raw.split(',') if s.strip()}
1054
1055
1056	def _parse_exclude_sources(config: dict[str, Any]) -> set[str]:
1057	"""Parse EXCLUDE_SOURCES config value into a set of lowercase source names."""
1058	raw = config.get('EXCLUDE_SOURCES') or ''
1059	return {s.strip().lower() for s in raw.split(',') if s.strip()}
1060
1061
1062	def include_sources(config: dict[str, Any]) -> set[str]:
1063	"""Public view of the parsed INCLUDE_SOURCES set.
1064
1065	Thin wrapper over ``_parse_include_sources`` so other modules (doctor,
1066	etc.) don't reach into env's privates.
1067	"""
1068	return _parse_include_sources(config)
1069
1070
1071	def is_setup_complete(config: dict[str, Any]) -> bool:
1072	"""Whether guided setup marked this config complete (SETUP_COMPLETE truthy).
1073
1074	Thin wrapper over ``_truthy`` so other modules don't reach into env's
1075	privates.
1076	"""
1077	return _truthy(config.get('SETUP_COMPLETE'))
1078
1079
1080	def is_threads_available(config: dict[str, Any]) -> bool:
1081	"""Check if the Threads credential is available.
1082
1083	Returns True when SCRAPECREATORS_API_KEY is set. This is an availability
1084	predicate only: whether Threads is actually scheduled is gated in the
1085	pipeline's ``available_sources`` by an ``INCLUDE_SOURCES=threads`` opt-in
1086	(the onboarding "Everything" tier), so a key alone no longer runs Threads.
1087	"""
1088	return bool(config.get('SCRAPECREATORS_API_KEY'))
1089
1090
1091	def is_instagram_available(config: dict[str, Any]) -> bool:
1092	"""Check if Instagram source is available (ScrapeCreators).
1093
1094	Returns True if SCRAPECREATORS_API_KEY is set.
1095	Instagram uses the same key as TikTok.
1096	"""
1097	return bool(config.get('SCRAPECREATORS_API_KEY'))
1098
1099
1100	def get_instagram_token(config: dict[str, Any]) -> str:
1101	"""Get Instagram API token (same ScrapeCreators key as TikTok)."""
1102	return config.get('SCRAPECREATORS_API_KEY') or ''
1103
1104
1105	def get_xiaohongshu_api_base(config: dict[str, Any]) -> str:
1106	"""Get Xiaohongshu HTTP API base URL.
1107
1108	The availability probe caches the first logged-in local service it finds so
1109	the later search request uses the same browser-backed session endpoint.
1110	"""
1111	cached = config.get(XIAOHONGSHU_RESOLVED_API_BASE_KEY)
1112	if cached:
1113	return str(cached).rstrip("/")
1114
1115	explicit = config.get("XIAOHONGSHU_API_BASE")
1116	if explicit:
1117	return str(explicit).rstrip("/")
1118
1119	return XIAOHONGSHU_DEFAULT_API_BASES[0]
1120
1121
1122	def _xiaohongshu_api_base_candidates(config: dict[str, Any]) -> list[str]:
1123	explicit = config.get("XIAOHONGSHU_API_BASE")
1124	if explicit:
1125	return [str(explicit).rstrip("/")]
1126
1127	candidates: list[str] = []
1128	cached = config.get(XIAOHONGSHU_RESOLVED_API_BASE_KEY)
1129	if cached:
1130	candidates.append(str(cached).rstrip("/"))
1131
1132	for base in XIAOHONGSHU_DEFAULT_API_BASES:
1133	if base not in candidates:
1134	candidates.append(base)
1135	return candidates
1136
1137
1138	def _xiaohongshu_base_logged_in(base: str, http_module: Any) -> bool:
1139	# Keep the health probe snappy, but allow one retry for transient hiccups.
1140	health = http_module.get(f"{base}/health", timeout=3, retries=2)
1141	if not isinstance(health, dict):
1142	return False
1143	if not health.get("success"):
1144	return False
1145
1146	# Login checks can be slower because some services consult the browser
1147	# profile/session, so use a slightly longer timeout than the health probe.
1148	login = http_module.get(f"{base}/api/v1/login/status", timeout=8, retries=2)
1149	is_logged_in = (
1150	login.get("data", {}).get("is_logged_in")
1151	if isinstance(login, dict) else False
1152	)
1153	return bool(is_logged_in)
1154
1155
1156	def is_xiaohongshu_available(config: dict[str, Any]) -> bool:
1157	"""Check whether Xiaohongshu HTTP API is reachable and logged in."""
1158	# Import here to avoid heavy imports at module load.
1159	from . import http
1160
1161	for base in _xiaohongshu_api_base_candidates(config):
1162	try:
1163	if _xiaohongshu_base_logged_in(base, http):
1164	config[XIAOHONGSHU_RESOLVED_API_BASE_KEY] = base
1165	return True
1166	except (OSError, http.HTTPError):
1167	continue
1168	except Exception as exc:
1169	sys.stderr.write(
1170	f"[last30days] WARNING: unexpected error checking Xiaohongshu "
1171	f"at {base}: {type(exc).__name__}: {exc}\n"
1172	)
1173	sys.stderr.flush()
1174	return False
1175
1176
1177	# Backward compat alias
1178	is_apify_available = is_tiktok_available
1179
1180
1181	def get_x_source_status(config: dict[str, Any], probe: bool = False) -> dict[str, Any]:
1182	"""Get detailed X source status for UI decisions.
1183
1184	Args:
1185	probe: when True, run a cheap 1-tweet bird probe and downgrade
1186	``bird_authenticated`` to False when X clearly returns nothing,
1187	so ``--diagnose`` reflects runtime reality instead of static
1188	credential presence. A transient timeout leaves the status
1189	unchanged (fail open). When False (the safe/diagnose path that
1190	doctor uses), NO network is touched: xurl availability comes
1191	from local evidence (``xurl_x.has_stored_auth``), never the
1192	live ``xurl whoami`` call.
1193
1194	Returns:
1195	Dict with keys: source, bird_installed, bird_authenticated,
1196	bird_username, xai_available, can_install_bird
1197	"""
1198	from . import bird_x
1199
1200	if config.get('AUTH_TOKEN') and config.get('CT0'):
1201	bird_x.set_credentials(config.get('AUTH_TOKEN'), config.get('CT0'))
1202	bird_status = bird_x.get_bird_status()
1203	xai_available = bool(config.get('XAI_API_KEY'))
1204
1205	# Report the TRUE auth lane (browser / env / keychain) rather than the static
1206	# "env AUTH_TOKEN" label — tokens usually come from live browser cookies, and
1207	# mislabeling the lane sent past debugging down a 30-minute wrong path.
1208	if bird_status["authenticated"]:
1209	lane = config.get('_AUTH_TOKEN_SOURCE') or 'env'
1210	bird_status["username"] = f"{lane} AUTH_TOKEN"
1211
1212	# Optional runtime probe: don't show X green when it's effectively dead.
1213	if probe and bird_status["authenticated"]:
1214	if bird_x.probe_works() is False:
1215	bird_status["authenticated"] = False
1216	bird_status["username"] = "probe failed (no working X auth)"
1217
1218	# Xquik: the key-based X source used when bird's cookie auth isn't available.
1219	# Probe so --diagnose reports the true state — funded, or configured-but-
1220	# unpaid (402) — instead of false-green on mere key presence.
1221	xquik_available = is_xquik_available(config)
1222	xquik_working: bool \| None = None
1223	xquik_status = ""
1224	if xquik_available:
1225	if probe:
1226	from . import xquik
1227	xquik_working = xquik.probe_works(get_xquik_token(config))
1228	xquik_status = xquik.probe_reason()
1229	else:
1230	xquik_status = "configured (not probed)"
1231
1232	# Xurl availability, computed ONCE. probe=True (a live diagnose) may run
1233	# the real `xurl whoami`; probe=False is the safe path (doctor,
1234	# --diagnose, --preflight) and must stay local-only — the live check is
1235	# an authenticated X API network call.
1236	from . import xurl_x as _xurl_x
1237	xurl_available = _xurl_x.is_available() if probe else _xurl_x.has_stored_auth()
1238
1239	# Determine active source. bird (browser cookies) and xAI win when present;
1240	# when neither is available, xquik is the active X source. A probe that
1241	# clearly failed (False) means xquik is not actually usable.
1242	if bird_status["authenticated"]:
1243	source = 'bird'
1244	elif xai_available:
1245	source = 'xai'
1246	else:
1247	if xurl_available:
1248	source = 'xurl'
1249	elif xquik_available and xquik_working is not False:
1250	source = 'xquik'
1251	else:
1252	source = None
1253
1254	return {
1255	"source": source,
1256	"bird_installed": bird_status["installed"],
1257	"bird_authenticated": bird_status["authenticated"],
1258	"bird_username": bird_status["username"],
1259	"xai_available": xai_available,
1260	"xurl_available": xurl_available,
1261	"xquik_available": xquik_available,
1262	"xquik_working": xquik_working,
1263	"xquik_status": xquik_status,
1264	"can_install_bird": bird_status["can_install"],
1265	}
1266
1267
1268	# Pinterest
1269	def is_pinterest_available(config: dict[str, Any]) -> bool:
1270	"""Check if Pinterest source is available.
1271
1272	Returns True when SCRAPECREATORS_API_KEY is set AND 'pinterest' is in
1273	INCLUDE_SOURCES (or requested_sources at the pipeline level). Pinterest
1274	is opt-in because not every topic benefits from visual pin results.
1275	"""
1276	return bool(config.get('SCRAPECREATORS_API_KEY'))
1277
1278
1279	def get_pinterest_token(config: dict[str, Any]) -> str:
1280	"""Get Pinterest API token (same ScrapeCreators key as TikTok/Instagram)."""
1281	return config.get('SCRAPECREATORS_API_KEY') or ''
1282
1283
1284	# Xquik
1285	def is_xquik_available(config: dict[str, Any]) -> bool:
1286	"""Check if Xquik X search source is available.
1287
1288	Requires XQUIK_API_KEY (API key from xquik.com).
1289	"""
1290	return bool(config.get('XQUIK_API_KEY'))
1291
1292
1293	def get_xquik_token(config: dict[str, Any]) -> str:
1294	"""Get Xquik API key."""
1295	return config.get('XQUIK_API_KEY') or ''
1296

1296 lines PYTHON