Matt Van Horn
09b09946c0
- New scripts/lib/reddit.py: multi-query expansion, global search, subreddit discovery, targeted subreddit search, comment enrichment - 68 results in 17s vs ~15 results in 60-90s (OpenAI) - Cost: ~$0.02/search vs $0.03-0.10 (15-50x cheaper) - Real engagement data (score, comments, dates) from API - No more 429 rate limits on comment enrichment - Falls back to OpenAI if SCRAPECREATORS_API_KEY missing - Registered as last30daysbeta for parallel local testing Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
494 lines
15 KiB
Python
494 lines
15 KiB
Python
"""Environment and API key management for last30days skill."""
|
|
|
|
import base64
|
|
import json
|
|
import os
|
|
import time
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Optional, Dict, Any, Literal
|
|
|
|
# Allow override via environment variable for testing
|
|
# Set LAST30DAYS_CONFIG_DIR="" for clean/no-config mode
|
|
# Set LAST30DAYS_CONFIG_DIR="/path/to/dir" for custom config location
|
|
_config_override = os.environ.get('LAST30DAYS_CONFIG_DIR')
|
|
if _config_override == "":
|
|
# Empty string = no config file (clean mode)
|
|
CONFIG_DIR = None
|
|
CONFIG_FILE = None
|
|
elif _config_override:
|
|
CONFIG_DIR = Path(_config_override)
|
|
CONFIG_FILE = CONFIG_DIR / ".env"
|
|
else:
|
|
CONFIG_DIR = Path.home() / ".config" / "last30days"
|
|
CONFIG_FILE = CONFIG_DIR / ".env"
|
|
|
|
CODEX_AUTH_FILE = Path(os.environ.get("CODEX_AUTH_FILE", str(Path.home() / ".codex" / "auth.json")))
|
|
|
|
AuthSource = Literal["api_key", "codex", "none"]
|
|
AuthStatus = Literal["ok", "missing", "expired", "missing_account_id"]
|
|
|
|
AUTH_SOURCE_API_KEY: AuthSource = "api_key"
|
|
AUTH_SOURCE_CODEX: AuthSource = "codex"
|
|
AUTH_SOURCE_NONE: AuthSource = "none"
|
|
|
|
AUTH_STATUS_OK: AuthStatus = "ok"
|
|
AUTH_STATUS_MISSING: AuthStatus = "missing"
|
|
AUTH_STATUS_EXPIRED: AuthStatus = "expired"
|
|
AUTH_STATUS_MISSING_ACCOUNT_ID: AuthStatus = "missing_account_id"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class OpenAIAuth:
|
|
token: Optional[str]
|
|
source: AuthSource
|
|
status: AuthStatus
|
|
account_id: Optional[str]
|
|
codex_auth_file: str
|
|
|
|
|
|
def load_env_file(path: Path) -> Dict[str, str]:
|
|
"""Load environment variables from a file."""
|
|
env = {}
|
|
if not path.exists():
|
|
return env
|
|
|
|
with open(path, 'r') as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if not line or line.startswith('#'):
|
|
continue
|
|
if '=' in line:
|
|
key, _, value = line.partition('=')
|
|
key = key.strip()
|
|
value = value.strip()
|
|
# Remove quotes if present
|
|
if value and value[0] in ('"', "'") and value[-1] == value[0]:
|
|
value = value[1:-1]
|
|
if key and value:
|
|
env[key] = value
|
|
return env
|
|
|
|
|
|
def _decode_jwt_payload(token: str) -> Optional[Dict[str, Any]]:
|
|
"""Decode JWT payload without verification."""
|
|
try:
|
|
parts = token.split(".")
|
|
if len(parts) < 2:
|
|
return None
|
|
payload_b64 = parts[1]
|
|
pad = "=" * (-len(payload_b64) % 4)
|
|
decoded = base64.urlsafe_b64decode(payload_b64 + pad)
|
|
return json.loads(decoded.decode("utf-8"))
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _token_expired(token: str, leeway_seconds: int = 60) -> bool:
|
|
"""Check if JWT token is expired."""
|
|
payload = _decode_jwt_payload(token)
|
|
if not payload:
|
|
return False
|
|
exp = payload.get("exp")
|
|
if not exp:
|
|
return False
|
|
return exp <= (time.time() + leeway_seconds)
|
|
|
|
|
|
def extract_chatgpt_account_id(access_token: str) -> Optional[str]:
|
|
"""Extract chatgpt_account_id from JWT token."""
|
|
payload = _decode_jwt_payload(access_token)
|
|
if not payload:
|
|
return None
|
|
auth_claim = payload.get("https://api.openai.com/auth", {})
|
|
if isinstance(auth_claim, dict):
|
|
return auth_claim.get("chatgpt_account_id")
|
|
return None
|
|
|
|
|
|
def load_codex_auth(path: Path = CODEX_AUTH_FILE) -> Dict[str, Any]:
|
|
"""Load Codex auth JSON."""
|
|
if not path.exists():
|
|
return {}
|
|
try:
|
|
with open(path, "r") as f:
|
|
return json.load(f)
|
|
except Exception:
|
|
return {}
|
|
|
|
|
|
def get_codex_access_token() -> tuple[Optional[str], str]:
|
|
"""Get Codex access token from auth.json.
|
|
|
|
Returns:
|
|
(token, status) where status is 'ok', 'missing', or 'expired'
|
|
"""
|
|
auth = load_codex_auth()
|
|
token = None
|
|
if isinstance(auth, dict):
|
|
tokens = auth.get("tokens") or {}
|
|
if isinstance(tokens, dict):
|
|
token = tokens.get("access_token")
|
|
if not token:
|
|
token = auth.get("access_token")
|
|
if not token:
|
|
return None, AUTH_STATUS_MISSING
|
|
if _token_expired(token):
|
|
return None, AUTH_STATUS_EXPIRED
|
|
return token, AUTH_STATUS_OK
|
|
|
|
|
|
def get_openai_auth(file_env: Dict[str, str]) -> OpenAIAuth:
|
|
"""Resolve OpenAI auth from API key or Codex login."""
|
|
api_key = os.environ.get('OPENAI_API_KEY') or file_env.get('OPENAI_API_KEY')
|
|
if api_key:
|
|
return OpenAIAuth(
|
|
token=api_key,
|
|
source=AUTH_SOURCE_API_KEY,
|
|
status=AUTH_STATUS_OK,
|
|
account_id=None,
|
|
codex_auth_file=str(CODEX_AUTH_FILE),
|
|
)
|
|
|
|
codex_token, codex_status = get_codex_access_token()
|
|
if codex_token:
|
|
account_id = extract_chatgpt_account_id(codex_token)
|
|
if account_id:
|
|
return OpenAIAuth(
|
|
token=codex_token,
|
|
source=AUTH_SOURCE_CODEX,
|
|
status=AUTH_STATUS_OK,
|
|
account_id=account_id,
|
|
codex_auth_file=str(CODEX_AUTH_FILE),
|
|
)
|
|
return OpenAIAuth(
|
|
token=None,
|
|
source=AUTH_SOURCE_CODEX,
|
|
status=AUTH_STATUS_MISSING_ACCOUNT_ID,
|
|
account_id=None,
|
|
codex_auth_file=str(CODEX_AUTH_FILE),
|
|
)
|
|
|
|
return OpenAIAuth(
|
|
token=None,
|
|
source=AUTH_SOURCE_NONE,
|
|
status=codex_status,
|
|
account_id=None,
|
|
codex_auth_file=str(CODEX_AUTH_FILE),
|
|
)
|
|
|
|
|
|
def get_config() -> Dict[str, Any]:
|
|
"""Load configuration from ~/.config/last30days/.env and environment."""
|
|
# Load from config file first (if configured)
|
|
file_env = load_env_file(CONFIG_FILE) if CONFIG_FILE else {}
|
|
|
|
openai_auth = get_openai_auth(file_env)
|
|
|
|
# Build config: Codex/OpenAI auth + process.env > .env file
|
|
config = {
|
|
'OPENAI_API_KEY': openai_auth.token,
|
|
'OPENAI_AUTH_SOURCE': openai_auth.source,
|
|
'OPENAI_AUTH_STATUS': openai_auth.status,
|
|
'OPENAI_CHATGPT_ACCOUNT_ID': openai_auth.account_id,
|
|
'CODEX_AUTH_FILE': openai_auth.codex_auth_file,
|
|
}
|
|
|
|
keys = [
|
|
('XAI_API_KEY', None),
|
|
('OPENROUTER_API_KEY', None),
|
|
('PARALLEL_API_KEY', None),
|
|
('BRAVE_API_KEY', None),
|
|
('OPENAI_MODEL_POLICY', 'auto'),
|
|
('OPENAI_MODEL_PIN', None),
|
|
('XAI_MODEL_POLICY', 'latest'),
|
|
('XAI_MODEL_PIN', None),
|
|
('SCRAPECREATORS_API_KEY', None),
|
|
('APIFY_API_TOKEN', None),
|
|
('AUTH_TOKEN', None),
|
|
('CT0', None),
|
|
]
|
|
|
|
for key, default in keys:
|
|
config[key] = os.environ.get(key) or file_env.get(key, default)
|
|
|
|
return config
|
|
|
|
|
|
def config_exists() -> bool:
|
|
"""Check if configuration file exists."""
|
|
return CONFIG_FILE.exists()
|
|
|
|
|
|
def is_reddit_available(config: Dict[str, Any]) -> bool:
|
|
"""Check if Reddit search is available.
|
|
|
|
Reddit can use either ScrapeCreators (preferred) or OpenAI.
|
|
"""
|
|
has_sc = bool(config.get('SCRAPECREATORS_API_KEY'))
|
|
has_openai = bool(config.get('OPENAI_API_KEY')) and config.get('OPENAI_AUTH_STATUS') == AUTH_STATUS_OK
|
|
return has_sc or has_openai
|
|
|
|
|
|
def get_reddit_source(config: Dict[str, Any]) -> Optional[str]:
|
|
"""Determine which Reddit backend to use.
|
|
|
|
Priority: ScrapeCreators (cheaper, faster) > OpenAI (legacy)
|
|
|
|
Returns: 'scrapecreators', 'openai', or None
|
|
"""
|
|
if config.get('SCRAPECREATORS_API_KEY'):
|
|
return 'scrapecreators'
|
|
if config.get('OPENAI_API_KEY') and config.get('OPENAI_AUTH_STATUS') == AUTH_STATUS_OK:
|
|
return 'openai'
|
|
return None
|
|
|
|
|
|
def get_available_sources(config: Dict[str, Any]) -> str:
|
|
"""Determine which sources are available based on API keys.
|
|
|
|
Returns: 'all', 'both', 'reddit', 'reddit-web', 'x', 'x-web', 'web', or 'none'
|
|
"""
|
|
has_reddit = is_reddit_available(config)
|
|
has_xai = bool(config.get('XAI_API_KEY'))
|
|
has_web = has_web_search_keys(config)
|
|
|
|
if has_reddit and has_xai:
|
|
return 'all' if has_web else 'both'
|
|
elif has_reddit:
|
|
return 'reddit-web' if has_web else 'reddit'
|
|
elif has_xai:
|
|
return 'x-web' if has_web else 'x'
|
|
elif has_web:
|
|
return 'web'
|
|
else:
|
|
return 'web' # Fallback: assistant WebSearch (no API keys needed)
|
|
|
|
|
|
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'))
|
|
|
|
|
|
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
|
|
|
|
Returns: 'parallel', 'brave', 'openrouter', or None
|
|
"""
|
|
if config.get('PARALLEL_API_KEY'):
|
|
return 'parallel'
|
|
if config.get('BRAVE_API_KEY'):
|
|
return 'brave'
|
|
if config.get('OPENROUTER_API_KEY'):
|
|
return 'openrouter'
|
|
return None
|
|
|
|
|
|
def get_missing_keys(config: Dict[str, Any]) -> str:
|
|
"""Determine which sources are missing (accounting for Bird and ScrapeCreators).
|
|
|
|
Returns: 'all', 'both', 'reddit', 'x', 'web', or 'none'
|
|
"""
|
|
has_reddit = is_reddit_available(config)
|
|
has_xai = bool(config.get('XAI_API_KEY'))
|
|
has_web = has_web_search_keys(config)
|
|
|
|
# Check if Bird provides X access (import here to avoid circular dependency)
|
|
from . import bird_x
|
|
has_bird = bird_x.is_bird_installed() and bird_x.is_bird_authenticated()
|
|
|
|
has_x = has_xai or has_bird
|
|
|
|
if has_reddit and has_x and has_web:
|
|
return 'none'
|
|
elif has_reddit and has_x:
|
|
return 'web' # Missing web search keys
|
|
elif has_reddit:
|
|
return 'x' # Missing X source (and possibly web)
|
|
elif has_x:
|
|
return 'reddit' # Missing Reddit source (and possibly web)
|
|
else:
|
|
return 'all' # Missing everything
|
|
|
|
|
|
def validate_sources(requested: str, available: str, include_web: bool = False) -> tuple[str, Optional[str]]:
|
|
"""Validate requested sources against available keys.
|
|
|
|
Args:
|
|
requested: 'auto', 'reddit', 'x', 'both', or 'web'
|
|
available: Result from get_available_sources()
|
|
include_web: If True, add WebSearch to available sources
|
|
|
|
Returns:
|
|
Tuple of (effective_sources, error_message)
|
|
"""
|
|
# No API keys at all
|
|
if available == 'none':
|
|
if requested == 'auto':
|
|
return 'web', "No API keys configured. The assistant can still search the web if it has a search tool."
|
|
elif requested == 'web':
|
|
return 'web', None
|
|
else:
|
|
return 'web', f"No API keys configured. Add keys to ~/.config/last30days/.env for Reddit/X."
|
|
|
|
# Web-only mode (only web search API keys)
|
|
if available == 'web':
|
|
if requested == 'auto':
|
|
return 'web', None
|
|
elif requested == 'web':
|
|
return 'web', None
|
|
else:
|
|
return 'web', "Only web search keys configured. Add OPENAI_API_KEY (or run codex login) for Reddit, XAI_API_KEY for X."
|
|
|
|
if requested == 'auto':
|
|
# Add web to sources if include_web is set
|
|
if include_web:
|
|
if available == 'both':
|
|
return 'all', None # reddit + x + web
|
|
elif available == 'reddit':
|
|
return 'reddit-web', None
|
|
elif available == 'x':
|
|
return 'x-web', None
|
|
return available, None
|
|
|
|
if requested == 'web':
|
|
return 'web', None
|
|
|
|
if requested == 'both':
|
|
if available not in ('both',):
|
|
missing = 'xAI' if available == 'reddit' else 'OpenAI'
|
|
return 'none', f"Requested both sources but {missing} key is missing. Use --sources=auto to use available keys."
|
|
if include_web:
|
|
return 'all', None
|
|
return 'both', None
|
|
|
|
if requested == 'reddit':
|
|
if available == 'x':
|
|
return 'none', "Requested Reddit but only xAI key is available."
|
|
if include_web:
|
|
return 'reddit-web', None
|
|
return 'reddit', None
|
|
|
|
if requested == 'x':
|
|
if available == 'reddit':
|
|
return 'none', "Requested X but only OpenAI key is available."
|
|
if include_web:
|
|
return 'x-web', None
|
|
return 'x', None
|
|
|
|
return requested, None
|
|
|
|
|
|
def get_x_source(config: Dict[str, Any]) -> Optional[str]:
|
|
"""Determine the best available X/Twitter source.
|
|
|
|
Priority: Bird (free) → xAI (paid API)
|
|
|
|
Args:
|
|
config: Configuration dict from get_config()
|
|
|
|
Returns:
|
|
'bird' if Bird is installed and authenticated,
|
|
'xai' if XAI_API_KEY is configured,
|
|
None if no X source available.
|
|
"""
|
|
# Import here to avoid circular dependency
|
|
from . import bird_x
|
|
|
|
# Check Bird first (free option)
|
|
if bird_x.is_bird_installed():
|
|
username = bird_x.is_bird_authenticated()
|
|
if username:
|
|
return 'bird'
|
|
|
|
# Fall back to xAI if key exists
|
|
if config.get('XAI_API_KEY'):
|
|
return 'xai'
|
|
|
|
return None
|
|
|
|
|
|
def is_ytdlp_available() -> bool:
|
|
"""Check if yt-dlp is installed for YouTube search."""
|
|
from . import youtube_yt
|
|
return youtube_yt.is_ytdlp_installed()
|
|
|
|
|
|
def is_hackernews_available() -> bool:
|
|
"""Check if Hacker News source is available.
|
|
|
|
Always returns True - HN uses free Algolia API, no key needed.
|
|
"""
|
|
return True
|
|
|
|
|
|
def is_polymarket_available() -> bool:
|
|
"""Check if Polymarket source is available.
|
|
|
|
Always returns True - Gamma API is free, no key needed.
|
|
"""
|
|
return True
|
|
|
|
|
|
def is_tiktok_available(config: Dict[str, Any]) -> bool:
|
|
"""Check if TikTok source is available (ScrapeCreators or legacy Apify).
|
|
|
|
Returns True if SCRAPECREATORS_API_KEY or APIFY_API_TOKEN is set.
|
|
"""
|
|
return bool(config.get('SCRAPECREATORS_API_KEY') or config.get('APIFY_API_TOKEN'))
|
|
|
|
|
|
def get_tiktok_token(config: Dict[str, Any]) -> str:
|
|
"""Get TikTok API token, preferring ScrapeCreators over legacy Apify."""
|
|
return config.get('SCRAPECREATORS_API_KEY') or config.get('APIFY_API_TOKEN') or ''
|
|
|
|
|
|
def is_instagram_available(config: Dict[str, Any]) -> bool:
|
|
"""Check if Instagram source is available (ScrapeCreators).
|
|
|
|
Returns True if SCRAPECREATORS_API_KEY is set.
|
|
Instagram uses the same key as TikTok.
|
|
"""
|
|
return bool(config.get('SCRAPECREATORS_API_KEY'))
|
|
|
|
|
|
def get_instagram_token(config: Dict[str, Any]) -> str:
|
|
"""Get Instagram API token (same ScrapeCreators key as TikTok)."""
|
|
return config.get('SCRAPECREATORS_API_KEY') or ''
|
|
|
|
|
|
# Backward compat alias
|
|
is_apify_available = is_tiktok_available
|
|
|
|
|
|
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,
|
|
bird_username, xai_available, can_install_bird
|
|
"""
|
|
from . import bird_x
|
|
|
|
bird_status = bird_x.get_bird_status()
|
|
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
|
|
|
|
return {
|
|
"source": source,
|
|
"bird_installed": bird_status["installed"],
|
|
"bird_authenticated": bird_status["authenticated"],
|
|
"bird_username": bird_status["username"],
|
|
"xai_available": xai_available,
|
|
"can_install_bird": bird_status["can_install"],
|
|
}
|