sawz/last30days-skill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: v2.9.6 — free-first NUX, cookie extraction, quality scoring

Browse Source 
Setup wizard with consent-first cookie extraction (Chrome/Firefox/Safari),
yt-dlp auto-install, ScrapeCreators push, quality scoring (5 core sources),
status banner redesign, honest Reddit labeling, inline YouTube transcripts,
Exa free web search, Reddit public fallback, and post-research quality nudge.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Matt Van Horn
2026-03-29 14:32:52 -07:00
parent 4d6224f79a
commit 3803f3df4b
34 changed files with 6238 additions and 320 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/last30days.py
+124 -79
View File
@@ -149,17 +149,20 @@ from lib import (
normalize,
openai_reddit,
reddit,
reddit_public,
reddit_enrich,
render,
schema,
score,
scrapecreators_x,
setup_wizard,
ui,
tiktok,
instagram,
websearch,
xai_x,
youtube_yt,
quality_nudge,
query_type as qt,
)
@@ -184,8 +187,10 @@ def _search_reddit(
) -> tuple:
"""Search Reddit (runs in thread).
Uses ScrapeCreators when SCRAPECREATORS_API_KEY is available (preferred).
Falls back to OpenAI Responses API otherwise.
Hierarchy:
1. ScrapeCreators (if SCRAPECREATORS_API_KEY exists) — premium, best quality
2. Public Reddit JSON (always available) — free, good for thread discovery
3. OpenAI Responses API — legacy fallback for backwards compatibility
Returns:
Tuple of (reddit_items, raw_response, error, used_scrapecreators)
@@ -199,7 +204,7 @@ def _search_reddit(
if mock:
raw_response = load_fixture("openai_sample.json")
elif sc_token:
# === ScrapeCreators path (preferred) ===
# === Tier 1: ScrapeCreators path (preferred) ===
used_scrapecreators = True
try:
sys.stderr.write("[Reddit] Using ScrapeCreators API\n")
@@ -216,61 +221,56 @@ def _search_reddit(
reddit_error = f"ScrapeCreators: {type(e).__name__}: {e}"
sys.stderr.write(f"[Reddit] ScrapeCreators failed: {e}\n")
sys.stderr.flush()
# Fall through to OpenAI if we have that key
if not config.get("OPENAI_API_KEY"):
# No OpenAI either: try public Reddit fallback.
try:
reddit_items = openai_reddit.search_reddit_public(
topic, from_date, to_date, depth=depth,
)
raw_response = {"source": "reddit_public", "items": reddit_items}
return reddit_items, raw_response, None, False
except Exception as e2:
return [], {"error": str(e)}, reddit_error, used_scrapecreators
used_scrapecreators = False
sys.stderr.write("[Reddit] Falling back to OpenAI\n")
sys.stderr.flush()
# Fall through to Tier 2 (public JSON)
# === OpenAI path (fallback) ===
# === Tier 2: Public Reddit JSON (free, always available) ===
if not mock:
if config.get("OPENAI_API_KEY"):
try:
raw_response = openai_reddit.search_reddit(
config["OPENAI_API_KEY"],
selected_models["openai"],
topic,
from_date,
to_date,
depth=depth,
auth_source=config.get("OPENAI_AUTH_SOURCE", "api_key"),
account_id=config.get("OPENAI_CHATGPT_ACCOUNT_ID"),
)
except http.HTTPError as e:
raw_response = {"error": str(e)}
reddit_error = f"API error: {e}"
except Exception as e:
raw_response = {"error": str(e)}
reddit_error = f"{type(e).__name__}: {e}"
else:
# No OpenAI auth: direct Reddit public JSON fallback.
try:
reddit_items = openai_reddit.search_reddit_public(
topic, from_date, to_date, depth=depth,
)
try:
sys.stderr.write("[Reddit] Trying public Reddit JSON\n")
sys.stderr.flush()
reddit_items = reddit_public.search_reddit_public(
topic, from_date, to_date, depth=depth,
)
if reddit_items:
raw_response = {"source": "reddit_public", "items": reddit_items}
except http.HTTPError as e:
reddit_items = []
raw_response = {"error": str(e), "source": "reddit_public"}
reddit_error = f"Reddit public API error: {e}"
except Exception as e:
reddit_items = []
raw_response = {"error": str(e), "source": "reddit_public"}
reddit_error = f"Reddit public search error: {type(e).__name__}: {e}"
sys.stderr.write(f"[Reddit] Public JSON returned {len(reddit_items)} results\n")
sys.stderr.flush()
return reddit_items, raw_response, None, False
# Empty results — fall through to Tier 3
sys.stderr.write("[Reddit] Public JSON returned 0 results, trying OpenAI\n")
sys.stderr.flush()
except Exception as e:
sys.stderr.write(f"[Reddit] Public JSON failed: {e}\n")
sys.stderr.flush()
# Fall through to Tier 3
# Parse response
# === Tier 3: OpenAI Responses API (legacy fallback) ===
if not mock and config.get("OPENAI_API_KEY"):
try:
sys.stderr.write("[Reddit] Falling back to OpenAI Responses API\n")
sys.stderr.flush()
raw_response = openai_reddit.search_reddit(
config["OPENAI_API_KEY"],
selected_models["openai"],
topic,
from_date,
to_date,
depth=depth,
auth_source=config.get("OPENAI_AUTH_SOURCE", "api_key"),
account_id=config.get("OPENAI_CHATGPT_ACCOUNT_ID"),
)
except http.HTTPError as e:
raw_response = {"error": str(e)}
reddit_error = f"API error: {e}"
except Exception as e:
raw_response = {"error": str(e)}
reddit_error = f"{type(e).__name__}: {e}"
# Parse response (OpenAI path)
reddit_items = openai_reddit.parse_reddit_response(raw_response or {})
# Quick retry with simpler query if few results
# Quick retry with simpler query if few results (OpenAI path only)
if len(reddit_items) < 5 and not mock and not reddit_error and config.get("OPENAI_API_KEY"):
core = openai_reddit._extract_core_subject(topic)
if core.lower() != topic.lower():
@@ -292,7 +292,7 @@ def _search_reddit(
except Exception:
pass
# Subreddit-targeted fallback if still < 3 results
# Subreddit-targeted fallback if still < 3 results (OpenAI path only)
if len(reddit_items) < 3 and not mock and not reddit_error and config.get("OPENAI_API_KEY"):
sub_query = openai_reddit._build_subreddit_query(topic)
try:
@@ -617,7 +617,7 @@ def _search_web(
Tuple of (web_items, web_error)
web_items are raw dicts ready for websearch.normalize_websearch_items()
"""
from lib import brave_search, parallel_search, openrouter_search
from lib import brave_search, parallel_search, openrouter_search, exa_search
backend = env.get_web_search_source(config)
if not backend:
@@ -627,7 +627,11 @@ def _search_web(
raw_results = []
try:
if backend == "parallel":
if backend == "exa":
raw_results = exa_search.search_web(
topic, from_date, to_date, config["EXA_API_KEY"], depth=depth,
)
elif backend == "parallel":
raw_results = parallel_search.search_web(
topic, from_date, to_date, config["PARALLEL_API_KEY"], depth=depth,
)
@@ -1537,12 +1541,29 @@ def main():
# Load config
config = env.get_config()
# Inject .env credentials into Bird module before auth check
bird_x.set_credentials(config.get('AUTH_TOKEN'), config.get('CT0'))
# Detect first run (no SETUP_COMPLETE in config)
first_run = setup_wizard.is_first_run(config)
# On first run, block Bird's Node.js sweet-cookie scanner from probing
# browser cookies before the user has given consent via the setup wizard.
# Explicit AUTH_TOKEN (from env var) is fine — only block browser scanning.
if first_run and config.get('_AUTH_TOKEN_SOURCE') != 'env':
os.environ['BIRD_DISABLE_BROWSER_COOKIES'] = '1'
# Inject .env credentials into Bird module before auth check.
# On first run (no SETUP_COMPLETE), only inject explicit env var credentials —
# skip if AUTH_TOKEN came from browser cookies (no consent yet).
if first_run:
auth_source = config.get('_AUTH_TOKEN_SOURCE')
if auth_source == 'env':
bird_x.set_credentials(config.get('AUTH_TOKEN'), config.get('CT0'))
else:
bird_x.set_credentials(config.get('AUTH_TOKEN'), config.get('CT0'))
# Auto-detect Bird (no prompts - just use it if available)
x_source_status = env.get_x_source_status(config)
x_source = x_source_status["source"] # 'bird', 'xai', or None
x_method = x_source_status.get("method") # 'env', 'browser-firefox', 'api', etc.
# Auto-detect yt-dlp for YouTube search
has_ytdlp = env.is_ytdlp_available()
@@ -1570,6 +1591,7 @@ def main():
"reddit_public": True,
"xai": bool(config.get("XAI_API_KEY")),
"x_source": x_source_status["source"],
"x_method": x_source_status.get("method"),
"bird_installed": x_source_status["bird_installed"],
"bird_authenticated": x_source_status["bird_authenticated"],
"bird_username": x_source_status.get("bird_username"),
@@ -1583,6 +1605,7 @@ def main():
"truthsocial": has_truthsocial,
"polymarket": True,
"web_search_backend": web_source,
"exa": bool(config.get("EXA_API_KEY")),
"parallel_ai": bool(config.get("PARALLEL_API_KEY")),
"brave": bool(config.get("BRAVE_API_KEY")),
"openrouter": bool(config.get("OPENROUTER_API_KEY")),
@@ -1590,6 +1613,19 @@ def main():
print(json.dumps(diag, indent=2))
sys.exit(0)
# Handle 'setup' subcommand
if args.topic and args.topic.strip().lower() == "setup":
results = setup_wizard.run_auto_setup(config)
# Write config
env_path = env.CONFIG_FILE
if env_path:
written = setup_wizard.write_setup_config(env_path)
results["env_written"] = written
else:
results["env_written"] = False
print(setup_wizard.get_setup_status_text(results))
sys.exit(0)
# Validate topic (--diagnose doesn't need one)
if not args.topic:
print("Error: Please provide a topic to research.", file=sys.stderr)
@@ -1599,40 +1635,30 @@ def main():
# Initialize progress display with topic
progress = ui.ProgressDisplay(args.topic, show_banner=True)
# Show diagnostic banner when sources are missing
# Show status banner (free-first design — lead with what works)
web_source = env.get_web_search_source(config)
reddit_source = env.get_reddit_source(config)
diag = {
"openai": bool(config.get("OPENAI_API_KEY")),
"reddit_public": True,
"xai": bool(config.get("XAI_API_KEY")),
"setup_complete": bool(config.get("SETUP_COMPLETE")),
"reddit_source": reddit_source, # 'scrapecreators', 'openai', or None
"x_source": x_source_status["source"],
"bird_installed": x_source_status["bird_installed"],
"bird_authenticated": x_source_status["bird_authenticated"],
"bird_username": x_source_status.get("bird_username"),
"x_method": x_source_status.get("method"),
"youtube": has_ytdlp,
"tiktok": has_tiktok,
"instagram": has_instagram,
"xiaohongshu": has_xiaohongshu,
"hackernews": True,
"bluesky": True,
"truthsocial": has_truthsocial,
"polymarket": True,
"bluesky": has_bluesky,
"truthsocial": has_truthsocial,
"xiaohongshu": has_xiaohongshu,
"scrapecreators": bool(config.get("SCRAPECREATORS_API_KEY")),
"web_search_backend": "deferred to assistant" if args.no_native_web else web_source,
}
ui.show_diagnostic_banner(diag)
# Check available sources (accounting for Bird auto-detection)
# Check available sources (now accounts for Bird/cookie auth automatically)
available = env.get_available_sources(config)
# Override available if Bird provides X
if x_source == 'bird':
if available == 'reddit':
available = 'both' # Now have both Reddit + X
elif available == 'reddit-web':
available = 'all' # Reddit + X + Web
elif available == 'web':
available = 'x-web' # X + Web
# Mock mode can work without keys
if args.mock:
if args.sources == "auto":
@@ -1902,9 +1928,9 @@ def main():
elif has_ytdlp and not report.youtube:
source_info["youtube_skip_reason"] = "0 results (query may be too specific)"
if not has_tiktok:
source_info["tiktok_skip_reason"] = "No SCRAPECREATORS_API_KEY - sign up at scrapecreators.com (100 free credits)"
source_info["tiktok_skip_reason"] = "No SCRAPECREATORS_API_KEY - sign up at scrapecreators.com (100 free API calls, no credit card)"
if not has_instagram:
source_info["instagram_skip_reason"] = "No SCRAPECREATORS_API_KEY - sign up at scrapecreators.com (100 free credits)"
source_info["instagram_skip_reason"] = "No SCRAPECREATORS_API_KEY - sign up at scrapecreators.com (100 free API calls, no credit card)"
if not has_xiaohongshu:
source_info["xiaohongshu_skip_reason"] = (
f"Xiaohongshu API unavailable or not logged in - start xiaohongshu-mcp and login "
@@ -1913,8 +1939,16 @@ def main():
if not web_source:
source_info["web_skip_reason"] = "assistant will use WebSearch (add BRAVE_API_KEY for native search)"
# Compute quality score and upgrade nudge
research_results = {
"x_error": x_error,
"youtube_error": youtube_error,
"reddit_error": reddit_error,
}
quality = quality_nudge.compute_quality_score(config, research_results)
# Output result
output_result(report, args.emit, web_needed, args.topic, from_date, to_date, missing_keys, args.days, source_info)
output_result(report, args.emit, web_needed, args.topic, from_date, to_date, missing_keys, args.days, source_info, first_run=first_run, quality=quality)
# Auto-save raw research to file if --save-dir is set
if args.save_dir:
@@ -1926,6 +1960,8 @@ def main():
if save_path.exists():
save_path = save_dir / f"{slug}-raw-{datetime.now().strftime('%Y-%m-%d')}.md"
content = render.render_compact(report, missing_keys=missing_keys)
if quality and quality.get("nudge_text"):
content += "\n" + render.render_quality_nudge(quality)
content += "\n" + render.render_source_status(report, source_info)
save_path.write_text(content, encoding="utf-8")
print(f"📎 {save_path}", file=sys.stderr)
@@ -2043,10 +2079,19 @@ def output_result(
missing_keys: str = "none",
days: int = 30,
source_info: dict = None,
first_run: bool = False,
quality: dict = None,
):
"""Output the result based on emit mode."""
if emit_mode == "compact":
# Emit first-run flag before research output so SKILL.md can detect it
if first_run:
print("FIRST_RUN: true")
print("")
print(render.render_compact(report, missing_keys=missing_keys))
# Quality nudge (right before source status/stats block)
if quality and quality.get("nudge_text"):
print(render.render_quality_nudge(quality))
# Append source status footer
print(render.render_source_status(report, source_info))
elif emit_mode == "json":
scripts/lib/chrome_cookies.py
+265
View File
@@ -0,0 +1,265 @@
"""Chrome cookie extraction for macOS.
Extracts cookies from Chrome's encrypted SQLite database using only stdlib
modules and the system openssl CLI (ships with macOS). Zero pip dependencies.
Chrome on macOS uses v10 encryption (AES-128-CBC with Keychain-stored key).
This is NOT affected by Windows App-Bound Encryption (v20).
"""
import hashlib
import logging
import shutil
import sqlite3
import subprocess
import tempfile
from pathlib import Path
from typing import Optional
logger = logging.getLogger(__name__)
# Chrome cookie DB location on macOS
CHROME_COOKIES_DB = Path.home() / "Library" / "Application Support" / "Google" / "Chrome" / "Default" / "Cookies"
# Chrome v10 encryption constants
CHROME_SALT = b"saltysalt"
CHROME_PBKDF2_ITERATIONS = 1003
CHROME_KEY_LENGTH = 16
# IV is 16 space characters (0x20)
CHROME_IV_HEX = "20" * 16
def _get_chrome_encryption_key() -> Optional[bytes]:
"""Retrieve Chrome's encryption passphrase from macOS Keychain.
Calls `security find-generic-password` which may trigger a system dialog
on first access.
Returns the raw passphrase bytes, or None on failure.
"""
try:
result = subprocess.run(
["security", "find-generic-password", "-w", "-s", "Chrome Safe Storage"],
capture_output=True,
text=True,
timeout=10,
)
if result.returncode != 0:
logger.info("Chrome Keychain access denied or Chrome not installed: %s", result.stderr.strip())
return None
passphrase = result.stdout.strip()
if not passphrase:
logger.info("Chrome Keychain returned empty passphrase")
return None
return passphrase.encode("utf-8")
except FileNotFoundError:
logger.info("'security' command not found — not on macOS?")
return None
except subprocess.TimeoutExpired:
logger.info("Chrome Keychain access timed out")
return None
except Exception as e:
logger.info("Failed to get Chrome encryption key: %s", e)
return None
def _derive_aes_key(passphrase: bytes) -> bytes:
"""Derive 16-byte AES key from Chrome's Keychain passphrase via PBKDF2."""
return hashlib.pbkdf2_hmac(
"sha1",
passphrase,
CHROME_SALT,
CHROME_PBKDF2_ITERATIONS,
dklen=CHROME_KEY_LENGTH,
)
def _decrypt_v10_value(encrypted_value: bytes, aes_key: bytes, db_version: int) -> Optional[str]:
"""Decrypt a Chrome v10-encrypted cookie value.
Uses system openssl CLI for AES-128-CBC decryption (zero pip deps).
For Chrome 130+ (db_version >= 24), strips 32-byte SHA-256 prefix after decryption.
Returns decrypted string or None on failure.
"""
# Strip the 'v10' prefix
ciphertext = encrypted_value[3:]
if not ciphertext:
return None
hex_key = aes_key.hex()
try:
result = subprocess.run(
[
"openssl", "enc", "-aes-128-cbc", "-d",
"-K", hex_key,
"-iv", CHROME_IV_HEX,
"-nopad",
],
input=ciphertext,
capture_output=True,
timeout=5,
)
if result.returncode != 0:
logger.debug("openssl decryption failed: %s", result.stderr.decode(errors="replace").strip())
return None
decrypted = result.stdout
if not decrypted:
return None
# Remove PKCS7 padding
decrypted = _remove_pkcs7_padding(decrypted)
if decrypted is None:
return None
# Chrome 130+ (db version >= 24): strip 32-byte SHA-256 prefix
if db_version >= 24 and len(decrypted) > 32:
decrypted = decrypted[32:]
return decrypted.decode("utf-8", errors="replace")
except FileNotFoundError:
logger.info("openssl not found — cannot decrypt Chrome cookies")
return None
except subprocess.TimeoutExpired:
logger.info("openssl decryption timed out")
return None
except Exception as e:
logger.debug("Chrome cookie decryption error: %s", e)
return None
def _remove_pkcs7_padding(data: bytes) -> Optional[bytes]:
"""Remove PKCS7 padding from decrypted data.
The last byte indicates the number of padding bytes added.
All padding bytes must have the same value.
Returns unpadded data or None if padding is invalid.
"""
if not data:
return None
pad_len = data[-1]
if pad_len < 1 or pad_len > 16:
return None
# Verify all padding bytes match
if data[-pad_len:] != bytes([pad_len]) * pad_len:
return None
return data[:-pad_len]
def _get_db_version(cursor: sqlite3.Cursor) -> int:
"""Get Chrome cookie database version from the meta table.
Returns 0 if meta table doesn't exist or version can't be read.
"""
try:
cursor.execute("SELECT value FROM meta WHERE key = 'version'")
row = cursor.fetchone()
if row:
return int(row[0])
except Exception:
pass
return 0
def extract_chrome_cookies_macos(domain: str, cookie_names: list[str]) -> Optional[dict[str, str]]:
"""Extract cookies from Chrome on macOS.
Copies the locked Cookies database to a temp file, reads specified cookies,
and decrypts v10-encrypted values using the Keychain-stored key.
Args:
domain: Cookie domain to match (e.g., ".twitter.com", ".x.com")
cookie_names: List of cookie names to extract
Returns:
Dict mapping cookie name to decrypted value, or None on failure.
Only includes cookies that were successfully found and decrypted.
"""
if not CHROME_COOKIES_DB.exists():
logger.info("Chrome cookies database not found at %s", CHROME_COOKIES_DB)
return None
# Get encryption key from Keychain
passphrase = _get_chrome_encryption_key()
aes_key = _derive_aes_key(passphrase) if passphrase else None
# Copy DB to temp file (Chrome locks the original)
tmp_fd = None
tmp_path = None
try:
tmp_fd, tmp_path = tempfile.mkstemp(suffix=".sqlite")
shutil.copy2(str(CHROME_COOKIES_DB), tmp_path)
except Exception as e:
logger.info("Failed to copy Chrome cookies database: %s", e)
if tmp_path:
try:
Path(tmp_path).unlink(missing_ok=True)
except Exception:
pass
return None
finally:
if tmp_fd is not None:
import os
os.close(tmp_fd)
try:
conn = sqlite3.connect(tmp_path)
cursor = conn.cursor()
db_version = _get_db_version(cursor)
logger.debug("Chrome cookie DB version: %d", db_version)
# Build query with placeholders for cookie names
placeholders = ",".join("?" for _ in cookie_names)
query = (
f"SELECT name, value, encrypted_value FROM cookies "
f"WHERE host_key LIKE ? AND name IN ({placeholders})"
)
# Use LIKE for domain matching (e.g., %.twitter.com matches .twitter.com)
params = [f"%{domain}"] + list(cookie_names)
cursor.execute(query, params)
results: dict[str, str] = {}
for name, value, encrypted_value in cursor.fetchall():
# Prefer unencrypted value if present
if value:
results[name] = value
continue
# Handle encrypted value
if encrypted_value and encrypted_value[:3] == b"v10":
if aes_key is None:
logger.debug("Skipping encrypted cookie %s — no Keychain access", name)
continue
decrypted = _decrypt_v10_value(encrypted_value, aes_key, db_version)
if decrypted:
results[name] = decrypted
else:
logger.debug("Failed to decrypt cookie %s", name)
elif encrypted_value:
# Unknown encryption version
logger.debug("Unknown encryption for cookie %s (prefix: %r)", name, encrypted_value[:3])
conn.close()
if not results:
logger.info("No matching cookies found in Chrome for domain %s", domain)
return None
return results
except sqlite3.Error as e:
logger.info("Failed to read Chrome cookies database: %s", e)
return None
except Exception as e:
logger.info("Unexpected error reading Chrome cookies: %s", e)
return None
finally:
try:
Path(tmp_path).unlink(missing_ok=True)
except Exception:
pass
scripts/lib/cookie_extract.py
+295
View File
@@ -0,0 +1,295 @@
"""Browser cookie extraction for last30days.
Extracts cookies from local browser databases (Firefox, Chrome, Safari)
to enable zero-config authentication for services like X/Twitter.
Only uses Python stdlib — no external dependencies.
"""
import configparser
import logging
import platform
import shutil
import sqlite3
import tempfile
from pathlib import Path
from typing import Dict, List, Optional
logger = logging.getLogger(__name__)
def _get_firefox_profiles_dir() -> Optional[Path]:
"""Return the Firefox profiles directory for the current platform, or None."""
system = platform.system()
if system == "Darwin":
path = Path.home() / "Library" / "Application Support" / "Firefox"
elif system == "Linux":
path = Path.home() / ".mozilla" / "firefox"
else:
# Windows: %APPDATA%\Mozilla\Firefox — best-effort
appdata = Path.home() / "AppData" / "Roaming" / "Mozilla" / "Firefox"
path = appdata
return path if path.is_dir() else None
def _find_default_profile(profiles_dir: Path) -> Optional[Path]:
"""Parse profiles.ini to find the default profile directory.
Looks for a section with Default=1. Falls back to the first profile
directory found on disk if profiles.ini is missing or malformed.
"""
ini_path = profiles_dir / "profiles.ini"
if ini_path.is_file():
try:
config = configparser.ConfigParser()
config.read(str(ini_path), encoding="utf-8")
# First pass: look for Default=1
for section in config.sections():
if config.has_option(section, "Default") and config.get(section, "Default") == "1":
return _resolve_profile_path(profiles_dir, config, section)
# Second pass: first Install* section with Default key (Firefox >= 67 format)
for section in config.sections():
if section.startswith("Install") and config.has_option(section, "Default"):
raw = config.get(section, "Default")
candidate = profiles_dir / raw
if candidate.is_dir():
return candidate
# Third pass: first Profile section that exists on disk
for section in config.sections():
if section.startswith("Profile"):
resolved = _resolve_profile_path(profiles_dir, config, section)
if resolved and resolved.is_dir():
return resolved
except (configparser.Error, OSError) as exc:
logger.debug("Failed to parse profiles.ini: %s", exc)
# Fallback: scan directory for anything that looks like a profile
return _fallback_find_profile(profiles_dir)
def _resolve_profile_path(
profiles_dir: Path, config: configparser.ConfigParser, section: str
) -> Optional[Path]:
"""Resolve a profile path from a ConfigParser section."""
if not config.has_option(section, "Path"):
return None
raw_path = config.get(section, "Path")
is_relative = config.has_option(section, "IsRelative") and config.get(section, "IsRelative") == "1"
if is_relative:
candidate = profiles_dir / raw_path
else:
candidate = Path(raw_path)
return candidate if candidate.is_dir() else None
def _fallback_find_profile(profiles_dir: Path) -> Optional[Path]:
"""Find the first directory that contains cookies.sqlite."""
try:
for child in sorted(profiles_dir.iterdir()):
if child.is_dir() and (child / "cookies.sqlite").is_file():
return child
except OSError:
pass
return None
def _query_cookies_db(
db_path: Path, domain: str, cookie_names: List[str]
) -> Optional[Dict[str, str]]:
"""Copy the cookies database to a temp file and query it.
Firefox locks cookies.sqlite while running, so we copy first.
Returns {name: value} dict or None if no matching cookies found.
"""
if not db_path.is_file():
return None
tmp_fd = None
tmp_path = None
try:
tmp_fd, tmp_path = tempfile.mkstemp(suffix=".sqlite")
shutil.copy2(str(db_path), tmp_path)
conn = sqlite3.connect(tmp_path)
try:
# Build parameterized query — SQLite doesn't support array params,
# so we build the IN clause with individual placeholders.
placeholders = ",".join("?" for _ in cookie_names)
query = (
f"SELECT name, value FROM moz_cookies "
f"WHERE host LIKE ? AND name IN ({placeholders})"
)
# domain pattern: match .x.com, x.com, etc.
domain_pattern = f"%{domain}"
params = [domain_pattern] + list(cookie_names)
cursor = conn.execute(query, params)
rows = cursor.fetchall()
finally:
conn.close()
if not rows:
return None
return {name: value for name, value in rows}
except (sqlite3.Error, OSError) as exc:
logger.debug("Failed to query cookies database %s: %s", db_path, exc)
return None
finally:
if tmp_path:
try:
Path(tmp_path).unlink(missing_ok=True)
except OSError:
pass
if tmp_fd is not None:
try:
import os
os.close(tmp_fd)
except OSError:
pass
def extract_firefox_cookies(
domain: str, cookie_names: List[str]
) -> Optional[Dict[str, str]]:
"""Extract cookies from Firefox for the given domain and cookie names.
Finds the default Firefox profile, copies cookies.sqlite to a temp file
(to avoid lock conflicts), and queries for the requested cookies.
Args:
domain: The cookie domain to match (e.g. ".x.com"). Matched with LIKE %domain.
cookie_names: List of cookie names to extract (e.g. ["auth_token", "ct0"]).
Returns:
Dict of {cookie_name: cookie_value} or None if extraction fails.
"""
profiles_dir = _get_firefox_profiles_dir()
if profiles_dir is None:
logger.debug("Firefox profiles directory not found")
return None
profile_path = _find_default_profile(profiles_dir)
if profile_path is None:
logger.debug("No Firefox profile found in %s", profiles_dir)
return None
db_path = profile_path / "cookies.sqlite"
return _query_cookies_db(db_path, domain, cookie_names)
def extract_chrome_cookies(
domain: str, cookie_names: List[str]
) -> Optional[Dict[str, str]]:
"""Extract cookies from Chrome for the given domain and cookie names.
macOS only — uses Keychain + system openssl for AES-128-CBC decryption.
Linux/Windows not supported (Chrome uses platform-specific encryption).
Returns:
Dict of {cookie_name: cookie_value} or None if extraction fails.
"""
if platform.system() != "Darwin":
logger.debug("Chrome cookie extraction only supported on macOS")
return None
try:
from .chrome_cookies import extract_chrome_cookies_macos
return extract_chrome_cookies_macos(domain, cookie_names)
except Exception as exc:
logger.debug("Chrome cookie extraction failed: %s", exc)
return None
def extract_safari_cookies(
domain: str, cookie_names: List[str]
) -> Optional[Dict[str, str]]:
"""Extract cookies from Safari for the given domain and cookie names.
macOS only — parses the unencrypted binary cookie file.
Returns:
Dict of {cookie_name: cookie_value} or None if extraction fails.
"""
if platform.system() != "Darwin":
logger.debug("Safari cookie extraction only supported on macOS")
return None
try:
from .safari_cookies import extract_safari_cookies_macos
return extract_safari_cookies_macos(domain, cookie_names)
except Exception as exc:
logger.debug("Safari cookie extraction failed: %s", exc)
return None
def extract_cookies(
browser: str, domain: str, cookie_names: list[str]
) -> Optional[dict[str, str]]:
"""Extract cookies from the specified browser.
Args:
browser: One of 'firefox', 'chrome', 'safari', or 'auto'.
'auto' tries browsers in platform-appropriate order:
- macOS: Chrome -> Firefox -> Safari
- Linux: Firefox only
domain: The cookie domain to match (e.g. ".x.com").
cookie_names: List of cookie names to extract.
Returns:
Dict of {cookie_name: cookie_value} or None if extraction fails.
"""
result = extract_cookies_with_source(browser, domain, cookie_names)
if result is None:
return None
cookies, _browser_name = result
return cookies
def extract_cookies_with_source(
browser: str, domain: str, cookie_names: list[str]
) -> Optional[tuple[dict[str, str], str]]:
"""Extract cookies and report which browser they came from.
Same as extract_cookies() but returns a (cookies, browser_name) tuple
so callers can track the source.
Args:
browser: One of 'firefox', 'chrome', 'safari', or 'auto'.
domain: The cookie domain to match (e.g. ".x.com").
cookie_names: List of cookie names to extract.
Returns:
Tuple of ({cookie_name: cookie_value}, browser_name) or None.
"""
extractors = {
"firefox": extract_firefox_cookies,
"chrome": extract_chrome_cookies,
"safari": extract_safari_cookies,
}
if browser != "auto":
extractor = extractors.get(browser)
if extractor is None:
logger.warning("Unknown browser: %s", browser)
return None
result = extractor(domain, cookie_names)
return (result, browser) if result is not None else None
# Auto mode: try browsers in platform-appropriate order
system = platform.system()
if system == "Darwin":
order = ["chrome", "firefox", "safari"]
elif system == "Linux":
order = ["firefox"]
else:
order = ["firefox"]
for name in order:
result = extractors[name](domain, cookie_names)
if result is not None:
return (result, name)
return None
scripts/lib/env.py
+229 -25
View File
@@ -2,11 +2,38 @@
import base64
import json
import logging
import os
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Optional, Dict, Any, Literal
from typing import Optional, Dict, Any, List, Literal
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Cookie domain registry: maps source names to browser cookie extraction params.
# Each entry: (domain, cookie_names, config_key_mapping)
# config_key_mapping: {cookie_name: config_key} so we know which config key
# each extracted cookie should populate.
# ---------------------------------------------------------------------------
COOKIE_DOMAINS: Dict[str, Dict[str, Any]] = {
"x": {
"domain": ".x.com",
"cookies": ["auth_token", "ct0"],
"mapping": {
"auth_token": "AUTH_TOKEN",
"ct0": "CT0",
},
},
"truthsocial": {
"domain": ".truthsocial.com",
"cookies": ["_session_id"],
"mapping": {
"_session_id": "TRUTHSOCIAL_TOKEN",
},
},
}
# Allow override via environment variable for testing
# Set LAST30DAYS_CONFIG_DIR="" for clean/no-config mode
@@ -212,6 +239,87 @@ def _find_project_env() -> Optional[Path]:
return None
def extract_browser_credentials(config: Dict[str, Any]) -> Dict[str, str]:
"""Extract credentials from browser cookies for sources that need them.
Checks the FROM_BROWSER config key to decide whether/how to extract:
- 'auto': try browsers in platform order
- 'firefox', 'chrome', 'safari': try only that browser
- 'off': skip extraction entirely
If SETUP_COMPLETE is not set AND FROM_BROWSER is not explicitly set,
defaults to 'off' (wizard hasn't run yet — no extraction without consent).
If SETUP_COMPLETE is set and FROM_BROWSER is not set, defaults to 'auto'.
Explicit env var/config values always take priority over extracted cookies.
Returns:
Dict of {config_key: value} for credentials discovered from cookies.
"""
setup_complete = config.get("SETUP_COMPLETE")
from_browser = config.get("FROM_BROWSER")
# Determine effective browser setting
if from_browser is None:
if setup_complete:
from_browser = "auto"
else:
from_browser = "off"
from_browser = from_browser.lower().strip() if isinstance(from_browser, str) else "off"
if from_browser == "off":
return {}
# Lazy import to avoid loading cookie_extract at module level
try:
from . import cookie_extract
except Exception:
logger.debug("cookie_extract module not available")
return {}
credentials: Dict[str, str] = {}
for source_name, spec in COOKIE_DOMAINS.items():
domain = spec["domain"]
cookie_names: List[str] = spec["cookies"]
mapping: Dict[str, str] = spec["mapping"]
# Skip if ALL mapped config keys already have values
all_present = all(config.get(config_key) for config_key in mapping.values())
if all_present:
logger.debug(
"Skipping cookie extraction for %s: credentials already set",
source_name,
)
continue
try:
result = cookie_extract.extract_cookies_with_source(from_browser, domain, cookie_names)
except Exception as exc:
logger.debug(
"Cookie extraction failed for %s: %s", source_name, exc
)
continue
if result is None:
continue
cookies, browser_name = result
filled_any = False
for cookie_name, config_key in mapping.items():
# Only fill in keys not already present
if not config.get(config_key) and cookie_name in cookies:
credentials[config_key] = cookies[cookie_name]
filled_any = True
# Track which browser provided the credentials for this source
if filled_any:
credentials[f"__{source_name.upper()}_BROWSER"] = browser_name
return credentials
def get_config() -> Dict[str, Any]:
"""Load configuration from multiple sources.
@@ -219,6 +327,7 @@ def get_config() -> Dict[str, Any]:
1. Environment variables (os.environ)
2. .claude/last30days.env (per-project config)
3. ~/.config/last30days/.env (global config)
4. Browser cookies (only fills in missing keys)
"""
# Load from global config file
file_env = load_env_file(CONFIG_FILE) if CONFIG_FILE else {}
@@ -249,6 +358,7 @@ def get_config() -> Dict[str, Any]:
('OPENROUTER_API_KEY', None),
('PARALLEL_API_KEY', None),
('BRAVE_API_KEY', None),
('EXA_API_KEY', None),
('XIAOHONGSHU_API_BASE', None),
('GEMINI_MODEL', None),
('OPENAI_MODEL_POLICY', 'auto'),
@@ -262,11 +372,31 @@ def get_config() -> Dict[str, Any]:
('BSKY_HANDLE', None),
('BSKY_APP_PASSWORD', None),
('TRUTHSOCIAL_TOKEN', None),
('FROM_BROWSER', None),
('SETUP_COMPLETE', None),
]
for key, default in keys:
config[key] = os.environ.get(key) or merged_env.get(key, default)
# Inject browser cookies for any credentials not already set
browser_creds = extract_browser_credentials(config)
for key, value in browser_creds.items():
if not config.get(key):
config[key] = value
# Track AUTH_TOKEN source for status reporting
if config.get('AUTH_TOKEN'):
if os.environ.get('AUTH_TOKEN') or merged_env.get('AUTH_TOKEN'):
config['_AUTH_TOKEN_SOURCE'] = 'env'
elif browser_creds.get('AUTH_TOKEN'):
browser_name = browser_creds.get('__X_BROWSER', 'unknown')
config['_AUTH_TOKEN_SOURCE'] = f'browser-{browser_name}'
else:
config['_AUTH_TOKEN_SOURCE'] = 'env' # fallback
else:
config['_AUTH_TOKEN_SOURCE'] = None
# Track which config source was used
if project_env_path:
config['_CONFIG_SOURCE'] = f'project:{project_env_path}'
@@ -314,14 +444,19 @@ def get_reddit_source(config: Dict[str, Any]) -> Optional[str]:
def get_available_sources(config: Dict[str, Any]) -> str:
"""Determine which sources are available.
X is available if ANY auth method works: AUTH_TOKEN/CT0 (env or cookies),
XAI_API_KEY, or Bird installed+authenticated.
Reddit is always available (public JSON fallback).
HN and Polymarket are always available.
YouTube available if yt-dlp installed.
Returns: 'all', 'both', 'reddit', 'reddit-web', 'x', 'x-web', 'web', or 'none'
"""
# Reddit is available via public JSON fallback even without OpenAI auth.
has_reddit = True
has_xai = bool(config.get('XAI_API_KEY'))
has_x = get_x_source(config) is not None
has_web = has_web_search_keys(config)
if has_reddit and has_xai:
if has_reddit and has_x:
return 'all' if has_web else 'both'
elif has_reddit:
return 'reddit-web' if has_web else 'reddit'
@@ -330,16 +465,18 @@ def get_available_sources(config: Dict[str, Any]) -> str:
def has_web_search_keys(config: Dict[str, Any]) -> bool:
"""Check if any web search API keys are configured."""
return bool(config.get('OPENROUTER_API_KEY') or config.get('PARALLEL_API_KEY') or config.get('BRAVE_API_KEY'))
return bool(config.get('EXA_API_KEY') or config.get('OPENROUTER_API_KEY') or config.get('PARALLEL_API_KEY') or config.get('BRAVE_API_KEY'))
def get_web_search_source(config: Dict[str, Any]) -> Optional[str]:
"""Determine the best available web search backend.
Priority: Parallel AI > Brave > OpenRouter/Sonar Pro
Priority: Exa (free) > Parallel AI > Brave > OpenRouter/Sonar Pro
Returns: 'parallel', 'brave', 'openrouter', or None
Returns: 'exa', 'parallel', 'brave', 'openrouter', or None
"""
if config.get('EXA_API_KEY'):
return 'exa'
if config.get('PARALLEL_API_KEY'):
return 'parallel'
if config.get('BRAVE_API_KEY'):
@@ -441,9 +578,13 @@ def validate_sources(requested: str, available: str, include_web: bool = False)
def get_x_source(config: Dict[str, Any]) -> Optional[str]:
"""Determine the best available X/Twitter source.
Priority: Bird (free) → xAI (paid API)
Priority chain:
1. AUTH_TOKEN/CT0 from env var or .env file → Bird with method "env"
2. AUTH_TOKEN/CT0 from browser cookie extraction → Bird with method "browser-{browser}"
3. XAI_API_KEY → xAI with method "api"
4. None
Keep X selection limited to documented, verified search backends.
Use get_x_source_with_method() to also get the method string.
Args:
config: Configuration dict from get_config()
@@ -453,20 +594,58 @@ def get_x_source(config: Dict[str, Any]) -> Optional[str]:
'xai' if XAI_API_KEY is configured,
None if no X source available.
"""
# Import here to avoid circular dependency
source, _method = get_x_source_with_method(config)
return source
def get_x_source_with_method(config: Dict[str, Any]) -> tuple[Optional[str], Optional[str]]:
"""Determine the best available X/Twitter source and auth method.
Priority chain:
1. AUTH_TOKEN/CT0 (env var or .env) → Bird with method "env"
2. AUTH_TOKEN/CT0 (browser cookies) → Bird with method "browser-{browser}"
3. XAI_API_KEY → xAI with method "api"
4. None
Args:
config: Configuration dict from get_config()
Returns:
Tuple of (source, method) where source is 'bird', 'xai', or None
and method is 'env', 'browser-chrome', 'browser-firefox', 'browser-safari', 'api', or None.
"""
from . import bird_x
# Check Bird first (free option)
setup_complete = config.get('SETUP_COMPLETE')
# Check Bird first (free option — uses AUTH_TOKEN/CT0 from any source)
if bird_x.is_bird_installed():
username = bird_x.is_bird_authenticated()
if username:
return 'bird'
auth_source = config.get('_AUTH_TOKEN_SOURCE')
# If SETUP_COMPLETE is not set, only allow explicit env var credentials.
# Do NOT call is_bird_authenticated() for browser-cookie probing —
# that requires user consent via the setup wizard.
if not setup_complete:
# Explicit AUTH_TOKEN from env var / .env file is always allowed
if auth_source == 'env' and config.get('AUTH_TOKEN'):
username = bird_x.is_bird_authenticated()
if username:
return 'bird', 'env'
else:
# SETUP_COMPLETE is set — normal flow, probe cookies if needed
username = bird_x.is_bird_authenticated()
if username:
if auth_source and auth_source.startswith('browser-'):
method = auth_source # e.g. "browser-firefox"
else:
method = 'env'
return 'bird', method
# Fall back to xAI if key exists
if config.get('XAI_API_KEY'):
return 'xai'
return 'xai', 'api'
return None
return None, None
def is_ytdlp_available() -> bool:
@@ -576,24 +755,49 @@ def get_x_source_status(config: Dict[str, Any]) -> Dict[str, Any]:
"""Get detailed X source status for UI decisions.
Returns:
Dict with keys: source, bird_installed, bird_authenticated,
Dict with keys: source, method, bird_installed, bird_authenticated,
bird_username, xai_available, can_install_bird
The ``method`` field indicates HOW the active source is authenticated:
- "env" — AUTH_TOKEN came from an env var or .env file
- "browser-chrome", "browser-firefox", "browser-safari" — from cookie extraction
- "api" — using xAI API key
- None — no X source available
"""
from . import bird_x
bird_status = bird_x.get_bird_status()
setup_complete = config.get('SETUP_COMPLETE')
xai_available = bool(config.get('XAI_API_KEY'))
# Determine active source
if bird_status["authenticated"]:
source = 'bird'
elif xai_available:
source = 'xai'
else:
source = None
if not setup_complete:
# Before consent: do NOT call get_bird_status() which probes cookies.
# Only check if Bird is installed (no cookie probing) and use the
# gated get_x_source_with_method() which blocks cookie detection.
bird_installed = bird_x.is_bird_installed()
source, method = get_x_source_with_method(config)
# Bird "authenticated" only if get_x_source_with_method found explicit creds
bird_authenticated = (source == 'bird')
return {
"source": source,
"method": method,
"bird_installed": bird_installed,
"bird_authenticated": bird_authenticated,
"bird_username": None if not bird_authenticated else "env AUTH_TOKEN",
"xai_available": xai_available,
"can_install_bird": True,
}
# SETUP_COMPLETE is set — normal flow
bird_status = bird_x.get_bird_status()
# Use the unified resolution function for source + method
source, method = get_x_source_with_method(config)
return {
"source": source,
"method": method,
"bird_installed": bird_status["installed"],
"bird_authenticated": bird_status["authenticated"],
"bird_username": bird_status["username"],
scripts/lib/exa_search.py
+176
View File
@@ -0,0 +1,176 @@
"""Exa AI web search for last30days skill.
Uses the Exa Search API as a free web search backend.
Free tier: 1,000 searches/month, semantic search, no credit card required.
API docs: https://docs.exa.ai/reference/search
"""
import sys
from typing import Any, Dict, List, Optional
from urllib.parse import urlparse
from . import http
ENDPOINT = "https://api.exa.ai/search"
# Domains to exclude (handled by Reddit/X search)
EXCLUDED_DOMAINS = {
"reddit.com", "www.reddit.com", "old.reddit.com",
"twitter.com", "www.twitter.com", "x.com", "www.x.com",
}
def search_web(
topic: str,
from_date: str,
to_date: str,
api_key: str,
depth: str = "default",
) -> List[Dict[str, Any]]:
"""Search the web via Exa AI Search API.
Args:
topic: Search topic
from_date: Start date (YYYY-MM-DD)
to_date: End date (YYYY-MM-DD)
api_key: Exa API key
depth: 'quick', 'default', or 'deep'
Returns:
List of result dicts with keys: url, title, snippet, source_domain, date, relevance
"""
num_results = {"quick": 8, "default": 15, "deep": 25}.get(depth, 15)
max_chars = {"quick": 1000, "default": 2000, "deep": 3000}.get(depth, 2000)
payload = {
"query": f"{topic} (from {from_date} to {to_date})",
"type": "auto",
"numResults": num_results,
"contents": {"text": {"maxCharacters": max_chars}},
}
# Add date filtering if dates are provided
if from_date:
payload["startPublishedDate"] = f"{from_date}T00:00:00.000Z"
if to_date:
payload["endPublishedDate"] = f"{to_date}T23:59:59.999Z"
sys.stderr.write(f"[Web] Searching Exa for: {topic}\n")
sys.stderr.flush()
try:
response = http.post(
ENDPOINT,
json_data=payload,
headers={
"x-api-key": api_key,
},
timeout=20,
retries=2,
)
except http.HTTPError as e:
status = e.status_code
if status == 401:
sys.stderr.write("[Web] Exa: invalid API key (401)\n")
sys.stderr.flush()
return []
if status == 429:
sys.stderr.write("[Web] Exa: rate limited (429)\n")
sys.stderr.flush()
return []
sys.stderr.write(f"[Web] Exa: HTTP error {status}: {e}\n")
sys.stderr.flush()
return []
except Exception as e:
sys.stderr.write(f"[Web] Exa: request failed: {e}\n")
sys.stderr.flush()
return []
return _normalize_results(response)
def _normalize_results(response: Dict[str, Any]) -> List[Dict[str, Any]]:
"""Convert Exa API response to websearch item schema.
Exa results have: title, url, text, publishedDate, score, author.
"""
items = []
results = response.get("results", [])
if not isinstance(results, list):
return items
for i, result in enumerate(results):
if not isinstance(result, dict):
continue
url = result.get("url", "")
if not url:
continue
# Skip excluded domains
try:
domain = urlparse(url).netloc.lower()
if domain in EXCLUDED_DOMAINS:
continue
if domain.startswith("www."):
domain = domain[4:]
except Exception:
domain = ""
title = str(result.get("title", "")).strip()
# Exa returns page content in "text" field
snippet = str(result.get("text", "")).strip()
if not title and not snippet:
continue
# Parse publishedDate (ISO format from Exa: "2026-03-15T00:00:00.000Z")
date = _parse_exa_date(result.get("publishedDate"))
date_confidence = "med" if date else "low"
# Exa provides a relevance score
relevance = result.get("score", 0.6)
try:
relevance = min(1.0, max(0.0, float(relevance)))
except (TypeError, ValueError):
relevance = 0.6
items.append({
"id": f"W{i+1}",
"title": title[:200],
"url": url,
"source_domain": domain,
"snippet": snippet[:500],
"date": date,
"date_confidence": date_confidence,
"relevance": relevance,
"why_relevant": "",
})
sys.stderr.write(f"[Web] Exa: {len(items)} results\n")
sys.stderr.flush()
return items
def _parse_exa_date(published_date: Optional[str]) -> Optional[str]:
"""Parse Exa's publishedDate to YYYY-MM-DD.
Exa returns ISO format like "2026-03-15T00:00:00.000Z".
"""
if not published_date:
return None
try:
# Extract YYYY-MM-DD from ISO datetime
if "T" in published_date:
return published_date.split("T")[0]
# Already YYYY-MM-DD
if len(published_date) >= 10:
return published_date[:10]
except Exception:
pass
return None
scripts/lib/instagram.py
+1 -1
View File
@@ -3,7 +3,7 @@
Uses ScrapeCreators REST API to search Instagram Reels by keyword, extract
engagement metrics (views, likes, comments), and fetch video transcripts.
Requires SCRAPECREATORS_API_KEY in config. 100 free credits, then PAYG.
Requires SCRAPECREATORS_API_KEY in config. 100 free API calls, then PAYG.
API docs: https://scrapecreators.com/docs
"""
scripts/lib/quality_nudge.py
+201
View File
@@ -0,0 +1,201 @@
"""Post-research quality score and upgrade nudge.
Computes a quality score based on 5 core sources and builds
a nudge message describing what the user missed and how to fix it.
"""
from typing import Any, Dict, List, Optional
# The 5 core sources
CORE_SOURCES = ["hn", "polymarket", "x", "youtube", "reddit_comments"]
# Labels for display
SOURCE_LABELS = {
"hn": "Hacker News",
"polymarket": "Polymarket",
"x": "X/Twitter",
"youtube": "YouTube",
"reddit_comments": "Reddit with comments",
}
def _is_x_active(config: dict, research_results: dict) -> bool:
"""Check if X source is active (has credentials AND didn't error)."""
has_creds = bool(config.get("AUTH_TOKEN") or config.get("XAI_API_KEY"))
if not has_creds:
return False
# If X errored this run, it's configured but broken
if research_results.get("x_error"):
return False
return True
def _is_youtube_active(config: dict, research_results: dict) -> bool:
"""Check if YouTube source is active (yt-dlp installed)."""
try:
from . import youtube_yt
has_ytdlp = youtube_yt.is_ytdlp_installed()
except Exception:
has_ytdlp = False
if not has_ytdlp:
return False
if research_results.get("youtube_error"):
return False
return True
def _is_reddit_comments_active(config: dict, research_results: dict) -> bool:
"""Check if Reddit with comments is active (ScrapeCreators)."""
has_sc = bool(config.get("SCRAPECREATORS_API_KEY"))
if not has_sc:
return False
if research_results.get("reddit_error"):
return False
return True
def compute_quality_score(config: dict, research_results: dict) -> dict:
"""Compute research quality score based on 5 core sources.
Args:
config: Configuration dict from env.get_config()
research_results: Dict with keys like x_error, youtube_error,
reddit_error reflecting what happened this run.
Returns:
{
"score_pct": 40-100,
"core_active": ["hn", "polymarket", ...],
"core_missing": ["x", "youtube", "reddit_comments"],
"core_errored": ["reddit_comments"], # configured but errored
"nudge_text": "..." or None if 100%
}
"""
core_active: List[str] = []
core_missing: List[str] = []
core_errored: List[str] = []
# HN and Polymarket are always active
core_active.append("hn")
core_active.append("polymarket")
# X
has_x_creds = bool(config.get("AUTH_TOKEN") or config.get("XAI_API_KEY"))
if _is_x_active(config, research_results):
core_active.append("x")
else:
core_missing.append("x")
if has_x_creds and research_results.get("x_error"):
core_errored.append("x")
# YouTube
try:
from . import youtube_yt
has_ytdlp = youtube_yt.is_ytdlp_installed()
except Exception:
has_ytdlp = False
if _is_youtube_active(config, research_results):
core_active.append("youtube")
else:
core_missing.append("youtube")
if has_ytdlp and research_results.get("youtube_error"):
core_errored.append("youtube")
# Reddit with comments (ScrapeCreators)
has_sc = bool(config.get("SCRAPECREATORS_API_KEY"))
if _is_reddit_comments_active(config, research_results):
core_active.append("reddit_comments")
else:
core_missing.append("reddit_comments")
if has_sc and research_results.get("reddit_error"):
core_errored.append("reddit_comments")
score_pct = int(len(core_active) / 5 * 100)
nudge_text = _build_nudge_text(core_missing, core_errored) if core_missing else None
return {
"score_pct": score_pct,
"core_active": core_active,
"core_missing": core_missing,
"core_errored": core_errored,
"nudge_text": nudge_text,
}
def _build_nudge_text(core_missing: List[str], core_errored: List[str]) -> str:
"""Build human-readable nudge text describing what was missed.
Prioritizes free suggestions before paid ones.
"""
lines: List[str] = []
# Describe what was missed
missed_parts: List[str] = []
for src in core_missing:
label = SOURCE_LABELS[src]
if src in core_errored:
missed_parts.append(f"{label} (errored this run)")
else:
missed_parts.append(label)
active_count = 5 - len(core_missing)
lines.append(f"Research quality: {active_count}/5 core sources.")
lines.append(f"Missing: {', '.join(missed_parts)}.")
lines.append("")
# Free suggestions first
free_suggestions: List[str] = []
paid_suggestions: List[str] = []
if "x" in core_missing:
if "x" in core_errored:
free_suggestions.append(
"X errored — try refreshing your browser cookies "
"(log into x.com, then re-run)."
)
else:
free_suggestions.append(
"X/Twitter: scan browser cookies automatically — "
"just log into x.com in any browser and re-run."
)
if "youtube" in core_missing:
if "youtube" in core_errored:
free_suggestions.append(
"YouTube errored — check that yt-dlp is up to date: "
"brew upgrade yt-dlp"
)
else:
free_suggestions.append(
"YouTube: install yt-dlp — brew install yt-dlp"
)
if "reddit_comments" in core_missing:
if "reddit_comments" in core_errored:
paid_suggestions.append(
"Reddit comments errored — check your ScrapeCreators API key "
"at scrapecreators.com."
)
else:
paid_suggestions.append(
"Reddit with comments: add SCRAPECREATORS_API_KEY — "
"100 free API calls, no credit card — scrapecreators.com"
)
if free_suggestions:
lines.append("Free fixes:")
for s in free_suggestions:
lines.append(f" - {s}")
lines.append("")
if paid_suggestions:
lines.append("Paid options:")
for s in paid_suggestions:
lines.append(f" - {s}")
lines.append("")
lines.append("last30days has no affiliation with any API provider.")
return "\n".join(lines)
scripts/lib/reddit_public.py
+259
View File
@@ -0,0 +1,259 @@
"""Standalone Reddit public JSON search module.
Searches Reddit using the free public JSON endpoints (no API key required).
Promoted from last-resort fallback to robust primary free path.
Endpoints:
- Global: https://www.reddit.com/search.json?q={query}&sort=relevance&t=month&limit={limit}
- Subreddit: https://www.reddit.com/r/{sub}/search.json?q={query}&restrict_sr=on&sort=relevance&t=month
Handles 429 rate limits with exponential backoff, HTML anti-bot responses,
network timeouts, and missing subreddits.
"""
import json
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from typing import Any, Dict, List, Optional
USER_AGENT = "last30days/3.0 (research tool)"
# Depth-aware limits for thread counts
DEPTH_LIMITS = {
"quick": 10,
"default": 25,
"deep": 50,
}
MAX_RETRIES = 3
BASE_BACKOFF = 2.0 # seconds
def _log(msg: str):
"""Log to stderr."""
sys.stderr.write(f"[RedditPublic] {msg}\n")
sys.stderr.flush()
def _url_encode(text: str) -> str:
"""URL-encode a query string."""
return urllib.parse.quote_plus(text)
def _fetch_json(url: str, timeout: int = 15) -> Optional[Dict[str, Any]]:
"""Fetch JSON from a URL with retry on 429 and error handling.
Returns parsed JSON dict, or None on unrecoverable failure.
"""
headers = {
"User-Agent": USER_AGENT,
"Accept": "application/json",
}
req = urllib.request.Request(url, headers=headers)
for attempt in range(MAX_RETRIES):
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
content_type = resp.headers.get("Content-Type", "")
if "json" not in content_type and "text/html" in content_type:
_log(f"Anti-bot HTML response (Content-Type: {content_type})")
return None
body = resp.read().decode("utf-8")
return json.loads(body)
except urllib.error.HTTPError as e:
if e.code == 429:
delay = BASE_BACKOFF * (2 ** attempt)
retry_after = None
if hasattr(e, "headers"):
retry_after = e.headers.get("Retry-After")
if retry_after:
try:
delay = float(retry_after)
except ValueError:
pass
_log(f"429 rate limited, retry {attempt + 1}/{MAX_RETRIES} after {delay:.1f}s")
if attempt < MAX_RETRIES - 1:
time.sleep(delay)
continue
# Last attempt exhausted
_log("429 retries exhausted")
return None
elif e.code == 404:
_log(f"404 not found: {url}")
return None
elif e.code == 403:
_log(f"403 forbidden: {url}")
return None
else:
_log(f"HTTP {e.code}: {e.reason}")
return None
except (urllib.error.URLError, OSError, TimeoutError) as e:
_log(f"Network error: {e}")
return None
except json.JSONDecodeError as e:
_log(f"JSON decode error: {e}")
return None
return None
def _parse_posts(data: Optional[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Parse Reddit listing JSON into normalized post dicts."""
if not data:
return []
children = data.get("data", {}).get("children", [])
posts = []
for child in children:
if child.get("kind") != "t3":
continue
post = child.get("data", {})
permalink = str(post.get("permalink", "")).strip()
if not permalink or "/comments/" not in permalink:
continue
score = int(post.get("score", 0) or 0)
num_comments = int(post.get("num_comments", 0) or 0)
selftext = str(post.get("selftext", ""))
author = str(post.get("author", "[deleted]"))
created_utc = post.get("created_utc")
# Parse date
date_str = None
if created_utc:
try:
from datetime import datetime, timezone
dt = datetime.fromtimestamp(float(created_utc), tz=timezone.utc)
date_str = dt.strftime("%Y-%m-%d")
except (ValueError, TypeError, OSError):
pass
posts.append({
"id": "", # Will be assigned after dedup
"title": str(post.get("title", "")).strip(),
"url": f"https://www.reddit.com{permalink}",
"score": score,
"num_comments": num_comments,
"subreddit": str(post.get("subreddit", "")).strip(),
"created_utc": float(created_utc) if created_utc else None,
"author": author if author not in ("[deleted]", "[removed]") else "[deleted]",
"selftext": selftext[:500] if selftext else "",
# Normalized fields matching ScrapeCreators output
"date": date_str,
"engagement": {
"score": score,
"num_comments": num_comments,
"upvote_ratio": post.get("upvote_ratio"),
},
"relevance": _compute_relevance(score, num_comments),
"why_relevant": "Reddit public search",
})
return posts
def _compute_relevance(score: int, num_comments: int) -> float:
"""Estimate relevance from engagement signals."""
score_component = min(1.0, max(0.0, score / 500.0))
comments_component = min(1.0, max(0.0, num_comments / 200.0))
return round((score_component * 0.6) + (comments_component * 0.4), 3)
def search(
query: str,
depth: str = "default",
subreddit: Optional[str] = None,
timeout: int = 15,
) -> List[Dict[str, Any]]:
"""Search Reddit via the public JSON endpoint.
Args:
query: Search query string
depth: 'quick', 'default', or 'deep' — controls result limit
subreddit: Optional subreddit name (without r/) for scoped search
timeout: HTTP timeout in seconds
Returns:
List of normalized post dicts. Empty list on any failure.
"""
limit = DEPTH_LIMITS.get(depth, DEPTH_LIMITS["default"])
encoded_query = _url_encode(query)
if subreddit:
sub = subreddit.lstrip("r/").strip()
url = (
f"https://www.reddit.com/r/{sub}/search.json"
f"?q={encoded_query}&restrict_sr=on&sort=relevance&t=month&limit={limit}&raw_json=1"
)
else:
url = (
f"https://www.reddit.com/search.json"
f"?q={encoded_query}&sort=relevance&t=month&limit={limit}&raw_json=1"
)
data = _fetch_json(url, timeout=timeout)
posts = _parse_posts(data)
# Dedupe by URL and assign IDs
seen_urls = set()
unique = []
for post in posts:
if post["url"] not in seen_urls:
seen_urls.add(post["url"])
unique.append(post)
for i, post in enumerate(unique):
post["id"] = f"R{i + 1}"
return unique[:limit]
def search_reddit_public(
topic: str,
from_date: str,
to_date: str,
depth: str = "default",
) -> List[Dict[str, Any]]:
"""High-level Reddit public search matching the openai_reddit interface.
Runs global search, deduplicates, filters by date range, and sorts
by engagement. Compatible as a drop-in replacement in the fallback chain.
Args:
topic: Search topic
from_date: Start date (YYYY-MM-DD)
to_date: End date (YYYY-MM-DD)
depth: 'quick', 'default', or 'deep'
Returns:
List of normalized item dicts matching ScrapeCreators output format.
"""
results = search(topic, depth=depth)
# Date filter: keep posts in range or with unknown dates
filtered = []
for item in results:
d = item.get("date")
if d is None or (from_date <= d <= to_date):
filtered.append(item)
# Sort by engagement (score desc)
filtered.sort(
key=lambda x: x.get("engagement", {}).get("score", 0),
reverse=True,
)
# Re-index IDs
for i, item in enumerate(filtered):
item["id"] = f"R{i + 1}"
return filtered
scripts/lib/render.py
+24 -2
View File
@@ -115,7 +115,7 @@ def render_compact(report: schema.Report, limit: int = 15, missing_keys: str = "
lines.append("")
lines.append("---")
lines.append("**⚡ Want better results?** Add API keys to unlock Reddit, TikTok, Instagram & X data:")
lines.append("- `SCRAPECREATORS_API_KEY` → Reddit + TikTok + Instagram (one key, all three!) — real upvotes, comments, views")
lines.append("- `SCRAPECREATORS_API_KEY` → Reddit + TikTok + Instagram (one key, all three!) — 100 free calls, no CC — scrapecreators.com (no affiliation)")
lines.append("- `XAI_API_KEY` → X posts with real likes & reposts")
lines.append("- `OPENAI_API_KEY` (legacy) → Reddit threads (slower, higher cost)")
lines.append("- Edit `~/.config/last30days/.env` to add keys")
@@ -143,7 +143,7 @@ def render_compact(report: schema.Report, limit: int = 15, missing_keys: str = "
lines.append("*💡 Tip: Add an xAI key (`XAI_API_KEY`) for X/Twitter data and better triangulation.*")
lines.append("")
elif report.mode == "x-only" and missing_keys in ("reddit", "none"):
lines.append("*💡 Tip: Add `SCRAPECREATORS_API_KEY` for Reddit + TikTok + Instagram data (one key, all three) and better triangulation.*")
lines.append("*💡 Tip: Add `SCRAPECREATORS_API_KEY` for Reddit + TikTok + Instagram data (one key, all three) — 100 free calls, no CC — scrapecreators.com (no affiliation)*")
lines.append("")
# Reddit items
@@ -525,6 +525,28 @@ def render_compact(report: schema.Report, limit: int = 15, missing_keys: str = "
return "\n".join(lines)
def render_quality_nudge(quality: dict) -> str:
"""Render the quality score nudge block.
Args:
quality: Dict from quality_nudge.compute_quality_score()
Returns:
Markdown string with quality nudge, or empty string if no nudge.
"""
nudge_text = quality.get("nudge_text")
if not nudge_text:
return ""
lines = []
lines.append("---")
lines.append(f"**🔍 Research Coverage: {quality['score_pct']}%**")
lines.append("")
lines.append(nudge_text)
lines.append("")
return "\n".join(lines)
def render_source_status(report: schema.Report, source_info: dict = None) -> str:
"""Render source status footer showing what was used/skipped and why.
scripts/lib/safari_cookies.py
+182
View File
@@ -0,0 +1,182 @@
"""
Safari binary cookie extractor for macOS.
Parses ~/Library/Cookies/Cookies.binarycookies (unencrypted binary format)
using only stdlib. Zero pip dependencies.
Reference: github.com/mdegrazia/Safari-Binary-Cookie-Parser
"""
from __future__ import annotations
import io
import struct
import sys
from pathlib import Path
# Mac epoch: 2001-01-01 00:00:00 UTC (not used for filtering, but documented)
_MAC_EPOCH_OFFSET = 978307200 # seconds between Unix epoch and Mac epoch
_MAGIC = b"cook"
def _read_null_terminated(data: bytes, offset: int) -> str:
"""Read a null-terminated string from data starting at offset."""
end = data.find(b"\x00", offset)
if end == -1:
end = len(data)
return data[offset:end].decode("utf-8", errors="replace")
def _parse_cookie_record(data: bytes) -> dict | None:
"""Parse a single cookie record. Returns dict with url, name, value, path or None."""
if len(data) < 44:
return None
try:
(size,) = struct.unpack("<I", data[0:4])
# flags at offset 4 (4 bytes, little-endian) — not needed for extraction
(url_offset,) = struct.unpack("<I", data[16:20])
(name_offset,) = struct.unpack("<I", data[20:24])
(path_offset,) = struct.unpack("<I", data[24:28])
(value_offset,) = struct.unpack("<I", data[28:32])
# expiry at offset 40 (8-byte double, little-endian) — not needed for filtering
# creation at offset 48 (8-byte double, little-endian) — not needed
url = _read_null_terminated(data, url_offset)
name = _read_null_terminated(data, name_offset)
path = _read_null_terminated(data, path_offset)
value = _read_null_terminated(data, value_offset)
return {"url": url, "name": name, "value": value, "path": path}
except (struct.error, IndexError, UnicodeDecodeError):
return None
def _parse_page(page_data: bytes) -> list[dict]:
"""Parse a single page of cookies. Returns list of cookie dicts."""
cookies = []
if len(page_data) < 8:
return cookies
# Page header: 4 bytes (always 00 00 01 00), then 4-byte LE cookie count
try:
(num_cookies,) = struct.unpack("<I", page_data[4:8])
except struct.error:
return cookies
# Sanity check
if num_cookies > 10000:
return cookies
# Cookie offsets: array of 4-byte LE uint32 starting at offset 8
offsets_end = 8 + num_cookies * 4
if offsets_end > len(page_data):
return cookies
for i in range(num_cookies):
off_start = 8 + i * 4
try:
(cookie_offset,) = struct.unpack("<I", page_data[off_start : off_start + 4])
except struct.error:
continue
if cookie_offset >= len(page_data):
continue
cookie_data = page_data[cookie_offset:]
record = _parse_cookie_record(cookie_data)
if record:
cookies.append(record)
return cookies
def extract_safari_cookies_macos(
domain: str, cookie_names: list[str]
) -> dict[str, str] | None:
"""
Extract cookies from Safari on macOS.
Args:
domain: Domain to match (substring match, e.g. "x.com")
cookie_names: List of cookie names to extract (e.g. ["auth_token", "ct0"])
Returns:
Dict mapping cookie name to value for found cookies, or None on failure.
"""
if sys.platform != "darwin":
return None
cookie_path = Path.home() / "Library" / "Cookies" / "Cookies.binarycookies"
try:
raw = cookie_path.read_bytes()
except FileNotFoundError:
return None
except PermissionError:
print(
"[safari] Permission denied reading Cookies.binarycookies. "
"Enable Full Disk Access for Terminal in System Settings > "
"Privacy & Security > Full Disk Access.",
file=sys.stderr,
)
return None
except OSError:
return None
return _parse_binary_cookies(raw, domain, cookie_names)
def _parse_binary_cookies(
raw: bytes, domain: str, cookie_names: list[str]
) -> dict[str, str] | None:
"""Parse raw binary cookie data. Separated for testability."""
if len(raw) < 8:
return None
# Validate magic
if raw[:4] != _MAGIC:
return None
try:
(num_pages,) = struct.unpack(">I", raw[4:8])
except struct.error:
return None
if num_pages > 100000:
return None
# Read page sizes (big-endian uint32 array)
page_sizes_end = 8 + num_pages * 4
if page_sizes_end > len(raw):
return None
page_sizes = []
for i in range(num_pages):
off = 8 + i * 4
try:
(ps,) = struct.unpack(">I", raw[off : off + 4])
page_sizes.append(ps)
except struct.error:
return None
# Parse each page
names_set = set(cookie_names)
result: dict[str, str] = {}
offset = page_sizes_end
for ps in page_sizes:
if offset + ps > len(raw):
break
page_data = raw[offset : offset + ps]
cookies = _parse_page(page_data)
for c in cookies:
# Substring match on domain (handles leading dots like ".x.com")
if domain in c["url"] and c["name"] in names_set:
result[c["name"]] = c["value"]
offset += ps
if not result:
return None
return result
scripts/lib/setup_wizard.py
+186
View File
@@ -0,0 +1,186 @@
"""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 logging
import shutil
import subprocess
from pathlib import Path
from typing import Any, Dict, Optional
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)
scripts/lib/tiktok.py
+1 -1
View File
@@ -3,7 +3,7 @@
Uses ScrapeCreators REST API to search TikTok by keyword, extract engagement
metrics (views, likes, comments, shares), and fetch video transcripts.
Requires SCRAPECREATORS_API_KEY in config. 100 free credits, then PAYG.
Requires SCRAPECREATORS_API_KEY in config. 100 free API calls, then PAYG.
API docs: https://scrapecreators.com/docs
"""
scripts/lib/ui.py
+177 -107
View File
@@ -417,120 +417,190 @@ class ProgressDisplay:
sys.stderr.flush()
def show_diagnostic_banner(diag: dict):
"""Show pre-flight source status banner when sources are missing.
def _build_status_banner(diag: dict) -> list[str]:
"""Build the status banner lines (plain text, no ANSI).
Returns a list of strings, each being a line of the banner box.
Args:
diag: Dict from env diagnostics with keys:
openai, xai, x_source, bird_installed, bird_authenticated,
bird_username, youtube, web_search_backend
diag: Dict with keys:
setup_complete, reddit_source, x_source, x_method,
youtube, tiktok, instagram, hackernews, polymarket,
bluesky, truthsocial, xiaohongshu, scrapecreators,
web_search_backend
"""
has_openai = diag.get("openai", False)
has_reddit_public = diag.get("reddit_public", False)
has_reddit = has_openai or has_reddit_public
has_x = diag.get("x_source") is not None
has_youtube = diag.get("youtube", False)
has_xiaohongshu = diag.get("xiaohongshu", False)
has_web = diag.get("web_search_backend") is not None
setup_complete = diag.get("setup_complete", False)
has_sc = diag.get("scrapecreators", False)
# If everything is available, no banner needed
if has_reddit and has_x and has_youtube and has_web:
return
# --- Build active sources list: (label, method_label) ---
active: list[str] = []
lines = []
# Reddit — always available; what matters to users is comments or not
reddit_src = diag.get("reddit_source")
if reddit_src == "scrapecreators":
active.append("Reddit (with comments)")
else:
active.append("Reddit (threads only)")
# X/Twitter
x_source = diag.get("x_source")
x_method = diag.get("x_method")
if x_source:
if x_method and x_method.startswith("browser-"):
browser = x_method.split("-", 1)[1].capitalize()
active.append(f"X ({browser})")
elif x_method == "env":
active.append("X (env)")
elif x_method == "api":
active.append("X (xAI)")
else:
active.append("X")
# YouTube
if diag.get("youtube"):
active.append("YouTube")
# HN — always available
if diag.get("hackernews"):
active.append("HN")
# Polymarket — always available
if diag.get("polymarket"):
active.append("Polymarket")
# TikTok (requires SC or Apify)
if diag.get("tiktok"):
active.append("TikTok")
# Instagram (requires SC)
if diag.get("instagram"):
active.append("Instagram")
# Bluesky
if diag.get("bluesky"):
active.append("Bluesky")
# Truth Social
if diag.get("truthsocial"):
active.append("Truth Social")
# Xiaohongshu
if diag.get("xiaohongshu"):
active.append("Xiaohongshu")
# --- Format active sources into wrapped lines ---
BOX_INNER = 53 # characters inside the box (between │ and │)
PREFIX = " " # 2-space indent inside box
def _wrap_sources(sources: list[str]) -> list[str]:
"""Wrap source labels into lines that fit the box width."""
result_lines: list[str] = []
current = PREFIX
for i, s in enumerate(sources):
token = f"✅ {s}"
sep = " " if current != PREFIX else ""
if len(current) + len(sep) + len(token) > BOX_INNER:
result_lines.append(current)
current = PREFIX + token
else:
current += sep + token
if current.strip():
result_lines.append(current)
return result_lines
source_lines = _wrap_sources(active)
# --- Title ---
if not setup_complete:
title = "/last30days v3.0 — First Run"
else:
title = "/last30days v3.0 — Source Status"
# --- Build upgrade suggestions ---
suggestions: list[str] = []
if not setup_complete:
suggestions.append("Run /last30days setup to unlock more sources")
else:
# Recommend ScrapeCreators if missing
if not has_sc:
suggestions.append("⭐ Add SCRAPECREATORS_API_KEY for Reddit comments")
suggestions.append(" + TikTok + Instagram")
suggestions.append(" 100 free calls, no CC — scrapecreators.com (no affiliation)")
# --- Assemble box lines ---
# Collect all content lines, then determine box width dynamically.
content: list[str] = []
content.append(f" {title}")
content.append("") # blank line
for sl in source_lines:
content.append(sl)
if suggestions:
content.append("") # blank line
for sg in suggestions:
content.append(f" {sg}")
content.append("") # blank line
content.append(" Config: ~/.config/last30days/.env")
# Width = widest content line + 1 for right margin
width = max(len(line) for line in content) + 1
if width < 53:
width = 53
lines: list[str] = []
lines.append("\u250c" + "\u2500" * width + "\u2510")
for c in content:
lines.append("\u2502" + c.ljust(width) + "\u2502")
lines.append("\u2514" + "\u2500" * width + "\u2518")
return lines
def _colorize_banner(lines: list[str]) -> list[str]:
"""Apply ANSI colors to plain-text banner lines for TTY output."""
colored: list[str] = []
for line in lines:
if line.startswith("\u250c") or line.startswith("\u2514"):
colored.append(f"{Colors.DIM}{line}{Colors.RESET}")
elif line.startswith("\u2502"):
inner = line[1:-1] # strip box chars on both sides
inner_width = len(inner)
# Colorize check marks green, star yellow
inner = inner.replace("\u2705", f"{Colors.GREEN}\u2705{Colors.RESET}")
inner = inner.replace("\u2b50", f"{Colors.YELLOW}\u2b50{Colors.RESET}")
# Bold the title line
if "/last30days v3.0" in inner:
stripped = inner.strip()
inner = f" {Colors.BOLD}{stripped}{Colors.RESET}"
# Re-pad to original width (ANSI codes are zero-width)
visible_len = 1 + len(stripped)
inner = inner + " " * max(0, inner_width - visible_len)
colored.append(f"{Colors.DIM}\u2502{Colors.RESET}{inner}{Colors.DIM}\u2502{Colors.RESET}")
else:
colored.append(line)
return colored
def show_diagnostic_banner(diag: dict):
"""Show pre-flight source status banner.
Free-first design: leads with what's working (✅), not what's broken.
Shows upgrade suggestions only when relevant.
Args:
diag: Dict with keys:
setup_complete, reddit_source, x_source, x_method,
youtube, tiktok, instagram, hackernews, polymarket,
bluesky, truthsocial, xiaohongshu, scrapecreators,
web_search_backend
"""
lines = _build_status_banner(diag)
if IS_TTY:
lines.append(f"{Colors.DIM}┌─────────────────────────────────────────────────────┐{Colors.RESET}")
lines.append(f"{Colors.DIM}│{Colors.RESET} {Colors.BOLD}/last30days v2.1 — Source Status{Colors.RESET} {Colors.DIM}│{Colors.RESET}")
lines.append(f"{Colors.DIM}│{Colors.RESET} {Colors.DIM}│{Colors.RESET}")
# Reddit
if has_openai:
lines.append(f"{Colors.DIM}│{Colors.RESET} {Colors.GREEN}✅ Reddit{Colors.RESET} — OpenAI/Codex auth found {Colors.DIM}│{Colors.RESET}")
elif has_reddit_public:
lines.append(f"{Colors.DIM}│{Colors.RESET} {Colors.GREEN}✅ Reddit{Colors.RESET} — Public Reddit search (no key) {Colors.DIM}│{Colors.RESET}")
else:
lines.append(f"{Colors.DIM}│{Colors.RESET} {Colors.RED}❌ Reddit{Colors.RESET} — No OPENAI_API_KEY {Colors.DIM}│{Colors.RESET}")
lines.append(f"{Colors.DIM}│{Colors.RESET} └─ Add to ~/.config/last30days/.env {Colors.DIM}│{Colors.RESET}")
# X/Twitter
if has_x:
source = diag.get("x_source", "")
username = diag.get("bird_username", "")
label = f"Bird ({username})" if source == "bird" and username else source.upper()
lines.append(f"{Colors.DIM}│{Colors.RESET} {Colors.GREEN}✅ X/Twitter{Colors.RESET} — {label} {Colors.DIM}│{Colors.RESET}")
else:
lines.append(f"{Colors.DIM}│{Colors.RESET} {Colors.RED}❌ X/Twitter{Colors.RESET} — No X auth or fallback key {Colors.DIM}│{Colors.RESET}")
if diag.get("bird_installed"):
lines.append(f"{Colors.DIM}│{Colors.RESET} └─ Add AUTH_TOKEN/CT0 or XAI_API_KEY {Colors.DIM}│{Colors.RESET}")
else:
lines.append(f"{Colors.DIM}│{Colors.RESET} └─ Needs Node.js 22+ (Bird is bundled) {Colors.DIM}│{Colors.RESET}")
# YouTube
if has_youtube:
lines.append(f"{Colors.DIM}│{Colors.RESET} {Colors.GREEN}✅ YouTube{Colors.RESET} — yt-dlp found {Colors.DIM}│{Colors.RESET}")
else:
lines.append(f"{Colors.DIM}│{Colors.RESET} {Colors.RED}❌ YouTube{Colors.RESET} — yt-dlp not installed {Colors.DIM}│{Colors.RESET}")
lines.append(f"{Colors.DIM}│{Colors.RESET} └─ Fix: brew install yt-dlp (free) {Colors.DIM}│{Colors.RESET}")
# Xiaohongshu
if has_xiaohongshu:
lines.append(f"{Colors.DIM}│{Colors.RESET} {Colors.GREEN}✅ Xiaohongshu{Colors.RESET} — API connected + logged in {Colors.DIM}│{Colors.RESET}")
else:
lines.append(f"{Colors.DIM}│{Colors.RESET} {Colors.YELLOW}⚡ Xiaohongshu{Colors.RESET} — API not connected/logged in {Colors.DIM}│{Colors.RESET}")
# Web
if has_web:
backend = diag.get("web_search_backend", "")
lines.append(f"{Colors.DIM}│{Colors.RESET} {Colors.GREEN}✅ Web{Colors.RESET} — {backend} API {Colors.DIM}│{Colors.RESET}")
else:
lines.append(f"{Colors.DIM}│{Colors.RESET} {Colors.YELLOW}⚡ Web{Colors.RESET} — Using assistant's search tool {Colors.DIM}│{Colors.RESET}")
lines.append(f"{Colors.DIM}│{Colors.RESET} {Colors.DIM}│{Colors.RESET}")
lines.append(f"{Colors.DIM}│{Colors.RESET} Config: {Colors.BOLD}~/.config/last30days/.env{Colors.RESET} {Colors.DIM}│{Colors.RESET}")
lines.append(f"{Colors.DIM}└─────────────────────────────────────────────────────┘{Colors.RESET}")
else:
# Plain text for non-TTY (Claude Code / Codex)
lines.append("┌─────────────────────────────────────────────────────┐")
lines.append("│ /last30days v2.1 — Source Status │")
lines.append("│ │")
if has_openai:
lines.append("│ ✅ Reddit — OpenAI/Codex auth found │")
elif has_reddit_public:
lines.append("│ ✅ Reddit — Public Reddit search (no key) │")
else:
lines.append("│ ❌ Reddit — No OPENAI_API_KEY │")
lines.append("│ └─ Add to ~/.config/last30days/.env │")
if has_x:
lines.append("│ ✅ X/Twitter — available │")
else:
lines.append("│ ❌ X/Twitter — No X auth or fallback key │")
if diag.get("bird_installed"):
lines.append("│ └─ Add AUTH_TOKEN/CT0 or XAI_API_KEY │")
else:
lines.append("│ └─ Needs Node.js 22+ (Bird is bundled) │")
if has_youtube:
lines.append("│ ✅ YouTube — yt-dlp found │")
else:
lines.append("│ ❌ YouTube — yt-dlp not installed │")
lines.append("│ └─ Fix: brew install yt-dlp (free) │")
if has_xiaohongshu:
lines.append("│ ✅ Xiaohongshu — API connected + logged in │")
else:
lines.append("│ ⚡ Xiaohongshu — API not connected/logged in │")
if has_web:
lines.append("│ ✅ Web — API search available │")
else:
lines.append("│ ⚡ Web — Using assistant's search tool │")
lines.append("│ │")
lines.append("│ Config: ~/.config/last30days/.env │")
lines.append("└─────────────────────────────────────────────────────┘")
lines = _colorize_banner(lines)
sys.stderr.write("\n".join(lines) + "\n\n")
sys.stderr.flush()
scripts/lib/youtube_yt.py
+130 -5
View File
@@ -15,6 +15,8 @@ import shutil
import subprocess
import sys
import tempfile
import urllib.error
import urllib.request
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
from typing import Any, Dict, List, Optional, Set, Tuple
@@ -255,15 +257,114 @@ def _clean_vtt(vtt_text: str) -> str:
return re.sub(r'\s+', ' ', ' '.join(unique)).strip()
def fetch_transcript(video_id: str, temp_dir: str) -> Optional[str]:
"""Fetch auto-generated transcript for a YouTube video.
_YT_USER_AGENT = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36"
def _fetch_transcript_direct(video_id: str, timeout: int = 30) -> Optional[str]:
"""Fetch YouTube transcript via direct HTTP without yt-dlp.
Scrapes the watch page HTML for the captions track URL in
ytInitialPlayerResponse, then fetches the VTT subtitle file.
Args:
video_id: YouTube video ID
timeout: HTTP request timeout in seconds
Returns:
Raw VTT text, or None if captions are unavailable.
"""
watch_url = f"https://www.youtube.com/watch?v={video_id}"
headers = {
"User-Agent": _YT_USER_AGENT,
"Accept-Language": "en-US,en;q=0.9",
}
# Step 1: Fetch the watch page HTML
req = urllib.request.Request(watch_url, headers=headers)
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
html = resp.read().decode("utf-8", errors="replace")
except (urllib.error.URLError, urllib.error.HTTPError, OSError, TimeoutError) as exc:
_log(f"Direct transcript: failed to fetch watch page for {video_id}: {exc}")
return None
# Step 2: Extract captions URL from ytInitialPlayerResponse
# YouTube embeds this as a JS variable in the page HTML
match = re.search(
r'ytInitialPlayerResponse\s*=\s*(\{.+?\})\s*;(?:\s*var\s|\s*<\/script>)',
html,
)
if not match:
# Fallback: try the JSON embedded in the script tag
match = re.search(
r'var\s+ytInitialPlayerResponse\s*=\s*(\{.+?\})\s*;',
html,
)
if not match:
_log(f"Direct transcript: no ytInitialPlayerResponse found for {video_id}")
return None
try:
player_response = json.loads(match.group(1))
except json.JSONDecodeError:
_log(f"Direct transcript: failed to parse ytInitialPlayerResponse for {video_id}")
return None
# Navigate to caption tracks
captions = player_response.get("captions", {})
renderer = captions.get("playerCaptionsTracklistRenderer", {})
caption_tracks = renderer.get("captionTracks", [])
if not caption_tracks:
_log(f"Direct transcript: no caption tracks for {video_id}")
return None
# Find English track (prefer exact 'en', then any en variant, then first track)
base_url = None
for track in caption_tracks:
lang = track.get("languageCode", "")
if lang == "en":
base_url = track.get("baseUrl")
break
if not base_url:
for track in caption_tracks:
lang = track.get("languageCode", "")
if lang.startswith("en"):
base_url = track.get("baseUrl")
break
if not base_url:
# Fall back to first available track
base_url = caption_tracks[0].get("baseUrl")
if not base_url:
_log(f"Direct transcript: no baseUrl in caption tracks for {video_id}")
return None
# Step 3: Fetch the VTT subtitle file
sep = "&" if "?" in base_url else "?"
vtt_url = f"{base_url}{sep}fmt=vtt"
vtt_req = urllib.request.Request(vtt_url, headers=headers)
try:
with urllib.request.urlopen(vtt_req, timeout=timeout) as resp:
vtt_text = resp.read().decode("utf-8", errors="replace")
except (urllib.error.URLError, urllib.error.HTTPError, OSError, TimeoutError) as exc:
_log(f"Direct transcript: failed to fetch VTT for {video_id}: {exc}")
return None
if not vtt_text or not vtt_text.strip():
return None
return vtt_text
def _fetch_transcript_ytdlp(video_id: str, temp_dir: str) -> Optional[str]:
"""Fetch transcript using yt-dlp (original implementation).
Args:
video_id: YouTube video ID
temp_dir: Temporary directory for subtitle files
Returns:
Plaintext transcript string, or None if no captions available.
Raw VTT text, or None if no captions available.
"""
cmd = [
"yt-dlp",
@@ -311,11 +412,35 @@ def fetch_transcript(video_id: str, temp_dir: str) -> Optional[str]:
return None
try:
raw = vtt_path.read_text(encoding="utf-8", errors="replace")
return vtt_path.read_text(encoding="utf-8", errors="replace")
except OSError:
return None
transcript = _clean_vtt(raw)
def fetch_transcript(video_id: str, temp_dir: str) -> Optional[str]:
"""Fetch auto-generated transcript for a YouTube video.
Uses yt-dlp when available (preferred, more robust). Falls back to
direct HTTP transcript fetching when yt-dlp is not installed.
Args:
video_id: YouTube video ID
temp_dir: Temporary directory for subtitle files
Returns:
Plaintext transcript string, or None if no captions available.
"""
raw_vtt = None
if is_ytdlp_installed():
raw_vtt = _fetch_transcript_ytdlp(video_id, temp_dir)
else:
_log("yt-dlp not installed, using direct HTTP transcript fetch")
raw_vtt = _fetch_transcript_direct(video_id)
if not raw_vtt:
return None
transcript = _clean_vtt(raw_vtt)
# Truncate to max words
words = transcript.split()
scripts/test-v1-vs-v2.sh
+1 -1
View File
@@ -6,7 +6,7 @@ set -euo pipefail
# using `claude --print` to capture real end-to-end output.
SKILL_DIR="$HOME/.claude/skills/last30days"
REPO_DIR="/Users/mvanhorn/last30days-skill-private"
REPO_DIR="/Users/mvanhorn/last30days-skill"
# Safety: always restore V2 SKILL.md on exit/crash
cleanup() {