Matt Van Horn
5ca4829be4
Research topics across Reddit + X from the last 30 days using OpenAI and xAI APIs. Features: - Auto model selection (GPT-5.x, Grok-3) - Popularity-aware scoring (relevance + recency + engagement) - Reddit thread enrichment with real metrics - Near-duplicate detection - Multiple emit modes (compact, json, context, path) - 24h caching with --refresh bypass - NUX for API key setup - 87 passing unit tests Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
127 lines
3.7 KiB
Python
127 lines
3.7 KiB
Python
"""HTTP utilities for last30days skill (stdlib only)."""
|
|
|
|
import json
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from typing import Any, Dict, Optional
|
|
from urllib.parse import urlencode
|
|
|
|
DEFAULT_TIMEOUT = 30
|
|
MAX_RETRIES = 3
|
|
RETRY_DELAY = 1.0
|
|
USER_AGENT = "last30days-skill/1.0 (Claude Code Skill)"
|
|
|
|
|
|
class HTTPError(Exception):
|
|
"""HTTP request error with status code."""
|
|
def __init__(self, message: str, status_code: Optional[int] = None, body: Optional[str] = None):
|
|
super().__init__(message)
|
|
self.status_code = status_code
|
|
self.body = body
|
|
|
|
|
|
def request(
|
|
method: str,
|
|
url: str,
|
|
headers: Optional[Dict[str, str]] = None,
|
|
json_data: Optional[Dict[str, Any]] = None,
|
|
timeout: int = DEFAULT_TIMEOUT,
|
|
retries: int = MAX_RETRIES,
|
|
) -> Dict[str, Any]:
|
|
"""Make an HTTP request and return JSON response.
|
|
|
|
Args:
|
|
method: HTTP method (GET, POST, etc.)
|
|
url: Request URL
|
|
headers: Optional headers dict
|
|
json_data: Optional JSON body (for POST)
|
|
timeout: Request timeout in seconds
|
|
retries: Number of retries on failure
|
|
|
|
Returns:
|
|
Parsed JSON response
|
|
|
|
Raises:
|
|
HTTPError: On request failure
|
|
"""
|
|
headers = headers or {}
|
|
headers.setdefault("User-Agent", USER_AGENT)
|
|
|
|
data = None
|
|
if json_data is not None:
|
|
data = json.dumps(json_data).encode('utf-8')
|
|
headers.setdefault("Content-Type", "application/json")
|
|
|
|
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
|
|
|
last_error = None
|
|
for attempt in range(retries):
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=timeout) as response:
|
|
body = response.read().decode('utf-8')
|
|
return json.loads(body) if body else {}
|
|
except urllib.error.HTTPError as e:
|
|
body = None
|
|
try:
|
|
body = e.read().decode('utf-8')
|
|
except:
|
|
pass
|
|
last_error = HTTPError(f"HTTP {e.code}: {e.reason}", e.code, body)
|
|
|
|
# Don't retry client errors (4xx) except rate limits
|
|
if 400 <= e.code < 500 and e.code != 429:
|
|
raise last_error
|
|
|
|
if attempt < retries - 1:
|
|
time.sleep(RETRY_DELAY * (attempt + 1))
|
|
except urllib.error.URLError as e:
|
|
last_error = HTTPError(f"URL Error: {e.reason}")
|
|
if attempt < retries - 1:
|
|
time.sleep(RETRY_DELAY * (attempt + 1))
|
|
except json.JSONDecodeError as e:
|
|
last_error = HTTPError(f"Invalid JSON response: {e}")
|
|
raise last_error
|
|
|
|
if last_error:
|
|
raise last_error
|
|
raise HTTPError("Request failed with no error details")
|
|
|
|
|
|
def get(url: str, headers: Optional[Dict[str, str]] = None, **kwargs) -> Dict[str, Any]:
|
|
"""Make a GET request."""
|
|
return request("GET", url, headers=headers, **kwargs)
|
|
|
|
|
|
def post(url: str, json_data: Dict[str, Any], headers: Optional[Dict[str, str]] = None, **kwargs) -> Dict[str, Any]:
|
|
"""Make a POST request with JSON body."""
|
|
return request("POST", url, headers=headers, json_data=json_data, **kwargs)
|
|
|
|
|
|
def get_reddit_json(path: str) -> Dict[str, Any]:
|
|
"""Fetch Reddit thread JSON.
|
|
|
|
Args:
|
|
path: Reddit path (e.g., /r/subreddit/comments/id/title)
|
|
|
|
Returns:
|
|
Parsed JSON response
|
|
"""
|
|
# Ensure path starts with /
|
|
if not path.startswith('/'):
|
|
path = '/' + path
|
|
|
|
# Remove trailing slash and add .json
|
|
path = path.rstrip('/')
|
|
if not path.endswith('.json'):
|
|
path = path + '.json'
|
|
|
|
url = f"https://www.reddit.com{path}?raw_json=1"
|
|
|
|
headers = {
|
|
"User-Agent": USER_AGENT,
|
|
"Accept": "application/json",
|
|
}
|
|
|
|
return get(url, headers=headers)
|